chore: libwaku tweaks (#3233)

* make lightpush return msg hash after successful publish
* libwaku avoid the use of string
* library alloc.nim allocate memory when nil cstring is passed
* libwaku store_request remove extra destroyShared(self)
This commit is contained in:
Ivan FB 2025-01-08 20:52:44 +01:00 committed by GitHub
parent e81a5517be
commit 625c8ee51b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 45 additions and 59 deletions

View File

@ -4,6 +4,11 @@ type SharedSeq*[T] = tuple[data: ptr UncheckedArray[T], len: int]
proc alloc*(str: cstring): cstring = proc alloc*(str: cstring): cstring =
# Byte allocation from the given address. # Byte allocation from the given address.
# There should be the corresponding manual deallocation with deallocShared ! # There should be the corresponding manual deallocation with deallocShared !
if str.isNil():
var ret = cast[cstring](allocShared(1)) # Allocate memory for the null terminator
ret[0] = '\0' # Set the null terminator
return ret
let ret = cast[cstring](allocShared(len(str) + 1)) let ret = cast[cstring](allocShared(len(str) + 1))
copyMem(ret, str, len(str) + 1) copyMem(ret, str, len(str) + 1)
return ret return ret

View File

@ -313,16 +313,10 @@ proc waku_relay_publish(
defer: defer:
deallocShared(pst) deallocShared(pst)
let targetPubSubTopic =
if len(pst) == 0:
DefaultPubsubTopic
else:
$pst
handleRequest( handleRequest(
ctx, ctx,
RequestType.RELAY, RequestType.RELAY,
RelayRequest.createShared(RelayMsgType.PUBLISH, PubsubTopic($pst), nil, wakuMessage), RelayRequest.createShared(RelayMsgType.PUBLISH, pst, nil, wakuMessage),
callback, callback,
userData, userData,
) )
@ -370,9 +364,7 @@ proc waku_relay_subscribe(
handleRequest( handleRequest(
ctx, ctx,
RequestType.RELAY, RequestType.RELAY,
RelayRequest.createShared( RelayRequest.createShared(RelayMsgType.SUBSCRIBE, pst, WakuRelayHandler(cb)),
RelayMsgType.SUBSCRIBE, PubsubTopic($pst), WakuRelayHandler(cb)
),
callback, callback,
userData, userData,
) )
@ -398,7 +390,7 @@ proc waku_relay_add_protected_shard(
RelayMsgType.ADD_PROTECTED_SHARD, RelayMsgType.ADD_PROTECTED_SHARD,
clusterId = clusterId, clusterId = clusterId,
shardId = shardId, shardId = shardId,
publicKey = $pubk, publicKey = pubk,
), ),
callback, callback,
userData, userData,
@ -421,9 +413,7 @@ proc waku_relay_unsubscribe(
ctx, ctx,
RequestType.RELAY, RequestType.RELAY,
RelayRequest.createShared( RelayRequest.createShared(
RelayMsgType.UNSUBSCRIBE, RelayMsgType.UNSUBSCRIBE, pst, WakuRelayHandler(onReceivedMessage(ctx))
PubsubTopic($pst),
WakuRelayHandler(onReceivedMessage(ctx)),
), ),
callback, callback,
userData, userData,
@ -445,7 +435,7 @@ proc waku_relay_get_num_connected_peers(
handleRequest( handleRequest(
ctx, ctx,
RequestType.RELAY, RequestType.RELAY,
RelayRequest.createShared(RelayMsgType.LIST_CONNECTED_PEERS, PubsubTopic($pst)), RelayRequest.createShared(RelayMsgType.LIST_CONNECTED_PEERS, pst),
callback, callback,
userData, userData,
) )
@ -466,7 +456,7 @@ proc waku_relay_get_num_peers_in_mesh(
handleRequest( handleRequest(
ctx, ctx,
RequestType.RELAY, RequestType.RELAY,
RelayRequest.createShared(RelayMsgType.LIST_MESH_PEERS, PubsubTopic($pst)), RelayRequest.createShared(RelayMsgType.LIST_MESH_PEERS, pst),
callback, callback,
userData, userData,
) )
@ -557,18 +547,10 @@ proc waku_lightpush_publish(
callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData) callback(RET_ERR, unsafeAddr msg[0], cast[csize_t](len(msg)), userData)
return RET_ERR return RET_ERR
let targetPubSubTopic =
if len(pst) == 0:
DefaultPubsubTopic
else:
$pst
handleRequest( handleRequest(
ctx, ctx,
RequestType.LIGHTPUSH, RequestType.LIGHTPUSH,
LightpushRequest.createShared( LightpushRequest.createShared(LightpushMsgType.PUBLISH, pst, wakuMessage),
LightpushMsgType.PUBLISH, PubsubTopic($pst), wakuMessage
),
callback, callback,
userData, userData,
) )

View File

@ -32,7 +32,7 @@ type LightpushRequest* = object
proc createShared*( proc createShared*(
T: type LightpushRequest, T: type LightpushRequest,
op: LightpushMsgType, op: LightpushMsgType,
pubsubTopic: PubsubTopic, pubsubTopic: cstring,
m = WakuMessage(), m = WakuMessage(),
): ptr type T = ): ptr type T =
var ret = createShared(T) var ret = createShared(T)
@ -97,12 +97,12 @@ proc process*(
error "PUBLISH failed", error = errorMsg error "PUBLISH failed", error = errorMsg
return err(errorMsg) return err(errorMsg)
( let msgHashHex = (
await waku.node.wakuLightpushClient.publish( await waku.node.wakuLightpushClient.publish(
pubsubTopic, msg, peer = peerOpt.get() pubsubTopic, msg, peer = peerOpt.get()
) )
).isOkOr: ).valueOr:
error "PUBLISH failed", error = error error "PUBLISH failed", error = error
return err("LightpushRequest error publishing: " & $error) return err("LightpushRequest error publishing: " & $error)
return ok("") return ok(msgHashHex)

View File

@ -41,12 +41,12 @@ type RelayRequest* = object
proc createShared*( proc createShared*(
T: type RelayRequest, T: type RelayRequest,
op: RelayMsgType, op: RelayMsgType,
pubsubTopic: PubsubTopic = "", pubsubTopic: cstring = nil,
relayEventCallback: WakuRelayHandler = nil, relayEventCallback: WakuRelayHandler = nil,
m = WakuMessage(), m = WakuMessage(),
clusterId: cint = 0, clusterId: cint = 0,
shardId: cint = 0, shardId: cint = 0,
publicKey: string = "", publicKey: cstring = nil,
): ptr type T = ): ptr type T =
var ret = createShared(T) var ret = createShared(T)
ret[].operation = op ret[].operation = op

View File

@ -106,9 +106,6 @@ proc destroyShared(self: ptr StoreRequest) =
proc process_remote_query( proc process_remote_query(
self: ptr StoreRequest, waku: ptr Waku self: ptr StoreRequest, waku: ptr Waku
): Future[Result[string, string]] {.async.} = ): Future[Result[string, string]] {.async.} =
defer:
destroyShared(self)
let jsonContentRes = catch: let jsonContentRes = catch:
parseJson($self[].jsonQuery) parseJson($self[].jsonQuery)

View File

@ -972,7 +972,7 @@ proc lightpushPublish*(
pubsubTopic: Option[PubsubTopic], pubsubTopic: Option[PubsubTopic],
message: WakuMessage, message: WakuMessage,
peer: RemotePeerInfo, peer: RemotePeerInfo,
): Future[WakuLightPushResult[void]] {.async, gcsafe.} = ): Future[WakuLightPushResult[string]] {.async, gcsafe.} =
## Pushes a `WakuMessage` to a node which relays it further on PubSub topic. ## Pushes a `WakuMessage` to a node which relays it further on PubSub topic.
## Returns whether relaying was successful or not. ## Returns whether relaying was successful or not.
## `WakuMessage` should contain a `contentTopic` field for light node ## `WakuMessage` should contain a `contentTopic` field for light node
@ -986,7 +986,7 @@ proc lightpushPublish*(
pubsubTopic: PubsubTopic, pubsubTopic: PubsubTopic,
message: WakuMessage, message: WakuMessage,
peer: RemotePeerInfo, peer: RemotePeerInfo,
): Future[WakuLightPushResult[void]] {.async, gcsafe.} = ): Future[WakuLightPushResult[string]] {.async, gcsafe.} =
let msgHash = pubsubTopic.computeMessageHash(message).to0xHex() let msgHash = pubsubTopic.computeMessageHash(message).to0xHex()
if not node.wakuLightpushClient.isNil(): if not node.wakuLightpushClient.isNil():
notice "publishing message with lightpush", notice "publishing message with lightpush",
@ -1023,7 +1023,7 @@ proc lightpushPublish*(
# TODO: Move to application module (e.g., wakunode2.nim) # TODO: Move to application module (e.g., wakunode2.nim)
proc lightpushPublish*( proc lightpushPublish*(
node: WakuNode, pubsubTopic: Option[PubsubTopic], message: WakuMessage node: WakuNode, pubsubTopic: Option[PubsubTopic], message: WakuMessage
): Future[WakuLightPushResult[void]] {. ): Future[WakuLightPushResult[string]] {.
async, gcsafe, deprecated: "Use 'node.lightpushPublish()' instead" async, gcsafe, deprecated: "Use 'node.lightpushPublish()' instead"
.} = .} =
if node.wakuLightpushClient.isNil() and node.wakuLightPush.isNil(): if node.wakuLightpushClient.isNil() and node.wakuLightPush.isNil():
@ -1040,13 +1040,7 @@ proc lightpushPublish*(
elif not node.wakuLightPush.isNil(): elif not node.wakuLightPush.isNil():
peerOpt = some(RemotePeerInfo.init($node.switch.peerInfo.peerId)) peerOpt = some(RemotePeerInfo.init($node.switch.peerInfo.peerId))
let publishRes = return await node.lightpushPublish(pubsubTopic, message, peer = peerOpt.get())
await node.lightpushPublish(pubsubTopic, message, peer = peerOpt.get())
if publishRes.isErr():
error "failed to publish message", error = publishRes.error
return publishRes
## Waku RLN Relay ## Waku RLN Relay
proc mountRlnRelay*( proc mountRlnRelay*(

View File

@ -71,24 +71,23 @@ proc publish*(
wl: WakuLightPushClient, wl: WakuLightPushClient,
pubSubTopic: PubsubTopic, pubSubTopic: PubsubTopic,
message: WakuMessage, message: WakuMessage,
peer: PeerId | RemotePeerInfo, peer: RemotePeerInfo,
): Future[WakuLightPushResult[void]] {.async, gcsafe.} = ): Future[WakuLightPushResult[string]] {.async, gcsafe.} =
when peer is PeerId: ## On success, returns the msg_hash of the published message
info "publish", let msg_hash_hex_str = computeMessageHash(pubsubTopic, message).to0xHex()
peerId = shortLog(peer),
msg_hash = computeMessageHash(pubsubTopic, message).to0xHex
else:
info "publish",
peerId = shortLog(peer.peerId),
msg_hash = computeMessageHash(pubsubTopic, message).to0xHex
let pushRequest = PushRequest(pubSubTopic: pubSubTopic, message: message) let pushRequest = PushRequest(pubSubTopic: pubSubTopic, message: message)
?await wl.sendPushRequest(pushRequest, peer) ?await wl.sendPushRequest(pushRequest, peer)
for obs in wl.publishObservers: for obs in wl.publishObservers:
obs.onMessagePublished(pubSubTopic, message) obs.onMessagePublished(pubSubTopic, message)
return ok() 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*( proc publishToAny*(
wl: WakuLightPushClient, pubSubTopic: PubsubTopic, message: WakuMessage wl: WakuLightPushClient, pubSubTopic: PubsubTopic, message: WakuMessage

View File

@ -9,7 +9,7 @@
## which spawn a full service Waku node ## which spawn a full service Waku node
## that could be used also as a lightpush client, helping testing and development. ## that could be used also as a lightpush client, helping testing and development.
import results, chronos, std/options, metrics import results, chronos, chronicles, std/options, metrics, stew/byteutils
import import
../waku_core, ../waku_core,
./protocol, ./protocol,
@ -21,9 +21,10 @@ import
proc handleSelfLightPushRequest*( proc handleSelfLightPushRequest*(
self: WakuLightPush, pubSubTopic: PubsubTopic, message: WakuMessage self: WakuLightPush, pubSubTopic: PubsubTopic, message: WakuMessage
): Future[WakuLightPushResult[void]] {.async.} = ): Future[WakuLightPushResult[string]] {.async.} =
## Handles the lightpush requests made by the node to itself. ## Handles the lightpush requests made by the node to itself.
## Normally used in REST-lightpush requests ## Normally used in REST-lightpush requests
## On success, returns the msg_hash of the published message.
try: try:
# provide self peerId as now this node is used directly, thus there is no light client sender peer. # provide self peerId as now this node is used directly, thus there is no light client sender peer.
@ -45,6 +46,14 @@ proc handleSelfLightPushRequest*(
else: else:
return err("unknown failure") return err("unknown failure")
return ok() let msg_hash_hex_str = computeMessageHash(pubSubTopic, message).to0xHex()
notice "publishing message with self hosted lightpush",
pubsubTopic = pubsubTopic,
contentTopic = message.contentTopic,
self_peer_id = selfPeerId,
msg_hash = msg_hash_hex_str
return ok(msg_hash_hex_str)
except Exception: except Exception:
return err("exception in handleSelfLightPushRequest: " & getCurrentExceptionMsg()) return err("exception in handleSelfLightPushRequest: " & getCurrentExceptionMsg())