mirror of
https://github.com/logos-messaging/logos-messaging-nim.git
synced 2026-05-16 15:29:41 +00:00
1fd25355e0
* Introduce api/send Added events and requests for support. Reworked delivery_monitor into a featured devlivery_service, that - supports relay publish and lightpush depending on configuration but with fallback options - if available and configured it utilizes store api to confirm message delivery - emits message delivery events accordingly prepare for use in api_example * Fix edge mode config and test added * Fix some import issues, start and stop waku shall not throw exception but return with result properly * Utlize sync RequestBroker, adapt to non-async broker usage and gcsafe where appropriate, removed leftover * add api_example app to examples2 * Adapt after merge from master * Adapt code for using broker context * Fix brokerCtx settings for all usedbrokers, cover locked node init * Various fixes upon test failures. Added initial of subscribe API and auto-subscribe for send api * More test added * Fix multi propagate event emit, fix fail send test case * Fix rebase * Fix PushMessageHandlers in tests * adapt libwaku to api changes * Fix relay test by adapting publish return error in case NoPeersToPublish * Addressing all remaining review findings. Removed leftovers. Fixed loggings and typos * Fix rln relay broker, missed brokerCtx * Fix rest relay test failed, due to publish will fail if no peer avail * ignore anvil test state file * Make terst_wakunode_rln_relay broker context aware to fix * Fix waku rln tests by having them broker context aware * fix typo in test_app.nim
107 lines
3.3 KiB
Nim
107 lines
3.3 KiB
Nim
{.push raises: [].}
|
|
|
|
import std/options, results, chronicles, chronos, metrics, bearssl/rand, stew/byteutils
|
|
import libp2p/peerid
|
|
import
|
|
../waku_core/peers,
|
|
../node/peer_manager,
|
|
../utils/requests,
|
|
../waku_core,
|
|
./common,
|
|
./protocol_metrics,
|
|
./rpc,
|
|
./rpc_codec
|
|
|
|
logScope:
|
|
topics = "waku lightpush legacy client"
|
|
|
|
type WakuLegacyLightPushClient* = ref object
|
|
peerManager*: PeerManager
|
|
rng*: ref rand.HmacDrbgContext
|
|
|
|
proc new*(
|
|
T: type WakuLegacyLightPushClient,
|
|
peerManager: PeerManager,
|
|
rng: ref rand.HmacDrbgContext,
|
|
): T =
|
|
WakuLegacyLightPushClient(peerManager: peerManager, rng: rng)
|
|
|
|
proc sendPushRequest(
|
|
wl: WakuLegacyLightPushClient, req: PushRequest, peer: PeerId | RemotePeerInfo
|
|
): Future[WakuLightPushResult[void]] {.async, gcsafe.} =
|
|
let connOpt = await wl.peerManager.dialPeer(peer, WakuLegacyLightPushCodec)
|
|
if connOpt.isNone():
|
|
waku_lightpush_errors.inc(labelValues = [dialFailure])
|
|
return err(dialFailure)
|
|
let connection = connOpt.get()
|
|
|
|
defer:
|
|
await connection.closeWithEOF()
|
|
|
|
let rpc = PushRPC(requestId: generateRequestId(wl.rng), request: some(req))
|
|
await connection.writeLP(rpc.encode().buffer)
|
|
|
|
var buffer: seq[byte]
|
|
try:
|
|
buffer = await connection.readLp(DefaultMaxRpcSize.int)
|
|
except LPStreamRemoteClosedError:
|
|
return err("Exception reading: " & getCurrentExceptionMsg())
|
|
|
|
let pushResponseRes = PushRPC.decode(buffer).valueOr:
|
|
error "failed to decode response"
|
|
waku_lightpush_errors.inc(labelValues = [decodeRpcFailure])
|
|
return err(decodeRpcFailure)
|
|
|
|
if pushResponseRes.response.isNone():
|
|
waku_lightpush_errors.inc(labelValues = [emptyResponseBodyFailure])
|
|
return err(emptyResponseBodyFailure)
|
|
|
|
let response = pushResponseRes.response.get()
|
|
if not response.isSuccess:
|
|
if response.info.isSome():
|
|
return err(response.info.get())
|
|
else:
|
|
return err("unknown failure")
|
|
|
|
return ok()
|
|
|
|
proc publish*(
|
|
wl: WakuLegacyLightPushClient,
|
|
pubSubTopic: PubsubTopic,
|
|
wakuMessage: WakuMessage,
|
|
peer: RemotePeerInfo,
|
|
): Future[WakuLightPushResult[string]] {.async, gcsafe.} =
|
|
## On success, returns the msg_hash of the published message
|
|
|
|
var message = wakuMessage
|
|
if message.timestamp == 0:
|
|
message.timestamp = getNowInNanosecondTime()
|
|
|
|
let msg_hash_hex_str = computeMessageHash(pubsubTopic, message).to0xHex()
|
|
let pushRequest = PushRequest(pubSubTopic: pubSubTopic, message: message)
|
|
?await wl.sendPushRequest(pushRequest, peer)
|
|
|
|
notice "publishing message with lightpush",
|
|
pubsubTopic = pubsubTopic,
|
|
contentTopic = message.contentTopic,
|
|
target_peer_id = peer.peerId,
|
|
msg_hash = msg_hash_hex_str
|
|
|
|
return ok(msg_hash_hex_str)
|
|
|
|
proc publishToAny*(
|
|
wl: WakuLegacyLightPushClient, pubSubTopic: PubsubTopic, message: WakuMessage
|
|
): Future[WakuLightPushResult[void]] {.async, gcsafe.} =
|
|
## This proc is similar to the publish one but in this case
|
|
## we don't specify a particular peer and instead we get it from peer manager
|
|
|
|
info "publishToAny", msg_hash = computeMessageHash(pubsubTopic, message).to0xHex
|
|
|
|
let peer = wl.peerManager.selectPeer(WakuLegacyLightPushCodec).valueOr:
|
|
return err("could not retrieve a peer supporting WakuLegacyLightPushCodec")
|
|
|
|
let pushRequest = PushRequest(pubSubTopic: pubSubTopic, message: message)
|
|
?await wl.sendPushRequest(pushRequest, peer)
|
|
|
|
return ok()
|