No commit message

src/channel.ts

@@ -19,7 +19,10 @@ export function createWrpChannel(socket: Socket): WrpChannel {
     async *listen() {
       const bufReader = new BufReader(socket);
       while (true) {
-        const lengthU8s = await bufReader.readFull(new Uint8Array(4));
+        let lengthU8s: Uint8Array | null = null;
+        try {
+          lengthU8s = await bufReader.readFull(new Uint8Array(4));
+        } catch { /* error when socket closed */ }
         if (!lengthU8s) break;
         const length = new DataView(lengthU8s.buffer).getUint32(0, true);
         const payload = await bufReader.readFull(new Uint8Array(length));

src/glue/index.ts

@@ -23,7 +23,10 @@ export function createGlue(): Glue {
   let closed = false;
   let wait: Deferred<void> | undefined;
   return {
-    close: () => closed = true,
+    close() {
+      closed = true;
+      wait?.resolve();
+    },
     recv: (data) => {
       if (closed) throw new Error("Glue has been closed.");
       queue.push(typeof data === "string" ? str2u8s(data) : data);

src/react/useWrpServer.ts

@@ -8,7 +8,7 @@ import {
 } from "https://deno.land/x/[email protected]/core/runtime/async/event-emitter.ts";
 import { useEffect, useRef } from "react";
 import { WrpChannel } from "../channel.ts";
-import { createWrpHost } from "../host.ts";
+import { createWrpHost, WrpRequest } from "../host.ts";
 import { Metadata } from "../metadata.ts";
 import { createWrpServer, createWrpServerImplBuilder } from "../rpc/server.ts";
 
@@ -18,10 +18,10 @@ export type MethodImpl<TState extends Record<string, any>, TReq, TRes> = [
   (
     params: {
       req: Parameters<
-        MethodImplHandler<TReq, TRes, Metadata, Metadata, Metadata>
+        MethodImplHandler<TReq, TRes, WrpRequest, Metadata, Metadata>
       >[0];
       res: Parameters<
-        MethodImplHandler<TReq, TRes, Metadata, Metadata, Metadata>
+        MethodImplHandler<TReq, TRes, WrpRequest, Metadata, Metadata>
       >[1];
       getState: GetStateFn<TState>;
       stateChanges: EventEmitter<TState>;

src/rpc/server.ts

@@ -2,17 +2,17 @@ import {
   createServerImplBuilder,
   Method,
 } from "https://deno.land/x/[email protected]/core/runtime/rpc.ts";
-import type { WrpHost } from "../host.ts";
+import type { WrpHost, WrpRequest } from "../host.ts";
 import type { Metadata } from "../metadata.ts";
 import { getMethodName, mapAsyncGenerator } from "./misc.ts";
 
 export interface CreateWrpServerConfig {
   host: WrpHost;
-  methods: AsyncGenerator<Method<Metadata, Metadata, Metadata>>;
+  methods: AsyncGenerator<Method<WrpRequest, Metadata, Metadata>>;
 }
 export async function createWrpServer(config: CreateWrpServerConfig) {
   const methods: {
-    [methodName: string]: Method<Metadata, Metadata, Metadata>;
+    [methodName: string]: Method<WrpRequest, Metadata, Metadata>;
   } = {};
   for await (const method of config.methods) {
     const methodName = getMethodName(method[0]);
@@ -37,7 +37,7 @@ export async function createWrpServer(config: CreateWrpServerConfig) {
         const { requestType, responseType } = methodDescriptor;
         const [res, header, trailer] = methodImpl(
           mapAsyncGenerator(req, requestType.deserializeBinary),
-          request.metadata,
+          request,
         );
         (async function () {
           let _header: Metadata = {};
@@ -61,7 +61,7 @@ export async function createWrpServer(config: CreateWrpServerConfig) {
 }
 
 export function createWrpServerImplBuilder() {
-  return createServerImplBuilder<Metadata, Metadata, Metadata>();
+  return createServerImplBuilder<WrpRequest, Metadata, Metadata>();
 }
 
 function doOnce<T extends (...args: any) => any>(fn: T) {