mirror of https://github.com/waku-org/nwaku.git
Removed json-rpc leftovers (#2503)
This commit is contained in:
parent
9ef2eccb8a
commit
b151672143
|
@ -127,33 +127,6 @@ type
|
||||||
defaultValue: ""
|
defaultValue: ""
|
||||||
name: "lightpushnode" }: string
|
name: "lightpushnode" }: string
|
||||||
|
|
||||||
## JSON-RPC config
|
|
||||||
|
|
||||||
rpc* {.
|
|
||||||
desc: "Enable Waku JSON-RPC server: true|false",
|
|
||||||
defaultValue: true
|
|
||||||
name: "rpc" }: bool
|
|
||||||
|
|
||||||
rpcAddress* {.
|
|
||||||
desc: "Listening address of the JSON-RPC server.",
|
|
||||||
defaultValue: parseIpAddress("127.0.0.1")
|
|
||||||
name: "rpc-address" }: IpAddress
|
|
||||||
|
|
||||||
rpcPort* {.
|
|
||||||
desc: "Listening port of the JSON-RPC server.",
|
|
||||||
defaultValue: 8545
|
|
||||||
name: "rpc-port" }: uint16
|
|
||||||
|
|
||||||
rpcAdmin* {.
|
|
||||||
desc: "Enable access to JSON-RPC Admin API: true|false",
|
|
||||||
defaultValue: false
|
|
||||||
name: "rpc-admin" }: bool
|
|
||||||
|
|
||||||
rpcPrivate* {.
|
|
||||||
desc: "Enable access to JSON-RPC Private API: true|false",
|
|
||||||
defaultValue: false
|
|
||||||
name: "rpc-private" }: bool
|
|
||||||
|
|
||||||
## Metrics config
|
## Metrics config
|
||||||
|
|
||||||
metricsServer* {.
|
metricsServer* {.
|
||||||
|
|
|
@ -1,283 +0,0 @@
|
||||||
{.used.}
|
|
||||||
|
|
||||||
import
|
|
||||||
std/[options, sequtils, tempfiles],
|
|
||||||
stew/shims/net as stewNet,
|
|
||||||
testutils/unittests,
|
|
||||||
libp2p/crypto/crypto,
|
|
||||||
json_rpc/[rpcserver, rpcclient]
|
|
||||||
import
|
|
||||||
../../../waku/common/base64,
|
|
||||||
../../../waku/waku_core,
|
|
||||||
../../../waku/node/peer_manager,
|
|
||||||
../../../waku/waku_api/message_cache,
|
|
||||||
../../../waku/waku_node,
|
|
||||||
../../../waku/waku_api/jsonrpc/relay/handlers as relay_api,
|
|
||||||
../../../waku/waku_api/jsonrpc/relay/client as relay_api_client,
|
|
||||||
../../../waku/waku_relay,
|
|
||||||
../../../waku/waku_rln_relay,
|
|
||||||
../testlib/wakucore,
|
|
||||||
../testlib/wakunode
|
|
||||||
|
|
||||||
suite "Waku v2 JSON-RPC API - Relay":
|
|
||||||
|
|
||||||
asyncTest "subscribe and unsubscribe from topics":
|
|
||||||
## Setup
|
|
||||||
let node = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
||||||
|
|
||||||
await node.start()
|
|
||||||
await node.mountRelay(@[])
|
|
||||||
|
|
||||||
# JSON-RPC server
|
|
||||||
let
|
|
||||||
rpcPort = Port(8547)
|
|
||||||
ta = initTAddress(parseIpAddress("0.0.0.0"), rpcPort)
|
|
||||||
server = newRpcHttpServer([ta])
|
|
||||||
|
|
||||||
let cache = MessageCache.init(capacity=30)
|
|
||||||
installRelayApiHandlers(node, server, cache)
|
|
||||||
server.start()
|
|
||||||
|
|
||||||
# JSON-RPC client
|
|
||||||
let client = newRpcHttpClient()
|
|
||||||
await client.connect("127.0.0.1", rpcPort, false)
|
|
||||||
|
|
||||||
## Given
|
|
||||||
let newTopics = @["test-topic1","test-topic2","test-topic3"]
|
|
||||||
|
|
||||||
## When
|
|
||||||
# Subscribe to new topics
|
|
||||||
let subResp = await client.post_waku_v2_relay_v1_subscriptions(newTopics)
|
|
||||||
|
|
||||||
let subTopics = node.wakuRelay.subscribedTopics.toSeq()
|
|
||||||
|
|
||||||
# Unsubscribe from new topics
|
|
||||||
let unsubResp = await client.delete_waku_v2_relay_v1_subscriptions(newTopics)
|
|
||||||
|
|
||||||
let unsubTopics = node.wakuRelay.subscribedTopics.toSeq()
|
|
||||||
|
|
||||||
## Then
|
|
||||||
check:
|
|
||||||
subResp == true
|
|
||||||
check:
|
|
||||||
# Node is now subscribed to default + new topics
|
|
||||||
subTopics.len == newTopics.len
|
|
||||||
newTopics.allIt(it in subTopics)
|
|
||||||
|
|
||||||
check:
|
|
||||||
unsubResp == true
|
|
||||||
check:
|
|
||||||
# Node is now unsubscribed from new topics
|
|
||||||
unsubTopics.len == 0
|
|
||||||
newTopics.allIt(it notin unsubTopics)
|
|
||||||
|
|
||||||
await server.stop()
|
|
||||||
await server.closeWait()
|
|
||||||
|
|
||||||
await node.stop()
|
|
||||||
|
|
||||||
asyncTest "publish message to topic":
|
|
||||||
## Setup
|
|
||||||
let
|
|
||||||
pubSubTopic = "test-jsonrpc-pubsub-topic"
|
|
||||||
contentTopic = "test-jsonrpc-content-topic"
|
|
||||||
|
|
||||||
# Relay nodes setup
|
|
||||||
let
|
|
||||||
srcNode = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
||||||
dstNode = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
||||||
|
|
||||||
await allFutures(srcNode.start(), dstNode.start())
|
|
||||||
|
|
||||||
await srcNode.mountRelay(@[pubSubTopic])
|
|
||||||
await dstNode.mountRelay(@[pubSubTopic])
|
|
||||||
|
|
||||||
await srcNode.mountRlnRelay(WakuRlnConfig(rlnRelayDynamic: false,
|
|
||||||
rlnRelayCredIndex: some(1.uint),
|
|
||||||
rlnEpochSizeSec: 1,
|
|
||||||
rlnRelayTreePath: genTempPath("rln_tree", "wakunode_1")))
|
|
||||||
|
|
||||||
await dstNode.mountRlnRelay(WakuRlnConfig(rlnRelayDynamic: false,
|
|
||||||
rlnRelayCredIndex: some(2.uint),
|
|
||||||
rlnEpochSizeSec: 1,
|
|
||||||
rlnRelayTreePath: genTempPath("rln_tree", "wakunode_2")))
|
|
||||||
|
|
||||||
await srcNode.connectToNodes(@[dstNode.peerInfo.toRemotePeerInfo()])
|
|
||||||
|
|
||||||
# RPC server (source node)
|
|
||||||
let
|
|
||||||
rpcPort = Port(8548)
|
|
||||||
ta = initTAddress(parseIpAddress("0.0.0.0"), rpcPort)
|
|
||||||
server = newRpcHttpServer([ta])
|
|
||||||
|
|
||||||
let cache = MessageCache.init(capacity=30)
|
|
||||||
installRelayApiHandlers(srcNode, server, cache)
|
|
||||||
server.start()
|
|
||||||
|
|
||||||
# JSON-RPC client
|
|
||||||
let client = newRpcHttpClient()
|
|
||||||
await client.connect("127.0.0.1", rpcPort, false)
|
|
||||||
|
|
||||||
## Given
|
|
||||||
let message = fakeWakuMessage( payload= @[byte 72], contentTopic=contentTopic)
|
|
||||||
|
|
||||||
let dstHandlerFut = newFuture[(PubsubTopic, WakuMessage)]()
|
|
||||||
proc dstHandler(topic: PubsubTopic, msg: WakuMessage) {.async, gcsafe.} =
|
|
||||||
dstHandlerFut.complete((topic, msg))
|
|
||||||
|
|
||||||
dstNode.subscribe((kind: PubsubSub, topic: pubsubTopic), some(dstHandler))
|
|
||||||
|
|
||||||
## When
|
|
||||||
let rpcMessage = WakuMessageRPC(
|
|
||||||
payload: base64.encode(message.payload),
|
|
||||||
contentTopic: some(message.contentTopic),
|
|
||||||
timestamp: some(message.timestamp),
|
|
||||||
version: some(message.version)
|
|
||||||
)
|
|
||||||
let response = await client.post_waku_v2_relay_v1_message(pubSubTopic, rpcMessage)
|
|
||||||
|
|
||||||
## Then
|
|
||||||
require:
|
|
||||||
response == true
|
|
||||||
await dstHandlerFut.withTimeout(chronos.seconds(5))
|
|
||||||
|
|
||||||
var (topic, msg) = dstHandlerFut.read()
|
|
||||||
|
|
||||||
# proof is injected under the hood, we compare just the message
|
|
||||||
msg.proof = @[]
|
|
||||||
|
|
||||||
check:
|
|
||||||
topic == pubSubTopic
|
|
||||||
msg == message
|
|
||||||
|
|
||||||
|
|
||||||
## Cleanup
|
|
||||||
await server.stop()
|
|
||||||
await server.closeWait()
|
|
||||||
await allFutures(srcNode.stop(), dstNode.stop())
|
|
||||||
|
|
||||||
asyncTest "get latest messages received from pubsub topics cache":
|
|
||||||
## Setup
|
|
||||||
let
|
|
||||||
pubSubTopic = "test-jsonrpc-pubsub-topic"
|
|
||||||
contentTopic = "test-jsonrpc-content-topic"
|
|
||||||
|
|
||||||
# Relay nodes setup
|
|
||||||
let
|
|
||||||
srcNode = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
||||||
dstNode = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
||||||
|
|
||||||
await allFutures(srcNode.start(), dstNode.start())
|
|
||||||
|
|
||||||
await srcNode.mountRelay(@[pubSubTopic])
|
|
||||||
await dstNode.mountRelay(@[])
|
|
||||||
|
|
||||||
await srcNode.connectToNodes(@[dstNode.peerInfo.toRemotePeerInfo()])
|
|
||||||
|
|
||||||
# RPC server (destination node)
|
|
||||||
let
|
|
||||||
rpcPort = Port(8549)
|
|
||||||
ta = initTAddress(parseIpAddress("0.0.0.0"), rpcPort)
|
|
||||||
server = newRpcHttpServer([ta])
|
|
||||||
|
|
||||||
let cache = MessageCache.init(capacity=30)
|
|
||||||
installRelayApiHandlers(dstNode, server, cache)
|
|
||||||
server.start()
|
|
||||||
|
|
||||||
# JSON-RPC client
|
|
||||||
let client = newRpcHttpClient()
|
|
||||||
await client.connect("127.0.0.1", rpcPort, false)
|
|
||||||
|
|
||||||
discard await client.post_waku_v2_relay_v1_subscriptions(@[pubSubTopic])
|
|
||||||
|
|
||||||
## Given
|
|
||||||
let messages = @[
|
|
||||||
fakeWakuMessage(payload= @[byte 70], contentTopic=contentTopic),
|
|
||||||
fakeWakuMessage(payload= @[byte 71], contentTopic=contentTopic),
|
|
||||||
fakeWakuMessage(payload= @[byte 72], contentTopic=contentTopic),
|
|
||||||
fakeWakuMessage(payload= @[byte 73], contentTopic=contentTopic)
|
|
||||||
]
|
|
||||||
|
|
||||||
## When
|
|
||||||
for msg in messages:
|
|
||||||
discard await srcNode.publish(some(pubSubTopic), msg)
|
|
||||||
|
|
||||||
await sleepAsync(200.millis)
|
|
||||||
|
|
||||||
let dstMessages = await client.get_waku_v2_relay_v1_messages(pubSubTopic)
|
|
||||||
|
|
||||||
## Then
|
|
||||||
check:
|
|
||||||
dstMessages.len == 4
|
|
||||||
dstMessages[2].payload == base64.encode(messages[2].payload)
|
|
||||||
dstMessages[2].contentTopic.get() == messages[2].contentTopic
|
|
||||||
dstMessages[2].timestamp.get() == messages[2].timestamp
|
|
||||||
dstMessages[2].version.get() == messages[2].version
|
|
||||||
|
|
||||||
## Cleanup
|
|
||||||
await server.stop()
|
|
||||||
await server.closeWait()
|
|
||||||
await allFutures(srcNode.stop(), dstNode.stop())
|
|
||||||
|
|
||||||
asyncTest "get latest messages received from content topics cache":
|
|
||||||
## Setup
|
|
||||||
let contentTopic = DefaultContentTopic
|
|
||||||
|
|
||||||
# Relay nodes setup
|
|
||||||
let
|
|
||||||
srcNode = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
||||||
dstNode = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(0))
|
|
||||||
|
|
||||||
await allFutures(srcNode.start(), dstNode.start())
|
|
||||||
|
|
||||||
let shard = getShard(contentTopic).expect("Valid Shard")
|
|
||||||
|
|
||||||
await srcNode.mountRelay(@[shard])
|
|
||||||
await dstNode.mountRelay(@[])
|
|
||||||
|
|
||||||
await srcNode.connectToNodes(@[dstNode.peerInfo.toRemotePeerInfo()])
|
|
||||||
|
|
||||||
# RPC server (destination node)
|
|
||||||
let
|
|
||||||
rpcPort = Port(8550)
|
|
||||||
ta = initTAddress(parseIpAddress("0.0.0.0"), rpcPort)
|
|
||||||
server = newRpcHttpServer([ta])
|
|
||||||
|
|
||||||
let cache = MessageCache.init(capacity=30)
|
|
||||||
installRelayApiHandlers(dstNode, server, cache)
|
|
||||||
server.start()
|
|
||||||
|
|
||||||
# JSON-RPC client
|
|
||||||
let client = newRpcHttpClient()
|
|
||||||
await client.connect("127.0.0.1", rpcPort, false)
|
|
||||||
|
|
||||||
discard await client.post_waku_v2_relay_v1_auto_subscriptions(@[contentTopic])
|
|
||||||
|
|
||||||
## Given
|
|
||||||
let messages = @[
|
|
||||||
fakeWakuMessage(payload= @[byte 70], contentTopic=contentTopic),
|
|
||||||
fakeWakuMessage(payload= @[byte 71], contentTopic=contentTopic),
|
|
||||||
fakeWakuMessage(payload= @[byte 72], contentTopic=contentTopic),
|
|
||||||
fakeWakuMessage(payload= @[byte 73], contentTopic=contentTopic)
|
|
||||||
]
|
|
||||||
|
|
||||||
## When
|
|
||||||
for msg in messages:
|
|
||||||
discard await srcNode.publish(none(PubsubTopic), msg)
|
|
||||||
|
|
||||||
await sleepAsync(200.millis)
|
|
||||||
|
|
||||||
let dstMessages = await client.get_waku_v2_relay_v1_auto_messages(contentTopic)
|
|
||||||
|
|
||||||
## Then
|
|
||||||
check:
|
|
||||||
dstMessages.len == 4
|
|
||||||
dstMessages[2].payload == base64.encode(messages[2].payload)
|
|
||||||
dstMessages[2].contentTopic.get() == messages[2].contentTopic
|
|
||||||
dstMessages[2].timestamp.get() == messages[2].timestamp
|
|
||||||
dstMessages[2].version.get() == messages[2].version
|
|
||||||
|
|
||||||
## Cleanup
|
|
||||||
await server.stop()
|
|
||||||
await server.closeWait()
|
|
||||||
await allFutures(srcNode.stop(), dstNode.stop())
|
|
|
@ -1,237 +0,0 @@
|
||||||
when (NimMajor, NimMinor) < (1, 4):
|
|
||||||
{.push raises: [Defect].}
|
|
||||||
else:
|
|
||||||
{.push raises: [].}
|
|
||||||
|
|
||||||
import
|
|
||||||
std/sequtils,
|
|
||||||
chronicles,
|
|
||||||
json_rpc/rpcserver,
|
|
||||||
eth/keys,
|
|
||||||
nimcrypto/sysrand
|
|
||||||
import
|
|
||||||
../../../common/base64,
|
|
||||||
../../../waku_core,
|
|
||||||
../../../waku_relay,
|
|
||||||
../../../waku_rln_relay,
|
|
||||||
../../../waku_rln_relay/rln/wrappers,
|
|
||||||
../../../waku_node,
|
|
||||||
../../message_cache,
|
|
||||||
../../handlers,
|
|
||||||
../message
|
|
||||||
|
|
||||||
from std/times import getTime
|
|
||||||
from std/times import toUnix
|
|
||||||
|
|
||||||
|
|
||||||
logScope:
|
|
||||||
topics = "waku node jsonrpc relay_api"
|
|
||||||
|
|
||||||
|
|
||||||
const futTimeout* = 5.seconds # Max time to wait for futures
|
|
||||||
|
|
||||||
## Waku Relay JSON-RPC API
|
|
||||||
|
|
||||||
proc installRelayApiHandlers*(node: WakuNode, server: RpcServer, cache: MessageCache) =
|
|
||||||
server.rpc("post_waku_v2_relay_v1_subscriptions") do (pubsubTopics: seq[PubsubTopic]) -> bool:
|
|
||||||
if pubsubTopics.len == 0:
|
|
||||||
raise newException(ValueError, "No pubsub topic provided")
|
|
||||||
|
|
||||||
## Subscribes a node to a list of PubSub topics
|
|
||||||
debug "post_waku_v2_relay_v1_subscriptions"
|
|
||||||
|
|
||||||
# Subscribe to all requested topics
|
|
||||||
let newTopics = pubsubTopics.filterIt(not cache.isPubsubSubscribed(it))
|
|
||||||
|
|
||||||
for pubsubTopic in newTopics:
|
|
||||||
if pubsubTopic == "":
|
|
||||||
raise newException(ValueError, "Empty pubsub topic")
|
|
||||||
|
|
||||||
cache.pubsubSubscribe(pubsubTopic)
|
|
||||||
node.subscribe((kind: PubsubSub, topic: pubsubTopic), some(messageCacheHandler(cache)))
|
|
||||||
|
|
||||||
return true
|
|
||||||
|
|
||||||
server.rpc("delete_waku_v2_relay_v1_subscriptions") do (pubsubTopics: seq[PubsubTopic]) -> bool:
|
|
||||||
if pubsubTopics.len == 0:
|
|
||||||
raise newException(ValueError, "No pubsub topic provided")
|
|
||||||
|
|
||||||
## Unsubscribes a node from a list of PubSub topics
|
|
||||||
debug "delete_waku_v2_relay_v1_subscriptions"
|
|
||||||
|
|
||||||
# Unsubscribe all handlers from requested topics
|
|
||||||
let subscribedTopics = pubsubTopics.filterIt(cache.isPubsubSubscribed(it))
|
|
||||||
|
|
||||||
for pubsubTopic in subscribedTopics:
|
|
||||||
if pubsubTopic == "":
|
|
||||||
raise newException(ValueError, "Empty pubsub topic")
|
|
||||||
|
|
||||||
cache.pubsubUnsubscribe(pubsubTopic)
|
|
||||||
node.unsubscribe((kind: PubsubUnsub, topic: pubsubTopic))
|
|
||||||
|
|
||||||
return true
|
|
||||||
|
|
||||||
server.rpc("post_waku_v2_relay_v1_message") do (pubsubTopic: PubsubTopic, msg: WakuMessageRPC) -> bool:
|
|
||||||
if pubsubTopic == "":
|
|
||||||
raise newException(ValueError, "Empty pubsub topic")
|
|
||||||
|
|
||||||
## Publishes a WakuMessage to a PubSub topic
|
|
||||||
debug "post_waku_v2_relay_v1_message", pubsubTopic=pubsubTopic
|
|
||||||
|
|
||||||
let payloadRes = base64.decode(msg.payload)
|
|
||||||
if payloadRes.isErr():
|
|
||||||
raise newException(ValueError, "invalid payload format: " & payloadRes.error)
|
|
||||||
|
|
||||||
if msg.contentTopic.isNone():
|
|
||||||
raise newException(ValueError, "message has no content topic")
|
|
||||||
|
|
||||||
var message = WakuMessage(
|
|
||||||
payload: payloadRes.value,
|
|
||||||
contentTopic: msg.contentTopic.get(),
|
|
||||||
version: msg.version.get(0'u32),
|
|
||||||
timestamp: msg.timestamp.get(Timestamp(0)),
|
|
||||||
ephemeral: msg.ephemeral.get(false)
|
|
||||||
)
|
|
||||||
|
|
||||||
# ensure the node is subscribed to the pubsubTopic. otherwise it risks publishing
|
|
||||||
# to a topic with no connected peers
|
|
||||||
if pubsubTopic notin node.wakuRelay.subscribedTopics():
|
|
||||||
raise newException(
|
|
||||||
ValueError, "Failed to publish: Node not subscribed to pubsubTopic: " & pubsubTopic)
|
|
||||||
|
|
||||||
# if RLN is mounted, append the proof to the message
|
|
||||||
if not node.wakuRlnRelay.isNil():
|
|
||||||
# append the proof to the message
|
|
||||||
node.wakuRlnRelay.appendRLNProof(message,
|
|
||||||
float64(getTime().toUnix())).isOkOr:
|
|
||||||
raise newException(ValueError, "Failed to publish: error appending RLN proof to message")
|
|
||||||
# validate the message before sending it
|
|
||||||
let result = node.wakuRlnRelay.validateMessageAndUpdateLog(message)
|
|
||||||
if result == MessageValidationResult.Invalid:
|
|
||||||
raise newException(ValueError, "Failed to publish: invalid RLN proof")
|
|
||||||
elif result == MessageValidationResult.Spam:
|
|
||||||
raise newException(ValueError, "Failed to publish: limit exceeded, try again later")
|
|
||||||
elif result == MessageValidationResult.Valid:
|
|
||||||
debug "RLN proof validated successfully", pubSubTopic=pubsubTopic
|
|
||||||
else:
|
|
||||||
raise newException(ValueError, "Failed to publish: unknown RLN proof validation result")
|
|
||||||
|
|
||||||
# if we reach here its either a non-RLN message or a RLN message with a valid proof
|
|
||||||
debug "Publishing message", pubSubTopic=pubsubTopic, rln=not node.wakuRlnRelay.isNil()
|
|
||||||
let publishFut = node.publish(some(pubsubTopic), message)
|
|
||||||
if not await publishFut.withTimeout(futTimeout):
|
|
||||||
raise newException(ValueError, "Failed to publish: timed out")
|
|
||||||
|
|
||||||
return true
|
|
||||||
|
|
||||||
server.rpc("get_waku_v2_relay_v1_messages") do (pubsubTopic: PubsubTopic) -> seq[WakuMessageRPC]:
|
|
||||||
if pubsubTopic == "":
|
|
||||||
raise newException(ValueError, "Empty pubsub topic")
|
|
||||||
|
|
||||||
## Returns all WakuMessages received on a PubSub topic since the
|
|
||||||
## last time this method was called
|
|
||||||
debug "get_waku_v2_relay_v1_messages", topic=pubsubTopic
|
|
||||||
|
|
||||||
let msgRes = cache.getMessages(pubsubTopic, clear=true)
|
|
||||||
if msgRes.isErr():
|
|
||||||
raise newException(ValueError, "Not subscribed to pubsub topic: " & pubsubTopic)
|
|
||||||
|
|
||||||
return msgRes.value.map(toWakuMessageRPC)
|
|
||||||
|
|
||||||
# Autosharding API
|
|
||||||
|
|
||||||
server.rpc("post_waku_v2_relay_v1_auto_subscriptions") do (contentTopics: seq[ContentTopic]) -> bool:
|
|
||||||
if contentTopics.len == 0:
|
|
||||||
raise newException(ValueError, "No content topic provided")
|
|
||||||
|
|
||||||
## Subscribes a node to a list of Content topics
|
|
||||||
debug "post_waku_v2_relay_v1_auto_subscriptions"
|
|
||||||
|
|
||||||
let newTopics = contentTopics.filterIt(not cache.isContentSubscribed(it))
|
|
||||||
|
|
||||||
# Subscribe to all requested topics
|
|
||||||
for contentTopic in newTopics:
|
|
||||||
if contentTopic == "":
|
|
||||||
raise newException(ValueError, "Empty content topic")
|
|
||||||
|
|
||||||
cache.contentSubscribe(contentTopic)
|
|
||||||
node.subscribe((kind: ContentSub, topic: contentTopic), some(messageCacheHandler(cache)))
|
|
||||||
|
|
||||||
return true
|
|
||||||
|
|
||||||
server.rpc("delete_waku_v2_relay_v1_auto_subscriptions") do (contentTopics: seq[ContentTopic]) -> bool:
|
|
||||||
if contentTopics.len == 0:
|
|
||||||
raise newException(ValueError, "No content topic provided")
|
|
||||||
|
|
||||||
## Unsubscribes a node from a list of Content topics
|
|
||||||
debug "delete_waku_v2_relay_v1_auto_subscriptions"
|
|
||||||
|
|
||||||
let subscribedTopics = contentTopics.filterIt(cache.isContentSubscribed(it))
|
|
||||||
|
|
||||||
# Unsubscribe all handlers from requested topics
|
|
||||||
for contentTopic in subscribedTopics:
|
|
||||||
if contentTopic == "":
|
|
||||||
raise newException(ValueError, "Empty content topic")
|
|
||||||
|
|
||||||
cache.contentUnsubscribe(contentTopic)
|
|
||||||
node.unsubscribe((kind: ContentUnsub, topic: contentTopic))
|
|
||||||
|
|
||||||
return true
|
|
||||||
|
|
||||||
server.rpc("post_waku_v2_relay_v1_auto_message") do (msg: WakuMessageRPC) -> bool:
|
|
||||||
## Publishes a WakuMessage to a Content topic
|
|
||||||
debug "post_waku_v2_relay_v1_auto_message"
|
|
||||||
|
|
||||||
let payloadRes = base64.decode(msg.payload)
|
|
||||||
if payloadRes.isErr():
|
|
||||||
raise newException(ValueError, "invalid payload format: " & payloadRes.error)
|
|
||||||
|
|
||||||
if msg.contentTopic.isNone():
|
|
||||||
raise newException(ValueError, "message has no content topic")
|
|
||||||
|
|
||||||
var message = WakuMessage(
|
|
||||||
payload: payloadRes.value,
|
|
||||||
contentTopic: msg.contentTopic.get(),
|
|
||||||
version: msg.version.get(0'u32),
|
|
||||||
timestamp: msg.timestamp.get(Timestamp(0)),
|
|
||||||
ephemeral: msg.ephemeral.get(false)
|
|
||||||
)
|
|
||||||
|
|
||||||
# if RLN is mounted, append the proof to the message
|
|
||||||
if not node.wakuRlnRelay.isNil():
|
|
||||||
# append the proof to the message
|
|
||||||
node.wakuRlnRelay.appendRLNProof(message,
|
|
||||||
float64(getTime().toUnix())).isOkOr:
|
|
||||||
raise newException(ValueError, "Failed to publish: error appending RLN proof to message")
|
|
||||||
# validate the message before sending it
|
|
||||||
let result = node.wakuRlnRelay.validateMessageAndUpdateLog(message)
|
|
||||||
if result == MessageValidationResult.Invalid:
|
|
||||||
raise newException(ValueError, "Failed to publish: invalid RLN proof")
|
|
||||||
elif result == MessageValidationResult.Spam:
|
|
||||||
raise newException(ValueError, "Failed to publish: limit exceeded, try again later")
|
|
||||||
elif result == MessageValidationResult.Valid:
|
|
||||||
debug "RLN proof validated successfully", contentTopic=message.contentTopic
|
|
||||||
else:
|
|
||||||
raise newException(ValueError, "Failed to publish: unknown RLN proof validation result")
|
|
||||||
|
|
||||||
# if we reach here its either a non-RLN message or a RLN message with a valid proof
|
|
||||||
debug "Publishing message", contentTopic=message.contentTopic, rln=not node.wakuRlnRelay.isNil()
|
|
||||||
let publishFut = node.publish(none(PubsubTopic), message)
|
|
||||||
if not await publishFut.withTimeout(futTimeout):
|
|
||||||
raise newException(ValueError, "Failed to publish: timed out")
|
|
||||||
|
|
||||||
return true
|
|
||||||
|
|
||||||
server.rpc("get_waku_v2_relay_v1_auto_messages") do (contentTopic: ContentTopic) -> seq[WakuMessageRPC]:
|
|
||||||
if contentTopic == "":
|
|
||||||
raise newException(ValueError, "Empty content topic")
|
|
||||||
|
|
||||||
## Returns all WakuMessages received on a Content topic since the
|
|
||||||
## last time this method was called
|
|
||||||
debug "get_waku_v2_relay_v1_auto_messages", topic=contentTopic
|
|
||||||
|
|
||||||
let msgRes = cache.getAutoMessages(contentTopic, clear=true)
|
|
||||||
if msgRes.isErr():
|
|
||||||
raise newException(ValueError, "Not subscribed to content topic: " & contentTopic)
|
|
||||||
|
|
||||||
return msgRes.value.map(toWakuMessageRPC)
|
|
Loading…
Reference in New Issue