mirror of
https://github.com/logos-storage/nim-json-rpc.git
synced 2026-01-07 08:03:07 +00:00
4eb39203eb
* fixes * fix https://github.com/status-im/nimbus-eth2/issues/1650 * only one of `result` and `error` allowed in response * fix invalid `string` instances being created from byte sequences * fix large int64 parsing on 32-bit * fix exception inheritance * fix some dangling results * some cleanups * annotate exception issues, fix cancellation * more error handling cleanup * add rudimentary error tests * cleanups * simplify init * use nextId -> lastId to avoid =1 init * remove obsolete tests
64 lines
1.8 KiB
Nim
64 lines
1.8 KiB
Nim
import
|
|
std/[json, tables],
|
|
../client, chronos
|
|
|
|
type
|
|
RpcSocketClient* = ref object of RpcClient
|
|
transport*: StreamTransport
|
|
address*: TransportAddress
|
|
loop*: Future[void]
|
|
|
|
const defaultMaxRequestLength* = 1024 * 128
|
|
|
|
proc new*(T: type RpcSocketClient): T =
|
|
T()
|
|
|
|
proc newRpcSocketClient*: RpcSocketClient =
|
|
## Creates a new client instance.
|
|
RpcSocketClient.new()
|
|
|
|
method call*(self: RpcSocketClient, name: string,
|
|
params: JsonNode): Future[Response] {.async.} =
|
|
## Remotely calls the specified RPC method.
|
|
let id = self.getNextId()
|
|
var value = $rpcCallNode(name, params, id) & "\r\n"
|
|
if self.transport.isNil:
|
|
raise newException(ValueError,
|
|
"Transport is not initialised (missing a call to connect?)")
|
|
|
|
# completed by processMessage.
|
|
var newFut = newFuture[Response]()
|
|
# add to awaiting responses
|
|
self.awaiting[id] = newFut
|
|
|
|
let res = await self.transport.write(value)
|
|
# TODO: Add actions when not full packet was send, e.g. disconnect peer.
|
|
doAssert(res == len(value))
|
|
|
|
return await newFut
|
|
|
|
proc processData(client: RpcSocketClient) {.async.} =
|
|
while true:
|
|
while true:
|
|
var value = await client.transport.readLine(defaultMaxRequestLength)
|
|
if value == "":
|
|
# transmission ends
|
|
await client.transport.closeWait()
|
|
break
|
|
|
|
# TODO handle exceptions
|
|
client.processMessage(value)
|
|
|
|
# async loop reconnection and waiting
|
|
client.transport = await connect(client.address)
|
|
|
|
proc connect*(client: RpcSocketClient, address: string, port: Port) {.async.} =
|
|
let addresses = resolveTAddress(address, port)
|
|
client.transport = await connect(addresses[0])
|
|
client.address = addresses[0]
|
|
client.loop = processData(client)
|
|
|
|
method close*(client: RpcSocketClient) {.async.} =
|
|
# TODO: Stop the processData loop
|
|
await client.transport.closeWait()
|