Implement RPC batch call both in servers and clients

json_rpc/client.nim

@@ -24,15 +24,30 @@ export
   tables,
   jsonmarshal,
   RequestParamsTx,
+  RequestBatchTx,
+  ResponseBatchRx,
   results
 
 type
+  RpcBatchItem* = object
+    meth*: string
+    params*: RequestParamsTx
+
+  RpcBatchCallRef* = ref object of RootRef
+    client*: RpcClient
+    batch*: seq[RpcBatchItem]
+
+  RpcBatchResponse* = object
+    error*: Opt[string]
+    result*: JsonString
+
   RpcClient* = ref object of RootRef
     awaiting*: Table[RequestId, Future[JsonString]]
     lastId: int
     onDisconnect*: proc() {.gcsafe, raises: [].}
     onProcessMessage*: proc(client: RpcClient, line: string):
       Result[bool, string] {.gcsafe, raises: [].}
+    batchFut*: Future[ResponseBatchRx]
 
   GetJsonRpcRequestHeaders* = proc(): seq[(string, string)] {.gcsafe, raises: [].}
 
@@ -42,10 +57,65 @@ type
 # Public helpers
 # ------------------------------------------------------------------------------
 
+func validateResponse(resIndex: int, res: ResponseRx): Result[void, string] =
+  if res.jsonrpc.isNone:
+    return err("missing or invalid `jsonrpc` in response " & $resIndex)
+
+  if res.id.isNone:
+    if res.error.isSome:
+      let error = JrpcSys.encode(res.error.get)
+      return err(error)
+    else:
+      return err("missing or invalid response id in response " & $resIndex)
+
+  if res.error.isSome:
+    let error = JrpcSys.encode(res.error.get)
+    return err(error)
+
+  # Up to this point, the result should contains something
+  if res.result.string.len == 0:
+    return err("missing or invalid response result in response " & $resIndex)
+
+  ok()
+
+proc processResponse(resIndex: int,
+                     map: var Table[RequestId, int],
+                     responses: var seq[RpcBatchResponse],
+                     response: ResponseRx): Result[void, string] =
+  let r = validateResponse(resIndex, response)
+  if r.isErr:
+    if response.id.isSome:
+      let id = response.id.get
+      var index: int
+      if not map.pop(id, index):
+        return err("cannot find message id: " & $id & " in response " & $resIndex)
+      responses[index] = RpcBatchResponse(
+        error: Opt.some(r.error)
+      )
+    else:
+      return err(r.error)
+  else:
+    let id = response.id.get
+    var index: int
+    if not map.pop(id, index):
+      return err("cannot find message id: " & $id & " in response " & $resIndex)
+    responses[index] = RpcBatchResponse(
+      result:  response.result
+    )
+
+  ok()
+
+# ------------------------------------------------------------------------------
+# Public helpers
+# ------------------------------------------------------------------------------
+
 func requestTxEncode*(name: string, params: RequestParamsTx, id: RequestId): string =
   let req = requestTx(name, params, id)
   JrpcSys.encode(req)
 
+func requestBatchEncode*(calls: RequestBatchTx): string =
+  JrpcSys.encode(calls)
+
 # ------------------------------------------------------------------------------
 # Public functions
 # ------------------------------------------------------------------------------
@@ -68,6 +138,11 @@ method call*(client: RpcClient, name: string,
 method close*(client: RpcClient): Future[void] {.base, gcsafe, async.} =
   doAssert(false, "`RpcClient.close` not implemented")
 
+method callBatch*(client: RpcClient,
+                  calls: RequestBatchTx): Future[ResponseBatchRx]
+                    {.base, gcsafe, async.} =
+  doAssert(false, "`RpcClient.callBatch` not implemented")
+
 proc processMessage*(client: RpcClient, line: string): Result[void, string] =
   if client.onProcessMessage.isNil.not:
     let fallBack = client.onProcessMessage(client, line).valueOr:
@@ -78,8 +153,14 @@ proc processMessage*(client: RpcClient, line: string): Result[void, string] =
   # Note: this doesn't use any transport code so doesn't need to be
   # differentiated.
   try:
-    let response = JrpcSys.decode(line, ResponseRx)
+    let batch = JrpcSys.decode(line, ResponseBatchRx)
+    if batch.kind == rbkMany:
+      if client.batchFut.isNil or client.batchFut.finished():
+        client.batchFut = newFuture[ResponseBatchRx]()
+      client.batchFut.complete(batch)
+      return ok()
 
+    let response = batch.single
     if response.jsonrpc.isNone:
       return err("missing or invalid `jsonrpc`")
 
@@ -114,6 +195,41 @@ proc processMessage*(client: RpcClient, line: string): Result[void, string] =
   except CatchableError as exc:
     return err(exc.msg)
 
+proc prepareBatch*(client: RpcClient): RpcBatchCallRef =
+  RpcBatchCallRef(client: client)
+
+proc send*(batch: RpcBatchCallRef):
+            Future[Result[seq[RpcBatchResponse], string]] {.
+              async: (raises: []).} =
+  var
+    calls = RequestBatchTx(
+      kind: rbkMany,
+      many: newSeqOfCap[RequestTx](batch.batch.len),
+    )
+    responses = newSeq[RpcBatchResponse](batch.batch.len)
+    map = initTable[RequestId, int]()
+
+  for item in batch.batch:
+    let id = batch.client.getNextId()
+    map[id] = calls.many.len
+    calls.many.add requestTx(item.meth, item.params, id)
+
+  try:
+    let res = await batch.client.callBatch(calls)
+    if res.kind == rbkSingle:
+      let r = processResponse(0, map, responses, res.single)
+      if r.isErr:
+        return err(r.error)
+    else:
+      for i, z in res.many:
+        let r = processResponse(i, map, responses, z)
+        if r.isErr:
+          return err(r.error)
+  except CatchableError as exc:
+    return err(exc.msg)
+
+  return ok(responses)
+
 # ------------------------------------------------------------------------------
 # Signature processing
 # ------------------------------------------------------------------------------

json_rpc/clients/httpclient.nim

@@ -1,5 +1,5 @@
 # json-rpc
-# Copyright (c) 2019-2023 Status Research & Development GmbH
+# Copyright (c) 2019-2024 Status Research & Development GmbH
 # Licensed under either of
 #  * Apache License, version 2.0, ([LICENSE-APACHE](LICENSE-APACHE))
 #  * MIT license ([LICENSE-MIT](LICENSE-MIT))
@@ -38,6 +38,10 @@ const
 
 {.push gcsafe, raises: [].}
 
+# ------------------------------------------------------------------------------
+# Private helpers
+# ------------------------------------------------------------------------------
+
 proc new(
     T: type RpcHttpClient, maxBodySize = MaxHttpRequestSize, secure = false,
     getHeaders: GetJsonRpcRequestHeaders = nil, flags: HttpClientFlags = {}): T =
@@ -53,15 +57,24 @@ proc new(
     getHeaders: getHeaders
   )
 
-proc newRpcHttpClient*(
-    maxBodySize = MaxHttpRequestSize, secure = false,
-    getHeaders: GetJsonRpcRequestHeaders = nil,
-    flags: HttpClientFlags = {}): RpcHttpClient =
-  RpcHttpClient.new(maxBodySize, secure, getHeaders, flags)
+template closeRefs(req, res: untyped) =
+  # We can't trust try/finally in async/await in all nim versions, so we
+  # do it manually instead
+  if req != nil:
+    try:
+      await req.closeWait()
+    except CatchableError as exc: # shouldn't happen
+      debug "Error closing JSON-RPC HTTP resuest/response", err = exc.msg
+      discard exc
 
-method call*(client: RpcHttpClient, name: string,
-             params: RequestParamsTx): Future[JsonString]
-            {.async, gcsafe.} =
+  if res != nil:
+    try:
+      await res.closeWait()
+    except CatchableError as exc: # shouldn't happen
+      debug "Error closing JSON-RPC HTTP resuest/response", err = exc.msg
+      discard exc
+
+proc callImpl(client: RpcHttpClient, reqBody: string): Future[string] {.async.} =
   doAssert client.httpSession != nil
   if client.httpAddress.isErr:
     raise newException(RpcAddressUnresolvableError, client.httpAddress.error)
@@ -73,33 +86,9 @@ method call*(client: RpcHttpClient, name: string,
       @[]
   headers.add(("Content-Type", "application/json"))
 
-  let
-    id = client.getNextId()
-    reqBody = requestTxEncode(name, params, id)
-
   var req: HttpClientRequestRef
   var res: HttpClientResponseRef
 
-  template used(x: typed) =
-    # silence unused warning
-    discard
-
-  template closeRefs() =
-    # We can't trust try/finally in async/await in all nim versions, so we
-    # do it manually instead
-    if req != nil:
-      try:
-        await req.closeWait()
-      except CatchableError as exc: # shouldn't happen
-        used(exc)
-        debug "Error closing JSON-RPC HTTP resuest/response", err = exc.msg
-    if res != nil:
-      try:
-        await res.closeWait()
-      except CatchableError as exc: # shouldn't happen
-        used(exc)
-        debug "Error closing JSON-RPC HTTP resuest/response", err = exc.msg
-
   debug "Sending message to RPC server",
          address = client.httpAddress, msg_len = len(reqBody), name
   trace "Message", msg = reqBody
@@ -113,33 +102,52 @@ method call*(client: RpcHttpClient, name: string,
       await req.send()
     except CancelledError as e:
       debug "Cancelled POST Request with JSON-RPC", e = e.msg
-      closeRefs()
+      closeRefs(req, res)
       raise e
     except CatchableError as e:
       debug "Failed to send POST Request with JSON-RPC", e = e.msg
-      closeRefs()
+      closeRefs(req, res)
       raise (ref RpcPostError)(msg: "Failed to send POST Request with JSON-RPC: " & e.msg, parent: e)
 
   if res.status < 200 or res.status >= 300: # res.status is not 2xx (success)
     debug "Unsuccessful POST Request with JSON-RPC",
       status = res.status, reason = res.reason
-    closeRefs()
+    closeRefs(req, res)
     raise (ref ErrorResponse)(status: res.status, msg: res.reason)
 
   let resBytes =
     try:
       await res.getBodyBytes(client.maxBodySize)
     except CancelledError as e:
       debug "Cancelled POST Response for JSON-RPC", e = e.msg
-      closeRefs()
+      closeRefs(req, res)
       raise e
     except CatchableError as e:
       debug "Failed to read POST Response for JSON-RPC", e = e.msg
-      closeRefs()
+      closeRefs(req, res)
       raise (ref FailedHttpResponse)(msg: "Failed to read POST Response for JSON-RPC: " & e.msg, parent: e)
 
-  let resText = string.fromBytes(resBytes)
-  trace "Response", text = resText
+  result = string.fromBytes(resBytes)
+  trace "Response", text = result
+  closeRefs(req, res)
+
+# ------------------------------------------------------------------------------
+# Public functions
+# ------------------------------------------------------------------------------
+
+proc newRpcHttpClient*(
+    maxBodySize = MaxHttpRequestSize, secure = false,
+    getHeaders: GetJsonRpcRequestHeaders = nil,
+    flags: HttpClientFlags = {}): RpcHttpClient =
+  RpcHttpClient.new(maxBodySize, secure, getHeaders, flags)
+
+method call*(client: RpcHttpClient, name: string,
+             params: RequestParamsTx): Future[JsonString]
+            {.async, gcsafe.} =
+  let
+    id = client.getNextId()
+    reqBody = requestTxEncode(name, params, id)
+    resText = await client.callImpl(reqBody)
 
   # completed by processMessage - the flow is quite weird here to accomodate
   # socket and ws clients, but could use a more thorough refactoring
@@ -155,13 +163,10 @@ method call*(client: RpcHttpClient, name: string,
     let exc = newException(JsonRpcError, msgRes.error)
     newFut.fail(exc)
     client.awaiting.del(id)
-    closeRefs()
     raise exc
 
   client.awaiting.del(id)
 
-  closeRefs()
-
   # processMessage should have completed this future - if it didn't, `read` will
   # raise, which is reasonable
   if newFut.finished:
@@ -171,6 +176,34 @@ method call*(client: RpcHttpClient, name: string,
     debug "Invalid POST Response for JSON-RPC"
     raise newException(InvalidResponse, "Invalid response")
 
+method callBatch*(client: RpcHttpClient,
+                  calls: RequestBatchTx): Future[ResponseBatchRx]
+                    {.gcsafe, async.} =
+  let
+    reqBody = requestBatchEncode(calls)
+    resText = await client.callImpl(reqBody)
+
+  if client.batchFut.isNil or client.batchFut.finished():
+    client.batchFut = newFuture[ResponseBatchRx]()
+
+  # Might error for all kinds of reasons
+  let msgRes = client.processMessage(resText)
+  if msgRes.isErr:
+    # Need to clean up in case the answer was invalid
+    debug "Failed to process POST Response for JSON-RPC", msg = msgRes.error
+    let exc = newException(JsonRpcError, msgRes.error)
+    client.batchFut.fail(exc)
+    raise exc
+
+  # processMessage should have completed this future - if it didn't, `read` will
+  # raise, which is reasonable
+  if client.batchFut.finished:
+    return client.batchFut.read()
+  else:
+    # TODO: Provide more clarity regarding the failure here
+    debug "Invalid POST Response for JSON-RPC"
+    raise newException(InvalidResponse, "Invalid response")
+
 proc connect*(client: RpcHttpClient, url: string) {.async.} =
   client.httpAddress = client.httpSession.getAddress(url)
   if client.httpAddress.isErr: