@@ -316,6 +316,22 @@ class GatewaySession(
316316return RpcResult(ok = res.ok, payloadJson = res.payloadJson, error = res.error)
317317 }
318318319+suspend fun sendRequestFrame(
320+method: String,
321+paramsJson: String?,
322+timeoutMs: Long = 15_000,
323+onError: (ErrorShape) -> Unit = {},
324+ ) {
325+val conn = currentConnection ?: throw IllegalStateException("not connected")
326+val params =
327+if (paramsJson.isNullOrBlank()) {
328+null
329+ } else {
330+ json.parseToJsonElement(paramsJson)
331+ }
332+ conn.sendRequestFrame(method = method, params = params, timeoutMs = timeoutMs, onError = onError)
333+ }
334+319335private data class RpcResponse(
320336val id: String,
321337val ok: Boolean,
@@ -360,14 +376,12 @@ class GatewaySession(
360376val id = UUID.randomUUID().toString()
361377val deferred = CompletableDeferred<RpcResponse>()
362378 pending[id] = deferred
363-val frame =
364- buildJsonObject {
365- put("type", JsonPrimitive("req"))
366- put("id", JsonPrimitive(id))
367- put("method", JsonPrimitive(method))
368-if (params != null) put("params", params)
369- }
370- sendJson(frame)
379+try {
380+ sendJson(buildRequestFrame(id = id, method = method, params = params))
381+ } catch (err: Throwable) {
382+ pending.remove(id)
383+throw err
384+ }
371385return try {
372386 withTimeout(timeoutMs) { deferred.await() }
373387 } catch (err: TimeoutCancellationException) {
@@ -376,13 +390,57 @@ class GatewaySession(
376390 }
377391 }
378392393+suspend fun sendRequestFrame(
394+method: String,
395+params: JsonElement?,
396+timeoutMs: Long,
397+onError: (ErrorShape) -> Unit,
398+ ) {
399+val id = UUID.randomUUID().toString()
400+val deferred = CompletableDeferred<RpcResponse>()
401+ pending[id] = deferred
402+try {
403+ sendJson(buildRequestFrame(id = id, method = method, params = params))
404+ } catch (err: Throwable) {
405+ pending.remove(id)
406+throw err
407+ }
408+ scope.launch(Dispatchers.IO) {
409+val response =
410+try {
411+ withTimeout(timeoutMs) { deferred.await() }
412+ } catch (_: TimeoutCancellationException) {
413+ pending.remove(id)
414+ onError(ErrorShape("UNAVAILABLE", "request timeout"))
415+return@launch
416+ }
417+if (!response.ok) {
418+ onError(response.error ?: ErrorShape("UNAVAILABLE", "request failed"))
419+ }
420+ }
421+ }
422+379423suspend fun sendJson(obj: JsonObject) {
380424val jsonString = obj.toString()
381425 writeLock.withLock {
382- socket?.send(jsonString)
426+if (socket?.send(jsonString) != true) {
427+throw IllegalStateException("gateway send failed")
428+ }
383429 }
384430 }
385431432+private fun buildRequestFrame(
433+id: String,
434+method: String,
435+params: JsonElement?,
436+ ): JsonObject =
437+ buildJsonObject {
438+ put("type", JsonPrimitive("req"))
439+ put("id", JsonPrimitive(id))
440+ put("method", JsonPrimitive(method))
441+if (params != null) put("params", params)
442+ }
443+386444suspend fun awaitClose() = closedDeferred.await()
387445388446fun closeQuietly() {
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。