Skip to content

Add support for two-way communication #127

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add support for two-way communication #127

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 0 additions & 7 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,3 @@
# Ignore all files without extensions (unix executable files)
*
!*.*
!*/

# Ignore all binary outputs
*.so
*.dylib
Expand All @@ -16,5 +11,3 @@ nimcache/

# Ignore editor settings
.vscode


3 changes: 1 addition & 2 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -346,8 +346,7 @@ waitFor client.connect("localhost", Port(8545))

let response = waitFor client.call("hello", %[%"Daisy"])

# the call returns a `Response` type which contains the result
echo response.result
echo response
```

### `createRpcSigs`
Expand Down
9 changes: 7 additions & 2 deletions json_rpc.nimble
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ requires "nim >= 1.2.0",
"chronicles",
"https://github.com/status-im/news#status",
"websock",
"asynctools",
"faststreams",
"json_serialization"

proc buildBinary(name: string, srcDir = "./", params = "", cmdParams = "") =
Expand All @@ -24,7 +26,10 @@ proc buildBinary(name: string, srcDir = "./", params = "", cmdParams = "") =

task test, "run tests":
buildBinary "all", "tests/",
params = "-d:json_rpc_websocket_package=websock"
params = "-d:json_rpc_websocket_package=websock -d:asyncBackend=chronos"

buildBinary "all", "tests/",
params = "-d:json_rpc_websocket_package=news"
params = "-d:json_rpc_websocket_package=news -d:asyncBackend=chronos"

buildBinary "teststreamconnection", "tests/",
params = "-d:asyncBackend=asyncdispatch"
2 changes: 2 additions & 0 deletions json_rpc.nimble.cfg
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
--define:"async_backend=asyncdispatch"
--threads:on
Copy link
Contributor

@kdeme kdeme Feb 17, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

newline

Copy link
Contributor Author

@yyoncho yyoncho Feb 17, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

will be added.

11 changes: 7 additions & 4 deletions json_rpc/client.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
import
std/[tables, macros],
chronos,
faststreams/async_backend,
./jsonmarshal

from strutils import toLowerAscii, replace

export
chronos, jsonmarshal, tables
jsonmarshal, tables

type
ClientId* = int64
Expand All @@ -26,13 +26,16 @@ proc getNextId*(client: RpcClient): ClientId =
proc rpcCallNode*(path: string, params: JsonNode, id: ClientId): JsonNode =
%{"jsonrpc": %"2.0", "method": %path, "params": params, "id": %id}

proc rpcNotificationNode*(path: string, params: JsonNode): JsonNode =
%{"jsonrpc": %"2.0", "method": %path, "params": params}

method call*(client: RpcClient, name: string,
params: JsonNode): Future[Response] {.
base, async, gcsafe, raises: [Defect, CatchableError].} =
base, async, gcsafe, raises: [Defect, CatchableError, Exception].} =
Copy link
Member

@arnetheduck arnetheduck Feb 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

throughout, Exception should never appear as part of the raises list - if it "leaks" from somewhere, it should be contained in that place by correctly annotating that source, or worst case, surrounding the code with try: ... except CatchableError as exc: raise exc except Exception ass exc: raiseAssert exc.msg - usually Execption infections come from incorrect forward and/or proc declarations.

Copy link
Contributor Author

@yyoncho yyoncho Feb 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I will track down that. I believe it was caused by async when used in combination with asyncdispatch but I have to verify that.

Copy link
Member

@arnetheduck arnetheduck Feb 15, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

we generally don't support / work with asyncdispatch (it doesn't work in faststreams for example), so that would be surprising - importantly though, raises lists are infectious.

ah, I also see that this function in particular is async - those should not have raises lists at all (chronos does not yet fully support raises lists for async functions and adding them causes trouble - the correct annotation for an async function is raises: [Defect] which should be done with a push raises: [Defect] on top of every file

Copy link
Contributor Author

@yyoncho yyoncho Feb 17, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FYI, here is the root cause:

import std/asyncdispatch

proc echoString(params: string): Future[string]
    {.async, raises: [Defect, CatchableError].} =
  return params

Fails with

/home/yyoncho/Sources/nim/Nim/lib/pure/asyncmacro.nim(232, 33) Error: echoStringNimAsyncContinue_436207620() can raise an unlisted exception: Exception

Copy link
Member

@arnetheduck arnetheduck Feb 17, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yes, so what async does is turn the function into two functions - one "wrapper" that creates a future and one closure that contains the body that you wrote - the closure catches all exceptions and puts them in the future, generally - the raises annotation is applied to the "wrapper", not the "user code" (as it perhaps should be, but this has other problems), hence the correct raises annotation for async is raises: [Defect], with async functions - if asyncdispatch fails here, it's a deeper issue with AD that needs to be resolved in the std lib most likely - we require that no exceptions are raised from the "wrapper", and enforce that in most projects (nim-json-rpc is just a little bit behind in maintenance - more info:

discard

method close*(client: RpcClient): Future[void] {.
base, async, gcsafe, raises: [Defect, CatchableError].} =
base, async, gcsafe, raises: [Defect, CatchableError, Exception].} =
discard

template `or`(a: JsonNode, b: typed): JsonNode =
Expand Down
2 changes: 1 addition & 1 deletion json_rpc/clients/httpclient.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
import
std/[strutils, tables, uri],
std/[tables, uri],
stew/[byteutils, results],
chronos/apps/http/httpclient as chronosHttpClient,
chronicles, httputils, json_serialization/std/net,
Expand Down
2 changes: 1 addition & 1 deletion json_rpc/clients/socketclient.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ import
chronos,
../client

export client
export client, chronos

type
RpcSocketClient* = ref object of RpcClient
Expand Down
4 changes: 2 additions & 2 deletions json_rpc/clients/websocketclient.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import
std/[strtabs, tables],
std/[tables],
pkg/[chronos, chronicles],
stew/byteutils,
../client, ./config

export client
Expand All @@ -21,6 +20,7 @@ when useNews:

else:
import std/[uri, strutils]
import stew/byteutils
import pkg/websock/[websock, extensions/compression/deflate]

type
Expand Down
77 changes: 39 additions & 38 deletions json_rpc/router.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,19 +1,21 @@
import
std/[macros, options, strutils, tables],
chronicles, chronos, json_serialization/writer,
std/[macros, options, strutils, tables], sugar,
Copy link
Contributor

@kdeme kdeme Feb 17, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sugar can go in the std brackets

Copy link
Contributor Author

@yyoncho yyoncho Feb 17, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

sure.

chronicles, faststreams/async_backend, json_serialization/writer,
./jsonmarshal, ./errors

export
chronos, jsonmarshal
export jsonmarshal

type
StringOfJson* = JsonString

RpcResult* = Option[JsonString]

# Procedure signature accepted as an RPC call by server
RpcProc* = proc(input: JsonNode): Future[StringOfJson] {.gcsafe, raises: [Defect, CatchableError].}
RpcProc* = proc(input: JsonNode): Future[RpcResult] {.gcsafe, raises: [Defect, CatchableError, Exception].}

RpcRouter* = object
procs*: Table[string, RpcProc]
fullParams*: bool # if false send "params" to the handlers

const
methodField = "method"
Expand All @@ -34,7 +36,7 @@ proc newRpcRouter*: RpcRouter {.deprecated.} =
RpcRouter.init()

proc register*(router: var RpcRouter, path: string, call: RpcProc) =
router.procs.add(path, call)
router.procs[path] = call

proc clear*(router: var RpcRouter) =
router.procs.clear
Expand All @@ -59,48 +61,57 @@ proc wrapError*(code: int, msg: string, id: JsonNode = newJNull(),
$id, $code, escapeJson(msg), $data
] & "\r\n")

proc route*(router: RpcRouter, node: JsonNode): Future[StringOfJson] {.async, gcsafe.} =
proc hasReturnType(params: NimNode): bool =
if params != nil and params.len > 0 and params[0] != nil and
params[0].kind != nnkEmpty:
result = true

proc route*(router: RpcRouter, node: JsonNode): Future[RpcResult] {.async, gcsafe.} =
Copy link
Member

@arnetheduck arnetheduck Feb 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The change to RpcResult is also an API-breaking change that will upset existing consumers

Copy link
Contributor Author

@yyoncho yyoncho Feb 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is the router used outside of nim-json-rpc?

Copy link
Member

@arnetheduck arnetheduck Feb 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, it's public, so it can be used - this is a bit of a messy aspect of Nim, the module system is shifty - a first litmus test would be to check if nimbus-eth1 and nimbus-eth2 still compile after the change - if these to projects compile, it's mostly fine for a library like json-rpc where we don't yet carefully maintain backwards compat

if node{"jsonrpc"}.getStr() != "2.0":
return wrapError(INVALID_REQUEST, "'jsonrpc' missing or invalid")
return some(wrapError(INVALID_REQUEST, "'jsonrpc' missing or invalid"))

let id = node{"id"}
if id == nil:
return wrapError(INVALID_REQUEST, "'id' missing or invalid")

let methodName = node{"method"}.getStr()
if methodName.len == 0:
return wrapError(INVALID_REQUEST, "'method' missing or invalid")
return some(wrapError(INVALID_REQUEST, "'method' missing or invalid"))

let rpcProc = router.procs.getOrDefault(methodName)
let params = node.getOrDefault("params")

if rpcProc == nil:
return wrapError(METHOD_NOT_FOUND, "'" & methodName & "' is not a registered RPC method", id)
return some(wrapError(METHOD_NOT_FOUND, "'" & methodName & "' is not a registered RPC method", id))
else:
try:
let params = if router.fullParams:
node
else:
node.getOrDefault("params")

let res = await rpcProc(if params == nil: newJArray() else: params)
return wrapReply(id, res)

return res.map((s) => wrapReply(id, s));
except InvalidRequest as err:
return wrapError(err.code, err.msg)
debug "Error occurred within RPC", methodName = methodName, err = err.msg
return some(wrapError(err.code, err.msg))
except CatchableError as err:
debug "Error occurred within RPC", methodName = methodName, err = err.msg
return wrapError(
SERVER_ERROR, methodName & " raised an exception", id, newJString(err.msg))
return some(wrapError(
SERVER_ERROR, methodName & " raised an exception", id, newJString(err.msg)))

proc route*(router: RpcRouter, data: string): Future[string] {.async, gcsafe.} =
proc route*(router: RpcRouter, data: string): Future[RpcResult] {.async, gcsafe.} =
## Route to RPC from string data. Data is expected to be able to be converted to Json.
## Returns string of Json from RPC result/error node
let node =
try: parseJson(data)
except CatchableError as err:
return string(wrapError(JSON_PARSE_ERROR, err.msg))
return some(wrapError(JSON_PARSE_ERROR, err.msg))
except Exception as err:
# TODO https://github.com/status-im/nimbus-eth2/issues/2430
return string(wrapError(JSON_PARSE_ERROR, err.msg))
return some(wrapError(JSON_PARSE_ERROR, err.msg))

return string(await router.route(node))
return await router.route(node);

proc tryRoute*(router: RpcRouter, data: JsonNode, fut: var Future[StringOfJson]): bool =
proc tryRoute*(router: RpcRouter, data: JsonNode, fut: var Future[RpcResult]): bool =
## Route to RPC, returns false if the method or params cannot be found.
## Expects json input and returns json output.
let
Expand All @@ -116,16 +127,6 @@ proc tryRoute*(router: RpcRouter, data: JsonNode, fut: var Future[StringOfJson])
fut = rpc(jParams)
return true

proc makeProcName(s: string): string =
result = ""
for c in s:
if c.isAlphaNumeric: result.add c

proc hasReturnType(params: NimNode): bool =
if params != nil and params.len > 0 and params[0] != nil and
params[0].kind != nnkEmpty:
result = true

macro rpc*(server: RpcRouter, path: string, body: untyped): untyped =
## Define a remote procedure call.
## Input and return parameters are defined using the ``do`` notation.
Expand Down Expand Up @@ -159,16 +160,16 @@ macro rpc*(server: RpcRouter, path: string, body: untyped): untyped =
if ReturnType == ident"JsonNode":
# `JsonNode` results don't need conversion
result.add quote do:
proc `rpcProcWrapper`(`paramsIdent`: JsonNode): Future[StringOfJson] {.async, gcsafe.} =
return StringOfJson($(await `rpcProcImpl`(`paramsIdent`)))
proc `rpcProcWrapper`(`paramsIdent`: JsonNode): Future[RpcResult] {.async, raises: [Defect, CatchableError, Exception].} =
return some(StringOfJson($(await `rpcProcImpl`(`paramsIdent`))))
elif ReturnType == ident"StringOfJson":
result.add quote do:
proc `rpcProcWrapper`(`paramsIdent`: JsonNode): Future[StringOfJson] {.async, gcsafe.} =
return await `rpcProcImpl`(`paramsIdent`)
proc `rpcProcWrapper`(`paramsIdent`: JsonNode): Future[RpcResult] {.async, raises: [Defect, CatchableError, Exception].} =
return some(await `rpcProcImpl`(`paramsIdent`))
else:
result.add quote do:
proc `rpcProcWrapper`(`paramsIdent`: JsonNode): Future[StringOfJson] {.async, gcsafe.} =
return StringOfJson($(%(await `rpcProcImpl`(`paramsIdent`))))
proc `rpcProcWrapper`(`paramsIdent`: JsonNode): Future[RpcResult] {.async, raises: [Defect, CatchableError, Exception].} =
return some(StringOfJson($(%(await `rpcProcImpl`(`paramsIdent`)))))

result.add quote do:
`server`.register(`path`, `rpcProcWrapper`)
Expand Down
8 changes: 4 additions & 4 deletions json_rpc/rpcproxy.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,9 @@ proc getWebSocketClientConfig*(
ClientConfig(kind: WebSocket, wsUri: uri, compression: compression, flags: flags)

proc proxyCall(client: RpcClient, name: string): RpcProc =
return proc (params: JsonNode): Future[StringOfJson] {.async.} =
let res = await client.call(name, params)
return StringOfJson($res)
return proc (params: JsonNode): Future[RpcResult] {.async, gcsafe, raises: [Defect, CatchableError, Exception].} =
let res = await client.call(name, params)
return some(StringOfJson($res))

proc getClient(proxy: RpcProxy): RpcClient =
case proxy.kind
Expand Down Expand Up @@ -91,7 +91,7 @@ proc start*(proxy: RpcProxy) {.async.} =
template rpc*(server: RpcProxy, path: string, body: untyped): untyped =
server.rpcHttpServer.rpc(path, body)

proc registerProxyMethod*(proxy: var RpcProxy, methodName: string) =
proc registerProxyMethod*(proxy: var RpcProxy, methodName: string) {.gcsafe, raises: [Defect, CatchableError, Exception].} =
try:
proxy.rpcHttpServer.register(methodName, proxyCall(proxy.getClient(), methodName))
except CatchableError as err:
Expand Down
4 changes: 2 additions & 2 deletions json_rpc/rpcserver.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
import server
import server, chronos
import servers/[socketserver, httpserver, websocketserver]
export server, socketserver, httpserver, websocketserver
export server, socketserver, httpserver, websocketserver, chronos
10 changes: 5 additions & 5 deletions json_rpc/server.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,10 @@
import
std/tables,
chronos,
faststreams/async_backend,
Copy link
Member

@arnetheduck arnetheduck Feb 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why would fs be needed here?

Copy link
Contributor Author

@yyoncho yyoncho Feb 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

to support the ability to compile in chronos/asyncdispatch mode(i. e. providing async)

Copy link
Member

@arnetheduck arnetheduck Feb 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

but faststreams is not used in this module? perhaps you need an export somewhere else

Copy link
Contributor Author

@yyoncho yyoncho Feb 16, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

faststream/async_backend is used only to export conditionally chronos or asyncdispatch stuff based on asyncBackend.

Copy link
Member

@arnetheduck arnetheduck Feb 17, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

right - so with asyncBackend=none (the default), nothing will be exported by async_backend - this is how we use the library generally in projects.

./router,
./jsonmarshal

export chronos, jsonmarshal, router
export jsonmarshal, router

type
RpcServer* = ref object of RootRef
Expand All @@ -23,12 +23,12 @@ template hasMethod*(server: RpcServer, methodName: string): bool =

proc executeMethod*(server: RpcServer,
methodName: string,
args: JsonNode): Future[StringOfJson] =
server.router.procs[methodName](args)
args: JsonNode): Future[StringOfJson] {.async} =
return (await server.router.procs[methodName](args)).get

# Wrapper for message processing

proc route*(server: RpcServer, line: string): Future[string] {.gcsafe.} =
proc route*(server: RpcServer, line: string): Future[RpcResult] {.gcsafe.} =
server.router.route(line)

# Server registration
Expand Down
12 changes: 8 additions & 4 deletions json_rpc/servers/httpserver.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import
stew/byteutils,
std/[strutils],
chronicles, httputils, chronos,
chronos/apps/http/[httpserver, shttpserver],
".."/[errors, server]
Expand Down Expand Up @@ -30,9 +29,14 @@ proc processClientRpc(rpcServer: RpcServer): HttpProcessCallback =
return await request.respond(Http503, "Internal error while processing JSON-RPC call")
else:
var data = future.read()
let res = await request.respond(Http200, data)
trace "JSON-RPC result has been sent"
return res
if data.isSome:
let res = await request.respond(Http200, string(data.get))
trace "JSON-RPC result has been sent"
return res
else:
let res = await request.respond(Http204)
trace "No-content has been sent"
return res
else:
return dumbResponse()

Expand Down
6 changes: 4 additions & 2 deletions json_rpc/servers/socketserver.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import
chronicles,
json_serialization/std/net,
".."/[errors, server]
".."/[errors, server],
chronos

export errors, server

Expand All @@ -28,7 +29,8 @@ proc processClient(server: StreamServer, transport: StreamTransport) {.async, gc
debug "Processing message", address = transport.remoteAddress(), line = value

let res = await rpc.route(value)
discard await transport.write(res)
if res.isSome:
discard await transport.write(string(res.get))

# Utility functions for setting up servers using stream transport addresses

Expand Down
7 changes: 4 additions & 3 deletions json_rpc/servers/websocketserver.nim
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
import
chronicles, httputils, chronos, websock/websock,
chronicles, chronos, websock/websock,
websock/extensions/compression/deflate,
stew/byteutils, json_serialization/std/net,
".."/[errors, server]
".."/server

export server, net

Expand Down Expand Up @@ -53,7 +53,8 @@ proc handleRequest(rpc: RpcWebSocketServer, request: HttpRequest) {.async.} =
var data = future.read()
trace "RPC result has been sent", address = $request.uri

await ws.send(data)
if data.isSome:
await ws.send(string(data.get))

except WebSocketError as exc:
error "WebSocket error:", exception = exc.msg
Expand Down
Loading