@takker/endpoint-link

https://github.com/takker99/endpoint-link

https://jsr.io/@takker/endpoint-link

抽象化する方針の場合のアイデア

endpoint

MessagePortを受け取る

単純な関数

引数と関数名をrequestして返り値をrespond

入出力にiterableを含む

MessageChannelでiterableごとに新しい通信経路を作る

iterable→messageport変換で送信

messageport→ReadableStream変換で受信

AbortSignal

endpointをrpcにするものとendpointをstreamにするものの二つを作り、rpcでstreamが必要になったら後者を使う

denoがReadableStreamのTransferableに対応したから、変換は不要そう

接続から終了まで

接続確立

receiverが応答するか確認する

WebWorker内でtop-level awaitした時の挙動のように、receiverがaddEventListenerする前にmessageを送っても届かない

receiverが準備完了して通信できることを確認してからRPCを開始する

RPC

通常

1. senderがAPI callをreceiverにpostMessageする

2. receiverが実行して、結果をsenderにpostMessageする

receiverでエラーが発生したらsenderにエラーをpostMessageする

senderはcancel callを送れる

receiverはcancelを受信したら、処理が未完了であれば中断し、cancelしたことを返す

senderが(非同期)反復可能objectを引数に渡して呼び出すとき

(非同期)反復可能objectをMessagePortに変換して、通常通りAPI callに引数として含めてpostMessageする

receiverは受け取ったMessagePortをReadableStreamに変換して、指定された関数に渡す

receiverが(非同期)反復可能objectを返すとき

(非同期)反復可能objectをMessagePortに変換し、それをAPI callのresultとしてpostMessageする

senderは受け取ったMessagePortをReadableStreamに変換して、戻り値として返す

RPCのイメージ

exposeでうまいこと型推論して、receiverに型情報を含める

code:ts

// receiver.ts

const receiver = expose(port, {

async add(a: number, b: number) { return a + b; },

async *series(n: number, { signal }: { signal: AbortSignal }) {

for (let i=0;i<n;i++) {

if (signal.aborted) return;

yield i;

await new Promise(r=>setTimeout(r,10));

}

});

export type { receiver };

// sender.ts

import type { receiver } from "./receiver.ts";

{

await using api: RemoteCall<typeof receiver> = await wrap<typeof receiver>(port, "add", "series", {

closeOnDispose: true,

// signal: signal

});

const sum = await api.add(1,2); // Promise<number>

for await (const i of api.series(5)) // ReadableStream<number>

console.log(i);

}

// detach wrapper and close connection

}

streamとMessagePortの相互変換

code:ts

declare const encode: <T>(stream: Iterable<T> | AsyncIterable<T>) => MessagePort;

declare const decode: <T>(port: MessagePort) => ReadableStream<T>;

protocol

以下ができればいい

通信準備完了の確認

データの送信

backpressureの伝達

close,cancel,abort,errorの伝達

methodを生やすのやめる

軽量化が目的

signature変更

before: fn(...args, signal)

after: call("fn", [...args], { transfer: [...], signal })

理由

transferやsignalと、関数の引数リストとを区別するのが困難

特にoptional parametersがあると厄介

minlinkと同様、引数を[]で区切って渡したほうが面倒ごとが少ない

code:ts

// Fnは関数を表す型制約

interface RemoteProcedure<RemoteProcedureMap extends Record<string, Fn>> extends Disposable {

// 次のように使うためのsignature:

// using call = await wrap(endpoint, ...);

// console.log(call("add", 1, 2)); // 3

// RemoteReturnType<T>は、TがIterable<U> | AsyncIterable<U>のときReadableStream<U>を、それ以外の時Promise<Awaited<T>>を返すtype alias

<FnName extends keyof RemoteProcedureMap>(functionName: FnName, args: Parameters<RemoteProcedureMap<FnName>>, options?: RemoteProcedureOptions): RemoteReturnType<RemoteProcedureMap<FnName>>;

FnName in keyof RemoteProcedureMap: (args: Parameters<RemoteProcedureMap<FnName>>, options?: RemoteProcedureOptions): RemoteReturnType<RemoteProcedureMap<FnName>>;

}

型制約

(a: number, b: number) => numberは(a: number, b: number, opts?: RemoteCallOptions) => Promise<number>になる

RemoteCallOptionsにAbortSignalなどを渡せる

Transferable[]もRemoteCallOptionsに渡す

(data: ReadableStream<T>) => numberは(data: Iterable<T> | AsyncIterable<T>, opts?: RemoteCallOptions) => Promise<T>になる

(fn: () => void) => numberは作成不能

exposeの型制約でエラーになるようにする

(a: number) => Iterable<T> | AsyncIterable<T>は(a: number, opts?: RemoteCallOptions) => ReadableStream<T>になる

懸念

もともとMessagePortを受け取る函数のときはどうするか

MessagePortを引数に取れないようにするか