mirror of https://github.com/waku-org/nwaku.git
chore: remove json rpc (#2416)
This commit is contained in:
Alvaro Revuelta
GitHub
parent
55ff667478
commit
c994ee049b
|
|
@ -222,29 +222,7 @@ proc stop*(cmb: Chat2MatterBridge) {.async: (raises: [Exception]).} =
|
|||
when isMainModule:
|
||||
import
|
||||
../../../waku/common/utils/nat,
|
||||
../../waku/waku_api/message_cache,
|
||||
../../waku/waku_api/jsonrpc/debug/handlers as debug_api,
|
||||
../../waku/waku_api/jsonrpc/filter/handlers as filter_api,
|
||||
../../waku/waku_api/jsonrpc/relay/handlers as relay_api,
|
||||
../../waku/waku_api/jsonrpc/store/handlers as store_api
|
||||
|
||||
|
||||
proc startV2Rpc(node: WakuNode, rpcServer: RpcHttpServer, conf: Chat2MatterbridgeConf) {.raises: [Exception].} =
|
||||
installDebugApiHandlers(node, rpcServer)
|
||||
|
||||
# Install enabled API handlers:
|
||||
if conf.relay:
|
||||
let cache = MessageCache.init(capacity=30)
|
||||
installRelayApiHandlers(node, rpcServer, cache)
|
||||
|
||||
if conf.filter:
|
||||
let messageCache = MessageCache.init(capacity=30)
|
||||
installFilterApiHandlers(node, rpcServer, messageCache)
|
||||
|
||||
if conf.store:
|
||||
installStoreApiHandlers(node, rpcServer)
|
||||
|
||||
rpcServer.start()
|
||||
../../waku/waku_api/message_cache
|
||||
|
||||
let
|
||||
rng = newRng()
|
||||
|
|
@ -309,15 +287,6 @@ when isMainModule:
|
|||
else:
|
||||
error "Error parsing conf.filternode", error = filterPeer.error
|
||||
|
||||
if conf.rpc:
|
||||
let ta = initTAddress(conf.rpcAddress,
|
||||
Port(conf.rpcPort + conf.portsShift))
|
||||
var rpcServer = newRpcHttpServer([ta])
|
||||
# Waku v2 rpc
|
||||
startV2Rpc(bridge.nodev2, rpcServer, conf)
|
||||
|
||||
rpcServer.start()
|
||||
|
||||
if conf.metricsServer:
|
||||
let
|
||||
address = conf.metricsServerAddress
|
||||
|
|
|
|||
|
|
@ -35,21 +35,6 @@ type
|
|||
"Must be one of: any, none, upnp, pmp, extip:<IP>"
|
||||
defaultValue: "any" .}: string
|
||||
|
||||
rpc* {.
|
||||
desc: "Enable Waku RPC server"
|
||||
defaultValue: false
|
||||
name: "rpc" .}: bool
|
||||
|
||||
rpcAddress* {.
|
||||
desc: "Listening address of the RPC server",
|
||||
defaultValue: parseIpAddress("127.0.0.1")
|
||||
name: "rpc-address" }: IpAddress
|
||||
|
||||
rpcPort* {.
|
||||
desc: "Listening port of the RPC server"
|
||||
defaultValue: 8545
|
||||
name: "rpc-port" .}: uint16
|
||||
|
||||
metricsServer* {.
|
||||
desc: "Enable the metrics server"
|
||||
defaultValue: false
|
||||
|
|
|
|||
|
|
@ -15,7 +15,6 @@ import
|
|||
libp2p/protocols/pubsub/gossipsub,
|
||||
libp2p/peerid,
|
||||
eth/keys,
|
||||
json_rpc/rpcserver,
|
||||
presto,
|
||||
metrics,
|
||||
metrics/chronos_httpserver
|
||||
|
|
@ -42,11 +41,6 @@ import
|
|||
../../waku/waku_api/rest/store/handlers as rest_store_api,
|
||||
../../waku/waku_api/rest/health/handlers as rest_health_api,
|
||||
../../waku/waku_api/rest/admin/handlers as rest_admin_api,
|
||||
../../waku/waku_api/jsonrpc/admin/handlers as rpc_admin_api,
|
||||
../../waku/waku_api/jsonrpc/debug/handlers as rpc_debug_api,
|
||||
../../waku/waku_api/jsonrpc/filter/handlers as rpc_filter_api,
|
||||
../../waku/waku_api/jsonrpc/relay/handlers as rpc_relay_api,
|
||||
../../waku/waku_api/jsonrpc/store/handlers as rpc_store_api,
|
||||
../../waku/waku_archive,
|
||||
../../waku/waku_dnsdisc,
|
||||
../../waku/waku_enr/sharding,
|
||||
|
|
@ -83,7 +77,6 @@ type
|
|||
|
||||
node: WakuNode
|
||||
|
||||
rpcServer: Option[RpcHttpServer]
|
||||
restServer: Option[WakuRestServerRef]
|
||||
metricsServer: Option[MetricsHttpServerRef]
|
||||
|
||||
|
|
@ -793,46 +786,6 @@ proc startRestServer(app: App,
|
|||
|
||||
ok(server)
|
||||
|
||||
proc startRpcServer(app: App, address: IpAddress, port: Port, conf: WakuNodeConf): AppResult[RpcHttpServer] =
|
||||
let ta = initTAddress(address, port)
|
||||
|
||||
var server: RpcHttpServer
|
||||
try:
|
||||
server = newRpcHttpServer([ta])
|
||||
except CatchableError:
|
||||
return err("failed to init JSON-RPC server: " & getCurrentExceptionMsg())
|
||||
|
||||
installDebugApiHandlers(app.node, server)
|
||||
|
||||
if conf.relay:
|
||||
let cache = MessageCache.init(capacity=50)
|
||||
|
||||
let handler = messageCacheHandler(cache)
|
||||
|
||||
for pubsubTopic in conf.pubsubTopics:
|
||||
cache.pubsubSubscribe(pubsubTopic)
|
||||
app.node.subscribe((kind: PubsubSub, topic: pubsubTopic), some(handler))
|
||||
|
||||
for contentTopic in conf.contentTopics:
|
||||
cache.contentSubscribe(contentTopic)
|
||||
app.node.subscribe((kind: ContentSub, topic: contentTopic), some(handler))
|
||||
|
||||
installRelayApiHandlers(app.node, server, cache)
|
||||
|
||||
if conf.filternode != "":
|
||||
let filterMessageCache = MessageCache.init(capacity=50)
|
||||
installFilterApiHandlers(app.node, server, filterMessageCache)
|
||||
|
||||
installStoreApiHandlers(app.node, server)
|
||||
|
||||
if conf.rpcAdmin:
|
||||
installAdminApiHandlers(app.node, server)
|
||||
|
||||
server.start()
|
||||
info "RPC Server started", address=ta
|
||||
|
||||
ok(server)
|
||||
|
||||
proc startMetricsServer(serverIp: IpAddress, serverPort: Port): AppResult[MetricsHttpServerRef] =
|
||||
info "Starting metrics HTTP server", serverIp= $serverIp, serverPort= $serverPort
|
||||
|
||||
|
|
@ -854,13 +807,6 @@ proc startMetricsLogging(): AppResult[void] =
|
|||
ok()
|
||||
|
||||
proc setupMonitoringAndExternalInterfaces*(app: var App): AppResult[void] =
|
||||
if app.conf.rpc:
|
||||
let startRpcServerRes = startRpcServer(app, app.conf.rpcAddress, Port(app.conf.rpcPort + app.conf.portsShift), app.conf)
|
||||
if startRpcServerRes.isErr():
|
||||
error "6/7 Starting JSON-RPC server failed. Continuing in current state.", error=startRpcServerRes.error
|
||||
else:
|
||||
app.rpcServer = some(startRpcServerRes.value)
|
||||
|
||||
if app.conf.rest:
|
||||
let startRestServerRes = startRestServer(app, app.conf.restAddress, Port(app.conf.restPort + app.conf.portsShift), app.conf)
|
||||
if startRestServerRes.isErr():
|
||||
|
|
@ -890,9 +836,6 @@ proc stop*(app: App): Future[void] {.async: (raises: [Exception]).} =
|
|||
if app.restServer.isSome():
|
||||
await app.restServer.get().stop()
|
||||
|
||||
if app.rpcServer.isSome():
|
||||
await app.rpcServer.get().stop()
|
||||
|
||||
if app.metricsServer.isSome():
|
||||
await app.metricsServer.get().stop()
|
||||
|
||||
|
|
|
|||
|
|
@ -360,33 +360,6 @@ type
|
|||
defaultValue: ""
|
||||
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
|
||||
|
||||
## REST HTTP config
|
||||
|
||||
rest* {.
|
||||
|
|
|
|||
|
|
@ -88,10 +88,6 @@ proc resume*(node: WakuNode, peerList: Option[seq[PeerInfo]]) =
|
|||
##
|
||||
```
|
||||
|
||||
## JSON RPC
|
||||
|
||||
TODO To specify
|
||||
|
||||
|
||||
## REST API
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
## HTTP REST API
|
||||
|
||||
Similar to the JSON-RPC API, the HTTP REST API consists of a set of methods operating on the Waku Node remotely over HTTP.
|
||||
The HTTP REST API consists of a set of methods operating on the Waku Node remotely over HTTP.
|
||||
|
||||
This API is divided in different _namespaces_ which group a set of resources:
|
||||
|
||||
|
|
@ -41,14 +41,6 @@ A particular OpenAPI spec can be easily imported into [Postman](https://www.post
|
|||
|
||||
#### [`get_waku_v2_debug_v1_info`](https://rfc.vac.dev/spec/16/#get_waku_v2_debug_v1_info)
|
||||
|
||||
JSON-RPC call:
|
||||
|
||||
```bash
|
||||
curl -d '{"jsonrpc":"2.0","method":"get_waku_v2_debug_v1_info","params":[],"id":1}' -H 'Content-Type: application/json' localhost:8645 -s | jq
|
||||
```
|
||||
|
||||
Equivalent call for the REST API:
|
||||
|
||||
```bash
|
||||
curl http://localhost:8645/debug/v1/info -s | jq
|
||||
```
|
||||
|
|
|
|||
|
|
@ -6,8 +6,6 @@ Grep for "Listening on". It should be printed at INFO level at the beginning. E.
|
|||
|
||||
`Oct 7, 2020 @ 23:17:00.383INF 2020-10-07 23:17:00.375+00:00 Listening on topics="wakunode" tid=1 file=wakunode2.nim:140 full=/ip4/0.0.0.0/tcp/60000/p2p/16Uiu2HAmJb2e28qLXxT5kZxVUUoJt72EMzNGXB47Rxx5hw3q4YjS`
|
||||
|
||||
Or use the [JSON-RPC API](https://github.com/waku-org/nwaku/blob/master/docs/tutorial/jsonrpc-api.md#perform-a-health-check).
|
||||
|
||||
## How do I find out node addresses at the test cluster?
|
||||
|
||||
The easiest way is to use `jq` and query the fleets registry that Status operates:
|
||||
|
|
|
|||
|
|
@ -57,27 +57,10 @@ and websocket address
|
|||
```
|
||||
|
||||
You can also query a running node for its listening addresses
|
||||
using a [`get_waku_v2_debug_v1_info` JSON-RPC API](https://rfc.vac.dev/spec/16/#get_waku_v2_debug_v1_info) call.
|
||||
using the REST API.
|
||||
|
||||
For example
|
||||
|
||||
```sh
|
||||
curl -d '{"jsonrpc":"2.0","id":"id","method":"get_waku_v2_debug_v1_info", "params":[]}' --header "Content-Type: application/json" http://localhost:8545
|
||||
```
|
||||
|
||||
returns a response similar to
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"result": {
|
||||
"listenAddresses": [
|
||||
"/ip4/0.0.0.0/tcp/60000/p2p/16Uiu2HAmLU5Nwng9dWFZwM2DgJ5QGcUuDnefJyHJiXUCVaprhgL4"
|
||||
],
|
||||
"enrUri": "enr:-IO4QDxToTg86pPCK2KvMeVCXC2ADVZWrxXSvNZeaoa0JhShbM5qed69RQz1s1mWEEqJ3aoklo_7EU9iIBcPMVeKlCQBgmlkgnY0iXNlY3AyNTZrMaEDdBHK1Gx6y_zv5DVw5Qb3DtSOMmVHTZO1WSORrF2loL2DdWRwgiMohXdha3UyAw"
|
||||
}
|
||||
}
|
||||
```bash
|
||||
curl http://localhost:8645/debug/v1/info -s | jq
|
||||
```
|
||||
|
||||
## Finding your discoverable ENR address(es)
|
||||
|
|
@ -210,8 +193,4 @@ See our [store configuration tutorial](./configure-store.md) for more.
|
|||
|
||||
## Interact with a running nwaku node
|
||||
|
||||
A running nwaku node can be interacted with using the [Waku v2 JSON RPC API](https://rfc.vac.dev/spec/16/).
|
||||
|
||||
> **Note:** Private and Admin API functionality are disabled by default.
|
||||
To configure a nwaku node with these enabled,
|
||||
use the `--rpc-admin:true` and `--rpc-private:true` CLI options.
|
||||
A running nwaku node can be interacted with using the [REST API](https://github.com/waku-org/nwaku/blob/master/docs/api/rest-api.md).
|
||||
|
|
@ -29,18 +29,7 @@ or [configure](./how-to/configure.md) the node for more advanced use cases.
|
|||
|
||||
## 3. Interact
|
||||
|
||||
A running nwaku node can be interacted with using the [Waku v2 JSON RPC API](https://rfc.vac.dev/spec/16/).
|
||||
|
||||
> **Note:** Private and Admin API functionality are disabled by default.
|
||||
To configure a nwaku node with these enabled,
|
||||
use the `--rpc-admin:true` and `--rpc-private:true` CLI options.
|
||||
|
||||
```bash
|
||||
curl -d '{"jsonrpc":"2.0","method":"get_waku_v2_debug_v1_info","params":[],"id":1}' -H 'Content-Type: application/json' localhost:8546 -s | jq
|
||||
```
|
||||
|
||||
|
||||
Or using the [Waku v2 HTTP REST API](../api/v2/rest-api.md):
|
||||
A running nwaku node can be interacted with using the [REST API](../api/v2/rest-api.md).
|
||||
|
||||
> **Note:** REST API functionality is in ALPHA and therefore it is disabled by default. To configure a nwaku node with this enabled, use the `--rest:true` CLI option.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,246 +0,0 @@
|
|||
# Waku v2 JSON-RPC API Basic Tutorial
|
||||
|
||||
## Background
|
||||
|
||||
This tutorial provides step-by-step instructions on how to start a `wakunode2` with the [JSON-RPC API](https://rfc.vac.dev/spec/16/) for basic peer-to-peer messaging using publish-subscribe (pubsub) patterns.
|
||||
Libp2p pubsub-functionality is provided in Waku v2 and accessed via the [Relay API](https://rfc.vac.dev/spec/16/#relay-api).
|
||||
Debugging methods are accessed via the [Debug API](https://rfc.vac.dev/spec/16/#debug-api).
|
||||
|
||||
## Setup
|
||||
|
||||
Ensure you have built and run `wakunode2` as per [these instructions](https://github.com/waku-org/nwaku).
|
||||
|
||||
By default a running `wakunode2` will expose a JSON-RPC API on the localhost (`127.0.0.1`) at port `8545`.
|
||||
It is possible to change this configuration by setting the `rpc-address` and `rpc-port` options when running the node:
|
||||
|
||||
```
|
||||
./build/wakunode2 --rpc-address:127.0.1.1 --rpc-port:8546
|
||||
```
|
||||
|
||||
It is also possible to connect to one of our [testnets](https://github.com/waku-org/nwaku/blob/master/docs/tutorial/dingpu.md) by specifying a `staticnode` when running the node:
|
||||
|
||||
```
|
||||
./build/wakunode2 --staticnode:<multiaddr>
|
||||
```
|
||||
|
||||
where `<multiaddr>` is the multiaddress of a testnet node,
|
||||
e.g. the dingpu cluster node at
|
||||
|
||||
```
|
||||
/ip4/134.209.139.210/tcp/60000/p2p/16Uiu2HAmJb2e28qLXxT5kZxVUUoJt72EMzNGXB47Rxx5hw3q4YjS
|
||||
```
|
||||
|
||||
## Calling JSON-RPC methods
|
||||
|
||||
One way to access JSON-RPC methods is by using the `cURL` command line tool.
|
||||
For example:
|
||||
|
||||
```
|
||||
curl -d '{"jsonrpc":"2.0","id":"id","method":"<method-name>", "params":[<params>]}' --header "Content-Type: application/json" http://localhost:8545
|
||||
```
|
||||
|
||||
where `<method-name>` is the name of the JSON-RPC method to call and `<params>` is a comma-separated `Array` of parameters to pass as arguments to the selected method.
|
||||
This assumes that the API is exposed on the `localhost` at port `8545` (the default configuration).
|
||||
See [this page](https://rfc.vac.dev/spec/16/) for a full list of supported methods and parameters.
|
||||
|
||||
## Perform a health check
|
||||
|
||||
You can perform a basic health check to verify that the `wakunode2` and API is up and running by calling the [`get_waku_v2_debug_v1_info` method](https://rfc.vac.dev/spec/16/#get_waku_v2_debug_v1_info) with no parameters.
|
||||
A successful response contains the node's [`multiaddress`](https://docs.libp2p.io/concepts/addressing/).
|
||||
|
||||
### Example request
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"method": "get_waku_v2_debug_v1_info",
|
||||
"params": []
|
||||
}
|
||||
```
|
||||
|
||||
### Example response
|
||||
|
||||
The response
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"result": {
|
||||
"listenAddresses": [
|
||||
"/ip4/0.0.0.0/tcp/60000/p2p/16Uiu2HAmPv6Bq8kcmnM6gpLnjLwx3YuxN4B6AhWnN2bM9jutSS2Q"
|
||||
]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
indicates that the `wakunode2` is running and provides its multiaddress.
|
||||
|
||||
## Subscribe to a pubsub topic
|
||||
|
||||
You can subscribe to pubsub topics by calling the [`post_waku_v2_relay_v1_subscriptions` method](https://rfc.vac.dev/spec/16/#post_waku_v2_relay_v1_subscriptions) with an array of topic(s) as parameter.
|
||||
Pubsub topics are in `String` format.
|
||||
|
||||
### Example request
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"method": "post_waku_v2_relay_v1_subscriptions",
|
||||
"params": [
|
||||
[
|
||||
"my_topic_1",
|
||||
"my_topic_2",
|
||||
"my_topic_3"
|
||||
]
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### Example response
|
||||
|
||||
The response
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"result": true,
|
||||
"error": null
|
||||
}
|
||||
```
|
||||
|
||||
indicates that the `wakunode2` successfully subscribed to all three requested topics.
|
||||
|
||||
## Publish to a pubsub topic
|
||||
|
||||
To publish a message to a pubsub topic, call the [`post_waku_v2_relay_v1_message` method](https://rfc.vac.dev/spec/16/#post_waku_v2_relay_v1_message) with the target topic and publishable message as parameters.
|
||||
The message payload must be stringified as a hexadecimal string and wrapped in a [`WakuRelayMessage`](https://rfc.vac.dev/spec/16/#wakurelaymessage) `Object`.
|
||||
Besides the published payload, a `WakuRelayMessage` can also contain an optional `contentTopic` that falls outside the scope of this tutorial.
|
||||
See the [`WakuFilter` specification](https://rfc.vac.dev/spec/12/) for more information.
|
||||
|
||||
### Example request
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"method": "post_waku_v2_relay_v1_message",
|
||||
"params": [
|
||||
"my_topic_1",
|
||||
{
|
||||
"payload": "0x1a2b3c4d5e6f",
|
||||
"timestamp": 1626813243.916377
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### Example response
|
||||
|
||||
The response
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"result": true,
|
||||
"error": null
|
||||
}
|
||||
```
|
||||
|
||||
indicates that the message payload was successfully published to `"my_topic_1"`.
|
||||
|
||||
## Read new messages from a subscribed pubsub topic
|
||||
|
||||
Use the [`get_waku_v2_relay_v1_messages` method](https://rfc.vac.dev/spec/16/#get_waku_v2_relay_v1_messages) to retrieve the messages received on a subscribed pubsub topic.
|
||||
The queried topic is passed as parameter.
|
||||
This will only return new messages received after the last time this method was called.
|
||||
Repeated calls to `get_waku_v2_relay_v1_messages` can therefore be used to continuously poll a topic for new messages.
|
||||
|
||||
### Example request
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"method": "get_waku_v2_relay_v1_messages",
|
||||
"params": [
|
||||
"my_topic_1"
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### Example response
|
||||
|
||||
The response is an `Array` of [`WakuMessage` objects](https://rfc.vac.dev/spec/16/#wakumessage).
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"result": [
|
||||
{
|
||||
"payload": "0xaabbccddeeff",
|
||||
"contentTopic": 0,
|
||||
"version": 0
|
||||
},
|
||||
{
|
||||
"payload": "0x112233445566",
|
||||
"contentTopic": 0,
|
||||
"version": 0
|
||||
}
|
||||
],
|
||||
"error": null
|
||||
}
|
||||
```
|
||||
|
||||
Calling the same method again returns
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"result": [],
|
||||
"error": null
|
||||
}
|
||||
```
|
||||
|
||||
indicating that no new messages were received on the subscribed topic.
|
||||
|
||||
## Unsubscribe from a pubsub topic
|
||||
|
||||
To unsubscribe from pubsub topics, call the [`delete_waku_v2_relay_v1_subscriptions` method](https://rfc.vac.dev/spec/16/#delete_waku_v2_relay_v1_subscriptions) with an array of topic(s) as parameter.
|
||||
|
||||
### Example request
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"method": "delete_waku_v2_relay_v1_subscriptions",
|
||||
"params": [
|
||||
[
|
||||
"my_topic_1",
|
||||
"my_topic_2"
|
||||
]
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### Example response
|
||||
|
||||
The response
|
||||
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": "id",
|
||||
"result": true,
|
||||
"error": null
|
||||
}
|
||||
```
|
||||
|
||||
indicates that the `wakunode2` successfully unsubscribed from both topics.
|
||||
|
|
@ -66,14 +66,6 @@ import
|
|||
# Waku Keystore test suite
|
||||
import ./test_waku_keystore_keyfile, ./test_waku_keystore
|
||||
|
||||
## Wakunode JSON-RPC API test suite
|
||||
import
|
||||
./wakunode_jsonrpc/test_jsonrpc_admin,
|
||||
./wakunode_jsonrpc/test_jsonrpc_debug,
|
||||
./wakunode_jsonrpc/test_jsonrpc_filter,
|
||||
./wakunode_jsonrpc/test_jsonrpc_relay,
|
||||
./wakunode_jsonrpc/test_jsonrpc_store
|
||||
|
||||
## Wakunode Rest API test suite
|
||||
import
|
||||
./wakunode_rest/test_rest_debug,
|
||||
|
|
|
|||
|
|
@ -26,7 +26,6 @@ proc defaultTestWakuNodeConf*(): WakuNodeConf =
|
|||
tcpPort: Port(60000),
|
||||
websocketPort: Port(8000),
|
||||
listenAddress: parseIpAddress("0.0.0.0"),
|
||||
rpcAddress: parseIpAddress("127.0.0.1"),
|
||||
restAddress: parseIpAddress("127.0.0.1"),
|
||||
metricsServerAddress: parseIpAddress("127.0.0.1"),
|
||||
dnsAddrsNameServers: @[parseIpAddress("1.1.1.1"), parseIpAddress("1.0.0.1")],
|
||||
|
|
|
|||
|
|
@ -1,8 +0,0 @@
|
|||
{.used.}
|
||||
|
||||
import
|
||||
./test_jsonrpc_admin,
|
||||
./test_jsonrpc_debug,
|
||||
./test_jsonrpc_filter,
|
||||
./test_jsonrpc_relay,
|
||||
./test_jsonrpc_store
|
||||
|
|
@ -1,194 +0,0 @@
|
|||
{.used.}
|
||||
|
||||
import
|
||||
std/[options, sequtils],
|
||||
stew/shims/net as stewNet,
|
||||
testutils/unittests,
|
||||
chronicles,
|
||||
chronos,
|
||||
eth/keys,
|
||||
libp2p/crypto/crypto,
|
||||
json_rpc/[rpcserver, rpcclient]
|
||||
import
|
||||
../../../waku/waku_core,
|
||||
../../../waku/node/peer_manager,
|
||||
../../../waku/waku_node,
|
||||
../../../waku/waku_api/jsonrpc/admin/handlers as admin_api,
|
||||
../../../waku/waku_api/jsonrpc/admin/client as admin_api_client,
|
||||
../../../waku/waku_relay,
|
||||
../../../waku/waku_archive,
|
||||
../../../waku/waku_archive/driver/queue_driver,
|
||||
../../../waku/waku_store,
|
||||
../../../waku/waku_filter,
|
||||
../testlib/wakucore,
|
||||
../testlib/wakunode
|
||||
|
||||
|
||||
procSuite "Waku v2 JSON-RPC API - Admin":
|
||||
let
|
||||
bindIp = parseIpAddress("0.0.0.0")
|
||||
|
||||
asyncTest "connect to ad-hoc peers":
|
||||
# Create a couple of nodes
|
||||
let
|
||||
node1 = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("127.0.0.1"), Port(60600))
|
||||
node2 = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("127.0.0.1"), Port(60602))
|
||||
peerInfo2 = node2.switch.peerInfo
|
||||
node3 = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("127.0.0.1"), Port(60604))
|
||||
peerInfo3 = node3.switch.peerInfo
|
||||
|
||||
await allFutures([node1.start(), node2.start(), node3.start()])
|
||||
|
||||
await node1.mountRelay()
|
||||
await node2.mountRelay()
|
||||
await node3.mountRelay()
|
||||
|
||||
# RPC server setup
|
||||
let
|
||||
rpcPort = Port(8551)
|
||||
ta = initTAddress(bindIp, rpcPort)
|
||||
server = newRpcHttpServer([ta])
|
||||
|
||||
installAdminApiHandlers(node1, server)
|
||||
server.start()
|
||||
|
||||
let client = newRpcHttpClient()
|
||||
await client.connect("127.0.0.1", rpcPort, false)
|
||||
|
||||
# Connect to nodes 2 and 3 using the Admin API
|
||||
let postRes = await client.post_waku_v2_admin_v1_peers(@[constructMultiaddrStr(peerInfo2),
|
||||
constructMultiaddrStr(peerInfo3)])
|
||||
|
||||
check:
|
||||
postRes
|
||||
|
||||
# Verify that newly connected peers are being managed
|
||||
let getRes = await client.get_waku_v2_admin_v1_peers()
|
||||
|
||||
check:
|
||||
getRes.len == 2
|
||||
# Check peer 2
|
||||
getRes.anyIt(it.protocol == WakuRelayCodec and
|
||||
it.multiaddr == constructMultiaddrStr(peerInfo2))
|
||||
# Check peer 3
|
||||
getRes.anyIt(it.protocol == WakuRelayCodec and
|
||||
it.multiaddr == constructMultiaddrStr(peerInfo3))
|
||||
|
||||
# Verify that raises an exception if we can't connect to the peer
|
||||
let nonExistentPeer = "/ip4/0.0.0.0/tcp/10000/p2p/16Uiu2HAm6HZZr7aToTvEBPpiys4UxajCTU97zj5v7RNR2gbniy1D"
|
||||
expect(ValueError):
|
||||
discard await client.post_waku_v2_admin_v1_peers(@[nonExistentPeer])
|
||||
|
||||
let malformedPeer = "/malformed/peer"
|
||||
expect(ValueError):
|
||||
discard await client.post_waku_v2_admin_v1_peers(@[malformedPeer])
|
||||
|
||||
await server.stop()
|
||||
await server.closeWait()
|
||||
|
||||
await allFutures([node1.stop(), node2.stop(), node3.stop()])
|
||||
|
||||
asyncTest "get managed peer information":
|
||||
# Create 3 nodes and start them with relay
|
||||
let nodes = toSeq(0..<3).mapIt(newTestWakuNode(generateSecp256k1Key(), parseIpAddress("127.0.0.1"), Port(60220+it*2)))
|
||||
await allFutures(nodes.mapIt(it.start()))
|
||||
await allFutures(nodes.mapIt(it.mountRelay()))
|
||||
|
||||
# Dial nodes 2 and 3 from node1
|
||||
await nodes[0].connectToNodes(@[constructMultiaddrStr(nodes[1].peerInfo)])
|
||||
await nodes[0].connectToNodes(@[constructMultiaddrStr(nodes[2].peerInfo)])
|
||||
|
||||
# RPC server setup
|
||||
let
|
||||
rpcPort = Port(8552)
|
||||
ta = initTAddress(bindIp, rpcPort)
|
||||
server = newRpcHttpServer([ta])
|
||||
|
||||
installAdminApiHandlers(nodes[0], server)
|
||||
server.start()
|
||||
|
||||
let client = newRpcHttpClient()
|
||||
await client.connect("127.0.0.1", rpcPort, false)
|
||||
|
||||
let response = await client.get_waku_v2_admin_v1_peers()
|
||||
|
||||
check:
|
||||
response.len == 2
|
||||
# Check peer 2
|
||||
response.anyIt(it.protocol == WakuRelayCodec and
|
||||
it.multiaddr == constructMultiaddrStr(nodes[1].peerInfo))
|
||||
# Check peer 3
|
||||
response.anyIt(it.protocol == WakuRelayCodec and
|
||||
it.multiaddr == constructMultiaddrStr(nodes[2].peerInfo))
|
||||
|
||||
# Artificially remove the address from the book
|
||||
nodes[0].peerManager.peerStore[AddressBook][nodes[1].peerInfo.peerId] = @[]
|
||||
nodes[0].peerManager.peerStore[AddressBook][nodes[2].peerInfo.peerId] = @[]
|
||||
|
||||
# Verify that the returned addresses are empty
|
||||
let responseEmptyAdd = await client.get_waku_v2_admin_v1_peers()
|
||||
check:
|
||||
responseEmptyAdd[0].multiaddr == ""
|
||||
responseEmptyAdd[1].multiaddr == ""
|
||||
|
||||
await server.stop()
|
||||
await server.closeWait()
|
||||
|
||||
await allFutures(nodes.mapIt(it.stop()))
|
||||
|
||||
asyncTest "get unmanaged peer information":
|
||||
let node = newTestWakuNode(generateSecp256k1Key(), parseIpAddress("0.0.0.0"), Port(60523))
|
||||
|
||||
await node.start()
|
||||
|
||||
# RPC server setup
|
||||
let
|
||||
rpcPort = Port(8553)
|
||||
ta = initTAddress(bindIp, rpcPort)
|
||||
server = newRpcHttpServer([ta])
|
||||
|
||||
installAdminApiHandlers(node, server)
|
||||
server.start()
|
||||
|
||||
let client = newRpcHttpClient()
|
||||
await client.connect("127.0.0.1", rpcPort, false)
|
||||
|
||||
await node.mountFilter()
|
||||
await node.mountLegacyFilter()
|
||||
await node.mountFilterClient()
|
||||
let driver: ArchiveDriver = QueueDriver.new()
|
||||
let mountArchiveRes = node.mountArchive(driver)
|
||||
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
||||
|
||||
await node.mountStore()
|
||||
node.mountStoreClient()
|
||||
|
||||
# Create and set some peers
|
||||
let
|
||||
locationAddr = MultiAddress.init("/ip4/127.0.0.1/tcp/0").tryGet()
|
||||
|
||||
filterPeer = PeerInfo.new(generateEcdsaKey(), @[locationAddr])
|
||||
storePeer = PeerInfo.new(generateEcdsaKey(), @[locationAddr])
|
||||
|
||||
node.peerManager.addServicePeer(filterPeer.toRemotePeerInfo(), WakuLegacyFilterCodec)
|
||||
node.peerManager.addServicePeer(storePeer.toRemotePeerInfo(), WakuStoreCodec)
|
||||
|
||||
# Mock that we connected in the past so Identify populated this
|
||||
node.peerManager.peerStore[ProtoBook][filterPeer.peerId] = @[WakuLegacyFilterCodec]
|
||||
node.peerManager.peerStore[ProtoBook][storePeer.peerId] = @[WakuStoreCodec]
|
||||
|
||||
let response = await client.get_waku_v2_admin_v1_peers()
|
||||
|
||||
## Then
|
||||
check:
|
||||
response.len == 2
|
||||
# Check filter peer
|
||||
(response.filterIt(it.protocol == WakuLegacyFilterCodec)[0]).multiaddr == constructMultiaddrStr(filterPeer)
|
||||
# Check store peer
|
||||
(response.filterIt(it.protocol == WakuStoreCodec)[0]).multiaddr == constructMultiaddrStr(storePeer)
|
||||
|
||||
## Cleanup
|
||||
await server.stop()
|
||||
await server.closeWait()
|
||||
|
||||
await node.stop()
|
||||
|
|
@ -1,54 +0,0 @@
|
|||
{.used.}
|
||||
|
||||
import
|
||||
std/options,
|
||||
stew/shims/net as stewNet,
|
||||
testutils/unittests,
|
||||
chronicles,
|
||||
chronos,
|
||||
libp2p/crypto/crypto,
|
||||
json_rpc/[rpcserver, rpcclient]
|
||||
import
|
||||
../../../waku/node/peer_manager,
|
||||
../../../waku/waku_node,
|
||||
../../../waku/waku_api/jsonrpc/debug/handlers as debug_api,
|
||||
../../../waku/waku_api/jsonrpc/debug/client as debug_api_client,
|
||||
../testlib/common,
|
||||
../testlib/wakucore,
|
||||
../testlib/wakunode
|
||||
|
||||
|
||||
procSuite "Waku v2 JSON-RPC API - Debug":
|
||||
let
|
||||
privkey = generateSecp256k1Key()
|
||||
bindIp = parseIpAddress("0.0.0.0")
|
||||
extIp = parseIpAddress("127.0.0.1")
|
||||
port = Port(0)
|
||||
node = newTestWakuNode(privkey, bindIp, port, some(extIp), some(port))
|
||||
|
||||
asyncTest "get node info":
|
||||
await node.start()
|
||||
|
||||
await node.mountRelay()
|
||||
|
||||
# RPC server setup
|
||||
let
|
||||
rpcPort = Port(8546)
|
||||
ta = initTAddress(bindIp, rpcPort)
|
||||
server = newRpcHttpServer([ta])
|
||||
|
||||
installDebugApiHandlers(node, server)
|
||||
server.start()
|
||||
|
||||
let client = newRpcHttpClient()
|
||||
await client.connect("127.0.0.1", rpcPort, false)
|
||||
|
||||
let response = await client.get_waku_v2_debug_v1_info()
|
||||
|
||||
check:
|
||||
response.listenAddresses == @[$node.switch.peerInfo.addrs[^1] & "/p2p/" & $node.switch.peerInfo.peerId]
|
||||
|
||||
await server.stop()
|
||||
await server.closeWait()
|
||||
|
||||
await node.stop()
|
||||
|
|
@ -1,79 +0,0 @@
|
|||
{.used.}
|
||||
|
||||
import
|
||||
std/options,
|
||||
stew/shims/net as stewNet,
|
||||
testutils/unittests,
|
||||
json_rpc/[rpcserver, rpcclient]
|
||||
import
|
||||
../../../waku/waku_core,
|
||||
../../../waku/node/peer_manager,
|
||||
../../../waku/waku_node,
|
||||
../../../waku/waku_api/message_cache,
|
||||
../../../waku/waku_api/jsonrpc/filter/handlers as filter_api,
|
||||
../../../waku/waku_api/jsonrpc/filter/client as filter_api_client,
|
||||
../../../waku/waku_filter,
|
||||
../../../waku/waku_filter/rpc,
|
||||
../../../waku/waku_filter/client,
|
||||
../testlib/wakucore,
|
||||
../testlib/wakunode
|
||||
|
||||
procSuite "Waku v2 JSON-RPC API - Filter":
|
||||
let
|
||||
bindIp = parseIpAddress("0.0.0.0")
|
||||
|
||||
asyncTest "subscribe and unsubscribe":
|
||||
let
|
||||
nodeKey1 = generateSecp256k1Key()
|
||||
node1 = newTestWakuNode(nodeKey1, bindIp, Port(0))
|
||||
nodeKey2 = generateSecp256k1Key()
|
||||
node2 = newTestWakuNode(nodeKey2, bindIp, Port(0))
|
||||
|
||||
await allFutures(node1.start(), node2.start())
|
||||
|
||||
await node1.mountFilter()
|
||||
await node1.mountLegacyFilter()
|
||||
await node2.mountFilterClient()
|
||||
|
||||
node2.peerManager.addServicePeer(node1.peerInfo.toRemotePeerInfo(), WakuLegacyFilterCodec)
|
||||
|
||||
# RPC server setup
|
||||
let
|
||||
rpcPort = Port(8550)
|
||||
ta = initTAddress(bindIp, rpcPort)
|
||||
server = newRpcHttpServer([ta])
|
||||
|
||||
let cache = MessageCache.init(capacity=30)
|
||||
installFilterApiHandlers(node2, server, cache)
|
||||
server.start()
|
||||
|
||||
let client = newRpcHttpClient()
|
||||
await client.connect("127.0.0.1", rpcPort, false)
|
||||
|
||||
check:
|
||||
# Light node has not yet subscribed to any filters
|
||||
node2.wakuFilterClientLegacy.getSubscriptionsCount() == 0
|
||||
|
||||
let contentFilters = @[
|
||||
ContentFilter(contentTopic: DefaultContentTopic),
|
||||
ContentFilter(contentTopic: ContentTopic("/waku/2/default-content2/proto")),
|
||||
ContentFilter(contentTopic: ContentTopic("/waku/2/default-content3/proto")),
|
||||
ContentFilter(contentTopic: ContentTopic("/waku/2/default-content4/proto")),
|
||||
]
|
||||
var response = await client.post_waku_v2_filter_v1_subscription(contentFilters=contentFilters, topic=some(DefaultPubsubTopic))
|
||||
check:
|
||||
response == true
|
||||
# Light node has successfully subscribed to 4 content topics
|
||||
node2.wakuFilterClientLegacy.getSubscriptionsCount() == 4
|
||||
|
||||
response = await client.delete_waku_v2_filter_v1_subscription(contentFilters=contentFilters, topic=some(DefaultPubsubTopic))
|
||||
check:
|
||||
response == true
|
||||
# Light node has successfully unsubscribed from all filters
|
||||
node2.wakuFilterClientLegacy.getSubscriptionsCount() == 0
|
||||
|
||||
## Cleanup
|
||||
await server.stop()
|
||||
await server.closeWait()
|
||||
|
||||
await allFutures(node1.stop(), node2.stop())
|
||||
|
|
@ -1,194 +0,0 @@
|
|||
{.used.}
|
||||
|
||||
import
|
||||
std/[options, times, json],
|
||||
stew/shims/net as stewNet,
|
||||
testutils/unittests,
|
||||
eth/keys,
|
||||
libp2p/crypto/crypto,
|
||||
json_rpc/[rpcserver, rpcclient]
|
||||
import
|
||||
../../../waku/waku_core,
|
||||
../../../waku/waku_core/message/digest,
|
||||
../../../waku/node/peer_manager,
|
||||
../../../waku/waku_node,
|
||||
../../../waku/waku_api/jsonrpc/store/handlers as store_api,
|
||||
../../../waku/waku_api/jsonrpc/store/client as store_api_client,
|
||||
../../../waku/waku_archive,
|
||||
../../../waku/waku_archive/driver/queue_driver,
|
||||
../../../waku/waku_store,
|
||||
../../../waku/waku_store/rpc,
|
||||
../testlib/common,
|
||||
../testlib/wakucore,
|
||||
../testlib/wakunode
|
||||
|
||||
|
||||
proc put(store: ArchiveDriver, pubsubTopic: PubsubTopic, message: WakuMessage): Future[Result[void, string]] =
|
||||
let
|
||||
digest = waku_archive.computeDigest(message)
|
||||
msgHash = computeMessageHash(pubsubTopic, message)
|
||||
receivedTime = if message.timestamp > 0: message.timestamp
|
||||
else: getNanosecondTime(getTime().toUnixFloat())
|
||||
|
||||
store.put(pubsubTopic, message, digest, msgHash, receivedTime)
|
||||
|
||||
procSuite "Waku v2 JSON-RPC API - Store":
|
||||
|
||||
asyncTest "query a node and retrieve historical messages":
|
||||
let
|
||||
privkey = generateSecp256k1Key()
|
||||
bindIp = parseIpAddress("0.0.0.0")
|
||||
extIp = parseIpAddress("127.0.0.1")
|
||||
port = Port(0)
|
||||
node = newTestWakuNode(privkey, bindIp, port, some(extIp), some(port))
|
||||
|
||||
await node.start()
|
||||
|
||||
# RPC server setup
|
||||
let
|
||||
rpcPort = Port(8549)
|
||||
ta = initTAddress(bindIp, rpcPort)
|
||||
server = newRpcHttpServer([ta])
|
||||
|
||||
installStoreApiHandlers(node, server)
|
||||
server.start()
|
||||
|
||||
# WakuStore setup
|
||||
let
|
||||
key = generateEcdsaKey()
|
||||
peer = PeerInfo.new(key)
|
||||
|
||||
let driver: ArchiveDriver = QueueDriver.new()
|
||||
let mountArchiveRes = node.mountArchive(driver)
|
||||
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
||||
|
||||
await node.mountStore()
|
||||
node.mountStoreClient()
|
||||
|
||||
var listenSwitch = newStandardSwitch(some(key))
|
||||
await listenSwitch.start()
|
||||
|
||||
node.peerManager.addServicePeer(listenSwitch.peerInfo.toRemotePeerInfo(), WakuStoreCodec)
|
||||
|
||||
listenSwitch.mount(node.wakuStore)
|
||||
|
||||
# Now prime it with some history before tests
|
||||
let msgList = @[
|
||||
fakeWakuMessage(@[byte 0], contentTopic=ContentTopic("2"), ts=0),
|
||||
fakeWakuMessage(@[byte 1], ts=1),
|
||||
fakeWakuMessage(@[byte 2], ts=2),
|
||||
fakeWakuMessage(@[byte 3], ts=3),
|
||||
fakeWakuMessage(@[byte 4], ts=4),
|
||||
fakeWakuMessage(@[byte 5], ts=5),
|
||||
fakeWakuMessage(@[byte 6], ts=6),
|
||||
fakeWakuMessage(@[byte 7], ts=7),
|
||||
fakeWakuMessage(@[byte 8], ts=8),
|
||||
fakeWakuMessage(@[byte 9], contentTopic=ContentTopic("2"), ts=9)
|
||||
]
|
||||
|
||||
for msg in msgList:
|
||||
require (waitFor driver.put(DefaultPubsubTopic, msg)).isOk()
|
||||
|
||||
let client = newRpcHttpClient()
|
||||
await client.connect("127.0.0.1", rpcPort, false)
|
||||
|
||||
let response = await client.get_waku_v2_store_v1_messages(
|
||||
some(DefaultPubsubTopic),
|
||||
some(@[HistoryContentFilterRPC(contentTopic: DefaultContentTopic)]),
|
||||
some(Timestamp(0)),
|
||||
some(Timestamp(9)),
|
||||
some(StorePagingOptions()))
|
||||
check:
|
||||
response.messages.len == 8
|
||||
response.pagingOptions.isNone()
|
||||
|
||||
await server.stop()
|
||||
await server.closeWait()
|
||||
await node.stop()
|
||||
await listenSwitch.stop()
|
||||
|
||||
asyncTest "check error response when peer-store-node is not available":
|
||||
let
|
||||
privkey = generateSecp256k1Key()
|
||||
bindIp = parseIpAddress("0.0.0.0")
|
||||
extIp = parseIpAddress("127.0.0.1")
|
||||
port = Port(0)
|
||||
node = newTestWakuNode(privkey, bindIp, port, some(extIp), some(port))
|
||||
|
||||
await node.start()
|
||||
|
||||
# RPC server setup
|
||||
let
|
||||
rpcPort = Port(8549)
|
||||
ta = initTAddress(bindIp, rpcPort)
|
||||
server = newRpcHttpServer([ta])
|
||||
|
||||
installStoreApiHandlers(node, server)
|
||||
server.start()
|
||||
|
||||
let driver: ArchiveDriver = QueueDriver.new()
|
||||
let mountArchiveRes = node.mountArchive(driver)
|
||||
assert mountArchiveRes.isOk(), mountArchiveRes.error
|
||||
|
||||
await node.mountStore()
|
||||
node.mountStoreClient()
|
||||
|
||||
# Now prime it with some history before tests
|
||||
let msgList = @[
|
||||
fakeWakuMessage(@[byte 0], ts=0),
|
||||
fakeWakuMessage(@[byte 9], ts=9)
|
||||
]
|
||||
for msg in msgList:
|
||||
require (waitFor driver.put(DefaultPubsubTopic, msg)).isOk()
|
||||
|
||||
let client = newRpcHttpClient()
|
||||
await client.connect("127.0.0.1", rpcPort, false)
|
||||
|
||||
var response:StoreResponse
|
||||
var jsonError:JsonNode
|
||||
try:
|
||||
response = await client.get_waku_v2_store_v1_messages(
|
||||
some(DefaultPubsubTopic),
|
||||
some(@[HistoryContentFilterRPC(contentTopic: DefaultContentTopic)]),
|
||||
some(Timestamp(0)),
|
||||
some(Timestamp(9)),
|
||||
some(StorePagingOptions()))
|
||||
except ValueError:
|
||||
|
||||
try:
|
||||
jsonError = parseJson(getCurrentExceptionMsg())
|
||||
except Exception:
|
||||
assert false, "exception calling parseJson: " & getCurrentExceptionMsg()
|
||||
|
||||
check:
|
||||
$jsonError["code"] == "-32000"
|
||||
jsonError["message"].getStr() == "get_waku_v2_store_v1_messages raised an exception"
|
||||
jsonError["data"].getStr() == "no suitable remote store peers"
|
||||
|
||||
# Now configure a store-peer
|
||||
let
|
||||
key = generateEcdsaKey()
|
||||
peer = PeerInfo.new(key)
|
||||
|
||||
var listenSwitch = newStandardSwitch(some(key))
|
||||
await listenSwitch.start()
|
||||
|
||||
listenSwitch.mount(node.wakuStore)
|
||||
|
||||
node.peerManager.addServicePeer(listenSwitch.peerInfo.toRemotePeerInfo(),
|
||||
WakuStoreCodec)
|
||||
|
||||
response = await client.get_waku_v2_store_v1_messages(
|
||||
some(DefaultPubsubTopic),
|
||||
some(@[HistoryContentFilterRPC(contentTopic: DefaultContentTopic)]),
|
||||
some(Timestamp(0)),
|
||||
some(Timestamp(9)),
|
||||
some(StorePagingOptions()))
|
||||
check:
|
||||
response.messages.len == 2
|
||||
response.pagingOptions.isNone()
|
||||
|
||||
await server.stop()
|
||||
await server.closeWait()
|
||||
await node.stop()
|
||||
await listenSwitch.stop()
|
||||
|
|
@ -5,5 +5,4 @@ import
|
|||
|
||||
export
|
||||
message_cache,
|
||||
rest,
|
||||
jsonrpc
|
||||
rest
|
||||
|
|
@ -1,4 +0,0 @@
|
|||
# Admin API
|
||||
|
||||
proc get_waku_v2_admin_v1_peers(): seq[WakuPeer]
|
||||
proc post_waku_v2_admin_v1_peers(peers: seq[string]): bool
|
||||
|
|
@ -1,14 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/[os, strutils],
|
||||
json_rpc/rpcclient
|
||||
import
|
||||
./types
|
||||
|
||||
template sourceDir: string = currentSourcePath.rsplit(DirSep, 1)[0]
|
||||
|
||||
createRpcSigs(RpcHttpClient, sourceDir / "callsigs.nim")
|
||||
|
|
@ -1,71 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/sequtils,
|
||||
chronicles,
|
||||
json_rpc/rpcserver,
|
||||
libp2p/[peerinfo, switch]
|
||||
import
|
||||
../../../waku_store,
|
||||
../../../waku_filter,
|
||||
../../../waku_relay,
|
||||
../../../waku_node,
|
||||
../../../node/peer_manager,
|
||||
../../../waku_core,
|
||||
./types
|
||||
|
||||
|
||||
logScope:
|
||||
topics = "waku node jsonrpc admin_api"
|
||||
|
||||
proc installAdminApiHandlers*(node: WakuNode, rpcsrv: RpcServer) =
|
||||
|
||||
rpcsrv.rpc("post_waku_v2_admin_v1_peers") do (peers: seq[string]) -> bool:
|
||||
## Connect to a list of peers
|
||||
debug "post_waku_v2_admin_v1_peers"
|
||||
|
||||
for i, peer in peers:
|
||||
let peerInfo = parsePeerInfo(peer)
|
||||
if peerInfo.isErr():
|
||||
raise newException(ValueError, "Couldn't parse remote peer info: " & peerInfo.error)
|
||||
|
||||
let connOk = await node.peerManager.connectRelay(peerInfo.value, source="rpc")
|
||||
if not connOk:
|
||||
raise newException(ValueError, "Failed to connect to peer at index: " & $i & " " & $peer)
|
||||
|
||||
return true
|
||||
|
||||
rpcsrv.rpc("get_waku_v2_admin_v1_peers") do () -> seq[WakuPeer]:
|
||||
## Returns a list of peers registered for this node
|
||||
debug "get_waku_v2_admin_v1_peers"
|
||||
|
||||
var peers = newSeq[WakuPeer]()
|
||||
|
||||
if not node.wakuRelay.isNil():
|
||||
# Map managed peers to WakuPeers and add to return list
|
||||
let relayPeers = node.peerManager.peerStore.peers(WakuRelayCodec)
|
||||
.mapIt(WakuPeer(multiaddr: constructMultiaddrStr(it),
|
||||
protocol: WakuRelayCodec,
|
||||
connected: it.connectedness == Connectedness.Connected))
|
||||
peers.add(relayPeers)
|
||||
|
||||
if not node.wakuFilterLegacy.isNil():
|
||||
# Map WakuFilter peers to WakuPeers and add to return list
|
||||
let filterPeers = node.peerManager.peerStore.peers(WakuLegacyFilterCodec)
|
||||
.mapIt(WakuPeer(multiaddr: constructMultiaddrStr(it),
|
||||
protocol: WakuLegacyFilterCodec,
|
||||
connected: it.connectedness == Connectedness.Connected))
|
||||
peers.add(filterPeers)
|
||||
|
||||
if not node.wakuStore.isNil():
|
||||
# Map WakuStore peers to WakuPeers and add to return list
|
||||
let storePeers = node.peerManager.peerStore.peers(WakuStoreCodec)
|
||||
.mapIt(WakuPeer(multiaddr: constructMultiaddrStr(it),
|
||||
protocol: WakuStoreCodec,
|
||||
connected: it.connectedness == Connectedness.Connected))
|
||||
peers.add(storePeers)
|
||||
|
||||
return peers
|
||||
|
|
@ -1,10 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
|
||||
type WakuPeer* = object
|
||||
multiaddr*: string
|
||||
protocol*: string
|
||||
connected*: bool
|
||||
|
|
@ -1,3 +0,0 @@
|
|||
# Debug API
|
||||
|
||||
proc get_waku_v2_debug_v1_info(): WakuInfo
|
||||
|
|
@ -1,14 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/[os, strutils],
|
||||
json_rpc/rpcclient
|
||||
import
|
||||
../../../waku_node
|
||||
|
||||
template sourceDir: string = currentSourcePath.rsplit(DirSep, 1)[0]
|
||||
|
||||
createRpcSigs(RpcHttpClient, sourceDir / "callsigs.nim")
|
||||
|
|
@ -1,30 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
chronicles,
|
||||
json_rpc/rpcserver
|
||||
import
|
||||
../../../waku_node
|
||||
|
||||
logScope:
|
||||
topics = "waku node jsonrpc debug_api"
|
||||
|
||||
proc installDebugApiHandlers*(node: WakuNode, server: RpcServer) =
|
||||
|
||||
## Debug API version 1 definitions
|
||||
|
||||
server.rpc("get_waku_v2_debug_v1_info") do () -> WakuInfo:
|
||||
## Returns information about WakuNode
|
||||
debug "get_waku_v2_debug_v1_info"
|
||||
|
||||
return node.info()
|
||||
|
||||
server.rpc("get_waku_v2_debug_v1_version") do () -> string:
|
||||
## Returns information about WakuNode
|
||||
debug "get_waku_v2_debug_v1_version"
|
||||
|
||||
return git_version
|
||||
|
||||
|
|
@ -1,5 +0,0 @@
|
|||
# Filter API
|
||||
|
||||
proc get_waku_v2_filter_v1_messages(contentTopic: ContentTopic): seq[WakuMessage]
|
||||
proc post_waku_v2_filter_v1_subscription(contentFilters: seq[ContentFilter], topic: Option[string]): bool
|
||||
proc delete_waku_v2_filter_v1_subscription(contentFilters: seq[ContentFilter], topic: Option[string]): bool
|
||||
|
|
@ -1,15 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/[os, strutils],
|
||||
json_rpc/rpcclient
|
||||
import
|
||||
../../../waku_core,
|
||||
../../../waku_filter/rpc
|
||||
|
||||
template sourceDir: string = currentSourcePath.rsplit(DirSep, 1)[0]
|
||||
|
||||
createRpcSigs(RpcHttpClient, sourceDir / "callsigs.nim")
|
||||
|
|
@ -1,83 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/sequtils,
|
||||
chronicles,
|
||||
json_rpc/rpcserver
|
||||
import
|
||||
../../../waku_core,
|
||||
../../../waku_filter,
|
||||
../../../waku_filter/rpc,
|
||||
../../../waku_filter/client,
|
||||
../../../waku_node,
|
||||
../../../node/peer_manager,
|
||||
../../message_cache
|
||||
|
||||
|
||||
|
||||
logScope:
|
||||
topics = "waku node jsonrpc filter_api"
|
||||
|
||||
|
||||
const futTimeout* = 5.seconds # Max time to wait for futures
|
||||
|
||||
proc installFilterApiHandlers*(node: WakuNode, server: RpcServer, cache: MessageCache) =
|
||||
|
||||
server.rpc("post_waku_v2_filter_v1_subscription") do (contentFilters: seq[ContentFilter], pubsubTopic: Option[PubsubTopic]) -> bool:
|
||||
## Subscribes a node to a list of content filters
|
||||
debug "post_waku_v2_filter_v1_subscription"
|
||||
|
||||
let peerOpt = node.peerManager.selectPeer(WakuLegacyFilterCodec)
|
||||
if peerOpt.isNone():
|
||||
raise newException(ValueError, "no suitable remote filter peers")
|
||||
|
||||
let contentTopics: seq[ContentTopic] = contentFilters.mapIt(it.contentTopic)
|
||||
|
||||
let handler: FilterPushHandler = proc(pubsubTopic: PubsubTopic, msg: WakuMessage) {.async, gcsafe, closure.} =
|
||||
cache.addMessage(pubsubTopic, msg)
|
||||
|
||||
let subFut = node.legacyFilterSubscribe(pubsubTopic, contentTopics, handler, peerOpt.get())
|
||||
if not await subFut.withTimeout(futTimeout):
|
||||
raise newException(ValueError, "Failed to subscribe to contentFilters")
|
||||
|
||||
# Successfully subscribed to all content filters
|
||||
for cTopic in contentTopics:
|
||||
cache.contentSubscribe(cTopic)
|
||||
|
||||
return true
|
||||
|
||||
server.rpc("delete_waku_v2_filter_v1_subscription") do (contentFilters: seq[ContentFilter], pubsubTopic: Option[PubsubTopic]) -> bool:
|
||||
## Unsubscribes a node from a list of content filters
|
||||
debug "delete_waku_v2_filter_v1_subscription"
|
||||
|
||||
let contentTopics: seq[ContentTopic] = contentFilters.mapIt(it.contentTopic)
|
||||
|
||||
let peerOpt = node.peerManager.selectPeer(WakuLegacyFilterCodec)
|
||||
if peerOpt.isNone():
|
||||
raise newException(ValueError, "no suitable remote filter peers")
|
||||
|
||||
let unsubFut = node.legacyFilterUnsubscribe(pubsubTopic, contentTopics, peerOpt.get())
|
||||
if not await unsubFut.withTimeout(futTimeout):
|
||||
raise newException(ValueError, "Failed to unsubscribe from contentFilters")
|
||||
|
||||
for cTopic in contentTopics:
|
||||
cache.contentUnsubscribe(cTopic)
|
||||
|
||||
return true
|
||||
|
||||
server.rpc("get_waku_v2_filter_v1_messages") do (contentTopic: ContentTopic) -> seq[WakuMessage]:
|
||||
## Returns all WakuMessages received on a content topic since the
|
||||
## last time this method was called
|
||||
debug "get_waku_v2_filter_v1_messages", contentTopic=contentTopic
|
||||
|
||||
if not cache.isContentSubscribed(contentTopic):
|
||||
raise newException(ValueError, "Not subscribed to topic: " & contentTopic)
|
||||
|
||||
let msgRes = cache.getMessages(contentTopic, clear=true)
|
||||
if msgRes.isErr():
|
||||
raise newException(ValueError, "Not subscribed to topic: " & contentTopic)
|
||||
|
||||
return msgRes.value
|
||||
|
|
@ -1,6 +0,0 @@
|
|||
{.deprecated: "import 'waku/common/hexstrings' instead".}
|
||||
|
||||
import ../../common/hexstrings
|
||||
|
||||
export hexstrings
|
||||
|
||||
|
|
@ -1,19 +0,0 @@
|
|||
import
|
||||
stew/byteutils,
|
||||
json,
|
||||
json_rpc/rpcserver
|
||||
|
||||
|
||||
func invalidMsg*(name: string): string =
|
||||
"When marshalling from JSON, parameter \"" & name & "\" is not valid"
|
||||
|
||||
|
||||
## JSON marshalling
|
||||
|
||||
# seq[byte]
|
||||
|
||||
proc `%`*(value: seq[byte]): JsonNode =
|
||||
if value.len > 0:
|
||||
%("0x" & value.toHex())
|
||||
else:
|
||||
newJArray()
|
||||
|
|
@ -1,49 +0,0 @@
|
|||
import
|
||||
std/options,
|
||||
json,
|
||||
json_rpc/rpcserver
|
||||
import
|
||||
../../common/base64,
|
||||
../../waku_core
|
||||
|
||||
|
||||
type
|
||||
WakuMessageRPC* = object
|
||||
payload*: Base64String
|
||||
contentTopic*: Option[ContentTopic]
|
||||
version*: Option[uint32]
|
||||
timestamp*: Option[Timestamp]
|
||||
ephemeral*: Option[bool]
|
||||
|
||||
|
||||
## Type mappings
|
||||
|
||||
func toWakuMessageRPC*(msg: WakuMessage): WakuMessageRPC =
|
||||
WakuMessageRPC(
|
||||
payload: base64.encode(msg.payload),
|
||||
contentTopic: some(msg.contentTopic),
|
||||
version: some(msg.version),
|
||||
timestamp: some(msg.timestamp),
|
||||
ephemeral: some(msg.ephemeral)
|
||||
)
|
||||
|
||||
|
||||
## JSON-RPC type marshalling
|
||||
|
||||
# Base64String
|
||||
|
||||
proc `%`*(value: Base64String): JsonNode =
|
||||
%(value.string)
|
||||
|
||||
proc fromJson*(n: JsonNode, argName: string, value: var Base64String) =
|
||||
n.kind.expect(JString, argName)
|
||||
|
||||
value = Base64String(n.getStr())
|
||||
|
||||
# WakuMessageRpc (WakuMessage)
|
||||
|
||||
proc `%`*(value: WakuMessageRpc): JsonNode =
|
||||
let jObj = newJObject()
|
||||
for k, v in value.fieldPairs:
|
||||
jObj[k] = %v
|
||||
return jObj
|
||||
|
|
@ -1,14 +0,0 @@
|
|||
# Relay API
|
||||
proc post_waku_v2_relay_v1_subscriptions(topics: seq[PubsubTopic]): bool
|
||||
proc delete_waku_v2_relay_v1_subscriptions(topics: seq[PubsubTopic]): bool
|
||||
proc post_waku_v2_relay_v1_message(topic: PubsubTopic, message: WakuMessageRPC): bool
|
||||
proc get_waku_v2_relay_v1_messages(topic: PubsubTopic): seq[WakuMessageRPC]
|
||||
|
||||
proc post_waku_v2_relay_v1_auto_subscriptions(topics: seq[ContentTopic]): bool
|
||||
proc delete_waku_v2_relay_v1_auto_subscriptions(topics: seq[ContentTopic]): bool
|
||||
proc post_waku_v2_relay_v1_auto_message(message: WakuMessageRPC): bool
|
||||
proc get_waku_v2_relay_v1_auto_messages(topic: ContentTopic): seq[WakuMessageRPC]
|
||||
|
||||
|
||||
# Support for the Relay Private API has been deprecated.
|
||||
# This API existed for compatibility with the Waku v1 spec and encryption scheme.
|
||||
|
|
@ -1,17 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/[os, strutils],
|
||||
json_rpc/rpcclient
|
||||
import
|
||||
../../../waku_core,
|
||||
../message
|
||||
|
||||
export message
|
||||
|
||||
template sourceDir: string = currentSourcePath.rsplit(DirSep, 1)[0]
|
||||
|
||||
createRpcSigs(RpcHttpClient, sourceDir / "callsigs.nim")
|
||||
|
|
@ -1,4 +0,0 @@
|
|||
# Store API
|
||||
|
||||
proc get_waku_v2_store_v1_messages(pubsubTopicOption: Option[string], contentFiltersOption: Option[seq[HistoryContentFilterRPC]], startTime: Option[Timestamp], endTime: Option[Timestamp], pagingOptions: Option[StorePagingOptions]): StoreResponse
|
||||
|
||||
|
|
@ -1,19 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/[os, strutils],
|
||||
json_rpc/rpcclient
|
||||
import
|
||||
../../../waku_core,
|
||||
../../../waku_store/rpc,
|
||||
./types
|
||||
|
||||
export types
|
||||
|
||||
|
||||
template sourceDir: string = currentSourcePath.rsplit(DirSep, 1)[0]
|
||||
|
||||
createRpcSigs(RpcHttpClient, sourceDir / "callsigs.nim")
|
||||
|
|
@ -1,85 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/[options, sequtils],
|
||||
chronicles,
|
||||
json_rpc/rpcserver
|
||||
import
|
||||
../../../[
|
||||
waku_core,
|
||||
waku_store,
|
||||
waku_node
|
||||
],
|
||||
../../../waku_store/rpc,
|
||||
../../../node/peer_manager,
|
||||
../../../common/paging,
|
||||
./types
|
||||
|
||||
|
||||
logScope:
|
||||
topics = "waku node jsonrpc store_api"
|
||||
|
||||
|
||||
const futTimeout = 5.seconds
|
||||
|
||||
|
||||
proc toPagingInfo*(pagingOptions: StorePagingOptions): PagingInfoRPC =
|
||||
PagingInfoRPC(
|
||||
pageSize: some(pagingOptions.pageSize),
|
||||
cursor: pagingOptions.cursor,
|
||||
direction: some(pagingOptions.forward.into())
|
||||
)
|
||||
|
||||
proc toPagingOptions*(pagingInfo: PagingInfoRPC): StorePagingOptions =
|
||||
StorePagingOptions(
|
||||
pageSize: pagingInfo.pageSize.get(0'u64),
|
||||
cursor: pagingInfo.cursor,
|
||||
forward: pagingInfo.direction.into()
|
||||
)
|
||||
|
||||
proc toJsonRPCStoreResponse*(response: HistoryResponse): StoreResponse =
|
||||
StoreResponse(
|
||||
messages: response.messages.map(toWakuMessageRPC),
|
||||
pagingOptions: if response.cursor.isNone(): none(StorePagingOptions)
|
||||
else: some(StorePagingOptions(
|
||||
pageSize: uint64(response.messages.len), # This field will be deprecated soon
|
||||
forward: true, # Hardcoded. This field will be deprecated soon
|
||||
cursor: response.cursor.map(toRPC)
|
||||
))
|
||||
)
|
||||
|
||||
proc installStoreApiHandlers*(node: WakuNode, server: RpcServer) =
|
||||
|
||||
server.rpc("get_waku_v2_store_v1_messages") do (pubsubTopicOption: Option[string], contentFiltersOption: Option[seq[HistoryContentFilterRPC]], startTime: Option[Timestamp], endTime: Option[Timestamp], pagingOptions: Option[StorePagingOptions]) -> StoreResponse:
|
||||
## Returns history for a list of content topics with optional paging
|
||||
debug "get_waku_v2_store_v1_messages"
|
||||
|
||||
let peerOpt = node.peerManager.selectPeer(WakuStoreCodec)
|
||||
if peerOpt.isNone():
|
||||
raise newException(ValueError, "no suitable remote store peers")
|
||||
|
||||
let req = HistoryQuery(
|
||||
pubsubTopic: pubsubTopicOption,
|
||||
contentTopics: contentFiltersOption.get(@[]).mapIt(it.contentTopic),
|
||||
startTime: startTime,
|
||||
endTime: endTime,
|
||||
direction: if pagingOptions.isNone(): default()
|
||||
else: pagingOptions.get().forward.into(),
|
||||
pageSize: if pagingOptions.isNone(): DefaultPageSize
|
||||
else: min(pagingOptions.get().pageSize, MaxPageSize),
|
||||
cursor: if pagingOptions.isNone(): none(HistoryCursor)
|
||||
else: pagingOptions.get().cursor.map(toAPI)
|
||||
)
|
||||
|
||||
let queryFut = node.query(req, peerOpt.get())
|
||||
if not await queryFut.withTimeout(futTimeout):
|
||||
raise newException(ValueError, "No history response received (timeout)")
|
||||
|
||||
let res = queryFut.read()
|
||||
if res.isErr():
|
||||
raise newException(ValueError, $res.error)
|
||||
|
||||
return res.value.toJsonRPCStoreResponse()
|
||||
|
|
@ -1,24 +0,0 @@
|
|||
when (NimMajor, NimMinor) < (1, 4):
|
||||
{.push raises: [Defect].}
|
||||
else:
|
||||
{.push raises: [].}
|
||||
|
||||
import
|
||||
std/options
|
||||
import
|
||||
../../../waku_store/rpc,
|
||||
../message
|
||||
|
||||
export message
|
||||
|
||||
|
||||
type
|
||||
StoreResponse* = object
|
||||
messages*: seq[WakuMessageRPC]
|
||||
pagingOptions*: Option[StorePagingOptions]
|
||||
|
||||
StorePagingOptions* = object
|
||||
## This type holds some options for pagination
|
||||
pageSize*: uint64
|
||||
cursor*: Option[PagingIndexRPC]
|
||||
forward*: bool
|
||||
Loading…
Reference in New Issue