mirror of
https://github.com/logos-messaging/logos-delivery.git
synced 2026-08-07 12:23:25 +00:00
FFI: typed kernel_api operations (nim-ffi v0.2.0)
Convert the kernel_api operations — relay/filter/lightpush/store,
peer_manager, discovery, ping, debug/node-info — to typed {.ffi.} procs that
pass parameters directly and ride generic CBOR (no per-op request wrappers).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Ivan FB
parent
c64d156f2a
commit
53c83184d0
@ -1,53 +1,32 @@
|
||||
import std/json
|
||||
import
|
||||
chronicles,
|
||||
chronos,
|
||||
results,
|
||||
eth/p2p/discoveryv5/enr,
|
||||
strutils,
|
||||
libp2p/peerid,
|
||||
metrics,
|
||||
ffi
|
||||
import
|
||||
logos_delivery/waku/waku,
|
||||
logos_delivery/waku/node/waku_node,
|
||||
logos_delivery/waku/node/health_monitor,
|
||||
library/declare_lib
|
||||
## The waku api getters are synchronous and can't fail, so the bodies just wrap
|
||||
## the value; the `{.ffi.}` macro wraps it into the `Future` it must expose.
|
||||
|
||||
proc getMultiaddresses(node: WakuNode): seq[string] =
|
||||
return node.info().listenAddresses
|
||||
proc version*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
return ok(self.waku.version())
|
||||
|
||||
proc getMetrics(): string =
|
||||
{.gcsafe.}:
|
||||
return defaultRegistry.toText() ## defaultRegistry is {.global.} in metrics module
|
||||
proc listen_addresses*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
return ok(self.waku.listenAddresses().join(","))
|
||||
|
||||
proc waku_version(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
return ok(WakuNodeVersionString)
|
||||
proc get_my_enr*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
return ok(self.waku.myEnr())
|
||||
|
||||
proc waku_listen_addresses(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
## returns a comma-separated string of the listen addresses
|
||||
return ok(ctx.myLib[].waku.node.getMultiaddresses().join(","))
|
||||
proc get_my_peerid*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
return ok(self.waku.myPeerId())
|
||||
|
||||
proc waku_get_my_enr(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
return ok(ctx.myLib[].waku.node.enr.toURI())
|
||||
proc get_metrics*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
return ok(self.waku.metrics())
|
||||
|
||||
proc waku_get_my_peerid(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
return ok($ctx.myLib[].waku.node.peerId())
|
||||
|
||||
proc waku_get_metrics(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
return ok(getMetrics())
|
||||
|
||||
proc waku_is_online(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
return ok($ctx.myLib[].waku.healthMonitor.onlineMonitor.amIOnline())
|
||||
proc is_online*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
return ok($self.waku.isOnline())
|
||||
|
||||
@ -1,96 +1,31 @@
|
||||
import logos_delivery/waku/compat/option_valueor
|
||||
import std/json
|
||||
import chronos, chronicles, results, strutils, libp2p/multiaddress, ffi
|
||||
import
|
||||
logos_delivery/waku/waku,
|
||||
logos_delivery/waku/discovery/waku_dnsdisc,
|
||||
logos_delivery/waku/discovery/waku_discv5,
|
||||
logos_delivery/waku/waku_core/peers,
|
||||
logos_delivery/waku/waku_node,
|
||||
library/declare_lib
|
||||
proc discv5_update_bootnodes*(
|
||||
self: LogosDelivery, bootnodes: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
## `bootnodes` is a JSON array of ENRs, e.g. `["enr:...", "enr:..."]`.
|
||||
(await self.waku.discv5UpdateBootnodes(bootnodes)).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc retrieveBootstrapNodes(
|
||||
enrTreeUrl: string, ipDnsServer: string
|
||||
): Future[Result[seq[string], string]] {.async.} =
|
||||
let dnsNameServers = @[parseIpAddress(ipDnsServer)]
|
||||
let discoveredPeers: seq[RemotePeerInfo] = (
|
||||
await retrieveDynamicBootstrapNodes(enrTreeUrl, dnsNameServers)
|
||||
).valueOr:
|
||||
return err("failed discovering peers from DNS: " & $error)
|
||||
|
||||
var multiAddresses = newSeq[string]()
|
||||
|
||||
for discPeer in discoveredPeers:
|
||||
for address in discPeer.addrs:
|
||||
multiAddresses.add($address & "/p2p/" & $discPeer)
|
||||
|
||||
return ok(multiAddresses)
|
||||
|
||||
proc updateDiscv5BootstrapNodes(nodes: string, waku: Waku): Result[void, string] =
|
||||
waku.wakuDiscv5.updateBootstrapRecords(nodes).isOkOr:
|
||||
return err("error in updateDiscv5BootstrapNodes: " & $error)
|
||||
return ok()
|
||||
|
||||
proc performPeerExchangeRequestTo*(
|
||||
numPeers: uint64, waku: Waku
|
||||
): Future[Result[int, string]] {.async.} =
|
||||
let numPeersRecv = (await waku.node.fetchPeerExchangePeers(numPeers)).valueOr:
|
||||
return err($error)
|
||||
return ok(numPeersRecv)
|
||||
|
||||
proc waku_discv5_update_bootnodes(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
bootnodes: cstring,
|
||||
) {.ffi.} =
|
||||
## Updates the bootnode list used for discovering new peers via DiscoveryV5
|
||||
## bootnodes - JSON array containing the bootnode ENRs i.e. `["enr:...", "enr:..."]`
|
||||
|
||||
updateDiscv5BootstrapNodes($bootnodes, ctx.myLib[].waku).isOkOr:
|
||||
error "UPDATE_DISCV5_BOOTSTRAP_NODES failed", error = error
|
||||
return err($error)
|
||||
|
||||
return ok("discovery request processed correctly")
|
||||
|
||||
proc waku_dns_discovery(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
enrTreeUrl: cstring,
|
||||
nameDnsServer: cstring,
|
||||
timeoutMs: cint,
|
||||
) {.ffi.} =
|
||||
let nodes = (await retrieveBootstrapNodes($enrTreeUrl, $nameDnsServer)).valueOr:
|
||||
error "GET_BOOTSTRAP_NODES failed", error = error
|
||||
return err($error)
|
||||
|
||||
## returns a comma-separated string of bootstrap nodes' multiaddresses
|
||||
proc dns_discovery*(
|
||||
self: LogosDelivery, enrTreeUrl: string, nameDnsServer: string, timeoutMs: int
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let nodes = (await self.waku.dnsDiscovery(enrTreeUrl, nameDnsServer, timeoutMs)).valueOr:
|
||||
return err(error)
|
||||
return ok(nodes.join(","))
|
||||
|
||||
proc waku_start_discv5(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
(await ctx.myLib[].waku.wakuDiscv5.start()).isOkOr:
|
||||
error "START_DISCV5 failed", error = error
|
||||
return err("error starting discv5: " & $error)
|
||||
proc start_discv5*(self: LogosDelivery): Future[Result[string, string]] {.ffi.} =
|
||||
(await self.waku.startDiscv5()).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
return ok("discv5 started correctly")
|
||||
proc stop_discv5*(self: LogosDelivery): Future[Result[string, string]] {.ffi.} =
|
||||
(await self.waku.stopDiscv5()).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_stop_discv5(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
await ctx.myLib[].waku.wakuDiscv5.stop()
|
||||
return ok("discv5 stopped correctly")
|
||||
|
||||
proc waku_peer_exchange_request(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
numPeers: uint64,
|
||||
) {.ffi.} =
|
||||
let numValidPeers = (await performPeerExchangeRequestTo(numPeers, ctx.myLib[].waku)).valueOr:
|
||||
error "waku_peer_exchange_request failed", error = error
|
||||
return err("failed peer exchange: " & $error)
|
||||
|
||||
return ok($numValidPeers)
|
||||
proc peer_exchange_request*(
|
||||
self: LogosDelivery, numPeers: uint64
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let n = (await self.waku.peerExchangeRequest(numPeers)).valueOr:
|
||||
return err(error)
|
||||
return ok($n)
|
||||
|
||||
34
library/kernel_api/node_info_api.nim
Normal file
34
library/kernel_api/node_info_api.nim
Normal file
@ -0,0 +1,34 @@
|
||||
import std/json
|
||||
import logos_delivery/waku/factory/waku_state_info
|
||||
import tools/confutils/[cli_args, config_option_meta]
|
||||
|
||||
proc get_available_node_info_ids*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
## All node-info item ids that can be queried with `get_node_info`.
|
||||
return ok($self.waku.stateInfo.getAllPossibleInfoItemIds())
|
||||
|
||||
proc get_node_info*(
|
||||
self: LogosDelivery, nodeInfoId: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let infoItemIdEnum =
|
||||
try:
|
||||
parseEnum[NodeInfoId](nodeInfoId)
|
||||
except ValueError:
|
||||
return err("Invalid node info id: " & nodeInfoId)
|
||||
return ok(self.waku.stateInfo.getNodeInfoItem(infoItemIdEnum))
|
||||
|
||||
proc get_available_configs*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let optionMetas: seq[ConfigOptionMeta] = extractConfigOptionMeta(WakuNodeConf)
|
||||
var configOptionDetails = newJArray()
|
||||
for meta in optionMetas:
|
||||
configOptionDetails.add(
|
||||
%*{
|
||||
meta.fieldName: meta.typeName & "(" & meta.defaultValue & ")", "desc": meta.desc
|
||||
}
|
||||
)
|
||||
var jsonNode = newJObject()
|
||||
jsonNode["configOptions"] = configOptionDetails
|
||||
return ok(pretty(jsonNode))
|
||||
@ -1,133 +1,76 @@
|
||||
import logos_delivery/waku/compat/option_valueor
|
||||
import std/[sequtils, strutils, tables]
|
||||
import chronicles, chronos, results, options, json, ffi
|
||||
import
|
||||
logos_delivery/waku/waku,
|
||||
logos_delivery/waku/node/waku_node,
|
||||
logos_delivery/waku/node/peer_manager,
|
||||
library/declare_lib
|
||||
import std/sequtils
|
||||
|
||||
type PeerInfo = object
|
||||
protocols: seq[string]
|
||||
addresses: seq[string]
|
||||
type ConnectedPeersInfoResponse {.ffi.} = object
|
||||
peers: seq[PeerConnInfoFFI]
|
||||
|
||||
proc waku_get_peerids_from_peerstore(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
## returns a comma-separated string of peerIDs
|
||||
let peerIDs = ctx.myLib[].waku.node.peerManager.switch.peerStore
|
||||
.peers()
|
||||
.mapIt($it.peerId)
|
||||
.join(",")
|
||||
return ok(peerIDs)
|
||||
proc get_peerids_from_peerstore*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let ids = (await self.waku.peerIdsFromPeerstore()).valueOr:
|
||||
return err(error)
|
||||
return ok(ids.join(","))
|
||||
|
||||
proc waku_connect(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
peerMultiAddr: cstring,
|
||||
timeoutMs: cuint,
|
||||
) {.ffi.} =
|
||||
let peers = ($peerMultiAddr).split(",").mapIt(strip(it))
|
||||
await ctx.myLib[].waku.node.connectToNodes(peers, source = "static")
|
||||
proc connect_peers*(
|
||||
self: LogosDelivery, peers: seq[string], timeoutMs: uint32
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
## `peers` are multiaddrs.
|
||||
(await self.waku.connect(peers, timeoutMs)).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_disconnect_peer_by_id(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
peerId: cstring,
|
||||
) {.ffi.} =
|
||||
let pId = PeerId.init($peerId).valueOr:
|
||||
error "DISCONNECT_PEER_BY_ID failed", error = $error
|
||||
return err($error)
|
||||
await ctx.myLib[].waku.node.peerManager.disconnectNode(pId)
|
||||
proc disconnect_peer_by_id*(
|
||||
self: LogosDelivery, peerId: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
(await self.waku.disconnectPeerById(peerId)).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_disconnect_all_peers(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
await ctx.myLib[].waku.node.peerManager.disconnectAllPeers()
|
||||
proc disconnect_all_peers*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
(await self.waku.disconnectAllPeers()).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_dial_peer(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
peerMultiAddr: cstring,
|
||||
protocol: cstring,
|
||||
timeoutMs: cuint,
|
||||
) {.ffi.} =
|
||||
let remotePeerInfo = parsePeerInfo($peerMultiAddr).valueOr:
|
||||
error "DIAL_PEER failed", error = $error
|
||||
return err($error)
|
||||
let conn = await ctx.myLib[].waku.node.peerManager.dialPeer(remotePeerInfo, $protocol)
|
||||
if conn.isNone():
|
||||
let msg = "failed dialing peer"
|
||||
error "DIAL_PEER failed", error = msg, peerId = $remotePeerInfo.peerId
|
||||
return err(msg)
|
||||
proc dial_peer*(
|
||||
self: LogosDelivery, peer: string, protocol: string, timeoutMs: int
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
## `peer` is a multiaddr.
|
||||
(await self.waku.dialPeer(peer, protocol, timeoutMs)).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_dial_peer_by_id(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
peerId: cstring,
|
||||
protocol: cstring,
|
||||
timeoutMs: cuint,
|
||||
) {.ffi.} =
|
||||
let pId = PeerId.init($peerId).valueOr:
|
||||
error "DIAL_PEER_BY_ID failed", error = $error
|
||||
return err($error)
|
||||
let conn = await ctx.myLib[].waku.node.peerManager.dialPeer(pId, $protocol)
|
||||
if conn.isNone():
|
||||
let msg = "failed dialing peer"
|
||||
error "DIAL_PEER_BY_ID failed", error = msg, peerId = $peerId
|
||||
return err(msg)
|
||||
|
||||
proc dial_peer_by_id*(
|
||||
self: LogosDelivery, peer: string, protocol: string, timeoutMs: int
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
## `peer` is a peerId.
|
||||
(await self.waku.dialPeerById(peer, protocol, timeoutMs)).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_get_connected_peers_info(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
## returns a JSON string mapping peerIDs to objects with protocols and addresses
|
||||
|
||||
var peersMap = initTable[string, PeerInfo]()
|
||||
let peers = ctx.myLib[].waku.node.peerManager.switch.peerStore.peers().filterIt(
|
||||
it.connectedness == Connected
|
||||
proc get_connected_peers_info*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[ConnectedPeersInfoResponse, string]] {.ffi.} =
|
||||
let infos = (await self.waku.connectedPeersInfo()).valueOr:
|
||||
return err(error)
|
||||
return ok(
|
||||
ConnectedPeersInfoResponse(
|
||||
peers: infos.mapIt(
|
||||
PeerConnInfoFFI(
|
||||
peerId: it.peerId, protocols: it.protocols, addresses: it.addresses
|
||||
)
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
# Build a map of peer IDs to peer info objects
|
||||
for peer in peers:
|
||||
let peerIdStr = $peer.peerId
|
||||
peersMap[peerIdStr] =
|
||||
PeerInfo(protocols: peer.protocols, addresses: peer.addrs.mapIt($it))
|
||||
proc get_connected_peers*(self: LogosDelivery): Future[Result[string, string]] {.ffi.} =
|
||||
let ids = (await self.waku.connectedPeers()).valueOr:
|
||||
return err(error)
|
||||
return ok(ids.join(","))
|
||||
|
||||
# Convert the map to JSON string
|
||||
let jsonObj = %*peersMap
|
||||
let jsonStr = $jsonObj
|
||||
return ok(jsonStr)
|
||||
|
||||
proc waku_get_connected_peers(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
## returns a comma-separated string of peerIDs
|
||||
let
|
||||
(inPeerIds, outPeerIds) = ctx.myLib[].waku.node.peerManager.connectedPeers()
|
||||
connectedPeerids = concat(inPeerIds, outPeerIds)
|
||||
|
||||
return ok(connectedPeerids.mapIt($it).join(","))
|
||||
|
||||
proc waku_get_peerids_by_protocol(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
protocol: cstring,
|
||||
) {.ffi.} =
|
||||
## returns a comma-separated string of peerIDs that mount the given protocol
|
||||
let connectedPeers = ctx.myLib[].waku.node.peerManager.switch.peerStore
|
||||
.peers($protocol)
|
||||
.filterIt(it.connectedness == Connected)
|
||||
.mapIt($it.peerId)
|
||||
.join(",")
|
||||
return ok(connectedPeers)
|
||||
proc get_peerids_by_protocol*(
|
||||
self: LogosDelivery, protocol: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let ids = (await self.waku.peerIdsByProtocol(protocol)).valueOr:
|
||||
return err(error)
|
||||
return ok(ids.join(","))
|
||||
|
||||
@ -1,44 +1,7 @@
|
||||
import std/[json, strutils]
|
||||
import chronos, results, ffi
|
||||
import libp2p/[protocols/ping, switch, multiaddress, multicodec]
|
||||
import logos_delivery/waku/[waku, waku_core/peers, node/waku_node], library/declare_lib
|
||||
|
||||
proc waku_ping_peer(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
peerAddr: cstring,
|
||||
timeoutMs: cuint,
|
||||
) {.ffi.} =
|
||||
let peerInfo = peers.parsePeerInfo(($peerAddr).split(",")).valueOr:
|
||||
return err("PingRequest failed to parse peer addr: " & $error)
|
||||
|
||||
let timeout = chronos.milliseconds(timeoutMs)
|
||||
proc ping(): Future[Result[Duration, string]] {.async, gcsafe.} =
|
||||
try:
|
||||
let conn = await ctx.myLib[].waku.node.switch.dial(
|
||||
peerInfo.peerId, peerInfo.addrs, PingCodec
|
||||
)
|
||||
defer:
|
||||
await conn.close()
|
||||
|
||||
let pingRTT = await ctx.myLib[].waku.node.libp2pPing.ping(conn)
|
||||
if pingRTT == 0.nanos:
|
||||
return err("could not ping peer: rtt-0")
|
||||
return ok(pingRTT)
|
||||
except CatchableError as exc:
|
||||
return err("could not ping peer: " & exc.msg)
|
||||
|
||||
let pingFuture = ping()
|
||||
let pingRTT: Duration =
|
||||
if timeout == chronos.milliseconds(0): # No timeout expected
|
||||
(await pingFuture).valueOr:
|
||||
return err("ping failed, no timeout expected: " & error)
|
||||
else:
|
||||
let timedOut = not (await pingFuture.withTimeout(timeout))
|
||||
if timedOut:
|
||||
return err("ping timed out")
|
||||
pingFuture.read().valueOr:
|
||||
return err("failed to read ping future: " & error)
|
||||
|
||||
return ok($(pingRTT.nanos))
|
||||
proc ping_peer*(
|
||||
self: LogosDelivery, peerAddr: string, timeoutMs: int
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
## Returns the round-trip time in nanoseconds.
|
||||
let rtt = (await self.waku.pingPeer(peerAddr, timeoutMs)).valueOr:
|
||||
return err(error)
|
||||
return ok($rtt)
|
||||
|
||||
@ -1,109 +1,39 @@
|
||||
import logos_delivery/waku/compat/option_valueor
|
||||
import options, std/[strutils, sequtils]
|
||||
import chronicles, chronos, results, ffi
|
||||
import
|
||||
logos_delivery/waku/waku_filter_v2/client,
|
||||
logos_delivery/waku/waku_core/message/message,
|
||||
logos_delivery/waku/waku,
|
||||
logos_delivery/waku/waku_relay,
|
||||
logos_delivery/waku/waku_filter_v2/common,
|
||||
logos_delivery/waku/waku_core/subscription/push_handler,
|
||||
logos_delivery/waku/node/peer_manager/peer_manager,
|
||||
logos_delivery/waku/waku_node,
|
||||
logos_delivery/waku/waku_core/topics/pubsub_topic,
|
||||
logos_delivery/waku/waku_core/topics/content_topic,
|
||||
library/events/json_message_event,
|
||||
library/declare_lib
|
||||
import std/sequtils
|
||||
import logos_delivery/waku/waku_core/subscription/push_handler
|
||||
|
||||
const FilterOpTimeout = 5.seconds
|
||||
|
||||
proc checkFilterClientMounted(waku: Waku): Result[string, string] =
|
||||
if waku.node.wakuFilterClient.isNil():
|
||||
let errorMsg = "wakuFilterClient is not mounted"
|
||||
error "fail filter process", error = errorMsg
|
||||
return err(errorMsg)
|
||||
proc filter_subscribe*(
|
||||
self: LogosDelivery, pubsubTopic: string, contentTopics: seq[string]
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
# `filterSubscribe` re-registers the filter push handler, so it must keep
|
||||
# feeding MessageSeenEvent — the single source the ctor's listener delivers
|
||||
# to the foreign side (see liblogosdelivery.nim).
|
||||
let brokerCtx = self.waku.brokerCtx
|
||||
let pushHandler = proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} =
|
||||
MessageSeenEvent.emit(brokerCtx, pubsubTopic, msg)
|
||||
(
|
||||
await self.waku.filterSubscribe(
|
||||
PubsubTopic(pubsubTopic),
|
||||
contentTopics.mapIt(ContentTopic(it)),
|
||||
FilterPushHandler(pushHandler),
|
||||
)
|
||||
).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_filter_subscribe(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
contentTopics: cstring,
|
||||
) {.ffi.} =
|
||||
proc onReceivedMessage(ctx: ptr FFIContext): WakuRelayHandler =
|
||||
return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} =
|
||||
callEventCallback(ctx, "onReceivedMessage"):
|
||||
$JsonMessageEvent.new(pubsubTopic, msg)
|
||||
|
||||
checkFilterClientMounted(ctx.myLib[].waku).isOkOr:
|
||||
return err($error)
|
||||
|
||||
var filterPushEventCallback = FilterPushHandler(onReceivedMessage(ctx))
|
||||
ctx.myLib[].waku.node.wakuFilterClient.registerPushHandler(filterPushEventCallback)
|
||||
|
||||
let peer = ctx.myLib[].waku.node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr:
|
||||
let errorMsg = "could not find peer with WakuFilterSubscribeCodec when subscribing"
|
||||
error "fail filter subscribe", error = errorMsg
|
||||
return err(errorMsg)
|
||||
|
||||
let subFut = ctx.myLib[].waku.node.filterSubscribe(
|
||||
some(PubsubTopic($pubsubTopic)),
|
||||
($contentTopics).split(",").mapIt(ContentTopic(it)),
|
||||
peer,
|
||||
)
|
||||
if not await subFut.withTimeout(FilterOpTimeout):
|
||||
let errorMsg = "filter subscription timed out"
|
||||
error "fail filter unsubscribe", error = errorMsg
|
||||
|
||||
return err(errorMsg)
|
||||
|
||||
proc filter_unsubscribe*(
|
||||
self: LogosDelivery, pubsubTopic: string, contentTopics: seq[string]
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
(
|
||||
await self.waku.filterUnsubscribe(
|
||||
PubsubTopic(pubsubTopic), contentTopics.mapIt(ContentTopic(it))
|
||||
)
|
||||
).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_filter_unsubscribe(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
contentTopics: cstring,
|
||||
) {.ffi.} =
|
||||
checkFilterClientMounted(ctx.myLib[].waku).isOkOr:
|
||||
return err($error)
|
||||
|
||||
let peer = ctx.myLib[].waku.node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr:
|
||||
let errorMsg =
|
||||
"could not find peer with WakuFilterSubscribeCodec when unsubscribing"
|
||||
error "fail filter process", error = errorMsg
|
||||
return err(errorMsg)
|
||||
|
||||
let subFut = ctx.myLib[].waku.node.filterUnsubscribe(
|
||||
some(PubsubTopic($pubsubTopic)),
|
||||
($contentTopics).split(",").mapIt(ContentTopic(it)),
|
||||
peer,
|
||||
)
|
||||
if not await subFut.withTimeout(FilterOpTimeout):
|
||||
let errorMsg = "filter un-subscription timed out"
|
||||
error "fail filter unsubscribe", error = errorMsg
|
||||
return err(errorMsg)
|
||||
return ok("")
|
||||
|
||||
proc waku_filter_unsubscribe_all(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
checkFilterClientMounted(ctx.myLib[].waku).isOkOr:
|
||||
return err($error)
|
||||
|
||||
let peer = ctx.myLib[].waku.node.peerManager.selectPeer(WakuFilterSubscribeCodec).valueOr:
|
||||
let errorMsg =
|
||||
"could not find peer with WakuFilterSubscribeCodec when unsubscribing all"
|
||||
error "fail filter unsubscribe all", error = errorMsg
|
||||
return err(errorMsg)
|
||||
|
||||
let unsubFut = ctx.myLib[].waku.node.filterUnsubscribeAll(peer)
|
||||
|
||||
if not await unsubFut.withTimeout(FilterOpTimeout):
|
||||
let errorMsg = "filter un-subscription all timed out"
|
||||
error "fail filter unsubscribe all", error = errorMsg
|
||||
|
||||
return err(errorMsg)
|
||||
proc filter_unsubscribe_all*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
(await self.waku.filterUnsubscribeAll()).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
@ -1,52 +1,7 @@
|
||||
import logos_delivery/waku/compat/option_valueor
|
||||
import options, std/[json, strformat]
|
||||
import chronicles, chronos, results, ffi
|
||||
import
|
||||
logos_delivery/waku/waku_core/message/message,
|
||||
logos_delivery/waku/waku_core/codecs,
|
||||
logos_delivery/waku/waku,
|
||||
logos_delivery/waku/waku_core/message,
|
||||
logos_delivery/waku/waku_core/topics/pubsub_topic,
|
||||
logos_delivery/waku/waku_lightpush_legacy/client,
|
||||
logos_delivery/waku/node/peer_manager/peer_manager,
|
||||
library/events/json_message_event,
|
||||
library/declare_lib
|
||||
|
||||
proc waku_lightpush_publish(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
jsonWakuMessage: cstring,
|
||||
) {.ffi.} =
|
||||
if ctx.myLib[].waku.node.wakuLightpushClient.isNil():
|
||||
let errorMsg = "LightpushRequest waku.node.wakuLightpushClient is nil"
|
||||
error "PUBLISH failed", error = errorMsg
|
||||
return err(errorMsg)
|
||||
|
||||
var jsonMessage: JsonMessage
|
||||
try:
|
||||
let jsonContent = parseJson($jsonWakuMessage)
|
||||
jsonMessage = JsonMessage.fromJsonNode(jsonContent).valueOr:
|
||||
raise newException(JsonParsingError, $error)
|
||||
except JsonParsingError as exc:
|
||||
return err(fmt"Error parsing json message: {exc.msg}")
|
||||
|
||||
let msg = json_message_event.toWakuMessage(jsonMessage).valueOr:
|
||||
return err("Problem building the WakuMessage: " & $error)
|
||||
|
||||
let peerOpt = ctx.myLib[].waku.node.peerManager.selectPeer(WakuLightPushCodec)
|
||||
if peerOpt.isNone():
|
||||
let errorMsg = "failed to lightpublish message, no suitable remote peers"
|
||||
error "PUBLISH failed", error = errorMsg
|
||||
return err(errorMsg)
|
||||
|
||||
let msgHashHex = (
|
||||
await ctx.myLib[].waku.node.wakuLegacyLightpushClient.publish(
|
||||
$pubsubTopic, msg, peer = peerOpt.get()
|
||||
)
|
||||
).valueOr:
|
||||
error "PUBLISH failed", error = error
|
||||
return err($error)
|
||||
|
||||
return ok(msgHashHex)
|
||||
proc lightpush_publish*(
|
||||
self: LogosDelivery, pubsubTopic: string, message: WakuMessage
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
## Returns the published message hash.
|
||||
let hash = (await self.waku.lightpushPublish(PubsubTopic(pubsubTopic), message)).valueOr:
|
||||
return err(error)
|
||||
return ok(hash)
|
||||
|
||||
@ -1,175 +1,85 @@
|
||||
import logos_delivery/waku/compat/option_valueor
|
||||
import std/[net, sequtils, strutils, json], strformat
|
||||
import chronicles, chronos, stew/byteutils, results, ffi
|
||||
import
|
||||
logos_delivery/waku/waku_core/message/message,
|
||||
logos_delivery/waku/factory/validator_signed,
|
||||
logos_delivery/waku/waku,
|
||||
tools/confutils/cli_args,
|
||||
logos_delivery/waku/waku_core/message,
|
||||
logos_delivery/waku/waku_core/topics/pubsub_topic,
|
||||
logos_delivery/waku/waku_core/topics,
|
||||
logos_delivery/waku/node/waku_node/relay,
|
||||
logos_delivery/waku/waku_relay/protocol,
|
||||
logos_delivery/waku/node/peer_manager,
|
||||
library/events/json_message_event,
|
||||
library/declare_lib
|
||||
proc relay_get_peers_in_mesh*(
|
||||
self: LogosDelivery, pubsubTopic: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let peers = (await self.waku.relayPeersInMesh(PubsubTopic(pubsubTopic))).valueOr:
|
||||
return err(error)
|
||||
return ok(peers.join(","))
|
||||
|
||||
proc waku_relay_get_peers_in_mesh(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
) {.ffi.} =
|
||||
let meshPeers = ctx.myLib[].waku.node.wakuRelay.getPeersInMesh($pubsubTopic).valueOr:
|
||||
error "LIST_MESH_PEERS failed", error = error
|
||||
return err($error)
|
||||
## returns a comma-separated string of peerIDs
|
||||
return ok(meshPeers.mapIt($it).join(","))
|
||||
proc relay_get_num_peers_in_mesh*(
|
||||
self: LogosDelivery, pubsubTopic: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let n = (await self.waku.relayNumPeersInMesh(PubsubTopic(pubsubTopic))).valueOr:
|
||||
return err(error)
|
||||
return ok($n)
|
||||
|
||||
proc waku_relay_get_num_peers_in_mesh(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
) {.ffi.} =
|
||||
let numPeersInMesh = ctx.myLib[].waku.node.wakuRelay.getNumPeersInMesh($pubsubTopic).valueOr:
|
||||
error "NUM_MESH_PEERS failed", error = error
|
||||
return err($error)
|
||||
return ok($numPeersInMesh)
|
||||
proc relay_get_connected_peers*(
|
||||
self: LogosDelivery, pubsubTopic: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let peers = (await self.waku.relayConnectedPeers(PubsubTopic(pubsubTopic))).valueOr:
|
||||
return err(error)
|
||||
return ok(peers.join(","))
|
||||
|
||||
proc waku_relay_get_connected_peers(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
) {.ffi.} =
|
||||
## Returns the list of all connected peers to an specific pubsub topic
|
||||
let connPeers = ctx.myLib[].waku.node.wakuRelay.getConnectedPeers($pubsubTopic).valueOr:
|
||||
error "LIST_CONNECTED_PEERS failed", error = error
|
||||
return err($error)
|
||||
## returns a comma-separated string of peerIDs
|
||||
return ok(connPeers.mapIt($it).join(","))
|
||||
|
||||
proc waku_relay_get_num_connected_peers(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
) {.ffi.} =
|
||||
let numConnPeers = ctx.myLib[].waku.node.wakuRelay.getNumConnectedPeers($pubsubTopic).valueOr:
|
||||
error "NUM_CONNECTED_PEERS failed", error = error
|
||||
return err($error)
|
||||
return ok($numConnPeers)
|
||||
|
||||
proc waku_relay_add_protected_shard(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
clusterId: cint,
|
||||
shardId: cint,
|
||||
publicKey: cstring,
|
||||
) {.ffi.} =
|
||||
## Protects a shard with a public key
|
||||
try:
|
||||
let relayShard = RelayShard(clusterId: uint16(clusterId), shardId: uint16(shardId))
|
||||
let protectedShard = ProtectedShard.parseCmdArg($relayShard & ":" & $publicKey)
|
||||
ctx.myLib[].waku.node.wakuRelay.addSignedShardsValidator(
|
||||
@[protectedShard], uint16(clusterId)
|
||||
)
|
||||
except ValueError as exc:
|
||||
return err("ERROR in waku_relay_add_protected_shard: " & exc.msg)
|
||||
proc relay_get_num_connected_peers*(
|
||||
self: LogosDelivery, pubsubTopic: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let n = (await self.waku.relayNumConnectedPeers(PubsubTopic(pubsubTopic))).valueOr:
|
||||
return err(error)
|
||||
return ok($n)
|
||||
|
||||
proc relay_add_protected_shard*(
|
||||
self: LogosDelivery, clusterId: uint16, shardId: uint16, publicKey: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
(await self.waku.relayAddProtectedShard(clusterId, shardId, publicKey)).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_relay_subscribe(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
) {.ffi.} =
|
||||
echo "Subscribing to topic: " & $pubSubTopic & " ..."
|
||||
proc onReceivedMessage(ctx: ptr FFIContext[LogosDelivery]): WakuRelayHandler =
|
||||
return proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async.} =
|
||||
callEventCallback(ctx, "onReceivedMessage"):
|
||||
$JsonMessageEvent.new(pubsubTopic, msg)
|
||||
|
||||
var cb = onReceivedMessage(ctx)
|
||||
|
||||
ctx.myLib[].waku.node.subscribe(
|
||||
(kind: SubscriptionKind.PubsubSub, topic: $pubsubTopic),
|
||||
handler = WakuRelayHandler(cb),
|
||||
).isOkOr:
|
||||
error "SUBSCRIBE failed", error = error
|
||||
return err($error)
|
||||
proc relay_subscribe*(
|
||||
self: LogosDelivery, pubsubTopic: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
# Just establishes the subscription; delivery flows through the global
|
||||
# MessageSeenEvent listener (see the ctor in liblogosdelivery.nim).
|
||||
(await self.waku.relaySubscribe(PubsubTopic(pubsubTopic))).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_relay_unsubscribe(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
) {.ffi.} =
|
||||
ctx.myLib[].waku.node.unsubscribe(
|
||||
(kind: SubscriptionKind.PubsubSub, topic: $pubsubTopic)
|
||||
).isOkOr:
|
||||
error "UNSUBSCRIBE failed", error = error
|
||||
return err($error)
|
||||
|
||||
proc relay_unsubscribe*(
|
||||
self: LogosDelivery, pubsubTopic: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
(await self.waku.relayUnsubscribe(PubsubTopic(pubsubTopic))).isOkOr:
|
||||
return err(error)
|
||||
return ok("")
|
||||
|
||||
proc waku_relay_publish(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
pubSubTopic: cstring,
|
||||
jsonWakuMessage: cstring,
|
||||
timeoutMs: cuint,
|
||||
) {.ffi.} =
|
||||
var
|
||||
# https://rfc.vac.dev/spec/36/#extern-char-waku_relay_publishchar-messagejson-char-pubsubtopic-int-timeoutms
|
||||
jsonMessage: JsonMessage
|
||||
try:
|
||||
let jsonContent = parseJson($jsonWakuMessage)
|
||||
jsonMessage = JsonMessage.fromJsonNode(jsonContent).valueOr:
|
||||
raise newException(JsonParsingError, $error)
|
||||
except JsonParsingError as exc:
|
||||
return err(fmt"Error parsing json message: {exc.msg}")
|
||||
proc relay_publish*(
|
||||
self: LogosDelivery, pubsubTopic: string, message: WakuMessage, timeoutMs: uint32
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
## Returns the published message hash (0x-hex).
|
||||
let hash = (
|
||||
await self.waku.relayPublish(PubsubTopic(pubsubTopic), message, timeoutMs)
|
||||
).valueOr:
|
||||
return err(error)
|
||||
return ok(hash)
|
||||
|
||||
let msg = json_message_event.toWakuMessage(jsonMessage).valueOr:
|
||||
return err("Problem building the WakuMessage: " & $error)
|
||||
proc relay_default_pubsub_topic*(
|
||||
self: LogosDelivery
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
return ok(string(self.waku.defaultPubsubTopic()))
|
||||
|
||||
(await ctx.myLib[].waku.node.wakuRelay.publish($pubsubTopic, msg)).isOkOr:
|
||||
error "PUBLISH failed", error = error
|
||||
return err($error)
|
||||
proc relay_content_topic*(
|
||||
self: LogosDelivery,
|
||||
appName: string,
|
||||
appVersion: uint32,
|
||||
contentTopicName: string,
|
||||
encoding: string,
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let contentTopic = self.waku.buildContentTopic(
|
||||
appName, appVersion, contentTopicName, encoding
|
||||
).valueOr:
|
||||
return err(error)
|
||||
return ok(string(contentTopic))
|
||||
|
||||
let msgHash = computeMessageHash($pubSubTopic, msg).to0xHex
|
||||
return ok(msgHash)
|
||||
|
||||
proc waku_default_pubsub_topic(
|
||||
ctx: ptr FFIContext[LogosDelivery], callback: FFICallBack, userData: pointer
|
||||
) {.ffi.} =
|
||||
# https://rfc.vac.dev/spec/36/#extern-char-waku_default_pubsub_topic
|
||||
return ok(DefaultPubsubTopic)
|
||||
|
||||
proc waku_content_topic(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
appName: cstring,
|
||||
appVersion: cuint,
|
||||
contentTopicName: cstring,
|
||||
encoding: cstring,
|
||||
) {.ffi.} =
|
||||
# https://rfc.vac.dev/spec/36/#extern-char-waku_content_topicchar-applicationname-unsigned-int-applicationversion-char-contenttopicname-char-encoding
|
||||
|
||||
return ok(fmt"/{$appName}/{$appVersion}/{$contentTopicName}/{$encoding}")
|
||||
|
||||
proc waku_pubsub_topic(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
topicName: cstring,
|
||||
) {.ffi.} =
|
||||
# https://rfc.vac.dev/spec/36/#extern-char-waku_pubsub_topicchar-name-char-encoding
|
||||
return ok(fmt"/waku/2/{$topicName}")
|
||||
proc relay_pubsub_topic*(
|
||||
self: LogosDelivery, topicName: string
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let pubsubTopic = self.waku.buildPubsubTopic(topicName).valueOr:
|
||||
return err(error)
|
||||
return ok(string(pubsubTopic))
|
||||
|
||||
@ -1,17 +1,12 @@
|
||||
import logos_delivery/waku/compat/option_valueor
|
||||
import std/[json, sugar, strutils, options]
|
||||
import chronos, chronicles, results, stew/byteutils, ffi
|
||||
import
|
||||
logos_delivery/waku/waku,
|
||||
library/utils,
|
||||
logos_delivery/waku/waku_core/peers,
|
||||
logos_delivery/waku/waku_core/message/digest,
|
||||
logos_delivery/waku/waku_store/common,
|
||||
logos_delivery/waku/waku_store/client,
|
||||
logos_delivery/waku/common/paging,
|
||||
library/declare_lib
|
||||
## The query/response are complex types, so this keeps the JSON bridge: the
|
||||
## request carries the query as a JSON string, the response is returned as JSON.
|
||||
import std/[json, sugar, options]
|
||||
import logos_delivery/waku/waku_core/message/digest
|
||||
import logos_delivery/waku/waku_store/common
|
||||
import logos_delivery/waku/common/paging
|
||||
import library/utils
|
||||
|
||||
func fromJsonNode(jsonContent: JsonNode): Result[StoreQueryRequest, string] =
|
||||
func storeQueryFromJson(jsonContent: JsonNode): Result[StoreQueryRequest, string] =
|
||||
var contentTopics: seq[string]
|
||||
if jsonContent.contains("contentTopics"):
|
||||
contentTopics = collect(newSeq):
|
||||
@ -67,29 +62,19 @@ func fromJsonNode(jsonContent: JsonNode): Result[StoreQueryRequest, string] =
|
||||
)
|
||||
)
|
||||
|
||||
proc waku_store_query(
|
||||
ctx: ptr FFIContext[LogosDelivery],
|
||||
callback: FFICallBack,
|
||||
userData: pointer,
|
||||
jsonQuery: cstring,
|
||||
peerAddr: cstring,
|
||||
timeoutMs: cint,
|
||||
) {.ffi.} =
|
||||
let jsonContentRes = catch:
|
||||
parseJson($jsonQuery)
|
||||
proc store_query*(
|
||||
self: LogosDelivery, queryJson: string, peer: string, timeoutMs: int
|
||||
): Future[Result[string, string]] {.ffi.} =
|
||||
let jsonContent =
|
||||
try:
|
||||
parseJson(queryJson)
|
||||
except CatchableError as e:
|
||||
return err("StoreRequest failed parsing store request: " & e.msg)
|
||||
|
||||
if jsonContentRes.isErr():
|
||||
return err("StoreRequest failed parsing store request: " & jsonContentRes.error.msg)
|
||||
let storeQueryRequest = storeQueryFromJson(jsonContent).valueOr:
|
||||
return err(error)
|
||||
|
||||
let storeQueryRequest = ?fromJsonNode(jsonContentRes.get())
|
||||
let queryResponse = (await self.waku.storeQuery(storeQueryRequest, peer, timeoutMs)).valueOr:
|
||||
return err("StoreRequest failed store query: " & error)
|
||||
|
||||
let peer = peers.parsePeerInfo(($peerAddr).split(",")).valueOr:
|
||||
return err("StoreRequest failed to parse peer addr: " & $error)
|
||||
|
||||
let queryResponse = (
|
||||
await ctx.myLib[].waku.node.wakuStoreClient.query(storeQueryRequest, peer)
|
||||
).valueOr:
|
||||
return err("StoreRequest failed store query: " & $error)
|
||||
|
||||
let res = $(%*(queryResponse.toHex()))
|
||||
return ok(res) ## returning the response in json format
|
||||
return ok($(%*(queryResponse.toHex()))) ## response in json format
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user