d3/server/network.ts

44 lines
1.4 KiB
TypeScript

#!/bin/env -S deno run --allow-net
import { MsgToClient, MsgToServer, mts } from "../common/mod.ts";
import { channel, log_from, parsed_stream, Receiver, Sender, serialized_stream } from "../common/utils.ts";
const log = log_from(import.meta);
export class Gateway {
server;
constructor(port: number) {
this.server = Deno.listen({ port });
log("Listening on", port);
}
async *accept() {
for await (const connection of this.server) {
const session = await ClientInterface.init(connection);
log("New session.");
yield session;
}
}
}
export class ClientInterface {
inputs;
outputs;
constructor(inputs: Receiver<MsgToServer>, outputs: Sender<MsgToClient>) {
this.inputs = inputs;
this.outputs = outputs;
}
// deno-lint-ignore require-await
static async init(connection: Deno.Conn) {
// TODO : handshake ?
const [input_sender, input_receiver] = channel<MsgToServer>();
const [output_sender, output_receiver] = channel<MsgToClient>();
input_sender.send_all(parsed_stream(mts.message_to_server_parser())(connection.readable))
.catch((err) => log("Deserialisation failure :", err.message))
.finally(() => input_sender.send({ kind: "exit" }));
serialized_stream(output_receiver.iter())(connection.writable)
.catch((err) => log("Serialisation failure :", err.message))
.finally(() => input_sender.send({ kind: "exit" }));
return new ClientInterface(input_receiver, output_sender);
}
}