|
| 1 | +## This module provides a lightweight, thread‑safe JSON‑RPC channel that can be |
| 2 | +## used to connect a client and a server running in different threads, reusing |
| 3 | +## existing JSON-RPC infrastructure already present in the application. |
| 4 | + |
| 5 | +{.push raises: [], gcsafe.} |
| 6 | + |
| 7 | +import ./[client, errors, router, server], asyncchannels, ./private/jrpc_sys |
| 8 | +export client, errors, server |
| 9 | + |
| 10 | +type |
| 11 | + RpcChannel* = object |
| 12 | + ## An RPC channel represents a thread‑safe, bidirectional communications |
| 13 | + ## channel from which a single "server" and a single "client" can be formed. |
| 14 | + ## |
| 15 | + ## The channel can be allocated in any thread while the server and client |
| 16 | + ## instances should be created in the thread where they will be used, |
| 17 | + ## passing to them the `RpcChannelPtrs` instance returned from `open`. |
| 18 | + recv, send: AsyncChannel[seq[byte]] |
| 19 | + |
| 20 | + RpcChannelPtrs* = object ## Raw pointer pair that can be moved to another thread. |
| 21 | + recv, send: ptr AsyncChannel[seq[byte]] |
| 22 | + # The `recv` pointer is the channel that receives data, the `send` pointer |
| 23 | + # is the channel that sends data. The two pointers are swapped when |
| 24 | + # the channel is handed to the opposite side. |
| 25 | + |
| 26 | + RpcChannelClient* = ref object of RpcConnection |
| 27 | + channel: RpcChannelPtrs |
| 28 | + loop: Future[void] |
| 29 | + |
| 30 | + RpcChannelServer* = ref object of RpcServer |
| 31 | + client: RpcChannelClient |
| 32 | + |
| 33 | +proc open*(c: var RpcChannel): Result[RpcChannelPtrs, string] = |
| 34 | + ## Open the channel, returning a channel pair that can be passed to the |
| 35 | + ## server and client threads respectively. |
| 36 | + ## |
| 37 | + ## Only one server and client instance each may use the returned channel |
| 38 | + ## pairs. The returned `RpcChannelPtrs` are raw pointers that must be |
| 39 | + ## moved to the thread that will own the client or server. |
| 40 | + ?c.recv.open() |
| 41 | + |
| 42 | + c.send.open().isOkOr: |
| 43 | + c.recv.close() |
| 44 | + return err(error) |
| 45 | + |
| 46 | + ok (RpcChannelPtrs(recv: addr c.recv, send: addr c.send)) |
| 47 | + |
| 48 | +proc close*(c: var RpcChannel) = |
| 49 | + c.recv.close() |
| 50 | + c.recv.reset() |
| 51 | + c.send.close() |
| 52 | + c.send.reset() |
| 53 | + |
| 54 | +proc new*( |
| 55 | + T: type RpcChannelClient, channel: RpcChannelPtrs, router = default(ref RpcRouter) |
| 56 | +): T = |
| 57 | + ## Create a new `RpcChannelClient` that will use the supplied `channel`. |
| 58 | + ## If a `router` is supplied, it will be used to route incoming requests. |
| 59 | + ## The returned client is ready to be connected with `connect`. |
| 60 | + let router = |
| 61 | + if router != nil: |
| 62 | + proc( |
| 63 | + request: RequestBatchRx |
| 64 | + ): Future[seq[byte]] {.async: (raises: [], raw: true).} = |
| 65 | + router[].route(request) |
| 66 | + else: |
| 67 | + nil |
| 68 | + |
| 69 | + T(channel: channel, router: router, remote: "client") |
| 70 | + |
| 71 | +proc newRpcChannelClient*( |
| 72 | + channel: RpcChannelPtrs, router = default(ref RpcRouter) |
| 73 | +): RpcChannelClient = |
| 74 | + ## Convenience wrapper that creates a new `RpcChannelClient` from a |
| 75 | + ## `RpcChannelPtrs` pair. The client can be used immediately or after |
| 76 | + ## calling `connect`. |
| 77 | + RpcChannelClient.new(channel, router) |
| 78 | + |
| 79 | +method send*( |
| 80 | + client: RpcChannelClient, reqData: seq[byte] |
| 81 | +) {.async: (raises: [CancelledError, JsonRpcError]).} = |
| 82 | + ## Send a raw JSON‑RPC request to the remote side. |
| 83 | + ## The data is written synchronously to the underlying channel. |
| 84 | + client.channel.send[].sendSync(reqData) |
| 85 | + |
| 86 | +method request*( |
| 87 | + client: RpcChannelClient, reqData: seq[byte], id: int |
| 88 | +): Future[ResponseBatchRx] {.async: (raises: [CancelledError, JsonRpcError]).} = |
| 89 | + ## Send a request and wait for the corresponding response. |
| 90 | + ## The request is sent synchronously and the future returned by |
| 91 | + ## `client.processMessage` is awaited. |
| 92 | + client.withPendingFut(fut, id): |
| 93 | + client.channel.send[].sendSync(reqData) |
| 94 | + await fut |
| 95 | + |
| 96 | +proc processData(client: RpcChannelClient) {.async: (raises: []).} = |
| 97 | + ## Internal loop that receives data from the channel, processes it |
| 98 | + ## with `client.processMessage`, and sends back any response. |
| 99 | + ## The loop terminates when the channel is closed or a |
| 100 | + ## `CancelledError` is raised. |
| 101 | + var lastError: ref JsonRpcError |
| 102 | + try: |
| 103 | + while true: |
| 104 | + let |
| 105 | + data = await client.channel.recv.recv() |
| 106 | + resp = |
| 107 | + try: |
| 108 | + await client.processMessage(data) |
| 109 | + except JsonRpcError as exc: |
| 110 | + lastError = exc |
| 111 | + break |
| 112 | + |
| 113 | + if resp.len > 0: |
| 114 | + client.channel.send[].sendSync(resp) |
| 115 | + except CancelledError: |
| 116 | + discard # shutting down |
| 117 | + |
| 118 | + if lastError == nil: |
| 119 | + lastError = (ref RpcTransportError)(msg: "Connection closed") |
| 120 | + |
| 121 | + client.clearPending(lastError) |
| 122 | + |
| 123 | + if not client.onDisconnect.isNil: |
| 124 | + client.onDisconnect() |
| 125 | + |
| 126 | +proc connect*( |
| 127 | + client: RpcChannelClient |
| 128 | +) {.async: (raises: [CancelledError, JsonRpcError]).} = |
| 129 | + ## Start the client's background processing loop. |
| 130 | + ## After calling this, the client is ready to send requests. |
| 131 | + doAssert client.loop == nil, "Must not already be connected" |
| 132 | + client.loop = client.processData() |
| 133 | + |
| 134 | +method close*(client: RpcChannelClient) {.async: (raises: []).} = |
| 135 | + ## Gracefully shut down the client. |
| 136 | + ## Cancels the background loop and waits for it to finish. |
| 137 | + if client.loop != nil: |
| 138 | + let loop = move(client.loop) |
| 139 | + await loop.cancelAndWait() |
| 140 | + |
| 141 | +proc new*(T: type RpcChannelServer, channel: RpcChannelPtrs): T = |
| 142 | + ## Create a new `RpcChannelServer` that will listen on the supplied |
| 143 | + ## `channel`. The server owns a fresh `RpcRouter` instance. |
| 144 | + let |
| 145 | + res = T(router: RpcRouter.init()) |
| 146 | + # Compared to the client, swap the channels in the server |
| 147 | + channel = RpcChannelPtrs(recv: channel.send, send: channel.recv) |
| 148 | + router = proc( |
| 149 | + request: RequestBatchRx |
| 150 | + ): Future[seq[byte]] {.async: (raises: [], raw: true).} = |
| 151 | + res[].router.route(request) |
| 152 | + |
| 153 | + client = RpcChannelClient(channel: channel, router: router, remote: "server") |
| 154 | + |
| 155 | + res.client = client |
| 156 | + res |
| 157 | + |
| 158 | +proc start*(server: RpcChannelServer) = |
| 159 | + ## Start the RPC server. |
| 160 | + ## The server's background loop is started and the client is ready to |
| 161 | + ## receive requests. |
| 162 | + |
| 163 | + # `connect` for a thread channel is actually synchronous and cannot fail so |
| 164 | + # we can ignore the future being returned |
| 165 | + discard server.client.connect() |
| 166 | + server.connections.incl server.client |
| 167 | + |
| 168 | +proc stop*(server: RpcChannelServer) = |
| 169 | + discard |
| 170 | + |
| 171 | +proc closeWait*(server: RpcChannelServer) {.async: (raises: []).} = |
| 172 | + ## Gracefully shut down the server. |
| 173 | + server.connections.excl server.client |
| 174 | + await server.client.close() |
0 commit comments