2018-06-14 16:52:41 +01:00
|
|
|
import tables, json, macros
|
|
|
|
import asyncdispatch2
|
2018-06-22 19:05:32 +01:00
|
|
|
from strutils import toLowerAscii
|
2018-06-14 16:52:41 +01:00
|
|
|
import jsonmarshal
|
2018-06-22 19:05:32 +01:00
|
|
|
export asyncdispatch2
|
2018-06-14 16:52:41 +01:00
|
|
|
|
|
|
|
type
|
2018-06-22 19:05:32 +01:00
|
|
|
RpcClient*[T, A] = ref object
|
2018-06-14 16:52:41 +01:00
|
|
|
awaiting: Table[string, Future[Response]]
|
2018-07-06 17:47:43 +01:00
|
|
|
transport: T
|
2018-06-25 17:54:28 +01:00
|
|
|
address: A
|
2018-06-14 16:52:41 +01:00
|
|
|
nextId: int64
|
2018-06-22 19:05:32 +01:00
|
|
|
|
2018-06-14 16:52:41 +01:00
|
|
|
Response* = tuple[error: bool, result: JsonNode]
|
|
|
|
|
2018-06-22 19:05:32 +01:00
|
|
|
const defaultMaxRequestLength* = 1024 * 128
|
2018-06-14 16:52:41 +01:00
|
|
|
|
2018-07-06 17:47:43 +01:00
|
|
|
proc newRpcClient*[T, A]: RpcClient[T, A] =
|
2018-06-14 16:52:41 +01:00
|
|
|
## Creates a new ``RpcClient`` instance.
|
2018-06-22 19:05:32 +01:00
|
|
|
result = RpcClient[T, A](awaiting: initTable[string, Future[Response]](), nextId: 1)
|
|
|
|
|
2018-07-06 17:47:43 +01:00
|
|
|
proc call*(self: RpcClient, name: string,
|
|
|
|
params: JsonNode): Future[Response] {.async.} =
|
|
|
|
## Remotely calls the specified RPC method.
|
|
|
|
let id = $self.nextId
|
|
|
|
self.nextId.inc
|
|
|
|
var
|
|
|
|
value =
|
|
|
|
$ %{"jsonrpc": %"2.0",
|
|
|
|
"method": %name,
|
|
|
|
"params": params,
|
|
|
|
"id": %id} & "\c\l"
|
2018-07-12 14:06:35 +01:00
|
|
|
if self.transport.isNil:
|
|
|
|
var connectStr = ""
|
|
|
|
raise newException(ValueError, "Transport is not initialised (missing a call to connect?)")
|
2018-07-06 17:47:43 +01:00
|
|
|
let res = await self.transport.write(value)
|
|
|
|
# TODO: Add actions when not full packet was send, e.g. disconnect peer.
|
|
|
|
assert(res == len(value))
|
|
|
|
|
|
|
|
# completed by processMessage.
|
|
|
|
var newFut = newFuture[Response]()
|
|
|
|
# add to awaiting responses
|
|
|
|
self.awaiting[id] = newFut
|
|
|
|
result = await newFut
|
2018-06-14 16:52:41 +01:00
|
|
|
|
2018-06-25 17:54:28 +01:00
|
|
|
template asyncRaise[T](fut: Future[T], errType: typedesc, msg: string) =
|
|
|
|
fut.fail(newException(errType, msg))
|
2018-06-14 16:52:41 +01:00
|
|
|
|
|
|
|
macro checkGet(node: JsonNode, fieldName: string,
|
|
|
|
jKind: static[JsonNodeKind]): untyped =
|
|
|
|
let n = genSym(ident = "n") #`node`{`fieldName`}
|
|
|
|
result = quote:
|
|
|
|
let `n` = `node`{`fieldname`}
|
|
|
|
if `n`.isNil or `n`.kind == JNull:
|
|
|
|
raise newException(ValueError,
|
|
|
|
"Message is missing required field \"" & `fieldName` & "\"")
|
|
|
|
if `n`.kind != `jKind`.JsonNodeKind:
|
|
|
|
raise newException(ValueError,
|
|
|
|
"Expected " & $(`jKind`.JsonNodeKind) & ", got " & $`n`.kind)
|
|
|
|
case jKind
|
|
|
|
of JBool: result.add(quote do: `n`.getBool)
|
|
|
|
of JInt: result.add(quote do: `n`.getInt)
|
|
|
|
of JString: result.add(quote do: `n`.getStr)
|
|
|
|
of JFloat: result.add(quote do: `n`.getFloat)
|
|
|
|
of JObject: result.add(quote do: `n`.getObject)
|
|
|
|
else: discard
|
|
|
|
|
2018-07-06 17:47:43 +01:00
|
|
|
proc processMessage(self: RpcClient, line: string) =
|
2018-06-26 15:38:49 +01:00
|
|
|
# Note: this doesn't use any transport code so doesn't need to be differentiated.
|
2018-07-06 17:47:43 +01:00
|
|
|
let
|
|
|
|
node = parseJson(line) # TODO: Check errors
|
|
|
|
id = checkGet(node, "id", JString)
|
2018-06-14 16:52:41 +01:00
|
|
|
|
|
|
|
if not self.awaiting.hasKey(id):
|
|
|
|
raise newException(ValueError,
|
|
|
|
"Cannot find message id \"" & node["id"].str & "\"")
|
|
|
|
|
|
|
|
let version = checkGet(node, "jsonrpc", JString)
|
|
|
|
if version != "2.0":
|
2018-06-25 17:54:28 +01:00
|
|
|
self.awaiting[id].asyncRaise(ValueError,
|
2018-06-14 16:52:41 +01:00
|
|
|
"Unsupported version of JSON, expected 2.0, received \"" & version & "\"")
|
|
|
|
|
|
|
|
let errorNode = node{"error"}
|
|
|
|
if errorNode.isNil or errorNode.kind == JNull:
|
|
|
|
var res = node{"result"}
|
|
|
|
if not res.isNil:
|
|
|
|
self.awaiting[id].complete((false, res))
|
|
|
|
self.awaiting.del(id)
|
|
|
|
# TODO: actions on unable find result node
|
|
|
|
else:
|
2018-06-19 18:17:12 +01:00
|
|
|
self.awaiting[id].fail(newException(ValueError, $errorNode))
|
2018-06-14 16:52:41 +01:00
|
|
|
self.awaiting.del(id)
|
|
|
|
|
2018-07-06 17:47:43 +01:00
|
|
|
proc processData(client: RpcClient) {.async.} =
|
|
|
|
while true:
|
|
|
|
var value = await client.transport.readLine(defaultMaxRequestLength)
|
|
|
|
if value == "":
|
|
|
|
# transmission ends
|
|
|
|
client.transport.close
|
|
|
|
break
|
2018-06-22 19:05:32 +01:00
|
|
|
|
2018-07-06 17:47:43 +01:00
|
|
|
client.processMessage(value)
|
|
|
|
# async loop reconnection and waiting
|
|
|
|
client.transport = await connect(client.address)
|
2018-06-22 19:05:32 +01:00
|
|
|
|
2018-07-12 14:06:35 +01:00
|
|
|
type RpcSocketClient* = RpcClient[StreamTransport, TransportAddress]
|
2018-06-22 19:05:32 +01:00
|
|
|
|
2018-07-12 14:06:35 +01:00
|
|
|
proc connect*(client: RpcSocketClient, address: string, port: Port) {.async.} =
|
2018-07-06 17:47:43 +01:00
|
|
|
let addresses = resolveTAddress(address, port)
|
|
|
|
client.transport = await connect(addresses[0])
|
|
|
|
client.address = addresses[0]
|
|
|
|
asyncCheck processData(client)
|
|
|
|
|
2018-07-12 14:06:35 +01:00
|
|
|
proc newRpcSocketClient*(): RpcSocketClient =
|
2018-06-22 19:05:32 +01:00
|
|
|
## Create new server and assign it to addresses ``addresses``.
|
|
|
|
result = newRpcClient[StreamTransport, TransportAddress]()
|
|
|
|
|
2018-06-25 17:54:28 +01:00
|
|
|
# Signature processing
|
2018-06-14 16:52:41 +01:00
|
|
|
|
|
|
|
proc createRpcProc(procName, parameters, callBody: NimNode): NimNode =
|
|
|
|
# parameters come as a tree
|
|
|
|
var paramList = newSeq[NimNode]()
|
|
|
|
for p in parameters: paramList.add(p)
|
|
|
|
|
|
|
|
# build proc
|
|
|
|
result = newProc(procName, paramList, callBody)
|
|
|
|
# make proc async
|
|
|
|
result.addPragma ident"async"
|
|
|
|
# export this proc
|
|
|
|
result[0] = nnkPostFix.newTree(ident"*", newIdentNode($procName))
|
|
|
|
|
|
|
|
proc toJsonArray(parameters: NimNode): NimNode =
|
|
|
|
# outputs an array of jsonified parameters
|
|
|
|
# ie; %[%a, %b, %c]
|
|
|
|
parameters.expectKind nnkFormalParams
|
|
|
|
var items = newNimNode(nnkBracket)
|
|
|
|
for i in 2 ..< parameters.len:
|
|
|
|
let curParam = parameters[i][0]
|
|
|
|
if curParam.kind != nnkEmpty:
|
|
|
|
items.add(nnkPrefix.newTree(ident"%", curParam))
|
|
|
|
result = nnkPrefix.newTree(bindSym("%", brForceOpen), items)
|
|
|
|
|
|
|
|
proc createRpcFromSig*(rpcDecl: NimNode): NimNode =
|
|
|
|
# Each input parameter in the rpc signature is converted
|
|
|
|
# to json with `%`.
|
|
|
|
# Return types are then converted back to native Nim types.
|
|
|
|
let iJsonNode = newIdentNode("JsonNode")
|
|
|
|
|
|
|
|
var parameters = rpcDecl.findChild(it.kind == nnkFormalParams).copy
|
|
|
|
# ensure we have at least space for a return parameter
|
|
|
|
if parameters.isNil or parameters.kind == nnkEmpty or parameters.len == 0:
|
|
|
|
parameters = nnkFormalParams.newTree(iJsonNode)
|
|
|
|
|
|
|
|
let
|
|
|
|
procName = rpcDecl.name
|
|
|
|
pathStr = $procName
|
|
|
|
returnType =
|
|
|
|
# if no return type specified, defaults to JsonNode
|
|
|
|
if parameters[0].kind == nnkEmpty: iJsonNode
|
|
|
|
else: parameters[0]
|
|
|
|
customReturnType = returnType != iJsonNode
|
|
|
|
|
|
|
|
# insert rpc client as first parameter
|
|
|
|
parameters.insert(1, nnkIdentDefs.newTree(ident"client", ident"RpcClient",
|
|
|
|
newEmptyNode()))
|
|
|
|
|
|
|
|
let
|
|
|
|
# variable used to send json to the server
|
|
|
|
jsonParamIdent = genSym(nskVar, "jsonParam")
|
|
|
|
# json array of marshalled parameters
|
|
|
|
jsonParamArray = parameters.toJsonArray()
|
|
|
|
var
|
|
|
|
# populate json params - even rpcs with no parameters have an empty json
|
|
|
|
# array node sent
|
|
|
|
callBody = newStmtList().add(quote do:
|
|
|
|
var `jsonParamIdent` = `jsonParamArray`
|
|
|
|
)
|
|
|
|
|
|
|
|
# convert return type to Future
|
|
|
|
parameters[0] = nnkBracketExpr.newTree(ident"Future", returnType)
|
|
|
|
# create rpc proc
|
|
|
|
result = createRpcProc(procName, parameters, callBody)
|
|
|
|
|
|
|
|
let
|
|
|
|
# temporary variable to hold `Response` from rpc call
|
|
|
|
rpcResult = genSym(nskLet, "res")
|
|
|
|
clientIdent = newIdentNode("client")
|
|
|
|
# proc return variable
|
|
|
|
procRes = ident"result"
|
|
|
|
# actual return value, `rpcResult`.result
|
|
|
|
jsonRpcResult = nnkDotExpr.newTree(rpcResult, newIdentNode("result"))
|
|
|
|
|
|
|
|
# perform rpc call
|
|
|
|
callBody.add(quote do:
|
|
|
|
# `rpcResult` is of type `Response`
|
|
|
|
let `rpcResult` = await `clientIdent`.call(`pathStr`, `jsonParamIdent`)
|
|
|
|
if `rpcResult`.error: raise newException(ValueError, $`rpcResult`.result)
|
|
|
|
)
|
|
|
|
|
|
|
|
if customReturnType:
|
|
|
|
# marshal json to native Nim type
|
|
|
|
callBody.add(jsonToNim(procRes, returnType, jsonRpcResult, "result"))
|
|
|
|
else:
|
|
|
|
# native json expected so no work
|
|
|
|
callBody.add(quote do:
|
|
|
|
`procRes` = `rpcResult`.result
|
|
|
|
)
|
|
|
|
when defined(nimDumpRpcs):
|
|
|
|
echo pathStr, ":\n", result.repr
|
|
|
|
|
|
|
|
proc processRpcSigs(parsedCode: NimNode): NimNode =
|
|
|
|
result = newStmtList()
|
|
|
|
|
|
|
|
for line in parsedCode:
|
|
|
|
if line.kind == nnkProcDef:
|
|
|
|
var procDef = createRpcFromSig(line)
|
|
|
|
result.add(procDef)
|
|
|
|
|
|
|
|
macro createRpcSigs*(filePath: static[string]): untyped =
|
|
|
|
## Takes a file of forward declarations in Nim and builds them into RPC
|
|
|
|
## calls, based on their parameters.
|
|
|
|
## Inputs are marshalled to json, and results are put into the signature's
|
|
|
|
## Nim type.
|
|
|
|
result = processRpcSigs(staticRead($filePath).parseStmt())
|