From 3eb4cc6acf28799398ba5bded97f330221b22c4c Mon Sep 17 00:00:00 2001 From: Leo Date: Sat, 10 Oct 2026 01:38:37 +0200 Subject: [PATCH 1/3] feat(network): packet manager --- docs/docs/network/network-client-api.mdx | 66 ++++- docs/docs/network/network-client.mdx | 42 +++ docs/docs/network/network-server-api.mdx | 44 ++- docs/docs/network/network-server.mdx | 25 ++ docs/docs/network/packet-framing.mdx | 34 ++- example/pong-network/apps/client/src/main.ts | 14 +- .../apps/client/src/systems/systems.ts | 44 +-- example/pong-network/apps/server/src/main.ts | 10 +- .../apps/server/src/systems/systems.ts | 53 ++-- example/pong-network/package.json | 1 + example/pong-network/shared/protocol.ts | 13 + modules/network/README.md | 40 ++- .../src/client/client.network.library.ts | 84 +++++- modules/network/src/client/index.ts | 12 +- .../src/client/network-client-context.type.ts | 51 +++- modules/network/src/index.ts | 2 + modules/network/src/server/index.ts | 12 +- .../src/server/network-server-context.type.ts | 64 +++- .../src/server/server.network.library.ts | 105 ++++++- modules/network/src/shared/channels.ts | 7 +- modules/network/src/shared/packet.ts | 275 ++++++++++++++++++ .../client/client.network.library.spec.ts | 129 +++++++- .../server/server.network.library.spec.ts | 121 +++++++- modules/network/test/shared/packet.spec-d.ts | 35 +++ modules/network/test/shared/packet.spec.ts | 218 ++++++++++++++ modules/network/tsconfig.spec.json | 2 +- pnpm-lock.yaml | 3 + 27 files changed, 1353 insertions(+), 153 deletions(-) create mode 100644 example/pong-network/shared/protocol.ts create mode 100644 modules/network/src/shared/packet.ts create mode 100644 modules/network/test/shared/packet.spec-d.ts create mode 100644 modules/network/test/shared/packet.spec.ts diff --git a/docs/docs/network/network-client-api.mdx b/docs/docs/network/network-client-api.mdx index 86542c79..1e5eff6b 100644 --- a/docs/docs/network/network-client-api.mdx +++ b/docs/docs/network/network-client-api.mdx @@ -59,13 +59,73 @@ Returns: `void` ## getReceivedPackets(options?) -Return the packets received since the last call. Call it once per frame. +Return the packets received since the last call. Call it once per frame: +packets left unread for a whole tick are dropped. Typed packets never show up +here. Option `channel`: `Channel` to read from. Without it, the packets of every configured channel are returned, reliable channels first. Returns: [`NetworkData[]`](#networkdata) +## packets + +Typed packets, declared with `definePacket` (see below). +Packets of a type are routed to its listeners and to `read`, never to +`getReceivedPackets`. + +### packets.on(packet, listener) + +Call `listener(data, { channel, ctx })` for every packet of this type, each tick +before the ECS systems (not while paused). Packets are dispatched in the order +they arrived, reliable channels first. A listener that throws, or returns a +rejected promise, is logged and does not stop the others. + +Returns: `() => void`, a function that removes the listener. + +### packets.read(packet) + +Return the packets of this type received this tick, as `{ data, channel }[]`. +Meant for ECS systems, which run after the packets are dispatched. + +Returns: `{ data: T; channel: Channel }[]` + +### packets.send(packet, data, options?) + +Send a packet to the server. `data` must match the definition's payload type and +is encoded as JSON. + +Option `channel`: `Channel` to send on. The definition's channel by default. + +Throws a `TypeError` when `data` cannot be encoded as JSON. + +Returns: `void` + +## definePacket(name, options?) + +Declare a packet type, in a file both apps import. Also exported by +`@nanoforge-dev/network` itself, for that shared file. + +Parameter `name`: `string`, 1 to 255 UTF-8 bytes, unique in the game. It is +written in front of every packet of this type. + +Option `channel`: `Channel` used when `send` gets no `channel`. +`Channel.ReliableOrdered` by default. + +Option `validate`: `(data: unknown) => data is T`, run on every received payload +before listeners and `read`. Packets that fail it are dropped with a warning. + +Throws when the name is empty or too long, and later, in `packets.on` or `read`, +when another definition of the same name has another channel or `validate`. + +Returns: `PacketDefinition` + +```ts +export const Move = definePacket<{ id: number; x: number; y: number }>("move", { + channel: Channel.UnreliableOrdered, +}); +``` + ## isConnected(options?) Return `true` when packets can be sent and received. @@ -85,8 +145,8 @@ Type: `string | undefined` ## NetworkPayload -What `sendData` accepts. Packets carry no format tag: the receiver reads a -packet the way it was sent. +What `sendData` accepts. Its payload carries no format tag: the receiver reads +a packet the way it was sent. | Value | Encoded as | Read back with | | -------------------------------------------------- | --------------------- | ----------------------------- | diff --git a/docs/docs/network/network-client.mdx b/docs/docs/network/network-client.mdx index 310e3efb..b2e097ec 100644 --- a/docs/docs/network/network-client.mdx +++ b/docs/docs/network/network-client.mdx @@ -43,6 +43,48 @@ const latestPackets = network.getReceivedPackets(); const decodedPackets = latestPackets.map((packet) => packet.json()); ``` +## Typed packets + +Instead of switching on a `type` field, declare each packet type once, in a file +both the client and the server import, and let the library route packets to it. + +```ts +// shared/protocol.ts +import { Channel, definePacket } from "@nanoforge-dev/network"; + +export const Play = definePacket("play"); // payload: null +export const Input = definePacket<{ key: "up" | "down" }>("input"); +export const Move = definePacket<{ id: number; x: number; y: number }>("move", { + channel: Channel.UnreliableOrdered, // default channel of this type +}); +``` + +```ts +// client: listen once, after app.init() +const off = network.packets.on(Move, (move, { channel, ctx }) => { + applyMove(ctx.ecs.registry, move); // `move` is typed +}); + +// or read this tick's packets from an ECS system +export const applyMoves = (registry, ctx) => { + for (const { data } of ctx.network.packets.read(Move)) applyMove(registry, data); +}; + +// send: the payload is checked against the definition +ctx.network.packets.send(Input, { key: "up" }); // Channel.ReliableOrdered +ctx.network.packets.send(Play, null); +``` + +- Listeners are called every tick before the ECS systems, in the order the + packets arrived, reliable channels first. They are not called while the game + is paused: packets wait for the next tick. +- `read` returns the packets of the current tick; a system that does not read + them every tick misses some. +- Typed packets never appear in `getReceivedPackets`, which only returns the + packets sent with `sendData`. +- A packet type with no listener and no reader is dropped with a warning, + printed once per type. + ## Notes - See [Network Client API](network-client-api) for the full list of available functions, diff --git a/docs/docs/network/network-server-api.mdx b/docs/docs/network/network-server-api.mdx index b4a7d8f4..eccd2cf8 100644 --- a/docs/docs/network/network-server-api.mdx +++ b/docs/docs/network/network-server-api.mdx @@ -62,7 +62,9 @@ Returns: `void` ### getReceivedPackets(options?) -Return the packets each client sent since the last call. Call it once per frame. +Return the packets each client sent since the last call. Call it once per +frame: packets left unread for a whole tick are dropped. Typed packets never +show up here. Option `channel`: `Channel` to read from. Without it, the packets of every configured channel are merged per client, reliable channels first. @@ -88,10 +90,46 @@ Parameter `clientId`: `ClientId` Returns: `ClientInfo | undefined` (`undefined` once the client is gone) +## packets (`ctx.network.packets`) + +Typed packets, declared with `definePacket` (see the +[network-client API](network-client-api)). Packets of a type +are routed to its listeners and to `read`, never to `getReceivedPackets`. + +### on(packet, listener) + +Call `listener(data, { clientId, channel, ctx })` for every packet of this type, +each tick before the ECS systems (not while paused). Packets are dispatched in +the order they arrived, reliable channels first. A listener that throws, or +returns a rejected promise, is logged and does not stop the others. + +Returns: `() => void`, a function that removes the listener. + +### read(packet) + +Return the packets of this type received this tick, as +`{ data, clientId, channel }[]`. Meant for ECS systems, which run after the +packets are dispatched. + +Returns: `{ data: T; clientId: ClientId; channel: Channel }[]` + +### sendToAll(packet, data, options?) + +Send a packet to every client reachable on the channel. `data` must match the +definition's payload type and is encoded as JSON. + +Option `channel`: `Channel` to send on. The definition's channel by default. + +Throws a `TypeError` when `data` cannot be encoded as JSON. + +### sendToClient(clientId, packet, data, options?) + +Send a packet to one client, like `sendToAll`. + ## NetworkPayload -What `sendToAll` and `sendToClient` accept. Packets carry no format tag: the -receiver reads a packet the way it was sent. +What `sendToAll` and `sendToClient` accept. Their payload carries no format +tag: the receiver reads a packet the way it was sent. | Value | Encoded as | Read back with | | -------------------------------------------------- | --------------------- | ----------------------------- | diff --git a/docs/docs/network/network-server.mdx b/docs/docs/network/network-server.mdx index a1185c87..6350fc4c 100644 --- a/docs/docs/network/network-server.mdx +++ b/docs/docs/network/network-server.mdx @@ -76,6 +76,31 @@ const firstClientPackets = allPackets.get(connectedClients[0]); const decodedPackets = firstClientPackets.map((packet) => packet.json()); ``` +## Typed packets + +The server uses the same packet types as the client, from the shared protocol +file (see [Typed packets](network-client#typed-packets)). Its listeners and +`read` also get the sender. + +```ts +import { Input, Move, Play } from "../../shared/protocol"; + +// listen once, after app.init() +network.packets.on(Input, ({ key }, { clientId, ctx }) => { + handleInput(clientId, key, ctx); +}); + +// or read this tick's packets from an ECS system +for (const { data, clientId } of ctx.network.packets.read(Input)) handleInput(clientId, data.key); + +// send to everybody or to one client, on the definition's channel by default +ctx.network.packets.sendToAll(Move, { id: 1, x: 2, y: 3 }); +ctx.network.packets.sendToClient(clientId, Move, { id: 1, x: 2, y: 3 }); +``` + +Payloads come from the clients, so check the ones the game cannot trust with +the `validate` option of `definePacket`. + ## Notes - See [Network Server API](network-server-api) for the full list of available functions. diff --git a/docs/docs/network/packet-framing.mdx b/docs/docs/network/packet-framing.mdx index 33a4cec5..1f0be676 100644 --- a/docs/docs/network/packet-framing.mdx +++ b/docs/docs/network/packet-framing.mdx @@ -9,7 +9,8 @@ description: "The four delivery channels and how packets are framed on each" Every packet is sent on one of four channels, picked with the `channel` option of the network methods, for example `ctx.network.sendData(data, { channel: Channel.UnreliableOrdered })`. Without -it, packets are sent on `Channel.ReliableOrdered`. Pick the channel by what the +it, packets are sent on `Channel.ReliableOrdered`, and typed packets on the +channel of their definition. Pick the channel by what the packet needs: whether it may be lost, and whether an older packet is still useful once a newer one has arrived. @@ -32,13 +33,26 @@ promises that nothing is lost. Both transports deliver whole messages, so one message is one packet: there is no delimiter and payloads may contain any bytes. -The payload carries no format tag. Bytes are sent as they are, a string as its -UTF-8 text, and any other value as its `JSON.stringify` text. The receiver gets a -`NetworkData` and reads it the way it was sent, with `bytes()`, `text()` or -`json()`. +Every packet starts with the name of its packet type: one length byte, then the +name in UTF-8, then the payload. + +- **Typed packets** (`ctx.network.packets.send`, see + [Typed packets](network-client#typed-packets)) carry the name given to + `definePacket`, and their payload is the `JSON.stringify` text of the data. + The receiver routes them to the listeners and `read` of that type. +- **`sendData` packets** have a length of `0`, so no name. Their payload carries + no format tag: bytes are sent as they are, a string as its UTF-8 text, and any + other value as its `JSON.stringify` text. The receiver gets a `NetworkData` + from `getReceivedPackets` and reads it the way it was sent, with `bytes()`, + `text()` or `json()`. + +``` +typed : [4]["move"][{"id":1,"x":2}] +sendData : [0][payload] +``` - **TCP**: each binary WebSocket frame starts with one tag byte, `0` for - `ReliableOrdered` and `1` for `ReliableUnordered`, followed by the payload. + `ReliableOrdered` and `1` for `ReliableUnordered`, followed by the packet. Text frames are reserved for control messages such as the server `welcome`. - **WebRTC**: the client opens one data channel per unreliable channel, labelled `unreliable-ordered` and `unreliable-unordered`. Both use `ordered: false` and @@ -46,8 +60,12 @@ UTF-8 text, and any other value as its `JSON.stringify` text. The receiver gets ## Dropping late packets -`UnreliableOrdered` packets start with a big-endian `uint32` sequence number, -counted per peer and per direction. The receiver delivers a packet only if it is +Each `UnreliableOrdered` message starts with a big-endian `uint32` sequence +number, followed by the packet; the number is counted per peer and per +direction. The receiver delivers a packet only if it is newer than the last one it delivered, and drops it otherwise. Nothing ever waits for a lost packet: a gap is skipped as soon as a newer packet arrives. Numbers are compared as serial numbers, so the counter can wrap around. + +The sequence is shared by every packet type sent on the channel: a late packet +is dropped even when the newer one was of another type. diff --git a/example/pong-network/apps/client/src/main.ts b/example/pong-network/apps/client/src/main.ts index f7fcdc65..ad272119 100644 --- a/example/pong-network/apps/client/src/main.ts +++ b/example/pong-network/apps/client/src/main.ts @@ -5,14 +5,16 @@ import { Circle, Graphics2DLibrary, Layer, Rect } from "@nanoforge-dev/graphics- import { InputEnum, InputLibrary } from "@nanoforge-dev/input"; import { Channel, NetworkClientLibrary } from "@nanoforge-dev/network/client"; +import { AssignId, Play } from "../../../shared/protocol"; import { CircleComponent, Controller, + NetworkId, Position, RectangleComponent, Velocity, } from "./components/components"; -import { controlPlayer, draw, move, packetHandler } from "./systems/systems"; +import { applyMoves, controlPlayer, draw, move } from "./systems/systems"; export const layer = new Layer(); @@ -82,7 +84,13 @@ export const main = async (options: ClientRunOptions): Promise => { new RectangleComponent(new Rect({ fill: "blue", width: 30, height: 300 })), ); - registry.addSystem(packetHandler); + // Dispatched before the systems, so an entity gets its id before `applyMoves` reads its moves. + network.packets.on(AssignId, ({ assigned, id }) => { + const index = { ball: 2, paddle1: 4, paddle2: 5 }[assigned]; + registry.addComponent(registry.entityFromIndex(index), new NetworkId(id)); + }); + + registry.addSystem(applyMoves); registry.addSystem(move); registry.addSystem(controlPlayer); registry.addSystem(draw); @@ -91,7 +99,7 @@ export const main = async (options: ClientRunOptions): Promise => { await network.connect({ channels: [Channel.ReliableOrdered, Channel.UnreliableOrdered, Channel.UnreliableUnordered], }); - network.sendData({ type: "play" }); + network.packets.send(Play, null); await app.run(); }; diff --git a/example/pong-network/apps/client/src/systems/systems.ts b/example/pong-network/apps/client/src/systems/systems.ts index 2fdcaceb..9ae87e9b 100644 --- a/example/pong-network/apps/client/src/systems/systems.ts +++ b/example/pong-network/apps/client/src/systems/systems.ts @@ -1,6 +1,7 @@ import { type Context } from "@nanoforge-dev/common"; import { type Registry } from "@nanoforge-dev/ecs/client"; +import { Input, Move } from "../../../../shared/protocol"; import { CircleComponent, Controller, @@ -29,15 +30,15 @@ export const controlPlayer = (registry: Registry, ctx: Context) => { const downPressed = input.isKeyPressed(Controller.down); if (upPressed == downPressed) { if (Controller.lastPressedUp || Controller.lastPressedDown) { - network.sendData({ type: "input", key: "stop" }); + network.packets.send(Input, { key: "stop" }); Controller.lastPressedDown = false; Controller.lastPressedUp = false; } } else if (upPressed && !Controller.lastPressedUp) { - network.sendData({ type: "input", key: "up" }); + network.packets.send(Input, { key: "up" }); Controller.lastPressedUp = true; } else if (downPressed && !Controller.lastPressedDown) { - network.sendData({ type: "input", key: "down" }); + network.packets.send(Input, { key: "down" }); Controller.lastPressedDown = true; } }); @@ -55,32 +56,15 @@ export function draw(registry: Registry) { }); } -export function packetHandler(registry: Registry, ctx: Context) { - const network = ctx.network; - // Every channel, reliable first, so an entity's `assignId` is applied before its moves. - const jsonPackets = network.getReceivedPackets().map((packet) => packet.json()); +export function applyMoves(registry: Registry, ctx: Context) { + const zipper = registry.getZipper([NetworkId, Position, Velocity]); - if (!jsonPackets || jsonPackets.length === 0) return; - jsonPackets.forEach((packet) => { - const type = packet.type; - if (type === "move") { - const zipper = registry.getZipper([NetworkId, Position, Velocity]); - const it = zipper.find((entity) => { - return entity.NetworkId.id === packet.id; - }); - if (!it) return; - it.Position.x = packet.position.x; - it.Position.y = packet.position.y; - it.Velocity.x = packet.velocity.x; - it.Velocity.y = packet.velocity.y; - } else if (type === "assignId") { - if (packet.assigned === "ball") { - registry.addComponent(registry.entityFromIndex(2), new NetworkId(packet.id)); - } else if (packet.assigned === "paddle1") { - registry.addComponent(registry.entityFromIndex(4), new NetworkId(packet.id)); - } else if (packet.assigned === "paddle2") { - registry.addComponent(registry.entityFromIndex(5), new NetworkId(packet.id)); - } - } - }); + for (const { data } of ctx.network.packets.read(Move)) { + const it = zipper.find((entity) => entity.NetworkId.id === data.id); + if (!it) continue; + it.Position.x = data.position.x; + it.Position.y = data.position.y; + it.Velocity.x = data.velocity.x; + it.Velocity.y = data.velocity.y; + } } diff --git a/example/pong-network/apps/server/src/main.ts b/example/pong-network/apps/server/src/main.ts index 9b8e599a..62ec77ea 100644 --- a/example/pong-network/apps/server/src/main.ts +++ b/example/pong-network/apps/server/src/main.ts @@ -3,8 +3,9 @@ import { NanoforgeFactory } from "@nanoforge-dev/core"; import { EcsLibrary } from "@nanoforge-dev/ecs/server"; import { NetworkServerLibrary } from "@nanoforge-dev/network/server"; +import { Input, Play } from "../../../shared/protocol"; import { Position, Velocity } from "./components/components"; -import { bounce, move, onClientDisconnect, packetHandler } from "./systems/systems"; +import { bounce, handleClientInput, handlePlay, move, onClientDisconnect } from "./systems/systems"; export const main = async (options: RunOptions): Promise => { const app = NanoforgeFactory.createServer({ tickRate: 60 }); @@ -32,7 +33,12 @@ export const main = async (options: RunOptions): Promise => { registry.addComponent(paddle2, new Position(1850, 390)); registry.addComponent(paddle2, new Velocity(0, 0)); - registry.addSystem(packetHandler); + const zip = () => registry.getZipper([Position, Velocity]); + network.packets.on(Play, (_, { clientId, ctx }) => handlePlay(clientId, ctx.network, zip())); + network.packets.on(Input, ({ key }, { clientId, ctx }) => + handleClientInput(clientId, key, ctx.network, zip()), + ); + registry.addSystem(bounce); registry.addSystem(move); diff --git a/example/pong-network/apps/server/src/systems/systems.ts b/example/pong-network/apps/server/src/systems/systems.ts index dbcb5dd5..0f2c2ac7 100644 --- a/example/pong-network/apps/server/src/systems/systems.ts +++ b/example/pong-network/apps/server/src/systems/systems.ts @@ -4,10 +4,10 @@ import { Channel, type ClientId, type ClientInfo, - type NetworkData, type NetworkServerContextApi, } from "@nanoforge-dev/network/server"; +import { AssignId, Move } from "../../../../shared/protocol"; import { Position, Velocity } from "../components/components"; let cli1: ClientId | null = null; @@ -32,13 +32,9 @@ function sendMoveAll( network: NetworkServerContextApi, channel: Channel, ) { - network.sendToAll( - { - type: "move", - id: id, - position: { x: pos.x, y: pos.y }, - velocity: { x: vel.x, y: vel.y }, - }, + network.packets.sendToAll( + Move, + { id: id, position: { x: pos.x, y: pos.y }, velocity: { x: vel.x, y: vel.y } }, { channel }, ); } @@ -50,15 +46,15 @@ export function onClientDisconnect(info: ClientInfo) { } function connectNewClient(newCli: ClientId, network: NetworkServerContextApi, zip: any) { - network.sendToClient(newCli, { type: "assignId", assigned: "ball", id: 0 }); - network.sendToClient(newCli, { type: "assignId", assigned: "paddle1", id: 1 }); - network.sendToClient(newCli, { type: "assignId", assigned: "paddle2", id: 2 }); + network.packets.sendToClient(newCli, AssignId, { assigned: "ball", id: 0 }); + network.packets.sendToClient(newCli, AssignId, { assigned: "paddle1", id: 1 }); + network.packets.sendToClient(newCli, AssignId, { assigned: "paddle2", id: 2 }); sendMoveAll(0, zip[0].Velocity, zip[0].Position, network, Channel.ReliableOrdered); sendMoveAll(1, zip[1].Velocity, zip[1].Position, network, Channel.ReliableOrdered); sendMoveAll(2, zip[2].Velocity, zip[2].Position, network, Channel.ReliableOrdered); } -function handleClientInput( +export function handleClientInput( clientId: ClientId, key: string, network: NetworkServerContextApi, @@ -95,29 +91,16 @@ function handleClientInput( sendMoveAll(id, paddle.Velocity, paddle.Position, network, Channel.UnreliableOrdered); } -export function packetHandler(registry: Registry, ctx: Context) { - const zip = registry.getZipper([Position, Velocity]); - const network = ctx.network; - - const clientPackets: Map = network.getReceivedPackets(); - clientPackets.forEach((packets, client) => { - packets.forEach((packet) => { - const data = packet.json(); - if (data.type == "play") { - if (client == cli1 || client == cli2) return; - if (cli1 === null) { - cli1 = client; - } else if (cli2 === null) { - cli2 = client; - } else { - return; - } - connectNewClient(client, network, zip); - } else if (data.type == "input") { - handleClientInput(client, data.key, network, zip); - } - }); - }); +export function handlePlay(client: ClientId, network: NetworkServerContextApi, zip: any) { + if (client == cli1 || client == cli2) return; + if (cli1 === null) { + cli1 = client; + } else if (cli2 === null) { + cli2 = client; + } else { + return; + } + connectNewClient(client, network, zip); } function checkOutOfTerrain(id: number, paddle: any, network: NetworkServerContextApi) { diff --git a/example/pong-network/package.json b/example/pong-network/package.json index 28742f02..47eaf1a2 100644 --- a/example/pong-network/package.json +++ b/example/pong-network/package.json @@ -17,6 +17,7 @@ "@nanoforge-dev/config": "catalog:config", "@nanoforge-dev/core": "workspace:*", "@nanoforge-dev/env": "workspace:*", + "@nanoforge-dev/network": "workspace:*", "@trivago/prettier-plugin-sort-imports": "catalog:lint", "lint-staged": "catalog:ci", "prettier": "catalog:lint", diff --git a/example/pong-network/shared/protocol.ts b/example/pong-network/shared/protocol.ts new file mode 100644 index 00000000..555aa576 --- /dev/null +++ b/example/pong-network/shared/protocol.ts @@ -0,0 +1,13 @@ +import { Channel, definePacket } from "@nanoforge-dev/network"; + +type Vector = { x: number; y: number }; + +export const Play = definePacket("play"); +export const Input = definePacket<{ key: "up" | "down" | "stop" }>("input"); + +export const AssignId = definePacket<{ assigned: "ball" | "paddle1" | "paddle2"; id: number }>( + "assignId", +); +export const Move = definePacket<{ id: number; position: Vector; velocity: Vector }>("move", { + channel: Channel.UnreliableOrdered, +}); diff --git a/modules/network/README.md b/modules/network/README.md index df7fa4c3..6c9a46e2 100644 --- a/modules/network/README.md +++ b/modules/network/README.md @@ -52,7 +52,8 @@ port enables the reliable channels, the UDP port the unreliable ones. ### Channels Every method takes an optional `{ channel }`. Sending defaults to -`Channel.ReliableOrdered`; reading covers every configured channel. +`Channel.ReliableOrdered`, or to the definition's channel for +[typed packets](#typed-packets); reading covers every configured channel. | Channel | Lost? | Order | Transport | | --------------------- | -------- | ------------------------ | --------- | @@ -64,7 +65,7 @@ Every method takes an optional `{ channel }`. Sending defaults to One WebSocket per client carries both reliable channels: each binary frame starts with a one-byte channel tag. Each unreliable channel is its own data channel (`ordered: false`, `maxRetransmits: 0`); `UnreliableOrdered` packets -carry a `uint32` sequence number so the receiver drops any packet older than +are prefixed with a `uint32` sequence number so the receiver drops any packet older than the last one it delivered. One message is one packet on every channel, so payloads may contain any bytes. Over TCP, `ReliableUnordered` is delivered in order in practice. @@ -72,8 +73,9 @@ order in practice. ### Payloads The send methods take bytes, a string or any JSON value, and every received -packet is a `NetworkData` you read in the format you need. Packets carry no -format tag: read a packet the way it was sent. +packet is a `NetworkData` you read in the format you need. Their payload carries +no format tag: read a packet the way it was sent. Every packet starts with the +name of its [typed packet](#typed-packets) type, empty for these ones. | Sent value | Encoded as | Read back with | | -------------------------------------------------- | --------------------- | --------------------------------- | @@ -191,6 +193,36 @@ ctx.network.sendToClient(clientId, { type: "event" }); // Channel.ReliableOrdere transport fails or the channels are not open within `timeout` milliseconds (10 seconds by default). The server starts listening during `__init`. +### Typed packets + +Declare each packet type once, in a file both apps import, then listen for it, +read it from a system, or send it, with the payload type checked: + +```ts +// shared/protocol.ts +import { Channel, definePacket } from "@nanoforge-dev/network"; + +export const Input = definePacket<{ key: "up" | "down" }>("input"); +export const Move = definePacket<{ id: number; x: number; y: number }>("move", { + channel: Channel.UnreliableOrdered, // default channel of this type +}); +``` + +```ts +// Client: listeners run every tick before the ECS systems, not while paused. +network.packets.on(Move, (move, { channel, ctx }) => applyMove(ctx.ecs.registry, move)); +ctx.network.packets.send(Input, { key: "up" }); + +// Server: listeners and `read` also get the sender. +network.packets.on(Input, ({ key }, { clientId }) => handleInput(clientId, key)); +for (const { data, clientId } of ctx.network.packets.read(Input)) handleInput(clientId, data.key); // in a system +ctx.network.packets.sendToAll(Move, { id: 1, x: 2, y: 3 }); +``` + +`on` returns a function that removes the listener. Typed packets never appear in +`getReceivedPackets`, and a type nobody listens to or reads is dropped with a +warning. Use `definePacket`'s `validate` option to check payloads on the server. + ### Client identification Each client gets a `ClientId` (a random UUID string) shared by its TCP and UDP diff --git a/modules/network/src/client/client.network.library.ts b/modules/network/src/client/client.network.library.ts index 53655a2a..17ee7c5b 100644 --- a/modules/network/src/client/client.network.library.ts +++ b/modules/network/src/client/client.network.library.ts @@ -11,11 +11,26 @@ import { DEFAULT_CHANNEL, isReliableChannel, } from "../shared/channels"; -import { NetworkData, type NetworkPayload, encodeNetworkPayload } from "../shared/network-data"; +import { + type NetworkData, + type NetworkPayload, + encodeNetworkPayload, +} from "../shared/network-data"; +import { + PacketDispatcher, + type TypedPacket, + encodePacket, + encodeTypedPacket, + sortPackets, +} from "../shared/packet"; import { type Trace, createTrace, traceClient } from "../shared/trace"; import type { ClientSession } from "./client-session"; import { ClientConfigNetwork } from "./config.client.network"; -import type { NetworkClientContextApi } from "./network-client-context.type"; +import type { + ClientPacketInfo, + ClientPacketsApi, + NetworkClientContextApi, +} from "./network-client-context.type"; import { TCPClient } from "./tcp.client.network"; import { UDPClient } from "./udp.client.network"; @@ -45,6 +60,7 @@ const CONNECT_POLL_MS = 50; * ctx.network.sendData({ type: "join" }); // Channel.ReliableOrdered * ctx.network.sendData({ x: 1, y: 2 }, { channel: Channel.UnreliableOrdered }); * const messages = ctx.network.getReceivedPackets().map((packet) => packet.json()); // every channel + * ctx.network.packets.send(Input, { key: "up" }); // typed packet, see `definePacket` * ``` */ export class NetworkClientLibrary extends Library implements EditorAwareLibrary { @@ -55,6 +71,17 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary /** Pending or settled connection of each transport, so it is only opened once. */ private readonly _connecting = new Map>(); private readonly _session: ClientSession = {}; + private readonly _dispatcher = new PacketDispatcher(); + /** Typed packets drained from the transports, dispatched by the next `__run`. */ + private _typed: TypedPacket[] = []; + /** `sendData` packets drained from the transports, returned by `getReceivedPackets`. */ + private readonly _raw = new Map(); + private readonly _packets: ClientPacketsApi = { + on: (packet, listener) => this._dispatcher.on(packet, listener), + read: (packet) => this._dispatcher.read(packet), + send: (packet, data, { channel = packet.channel } = {}) => + this.sendBytes(channel, encodeTypedPacket(packet, data)), + }; /** Client id assigned by the server, `undefined` until transport is welcomed. */ public get clientId(): string | undefined { @@ -63,6 +90,16 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary private _trace: Trace | undefined; + constructor() { + // `runAfter` keys run after this library: packets are dispatched before the ECS systems. + super({ runAfter: ["ecs"] }); + } + + /** Typed packets: per-type listeners, `read` and `send`. */ + public get packets(): ClientPacketsApi { + return this._packets; + } + public override async __init(ctx: InitContext): Promise { const config = await registerEnv(ClientConfigNetwork, ctx.env); @@ -152,17 +189,16 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary * @throws When the channel's port was not configured */ public sendData(data: NetworkPayload, { channel = DEFAULT_CHANNEL }: ChannelOptions = {}): void { - const bytes = encodeNetworkPayload(data); - if (isReliableChannel(channel)) this.requireTcp(channel).sendData(channel, bytes); - else this.requireUdp(channel).sendData(channel, bytes); + this.sendBytes(channel, encodePacket(null, encodeNetworkPayload(data))); } /** * Return the packets received since the last call. * * @remarks - * Call this method once per frame. Without a `channel`, the packets of every - * configured channel are returned, reliable channels first. + * Call this method once per frame: packets left unread for a whole tick are + * dropped, and typed packets never show up here. Without a `channel`, the + * packets of every configured channel are returned, reliable channels first. * * @param options - `channel` to read from, every channel by default. * @returns Array of packets, each read with `bytes()`, `text()`, `json()`… @@ -172,10 +208,10 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary if (channel === undefined) { return this.configuredChannels().flatMap((c) => this.getReceivedPackets({ channel: c })); } - const packets = isReliableChannel(channel) - ? this.requireTcp(channel).getReceivedPackets(channel) - : this.requireUdp(channel).getReceivedPackets(channel); - return packets.map((packet) => new NetworkData(packet)); + this.drain(channel); + const packets = this._raw.get(channel) ?? []; + this._raw.delete(channel); + return packets; } /** @@ -210,6 +246,15 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary this._trace?.tick(); } + /** Calls the listeners of the typed packets received since the last tick. */ + public override async __run(ctx: Context): Promise { + this._raw.clear(); + for (const channel of this.configuredChannels()) this.drain(channel); + const typed = this._typed; + this._typed = []; + this._dispatcher.dispatch(typed, ctx); + } + public override expose(): NetworkClientContextApi { // eslint-disable-next-line @typescript-eslint/no-this-alias const library = this; @@ -218,6 +263,9 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary sendData: (data, options) => library.sendData(data, options), getReceivedPackets: (options) => library.getReceivedPackets(options), isConnected: (options) => library.isConnected(options), + get packets() { + return library.packets; + }, get clientId() { return library.clientId; }, @@ -230,6 +278,20 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary ); } + private sendBytes(channel: Channel, bytes: Uint8Array): void { + if (isReliableChannel(channel)) this.requireTcp(channel).sendData(channel, bytes); + else this.requireUdp(channel).sendData(channel, bytes); + } + + private drain(channel: Channel): void { + const packets = isReliableChannel(channel) + ? this.requireTcp(channel).getReceivedPackets(channel) + : this.requireUdp(channel).getReceivedPackets(channel); + const raw = this._raw.get(channel) ?? []; + sortPackets(packets, { channel }, raw, this._typed); + this._raw.set(channel, raw); + } + private requireTcp(channel: Channel): TCPClient { if (!this._tcp) throw new Error(`Channel ${channel} needs SERVER_TCP_PORT to be set`); return this._tcp; diff --git a/modules/network/src/client/index.ts b/modules/network/src/client/index.ts index 9e05745b..a32b67e6 100644 --- a/modules/network/src/client/index.ts +++ b/modules/network/src/client/index.ts @@ -3,6 +3,16 @@ import "./context-augmentation"; export type { NetworkConfig } from "../shared/config.network"; export { Channel, type ChannelOptions, type ConnectOptions } from "../shared/channels"; export { NetworkData, type NetworkPayload } from "../shared/network-data"; +export { + type PacketDefinition, + type PacketListener, + type PacketOptions, + definePacket, +} from "../shared/packet"; export { NetworkClientLibrary } from "./client.network.library"; export { ClientConfigNetwork } from "./config.client.network"; -export type { NetworkClientContextApi } from "./network-client-context.type"; +export type { + ClientPacketInfo, + ClientPacketsApi, + NetworkClientContextApi, +} from "./network-client-context.type"; diff --git a/modules/network/src/client/network-client-context.type.ts b/modules/network/src/client/network-client-context.type.ts index f6e21a27..14987bfc 100644 --- a/modules/network/src/client/network-client-context.type.ts +++ b/modules/network/src/client/network-client-context.type.ts @@ -1,5 +1,49 @@ -import type { ChannelOptions, ConnectOptions } from "../shared/channels"; +import type { Channel, ChannelOptions, ConnectOptions } from "../shared/channels"; import type { NetworkData, NetworkPayload } from "../shared/network-data"; +import type { PacketDefinition, PacketListener } from "../shared/packet"; + +/** Where a packet received by the client comes from. */ +export type ClientPacketInfo = { channel: Channel }; + +/** + * Typed packets of the client, exposed as `ctx.network.packets`. + * + * @example + * ```ts + * network.packets.on(Move, (move, { ctx }) => applyMove(ctx.ecs.registry, move)); + * for (const { data } of ctx.network.packets.read(Move)) applyMove(registry, data); + * ctx.network.packets.send(Input, { key: "up" }); + * ``` + */ +export interface ClientPacketsApi { + /** + * Be notified of every packet of a type, each tick before the ECS systems + * (not while paused). + * + * @returns A function that removes the listener. + * @throws When another definition with the same name has another channel or + * `validate`. + */ + on( + packet: PacketDefinition, + listener: PacketListener, ClientPacketInfo>, + ): () => void; + /** + * Return the packets of a type received this tick, for an ECS system. + * + * @throws When another definition with the same name has another channel or + * `validate`. + */ + read(packet: PacketDefinition): (ClientPacketInfo & { data: T })[]; + /** + * Send a packet to the server, on the definition's channel unless another + * `channel` is given. + * + * @throws When the channel's port was not configured, or `data` cannot be + * encoded as JSON. + */ + send(packet: PacketDefinition, data: NoInfer, options?: ChannelOptions): void; +} /** * Public surface of `NetworkClientLibrary`, exposed on `Context.network`. @@ -10,6 +54,8 @@ import type { NetworkData, NetworkPayload } from "../shared/network-data"; * port is not configured throws. */ export interface NetworkClientContextApi { + /** Typed packets: per-type listeners, `read` and `send`. */ + readonly packets: ClientPacketsApi; /** Client id assigned by the server, shared by TCP and UDP. `undefined` until welcomed. */ readonly clientId: string | undefined; /** @@ -24,7 +70,8 @@ export interface NetworkClientContextApi { sendData(data: NetworkPayload, options?: ChannelOptions): void; /** * Return the packets received since the last call, from `channel` or from - * every configured channel. Call it once per frame. + * every configured channel. Call it once per frame: + * packets left unread for a whole tick are dropped. */ getReceivedPackets(options?: ChannelOptions): NetworkData[]; /** Return `true` when `channel`, or every configured channel, is connected. */ diff --git a/modules/network/src/index.ts b/modules/network/src/index.ts index 34317cf2..629936ea 100644 --- a/modules/network/src/index.ts +++ b/modules/network/src/index.ts @@ -1 +1,3 @@ export type { NetworkConfig } from "./shared/config.network"; +export { Channel, type ChannelOptions } from "./shared/channels"; +export { type PacketDefinition, type PacketOptions, definePacket } from "./shared/packet"; diff --git a/modules/network/src/server/index.ts b/modules/network/src/server/index.ts index 00f0e922..5b21bee6 100644 --- a/modules/network/src/server/index.ts +++ b/modules/network/src/server/index.ts @@ -3,6 +3,12 @@ import "./context-augmentation"; export type { NetworkConfig } from "../shared/config.network"; export { Channel, type ChannelOptions } from "../shared/channels"; export { NetworkData, type NetworkPayload } from "../shared/network-data"; +export { + type PacketDefinition, + type PacketListener, + type PacketOptions, + definePacket, +} from "../shared/packet"; export type { ClientId, ClientInfo, @@ -12,6 +18,10 @@ export type { Transport, } from "./client-registry"; export { ServerConfigNetwork } from "./config.server.network"; -export type { NetworkServerContextApi } from "./network-server-context.type"; +export type { + NetworkServerContextApi, + ServerPacketInfo, + ServerPacketsApi, +} from "./network-server-context.type"; export { NetworkServerLibrary } from "./server.network.library"; export type { UDPServerIceConfig } from "./udp.server.network"; diff --git a/modules/network/src/server/network-server-context.type.ts b/modules/network/src/server/network-server-context.type.ts index 63050268..7a29072a 100644 --- a/modules/network/src/server/network-server-context.type.ts +++ b/modules/network/src/server/network-server-context.type.ts @@ -1,7 +1,64 @@ -import type { ChannelOptions } from "../shared/channels"; +import type { Channel, ChannelOptions } from "../shared/channels"; import type { NetworkData, NetworkPayload } from "../shared/network-data"; +import type { PacketDefinition, PacketListener } from "../shared/packet"; import type { ClientId, ClientInfo, ClientsApi } from "./client-registry"; +/** Where a packet received by the server comes from. */ +export type ServerPacketInfo = { clientId: ClientId; channel: Channel }; + +/** + * Typed packets of the server, exposed as `ctx.network.packets`. + * + * @example + * ```ts + * network.packets.on(Input, ({ key }, { clientId, ctx }) => handleInput(clientId, key, ctx)); + * for (const { data, clientId } of ctx.network.packets.read(Input)) handleInput(clientId, data.key); + * ctx.network.packets.sendToAll(Move, move); + * ``` + */ +export interface ServerPacketsApi { + /** + * Be notified of every packet of a type, each tick before the ECS systems + * (not while paused). + * + * @returns A function that removes the listener. + * @throws When another definition with the same name has another channel or + * `validate`. + */ + on( + packet: PacketDefinition, + listener: PacketListener, ServerPacketInfo>, + ): () => void; + /** + * Return the packets of a type received this tick, for an ECS system. + * + * @throws When another definition with the same name has another channel or + * `validate`. + */ + read(packet: PacketDefinition): (ServerPacketInfo & { data: T })[]; + /** + * Send a packet to every client reachable on the definition's channel, + * unless another `channel` is given. + * + * @throws When the channel's port was not configured, or `data` cannot be + * encoded as JSON. + */ + sendToAll(packet: PacketDefinition, data: NoInfer, options?: ChannelOptions): void; + /** + * Send a packet to one client, on the definition's channel unless another + * `channel` is given. + * + * @throws When the channel's port was not configured, or `data` cannot be + * encoded as JSON. + */ + sendToClient( + clientId: ClientId, + packet: PacketDefinition, + data: NoInfer, + options?: ChannelOptions, + ): void; +} + /** * Public surface of `NetworkServerLibrary`, exposed on `Context.network`. * @@ -11,6 +68,8 @@ import type { ClientId, ClientInfo, ClientsApi } from "./client-registry"; * whose port is not configured throws. */ export interface NetworkServerContextApi { + /** Typed packets: per-type listeners, `read` and sending. */ + readonly packets: ServerPacketsApi; /** Client sessions, shared by TCP and UDP: lookup and connect/disconnect callbacks. */ readonly clients: ClientsApi; /** @@ -25,7 +84,8 @@ export interface NetworkServerContextApi { sendToClient(clientId: ClientId, data: NetworkPayload, options?: ChannelOptions): void; /** * Return the packets each client sent since the last call, from `channel` or - * from every configured channel. Call it once per frame. + * from every configured channel. Call it once per frame: + * packets left unread for a whole tick are dropped. */ getReceivedPackets(options?: ChannelOptions): Map; /** Return the clients reachable on `channel`, or on any configured channel. */ diff --git a/modules/network/src/server/server.network.library.ts b/modules/network/src/server/server.network.library.ts index 1b107d70..0a06cf40 100644 --- a/modules/network/src/server/server.network.library.ts +++ b/modules/network/src/server/server.network.library.ts @@ -9,11 +9,26 @@ import { DEFAULT_CHANNEL, isReliableChannel, } from "../shared/channels"; -import { NetworkData, type NetworkPayload, encodeNetworkPayload } from "../shared/network-data"; +import { + type NetworkData, + type NetworkPayload, + encodeNetworkPayload, +} from "../shared/network-data"; +import { + PacketDispatcher, + type TypedPacket, + encodePacket, + encodeTypedPacket, + sortPackets, +} from "../shared/packet"; import { type Trace, createTrace, traceServer } from "../shared/trace"; import { type ClientId, type ClientInfo, ClientRegistry, type ClientsApi } from "./client-registry"; import { ServerConfigNetwork } from "./config.server.network"; -import type { NetworkServerContextApi } from "./network-server-context.type"; +import type { + NetworkServerContextApi, + ServerPacketInfo, + ServerPacketsApi, +} from "./network-server-context.type"; import { TCPServer } from "./tcp.server.network"; import { UDPServer } from "./udp.server.network"; @@ -38,6 +53,7 @@ import { UDPServer } from "./udp.server.network"; * ```ts * ctx.network.sendToAll({ type: "start" }); // Channel.ReliableOrdered * ctx.network.sendToAll({ x: 1, y: 2 }, { channel: Channel.UnreliableUnordered }); + * ctx.network.packets.sendToAll(Move, move); // typed packet, see `definePacket` * ctx.network.getReceivedPackets().forEach((packets, clientId) => { * const messages = packets.map((packet) => packet.json()); // every channel * }); @@ -50,6 +66,17 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary private _udp?: UDPServer; private readonly _registry = new ClientRegistry(); + private readonly _dispatcher = new PacketDispatcher(); + private _typed: TypedPacket[] = []; + private readonly _raw = new Map>(); + private readonly _packets: ServerPacketsApi = { + on: (packet, listener) => this._dispatcher.on(packet, listener), + read: (packet) => this._dispatcher.read(packet), + sendToAll: (packet, data, { channel = packet.channel } = {}) => + this.sendBytesToAll(channel, encodeTypedPacket(packet, data)), + sendToClient: (clientId, packet, data, { channel = packet.channel } = {}) => + this.sendBytesToClient(clientId, channel, encodeTypedPacket(packet, data)), + }; /** Client sessions shared by the TCP and UDP servers. */ public get clients(): ClientsApi { @@ -58,6 +85,15 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary private _trace: Trace | undefined; + constructor() { + super({ runAfter: ["ecs"] }); + } + + /** Typed packets: per-type listeners, `read` and sending. */ + public get packets(): ServerPacketsApi { + return this._packets; + } + public override async __init(ctx: InitContext): Promise { if (typeof Bun === "undefined") { throw new Error( @@ -125,9 +161,7 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary * encoded as JSON. */ public sendToAll(data: NetworkPayload, { channel = DEFAULT_CHANNEL }: ChannelOptions = {}): void { - const bytes = encodeNetworkPayload(data); - if (isReliableChannel(channel)) this.requireTcp(channel).sendToEverybody(channel, bytes); - else this.requireUdp(channel).sendToEverybody(channel, bytes); + this.sendBytesToAll(channel, encodePacket(null, encodeNetworkPayload(data))); } /** @@ -144,17 +178,16 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary data: NetworkPayload, { channel = DEFAULT_CHANNEL }: ChannelOptions = {}, ): void { - const bytes = encodeNetworkPayload(data); - if (isReliableChannel(channel)) this.requireTcp(channel).sendToClient(channel, clientId, bytes); - else this.requireUdp(channel).sendToClient(channel, clientId, bytes); + this.sendBytesToClient(clientId, channel, encodePacket(null, encodeNetworkPayload(data))); } /** * Return the packets each client sent since the last call. * * @remarks - * Call this method once per frame. Without a `channel`, the packets of every - * configured channel are merged per client, reliable channels first. + * Call this method once per frame: packets left unread for a whole tick are + * dropped, and typed packets never show up here. Without a `channel`, the + * packets of every configured channel are merged per client, reliable channels first. * * @param options - `channel` to read from, every channel by default. * @returns Map of client id to packets, each read with `bytes()`, `text()`, @@ -163,12 +196,11 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary */ public getReceivedPackets({ channel }: ChannelOptions = {}): Map { if (channel !== undefined) { - const packets = isReliableChannel(channel) - ? this.requireTcp(channel).getReceivedPackets(channel) - : this.requireUdp(channel).getReceivedPackets(channel); - return new Map( - [...packets].map(([clientId, raw]) => [clientId, raw.map((p) => new NetworkData(p))]), - ); + const clients = this.drain(channel); + const packets = this._raw.get(channel) ?? new Map(); + this._raw.delete(channel); + for (const clientId of clients) if (!packets.has(clientId)) packets.set(clientId, []); + return packets; } const merged = new Map(); @@ -228,6 +260,15 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary this._trace?.tick(); } + /** Calls the listeners of the typed packets received since the last tick. */ + public override async __run(ctx: Context): Promise { + this._raw.clear(); + for (const channel of this.configuredChannels()) this.drain(channel); + const typed = this._typed; + this._typed = []; + this._dispatcher.dispatch(typed, ctx); + } + public override expose(): NetworkServerContextApi { // eslint-disable-next-line @typescript-eslint/no-this-alias const library = this; @@ -240,6 +281,9 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary get clients() { return library.clients; }, + get packets() { + return library.packets; + }, }; } @@ -249,6 +293,35 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary ); } + private sendBytesToAll(channel: Channel, bytes: Uint8Array): void { + if (isReliableChannel(channel)) this.requireTcp(channel).sendToEverybody(channel, bytes); + else this.requireUdp(channel).sendToEverybody(channel, bytes); + } + + private sendBytesToClient(clientId: ClientId, channel: Channel, bytes: Uint8Array): void { + if (isReliableChannel(channel)) this.requireTcp(channel).sendToClient(channel, clientId, bytes); + else this.requireUdp(channel).sendToClient(channel, clientId, bytes); + } + + /** + * Move what each client sent on `channel` to the typed and raw queues. + * + * @returns The clients the transport reported, packets or not. + */ + private drain(channel: Channel): ClientId[] { + const received = isReliableChannel(channel) + ? this.requireTcp(channel).getReceivedPackets(channel) + : this.requireUdp(channel).getReceivedPackets(channel); + const raw = this._raw.get(channel) ?? new Map(); + received.forEach((packets, clientId) => { + const clientRaw = raw.get(clientId) ?? []; + sortPackets(packets, { clientId, channel }, clientRaw, this._typed); + if (clientRaw.length > 0) raw.set(clientId, clientRaw); + }); + this._raw.set(channel, raw); + return [...received.keys()]; + } + private requireTcp(channel: Channel): TCPServer { if (!this._tcp) throw new Error(`Channel ${channel} needs LISTENING_TCP_PORT to be set`); return this._tcp; diff --git a/modules/network/src/shared/channels.ts b/modules/network/src/shared/channels.ts index 649243fc..fe8dffd6 100644 --- a/modules/network/src/shared/channels.ts +++ b/modules/network/src/shared/channels.ts @@ -44,9 +44,10 @@ export const DEFAULT_CHANNEL = Channel.ReliableOrdered; */ export type ChannelOptions = { /** - * Channel to send on or read from. Sending defaults to - * `Channel.ReliableOrdered`; reading and connection checks cover every - * configured channel when omitted. + * Channel to send on or read from. Sending defaults to the packet + * definition's channel for typed packets and to `Channel.ReliableOrdered` + * otherwise; reading and connection checks cover every configured channel + * when omitted. */ channel?: Channel; }; diff --git a/modules/network/src/shared/packet.ts b/modules/network/src/shared/packet.ts new file mode 100644 index 00000000..3dabc9f1 --- /dev/null +++ b/modules/network/src/shared/packet.ts @@ -0,0 +1,275 @@ +import type { Context } from "@nanoforge-dev/common"; + +import { type Channel, DEFAULT_CHANNEL } from "./channels"; +import { NetworkData } from "./network-data"; + +const encoder = new TextEncoder(); +const decoder = new TextDecoder(); + +/** Longest packet name in UTF-8 bytes, as its length is written on one byte. */ +const MAX_NAME_BYTES = 255; + +/** Unknown packet names warned about: past it, they are dropped silently (peers choose them). */ +const MAX_WARNED_NAMES = 100; + +/** + * Options of `definePacket`. + * + * @typeParam T - Payload type of the packet. + */ +export type PacketOptions = { + /** + * Channel the packet is sent on when `send` gets no `channel` option. + * + * @defaultValue `Channel.ReliableOrdered` + */ + channel?: Channel; + /** + * Check a received payload before it reaches listeners or `read`. Packets + * that fail it, or make it throw, are dropped with a warning. + */ + validate?: (data: unknown) => data is T; +}; + +/** + * A packet type, shared by the client and the server. + * + * @remarks + * Create it with `definePacket`, in a file both apps import. Its `name` is + * written in front of every packet of this type, so both sides must use the + * same one. + * + * @typeParam T - Payload type of the packet. + */ +export interface PacketDefinition { + /** Name identifying the packet type on the wire. */ + readonly name: string; + /** Channel used when `send` gets no `channel` option. */ + readonly channel: Channel; + /** Check run on every received payload, see `PacketOptions.validate`. */ + readonly validate?: ((data: unknown) => data is T) | undefined; + /** Never set: only keeps definitions of different payloads apart. */ + readonly __payload?: T; +} + +/** + * Declare a packet type. + * + * @example + * ```ts + * // protocol.ts, imported by the client and the server + * export const Move = definePacket<{ id: number; x: number; y: number }>("move", { + * channel: Channel.UnreliableOrdered, + * }); + * export const Play = definePacket("play"); // payload: null + * ``` + * + * @param name - Name identifying the type on the wire, unique in the game. + * @param options - Default `channel` and payload `validate` check. + * @returns The packet definition, used by `packets.on`, `read` and `send`. + * @throws When the name is empty or longer than 255 UTF-8 bytes. + */ +export const definePacket = ( + name: string, + { channel = DEFAULT_CHANNEL, validate }: PacketOptions = {}, +): PacketDefinition => { + const length = encoder.encode(name).length; + if (length === 0 || length > MAX_NAME_BYTES) { + throw new Error(`Packet name must be 1 to ${MAX_NAME_BYTES} UTF-8 bytes long, got "${name}"`); + } + return { name, channel, validate }; +}; + +/** + * Prefix a payload with the name of its packet type. + * + * @param name - Packet type name, or `null` for a packet sent with `sendData`. + * @param payload - Payload bytes. + * @returns `[name length u8][name UTF-8][payload]`, the length being `0` when + * the name is `null`. + */ +export const encodePacket = (name: string | null, payload: Uint8Array): Uint8Array => { + const header = encoder.encode(name ?? ""); + const packet = new Uint8Array(1 + header.length + payload.length); + packet[0] = header.length; + packet.set(header, 1); + packet.set(payload, 1 + header.length); + return packet; +}; + +/** + * Encode a typed packet: its payload as JSON, prefixed with its type name. + * + * @throws `TypeError` When the payload cannot be encoded as JSON. + */ +export const encodeTypedPacket = ( + packet: PacketDefinition, + data: T, +): Uint8Array => { + const json: string | undefined = JSON.stringify(data); + if (json === undefined) + throw new TypeError(`Cannot send a ${typeof data} as packet "${packet.name}"`); + return encodePacket(packet.name, encoder.encode(json)); +}; + +/** + * Split a packet into the name of its type and its payload. + * + * @param packet - Packet built by `encodePacket`. + * @returns The name, `null` for a packet sent with `sendData`, and the + * payload; or `null` when the packet is empty or truncated. + */ +export const decodePacket = ( + packet: Uint8Array, +): { name: string | null; data: Uint8Array } | null => { + if (packet.length === 0) return null; + const end = 1 + (packet[0] ?? 0); + if (packet.length < end) return null; + return { + name: end === 1 ? null : decoder.decode(packet.subarray(1, end)), + data: packet.subarray(end), + }; +}; + +/** + * Callback of `packets.on`, called with the payload of each packet received. + * + * @typeParam T - Payload type of the packet. + * @typeParam Info - Where the packet comes from: its channel, and its + * sender on the server. + */ +export type PacketListener = (data: T, info: Info & { ctx: Context }) => void; + +/** A received packet with a type name, waiting to be dispatched. */ +export type TypedPacket = { name: string; data: Uint8Array; info: Info }; + +/** + * Sort packets received from a transport: `sendData` payloads go to `raw`, + * typed packets to `typed`. Empty or truncated packets are dropped. + */ +export const sortPackets = ( + packets: Uint8Array[], + info: Info, + raw: NetworkData[], + typed: TypedPacket[], +): void => { + for (const packet of packets) { + const decoded = decodePacket(packet); + if (!decoded) continue; + if (decoded.name === null) raw.push(new NetworkData(decoded.data)); + else typed.push({ name: decoded.name, data: decoded.data, info }); + } +}; + +type FramePacket = { name: string; value: unknown; info: Info; valid?: boolean }; + +/** + * Calls the listeners of each received packet type and keeps the packets of + * the current tick for `read`. Shared by the client and the server. + */ +export class PacketDispatcher { + private readonly _definitions = new Map>(); + private readonly _listeners = new Map>>(); + private readonly _warned = new Set(); + private _frame: FramePacket[] = []; + + /** + * Be notified of every packet of a type, each tick before the ECS systems. + * + * @returns A function that removes the listener. + * @throws When another definition with the same name has another channel or + * `validate`. + */ + public on( + packet: PacketDefinition, + listener: PacketListener, Info>, + ): () => void { + this.define(packet); + const listeners = this._listeners.get(packet.name) ?? new Set(); + listeners.add(listener); + this._listeners.set(packet.name, listeners); + return () => listeners.delete(listener); + } + + /** + * Return the packets of a type received this tick. + * + * @throws When another definition with the same name has another channel or + * `validate`. + */ + public read(packet: PacketDefinition): (Info & { data: T })[] { + this.define(packet); + return this._frame + .filter((received) => received.name === packet.name && this.check(received)) + .map((received) => ({ ...received.info, data: received.value as T })); + } + + /** + * Start a new tick: decode `packets` and call their listeners, in order. + * + * @param packets - Typed packets received since the last tick. + * @param ctx - Context handed to the listeners. + */ + public dispatch(packets: TypedPacket[], ctx: Context): void { + for (const { name } of this._frame) { + if (this._definitions.has(name) || this._warned.has(name)) continue; + if (this._warned.size >= MAX_WARNED_NAMES) break; + this._warned.add(name); + console.warn(`Network: no listener or reader for packet "${name}", dropping them`); + } + + this._frame = []; + for (const { name, data, info } of packets) { + let value: unknown; + try { + value = JSON.parse(decoder.decode(data)); + } catch (error) { + console.warn(`Network: packet "${name}" is not valid JSON, dropped`, { cause: error }); + continue; + } + const received: FramePacket = { name, value, info }; + this._frame.push(received); + + const listeners = this._listeners.get(name); + if (!listeners?.size || !this.check(received)) continue; + for (const listener of [...listeners]) this.call(listener, received, ctx); + } + } + + private define(packet: PacketDefinition): void { + const known = this._definitions.get(packet.name); + if (known && (known.channel !== packet.channel || known.validate !== packet.validate)) { + throw new Error(`Packet "${packet.name}" is defined twice, with another channel or validate`); + } + if (!known) this._definitions.set(packet.name, packet); + } + + /** Run the type's `validate` once per packet. */ + private check(received: FramePacket): boolean { + if (received.valid === undefined) { + const validate = this._definitions.get(received.name)?.validate; + try { + received.valid = !validate || validate(received.value); + } catch { + received.valid = false; + } + if (!received.valid) console.warn(`Network: packet "${received.name}" is invalid, dropped`); + } + return received.valid; + } + + private call( + listener: PacketListener, + received: FramePacket, + ctx: Context, + ): void { + const fail = (error: unknown) => + console.error(`Network: listener of packet "${received.name}" failed`, { cause: error }); + try { + const result: unknown = listener(received.value, { ...received.info, ctx }); + if (result instanceof Promise) result.catch(fail); + } catch (error) { + fail(error); + } + } +} diff --git a/modules/network/test/client/client.network.library.spec.ts b/modules/network/test/client/client.network.library.spec.ts index c2687371..f67d16e0 100644 --- a/modules/network/test/client/client.network.library.spec.ts +++ b/modules/network/test/client/client.network.library.spec.ts @@ -1,7 +1,8 @@ -import type { InitContext } from "@nanoforge-dev/common"; +import type { Context, InitContext } from "@nanoforge-dev/common"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { Channel, NetworkClientLibrary, NetworkData } from "../../src/client"; +import { Channel, NetworkClientLibrary, NetworkData, definePacket } from "../../src/client"; +import { encodeTypedPacket } from "../../src/shared/packet"; const makeInitContext = (env: Record): InitContext => ({ vars: { get: () => undefined, set: () => {} }, @@ -273,7 +274,7 @@ describe("NetworkClientLibrary", () => { await lib.connect(); lib.sendData(new Uint8Array([1])); - expect(tcpSocket().send).toHaveBeenCalledWith(new Uint8Array([0, 1])); + expect(tcpSocket().send).toHaveBeenCalledWith(new Uint8Array([0, 0, 1])); }); it("should send each reliable channel over the same TCP socket", async () => { @@ -285,8 +286,8 @@ describe("NetworkClientLibrary", () => { lib.sendData(new Uint8Array([2]), { channel: Channel.ReliableUnordered }); expect(tcpSocket().send.mock.calls).toStrictEqual([ - [new Uint8Array([0, 1])], - [new Uint8Array([1, 2])], + [new Uint8Array([0, 0, 1])], + [new Uint8Array([1, 0, 2])], ]); }); @@ -299,9 +300,11 @@ describe("NetworkClientLibrary", () => { lib.sendData("hi", { channel: Channel.UnreliableUnordered }); expect(tcpSocket().send).toHaveBeenCalledWith( - new Uint8Array([0, ...utf8('{"type":"play"}')]), + new Uint8Array([0, 0, ...utf8('{"type":"play"}')]), + ); + expect(dataChannel("unreliable-unordered").send).toHaveBeenCalledWith( + new Uint8Array([0, ...utf8("hi")]), ); - expect(dataChannel("unreliable-unordered").send).toHaveBeenCalledWith(utf8("hi")); }); it("should open one data channel per unreliable channel", async () => { @@ -321,7 +324,7 @@ describe("NetworkClientLibrary", () => { await lib.connect(); lib.sendData(new Uint8Array([3]), { channel: Channel.UnreliableUnordered }); - expect(dataChannel("unreliable-unordered").send).toHaveBeenCalledWith(new Uint8Array([3])); + expect(dataChannel("unreliable-unordered").send).toHaveBeenCalledWith(new Uint8Array([0, 3])); expect(dataChannel("unreliable-ordered").send).not.toHaveBeenCalled(); }); @@ -330,9 +333,9 @@ describe("NetworkClientLibrary", () => { await lib.__init(makeInitContext(both)); await lib.connect(); - dataChannel("unreliable-unordered").onmessage({ data: new Uint8Array([9]).buffer }); - tcpSocket().onmessage({ data: new Uint8Array([1, 2]).buffer }); - tcpSocket().onmessage({ data: new Uint8Array([0, 1]).buffer }); + dataChannel("unreliable-unordered").onmessage({ data: new Uint8Array([0, 9]).buffer }); + tcpSocket().onmessage({ data: new Uint8Array([1, 0, 2]).buffer }); + tcpSocket().onmessage({ data: new Uint8Array([0, 0, 1]).buffer }); expect(bytesOf(lib.getReceivedPackets({ channel: Channel.ReliableUnordered }))).toStrictEqual( [new Uint8Array([2])], @@ -349,7 +352,7 @@ describe("NetworkClientLibrary", () => { await lib.__init(makeInitContext(tcpOnly)); await lib.connect(); - tcpSocket().onmessage({ data: new Uint8Array([0, ...utf8('{"type":"move"}')]).buffer }); + tcpSocket().onmessage({ data: new Uint8Array([0, 0, ...utf8('{"type":"move"}')]).buffer }); const [packet] = lib.getReceivedPackets(); expect(packet).toBeInstanceOf(NetworkData); @@ -376,6 +379,105 @@ describe("NetworkClientLibrary", () => { }); }); + describe("packets", () => { + const ctx = {} as Context; + const Move = definePacket<{ x: number }>("move", { channel: Channel.UnreliableUnordered }); + const Play = definePacket("play"); + + /** Deliver a typed packet as the server would, on a reliable channel or a data channel. */ + const receive = (packet: Uint8Array, channel: Channel) => + channel === Channel.UnreliableUnordered + ? dataChannel(channel).onmessage({ data: packet.slice().buffer }) + : tcpSocket().onmessage({ data: new Uint8Array([0, ...packet]).buffer }); + + it("should dispatch typed packets in __run, reliable ones first", async () => { + const lib = new NetworkClientLibrary(); + await lib.__init(makeInitContext(both)); + await lib.connect(); + const calls: unknown[] = []; + lib.packets.on(Move, (move, info) => calls.push([move, info])); + lib.packets.on(Play, (play, info) => calls.push([play, info])); + + receive(encodeTypedPacket(Move, { x: 1 }), Channel.UnreliableUnordered); + receive(encodeTypedPacket(Play, null), Channel.ReliableOrdered); + tcpSocket().onmessage({ data: new Uint8Array([0, 0, 7]).buffer }); + expect(calls).toStrictEqual([]); + + await lib.__run(ctx); + expect(calls).toStrictEqual([ + [null, { channel: Channel.ReliableOrdered, ctx }], + [{ x: 1 }, { channel: Channel.UnreliableUnordered, ctx }], + ]); + expect(lib.packets.read(Move)).toStrictEqual([ + { channel: Channel.UnreliableUnordered, data: { x: 1 } }, + ]); + expect(bytesOf(lib.getReceivedPackets())).toStrictEqual([new Uint8Array([7])]); + }); + + it("should keep typed packets out of getReceivedPackets until __run", async () => { + const lib = new NetworkClientLibrary(); + await lib.__init(makeInitContext(tcpOnly)); + await lib.connect(); + const listener = vi.fn(); + lib.packets.on(Play, listener); + + receive(encodeTypedPacket(Play, null), Channel.ReliableOrdered); + expect(lib.getReceivedPackets()).toStrictEqual([]); + await lib.__run(ctx); + expect(listener).toHaveBeenCalledOnce(); + }); + + it("should drop sendData packets left unread for a whole tick", async () => { + const lib = new NetworkClientLibrary(); + await lib.__init(makeInitContext(tcpOnly)); + await lib.connect(); + + tcpSocket().onmessage({ data: new Uint8Array([0, 0, 1]).buffer }); + await lib.__run(ctx); + expect(bytesOf(lib.getReceivedPackets())).toStrictEqual([new Uint8Array([1])]); + + tcpSocket().onmessage({ data: new Uint8Array([0, 0, 2]).buffer }); + await lib.__run(ctx); + await lib.__run(ctx); + expect(lib.getReceivedPackets()).toStrictEqual([]); + }); + + it("should not lose packets drained by a listener", async () => { + const lib = new NetworkClientLibrary(); + await lib.__init(makeInitContext(tcpOnly)); + await lib.connect(); + const listener = vi.fn(() => { + receive(encodeTypedPacket(Play, null), Channel.ReliableOrdered); + lib.getReceivedPackets(); + }); + lib.packets.on(Play, listener); + + receive(encodeTypedPacket(Play, null), Channel.ReliableOrdered); + await lib.__run(ctx); + await lib.__run(ctx); + expect(listener).toHaveBeenCalledTimes(2); + }); + + it("should send on the definition's channel unless another is given", async () => { + const lib = new NetworkClientLibrary(); + await lib.__init(makeInitContext(both)); + await lib.connect(); + + lib.packets.send(Move, { x: 1 }); + lib.packets.send(Play, null, { channel: Channel.ReliableUnordered }); + expect(dataChannel("unreliable-unordered").send).toHaveBeenCalledWith( + encodeTypedPacket(Move, { x: 1 }), + ); + expect(tcpSocket().send).toHaveBeenCalledWith( + new Uint8Array([1, ...encodeTypedPacket(Play, null)]), + ); + }); + + it("should run before the ECS", () => { + expect(new NetworkClientLibrary().relationships.runAfter).toStrictEqual(["ecs"]); + }); + }); + describe("expose", () => { it("delegates to the library", async () => { const lib = new NetworkClientLibrary(); @@ -384,12 +486,13 @@ describe("NetworkClientLibrary", () => { const api = lib.expose(); api.sendData(new Uint8Array([5]), { channel: Channel.ReliableUnordered }); - expect(tcpSocket().send).toHaveBeenCalledWith(new Uint8Array([1, 5])); + expect(tcpSocket().send).toHaveBeenCalledWith(new Uint8Array([1, 0, 5])); expect(api.getReceivedPackets()).toStrictEqual([]); expect(() => api.isConnected({ channel: Channel.UnreliableOrdered })).toThrow( "SERVER_UDP_PORT", ); expect(api.clientId).toBe("client-0"); + expect(api.packets).toBe(lib.packets); }); }); }); diff --git a/modules/network/test/server/server.network.library.spec.ts b/modules/network/test/server/server.network.library.spec.ts index 4bbb8958..3d5e5efb 100644 --- a/modules/network/test/server/server.network.library.spec.ts +++ b/modules/network/test/server/server.network.library.spec.ts @@ -1,7 +1,8 @@ -import type { InitContext } from "@nanoforge-dev/common"; +import type { Context, InitContext } from "@nanoforge-dev/common"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { Channel, NetworkData, NetworkServerLibrary } from "../../src/server"; +import { Channel, NetworkData, NetworkServerLibrary, definePacket } from "../../src/server"; +import { encodeTypedPacket } from "../../src/shared/packet"; vi.mock("node-datachannel/polyfill", () => ({ RTCPeerConnection: vi.fn(function (this: any) { @@ -151,8 +152,8 @@ describe("NetworkServerLibrary", () => { lib.sendToAll(new Uint8Array([1])); lib.sendToClient(welcome.id, new Uint8Array([2])); expect(webSocket.send.mock.calls).toStrictEqual([ - [new Uint8Array([0, 1])], - [new Uint8Array([0, 2])], + [new Uint8Array([0, 0, 1])], + [new Uint8Array([0, 0, 2])], ]); }); @@ -169,8 +170,8 @@ describe("NetworkServerLibrary", () => { }); lib.sendToAll(new Uint8Array([2]), { channel: Channel.UnreliableUnordered }); - expect(tcp.webSocket.send).toHaveBeenCalledWith(new Uint8Array([1, 1])); - expect(unordered.send).toHaveBeenCalledWith(new Uint8Array([2])); + expect(tcp.webSocket.send).toHaveBeenCalledWith(new Uint8Array([1, 0, 1])); + expect(unordered.send).toHaveBeenCalledWith(new Uint8Array([0, 2])); }); it("should send strings as UTF-8 and other values as JSON", async () => { @@ -185,9 +186,9 @@ describe("NetworkServerLibrary", () => { lib.sendToAll("hi", { channel: Channel.UnreliableUnordered }); expect(tcp.webSocket.send).toHaveBeenCalledWith( - new Uint8Array([0, ...utf8('{"type":"assignId","id":0}')]), + new Uint8Array([0, 0, ...utf8('{"type":"assignId","id":0}')]), ); - expect(unordered.send).toHaveBeenCalledWith(utf8("hi")); + expect(unordered.send).toHaveBeenCalledWith(new Uint8Array([0, ...utf8("hi")])); }); it("should relay a received packet as it is", async () => { @@ -195,13 +196,15 @@ describe("NetworkServerLibrary", () => { await lib.__init(makeInitContext(tcpOnly)); const { webSocket, options, welcome } = openSocket(0, "http://localhost:9000/"); - options.websocket.message(webSocket, Buffer.from([0, ...utf8('{"type":"input"}')])); + options.websocket.message(webSocket, Buffer.from([0, 0, ...utf8('{"type":"input"}')])); const [packet] = lib.getReceivedPackets().get(welcome.id) ?? []; expect(packet).toBeInstanceOf(NetworkData); expect(packet?.json()).toStrictEqual({ type: "input" }); lib.sendToAll(packet!); - expect(webSocket.send).toHaveBeenCalledWith(new Uint8Array([0, ...utf8('{"type":"input"}')])); + expect(webSocket.send).toHaveBeenCalledWith( + new Uint8Array([0, 0, ...utf8('{"type":"input"}')]), + ); }); it("should read one channel, or merge every channel per client", async () => { @@ -213,9 +216,9 @@ describe("NetworkServerLibrary", () => { udp.webSocket.data.peerConnection.ondatachannel({ channel: unordered }); const id = tcp.welcome.id; - unordered.onmessage?.({ data: new Uint8Array([9]).buffer }); - tcp.options.websocket.message(tcp.webSocket, Buffer.from([1, 2])); - tcp.options.websocket.message(tcp.webSocket, Buffer.from([0, 1])); + unordered.onmessage?.({ data: new Uint8Array([0, 9]).buffer }); + tcp.options.websocket.message(tcp.webSocket, Buffer.from([1, 0, 2])); + tcp.options.websocket.message(tcp.webSocket, Buffer.from([0, 0, 1])); expect( bytesOf(lib.getReceivedPackets({ channel: Channel.ReliableUnordered }).get(id)), @@ -255,6 +258,93 @@ describe("NetworkServerLibrary", () => { }); }); + describe("packets", () => { + const ctx = {} as Context; + const Input = definePacket<{ key: string }>("input"); + const Move = definePacket<{ x: number }>("move", { channel: Channel.UnreliableUnordered }); + + it("should dispatch typed packets in __run with their sender", async () => { + const lib = new NetworkServerLibrary(); + await lib.__init(makeInitContext(both)); + const tcp = openSocket(0, "http://localhost:9000/"); + const udp = openSocket(1, `http://localhost:9001/?token=${tcp.welcome.token}`); + const unordered = makeDataChannel("unreliable-unordered"); + udp.webSocket.data.peerConnection.ondatachannel({ channel: unordered }); + const clientId = tcp.welcome.id; + const calls: unknown[] = []; + lib.packets.on(Input, (input, info) => calls.push([input, info])); + lib.packets.on(Move, (move, info) => calls.push([move, info])); + + unordered.onmessage?.({ data: encodeTypedPacket(Move, { x: 1 }).buffer }); + tcp.options.websocket.message( + tcp.webSocket, + Buffer.from([0, ...encodeTypedPacket(Input, { key: "up" })]), + ); + tcp.options.websocket.message(tcp.webSocket, Buffer.from([0, 0, 7])); + expect(calls).toStrictEqual([]); + + await lib.__run(ctx); + expect(calls).toStrictEqual([ + [{ key: "up" }, { clientId, channel: Channel.ReliableOrdered, ctx }], + [{ x: 1 }, { clientId, channel: Channel.UnreliableUnordered, ctx }], + ]); + expect(lib.packets.read(Input)).toStrictEqual([ + { clientId, channel: Channel.ReliableOrdered, data: { key: "up" } }, + ]); + expect(bytesOf(lib.getReceivedPackets().get(clientId))).toStrictEqual([new Uint8Array([7])]); + }); + + it("should send on the definition's channel unless another is given", async () => { + const lib = new NetworkServerLibrary(); + await lib.__init(makeInitContext(both)); + const tcp = openSocket(0, "http://localhost:9000/"); + const udp = openSocket(1, `http://localhost:9001/?token=${tcp.welcome.token}`); + const unordered = makeDataChannel("unreliable-unordered"); + udp.webSocket.data.peerConnection.ondatachannel({ channel: unordered }); + + lib.packets.sendToAll(Move, { x: 1 }); + lib.packets.sendToClient( + tcp.welcome.id, + Input, + { key: "up" }, + { + channel: Channel.ReliableUnordered, + }, + ); + expect(unordered.send).toHaveBeenCalledWith(encodeTypedPacket(Move, { x: 1 })); + expect(tcp.webSocket.send).toHaveBeenCalledWith( + new Uint8Array([1, ...encodeTypedPacket(Input, { key: "up" })]), + ); + }); + + it("should drop sendData packets left unread for a whole tick", async () => { + const lib = new NetworkServerLibrary(); + await lib.__init(makeInitContext(tcpOnly)); + const { webSocket, options, welcome } = openSocket(0, "http://localhost:9000/"); + + options.websocket.message(webSocket, Buffer.from([0, 0, 1])); + await lib.__run(ctx); + await lib.__run(ctx); + expect(lib.getReceivedPackets().get(welcome.id)).toStrictEqual([]); + }); + + it("should forget clients that left without sending raw packets", async () => { + const lib = new NetworkServerLibrary(); + await lib.__init(makeInitContext(tcpOnly)); + const { webSocket, options, welcome } = openSocket(0, "http://localhost:9000/"); + + await lib.__run(ctx); + expect(lib.getReceivedPackets().get(welcome.id)).toStrictEqual([]); + await lib.__run(ctx); + options.websocket.close(webSocket); + expect(lib.getReceivedPackets().has(welcome.id)).toBe(false); + }); + + it("should run before the ECS", () => { + expect(new NetworkServerLibrary().relationships.runAfter).toStrictEqual(["ecs"]); + }); + }); + describe("expose", () => { it("delegates to the library", async () => { const lib = new NetworkServerLibrary(); @@ -265,8 +355,8 @@ describe("NetworkServerLibrary", () => { api.sendToAll(new Uint8Array([1]), { channel: Channel.ReliableUnordered }); api.sendToClient(welcome.id, new Uint8Array([2])); expect(webSocket.send.mock.calls).toStrictEqual([ - [new Uint8Array([1, 1])], - [new Uint8Array([0, 2])], + [new Uint8Array([1, 0, 1])], + [new Uint8Array([0, 0, 2])], ]); expect(api.getConnectedClients()).toStrictEqual([welcome.id]); expect(api.getReceivedPackets().get(welcome.id)).toStrictEqual([]); @@ -279,6 +369,7 @@ describe("NetworkServerLibrary", () => { it("returns the client sessions", () => { const lib = new NetworkServerLibrary(); expect(lib.expose().clients).toBe(lib.clients); + expect(lib.expose().packets).toBe(lib.packets); }); }); diff --git a/modules/network/test/shared/packet.spec-d.ts b/modules/network/test/shared/packet.spec-d.ts new file mode 100644 index 00000000..fe98a315 --- /dev/null +++ b/modules/network/test/shared/packet.spec-d.ts @@ -0,0 +1,35 @@ +import { describe, it } from "vitest"; + +// From the type files: both entries augment `Context.network`, so they cannot share a program. +import type { ClientPacketsApi } from "../../src/client/network-client-context.type"; +import { definePacket } from "../../src/index"; +import type { ServerPacketsApi } from "../../src/server/network-server-context.type"; + +declare const client: ClientPacketsApi; +declare const server: ServerPacketsApi; + +const Move = definePacket<{ x: number }>("move"); +const Play = definePacket("play"); + +describe("packets (type-level)", () => { + it("accepts the payload of the definition", () => { + client.send(Move, { x: 1 }); + client.send(Play, null); + server.sendToAll(Move, { x: 1 }); + server.sendToClient("client-0", Move, { x: 1 }); + client.on(Move, (move) => move.x.toFixed()); + server.on(Move, (move, { clientId }) => `${clientId}${move.x}`); + client.read(Move).map(({ data }) => data.x); + }); + + it("rejects another payload", () => { + // @ts-expect-error x must be a number + client.send(Move, { x: "1" }); + // @ts-expect-error x must be a number + server.sendToAll(Move, { x: "1" }); + // @ts-expect-error Play carries null + server.sendToClient("client-0", Play, { x: 1 }); + // @ts-expect-error the listener must take the definition's payload + client.on(Move, (move: { y: number }) => move.y); + }); +}); diff --git a/modules/network/test/shared/packet.spec.ts b/modules/network/test/shared/packet.spec.ts new file mode 100644 index 00000000..fd1456fd --- /dev/null +++ b/modules/network/test/shared/packet.spec.ts @@ -0,0 +1,218 @@ +import type { Context } from "@nanoforge-dev/common"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { Channel } from "../../src/shared/channels"; +import { NetworkData } from "../../src/shared/network-data"; +import { + PacketDispatcher, + type TypedPacket, + decodePacket, + definePacket, + encodePacket, + encodeTypedPacket, + sortPackets, +} from "../../src/shared/packet"; + +const utf8 = (text: string) => new TextEncoder().encode(text); + +type Info = { channel: Channel }; +const ctx = {} as Context; +const info: Info = { channel: Channel.ReliableOrdered }; + +const Move = definePacket<{ x: number }>("move", { + channel: Channel.UnreliableOrdered, + validate: (data): data is { x: number } => typeof (data as { x?: unknown }).x === "number", +}); +const Play = definePacket("play"); + +/** A typed packet as received, its payload being `value` as JSON. */ +const received = (name: string, value: unknown): TypedPacket => ({ + name, + data: utf8(JSON.stringify(value)), + info, +}); + +describe("definePacket", () => { + it("should default to ReliableOrdered", () => { + expect(Play).toStrictEqual({ + name: "play", + channel: Channel.ReliableOrdered, + validate: undefined, + }); + expect(Move.channel).toBe(Channel.UnreliableOrdered); + }); + + it("should reject an empty name or one longer than 255 bytes", () => { + expect(() => definePacket("")).toThrow("1 to 255"); + expect(() => definePacket("é".repeat(128))).toThrow("1 to 255"); + expect(() => definePacket("a".repeat(255))).not.toThrow(); + }); +}); + +describe("packet header", () => { + it("should round-trip a typed packet", () => { + const packet = encodeTypedPacket(Move, { x: 1 }); + + expect(packet).toStrictEqual(new Uint8Array([4, ...utf8("move"), ...utf8('{"x":1}')])); + expect(decodePacket(packet)).toStrictEqual({ name: "move", data: utf8('{"x":1}') }); + }); + + it("should mark sendData packets with a zero length", () => { + const packet = encodePacket(null, new Uint8Array([7])); + + expect(packet).toStrictEqual(new Uint8Array([0, 7])); + expect(decodePacket(packet)).toStrictEqual({ name: null, data: new Uint8Array([7]) }); + }); + + it("should reject empty and truncated packets", () => { + expect(decodePacket(new Uint8Array())).toBeNull(); + expect(decodePacket(new Uint8Array([4, 1, 2]))).toBeNull(); + }); + + it("should refuse a payload JSON cannot encode", () => { + expect(() => encodeTypedPacket(Play as never, undefined as never)).toThrow(TypeError); + }); + + it("should sort raw and typed packets, dropping broken ones", () => { + const raw: NetworkData[] = []; + const typed: TypedPacket[] = []; + + sortPackets( + [encodePacket(null, new Uint8Array([1])), new Uint8Array(), encodeTypedPacket(Play, null)], + info, + raw, + typed, + ); + expect(raw).toStrictEqual([new NetworkData(new Uint8Array([1]))]); + expect(typed).toStrictEqual([{ name: "play", data: utf8("null"), info }]); + }); +}); + +describe("PacketDispatcher", () => { + let warn: ReturnType; + let error: ReturnType; + + beforeEach(() => { + warn = vi.spyOn(console, "warn").mockImplementation(() => {}); + error = vi.spyOn(console, "error").mockImplementation(() => {}); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it("should call every listener of a type, in order, with the context", () => { + const dispatcher = new PacketDispatcher(); + const calls: unknown[] = []; + dispatcher.on(Move, (move, packetInfo) => calls.push(["a", move, packetInfo])); + dispatcher.on(Move, (move) => calls.push(["b", move])); + + dispatcher.dispatch([received("move", { x: 1 }), received("move", { x: 2 })], ctx); + expect(calls).toStrictEqual([ + ["a", { x: 1 }, { ...info, ctx }], + ["b", { x: 1 }], + ["a", { x: 2 }, { ...info, ctx }], + ["b", { x: 2 }], + ]); + }); + + it("should stop calling a removed listener", () => { + const dispatcher = new PacketDispatcher(); + const listener = vi.fn(); + const off = dispatcher.on(Play, listener); + + off(); + dispatcher.dispatch([received("play", null)], ctx); + expect(listener).not.toHaveBeenCalled(); + }); + + it("should keep calling listeners when one throws or rejects", async () => { + const dispatcher = new PacketDispatcher(); + const listener = vi.fn(); + dispatcher.on(Play, () => { + throw new Error("boom"); + }); + dispatcher.on(Play, async () => { + throw new Error("async boom"); + }); + dispatcher.on(Play, listener); + + dispatcher.dispatch([received("play", null)], ctx); + await Promise.resolve(); + expect(listener).toHaveBeenCalledOnce(); + expect(error).toHaveBeenCalledTimes(2); + }); + + it("should return the packets of the current tick only", () => { + const dispatcher = new PacketDispatcher(); + + dispatcher.dispatch([received("move", { x: 1 }), received("play", null)], ctx); + expect(dispatcher.read(Move)).toStrictEqual([{ ...info, data: { x: 1 } }]); + expect(dispatcher.read(Move)).toStrictEqual([{ ...info, data: { x: 1 } }]); + + dispatcher.dispatch([], ctx); + expect(dispatcher.read(Move)).toStrictEqual([]); + }); + + it("should drop invalid payloads and bad JSON, for listeners and readers", () => { + const dispatcher = new PacketDispatcher(); + const listener = vi.fn(); + dispatcher.on(Move, listener); + + dispatcher.dispatch( + [received("move", { x: "1" }), { name: "move", data: utf8("{"), info }], + ctx, + ); + expect(listener).not.toHaveBeenCalled(); + expect(dispatcher.read(Move)).toStrictEqual([]); + expect(warn).toHaveBeenCalledTimes(2); + }); + + it("should treat a throwing validate as invalid", () => { + const dispatcher = new PacketDispatcher(); + const Strict = definePacket<{ x: number }>("strict", { + validate: (data): data is { x: number } => (data as { x: number }).x.toFixed() !== "", + }); + + dispatcher.dispatch([received("strict", null)], ctx); + expect(dispatcher.read(Strict)).toStrictEqual([]); + }); + + it("should reject a definition of a known name with another channel or validate", () => { + const dispatcher = new PacketDispatcher(); + dispatcher.on(Move, () => {}); + + expect(() => dispatcher.on(definePacket("move"), () => {})).toThrow("defined twice"); + expect(() => + dispatcher.read(definePacket("move", { channel: Channel.UnreliableOrdered })), + ).toThrow("defined twice"); + expect(() => + dispatcher.read(definePacket("move", { channel: Move.channel, validate: Move.validate })), + ).not.toThrow(); + }); + + it("should warn once, a tick late, about types nobody listens to or reads", () => { + const dispatcher = new PacketDispatcher(); + + dispatcher.dispatch([received("move", { x: 1 }), received("ghost", null)], ctx); + expect(warn).not.toHaveBeenCalled(); + dispatcher.read(Move); // read after the dispatch, like a system + + dispatcher.dispatch([received("ghost", null)], ctx); + expect(warn).toHaveBeenCalledOnce(); + expect(warn).toHaveBeenCalledWith(expect.stringContaining('"ghost"')); + + dispatcher.dispatch([], ctx); + expect(warn).toHaveBeenCalledOnce(); + }); + + it("should stop warning about unknown types past 100 names", () => { + const dispatcher = new PacketDispatcher(); + const names = Array.from({ length: 150 }, (_, i) => received(`ghost-${i}`, null)); + + dispatcher.dispatch(names, ctx); + dispatcher.dispatch(names, ctx); + dispatcher.dispatch([], ctx); + expect(warn).toHaveBeenCalledTimes(100); + }); +}); diff --git a/modules/network/tsconfig.spec.json b/modules/network/tsconfig.spec.json index 8270caba..c62afd58 100644 --- a/modules/network/tsconfig.spec.json +++ b/modules/network/tsconfig.spec.json @@ -5,6 +5,6 @@ "noEmit": true, "skipLibCheck": true }, - "include": ["test/**/*.spec.ts"], + "include": ["test/**/*.spec-d.ts"], "exclude": ["node_modules"] } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 7d130836..0fb63596 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -373,6 +373,9 @@ importers: '@nanoforge-dev/env': specifier: workspace:* version: link:../../packages/env + '@nanoforge-dev/network': + specifier: workspace:* + version: link:../../modules/network '@trivago/prettier-plugin-sort-imports': specifier: catalog:lint version: 6.0.2(prettier@3.9.6)(supports-color@10.2.2) From 15e5e81e658dcf5d4e057f85f8b756eebaca64e3 Mon Sep 17 00:00:00 2001 From: Leo Date: Sat, 10 Oct 2026 02:15:44 +0200 Subject: [PATCH 2/3] fix(network): remove null spcification for empty packets --- docs/docs/network/network-client-api.mdx | 8 +++-- docs/docs/network/network-client.mdx | 4 +-- docs/docs/network/network-server-api.mdx | 7 +++-- docs/docs/network/packet-framing.mdx | 3 +- example/pong-network/apps/client/src/main.ts | 3 +- .../src/client/client.network.library.ts | 4 +-- modules/network/src/client/index.ts | 1 + .../src/client/network-client-context.type.ts | 7 +++-- modules/network/src/server/index.ts | 1 + .../src/server/network-server-context.type.ts | 10 +++---- .../src/server/server.network.library.ts | 12 +++++--- modules/network/src/shared/packet.ts | 29 ++++++++++++++----- .../client/client.network.library.spec.ts | 14 ++++----- modules/network/test/shared/packet.spec-d.ts | 13 +++++++-- modules/network/test/shared/packet.spec.ts | 26 +++++++++++++++-- 15 files changed, 96 insertions(+), 46 deletions(-) diff --git a/docs/docs/network/network-client-api.mdx b/docs/docs/network/network-client-api.mdx index 1e5eff6b..b8b9d7d7 100644 --- a/docs/docs/network/network-client-api.mdx +++ b/docs/docs/network/network-client-api.mdx @@ -90,10 +90,11 @@ Meant for ECS systems, which run after the packets are dispatched. Returns: `{ data: T; channel: Channel }[]` -### packets.send(packet, data, options?) +### packets.send(packet, data?, options?) Send a packet to the server. `data` must match the definition's payload type and -is encoded as JSON. +is encoded as JSON. Leave it out for a packet without payload, declared without +a type: `send(Play)`. Option `channel`: `Channel` to send on. The definition's channel by default. @@ -118,7 +119,8 @@ before listeners and `read`. Packets that fail it are dropped with a warning. Throws when the name is empty or too long, and later, in `packets.on` or `read`, when another definition of the same name has another channel or `validate`. -Returns: `PacketDefinition` +Returns: `PacketDefinition`. Without a type, `T` is `void`: the packet +carries no payload. ```ts export const Move = definePacket<{ id: number; x: number; y: number }>("move", { diff --git a/docs/docs/network/network-client.mdx b/docs/docs/network/network-client.mdx index b2e097ec..48991520 100644 --- a/docs/docs/network/network-client.mdx +++ b/docs/docs/network/network-client.mdx @@ -52,7 +52,7 @@ both the client and the server import, and let the library route packets to it. // shared/protocol.ts import { Channel, definePacket } from "@nanoforge-dev/network"; -export const Play = definePacket("play"); // payload: null +export const Play = definePacket("play"); // no payload export const Input = definePacket<{ key: "up" | "down" }>("input"); export const Move = definePacket<{ id: number; x: number; y: number }>("move", { channel: Channel.UnreliableOrdered, // default channel of this type @@ -72,7 +72,7 @@ export const applyMoves = (registry, ctx) => { // send: the payload is checked against the definition ctx.network.packets.send(Input, { key: "up" }); // Channel.ReliableOrdered -ctx.network.packets.send(Play, null); +ctx.network.packets.send(Play); // no payload to give ``` - Listeners are called every tick before the ECS systems, in the order the diff --git a/docs/docs/network/network-server-api.mdx b/docs/docs/network/network-server-api.mdx index eccd2cf8..a0834c76 100644 --- a/docs/docs/network/network-server-api.mdx +++ b/docs/docs/network/network-server-api.mdx @@ -113,16 +113,17 @@ packets are dispatched. Returns: `{ data: T; clientId: ClientId; channel: Channel }[]` -### sendToAll(packet, data, options?) +### sendToAll(packet, data?, options?) Send a packet to every client reachable on the channel. `data` must match the -definition's payload type and is encoded as JSON. +definition's payload type and is encoded as JSON. Leave it out for a packet +without payload: `sendToAll(Start)`. Option `channel`: `Channel` to send on. The definition's channel by default. Throws a `TypeError` when `data` cannot be encoded as JSON. -### sendToClient(clientId, packet, data, options?) +### sendToClient(clientId, packet, data?, options?) Send a packet to one client, like `sendToAll`. diff --git a/docs/docs/network/packet-framing.mdx b/docs/docs/network/packet-framing.mdx index 1f0be676..d3dce2af 100644 --- a/docs/docs/network/packet-framing.mdx +++ b/docs/docs/network/packet-framing.mdx @@ -38,7 +38,8 @@ name in UTF-8, then the payload. - **Typed packets** (`ctx.network.packets.send`, see [Typed packets](network-client#typed-packets)) carry the name given to - `definePacket`, and their payload is the `JSON.stringify` text of the data. + `definePacket`, and their payload is the `JSON.stringify` text of the data, + or nothing for a packet without payload. The receiver routes them to the listeners and `read` of that type. - **`sendData` packets** have a length of `0`, so no name. Their payload carries no format tag: bytes are sent as they are, a string as its UTF-8 text, and any diff --git a/example/pong-network/apps/client/src/main.ts b/example/pong-network/apps/client/src/main.ts index ad272119..2b4ef183 100644 --- a/example/pong-network/apps/client/src/main.ts +++ b/example/pong-network/apps/client/src/main.ts @@ -95,11 +95,10 @@ export const main = async (options: ClientRunOptions): Promise => { registry.addSystem(controlPlayer); registry.addSystem(draw); - // Wait for every channel the game uses, so no early move is missed. await network.connect({ channels: [Channel.ReliableOrdered, Channel.UnreliableOrdered, Channel.UnreliableUnordered], }); - network.packets.send(Play, null); + network.packets.send(Play); await app.run(); }; diff --git a/modules/network/src/client/client.network.library.ts b/modules/network/src/client/client.network.library.ts index 17ee7c5b..ed9e33ca 100644 --- a/modules/network/src/client/client.network.library.ts +++ b/modules/network/src/client/client.network.library.ts @@ -79,8 +79,8 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary private readonly _packets: ClientPacketsApi = { on: (packet, listener) => this._dispatcher.on(packet, listener), read: (packet) => this._dispatcher.read(packet), - send: (packet, data, { channel = packet.channel } = {}) => - this.sendBytes(channel, encodeTypedPacket(packet, data)), + send: (packet, ...[data, options]) => + this.sendBytes(options?.channel ?? packet.channel, encodeTypedPacket(packet, data)), }; /** Client id assigned by the server, `undefined` until transport is welcomed. */ diff --git a/modules/network/src/client/index.ts b/modules/network/src/client/index.ts index a32b67e6..e4f7f13e 100644 --- a/modules/network/src/client/index.ts +++ b/modules/network/src/client/index.ts @@ -4,6 +4,7 @@ export type { NetworkConfig } from "../shared/config.network"; export { Channel, type ChannelOptions, type ConnectOptions } from "../shared/channels"; export { NetworkData, type NetworkPayload } from "../shared/network-data"; export { + type PacketArgs, type PacketDefinition, type PacketListener, type PacketOptions, diff --git a/modules/network/src/client/network-client-context.type.ts b/modules/network/src/client/network-client-context.type.ts index 14987bfc..9d210d3f 100644 --- a/modules/network/src/client/network-client-context.type.ts +++ b/modules/network/src/client/network-client-context.type.ts @@ -1,6 +1,6 @@ import type { Channel, ChannelOptions, ConnectOptions } from "../shared/channels"; import type { NetworkData, NetworkPayload } from "../shared/network-data"; -import type { PacketDefinition, PacketListener } from "../shared/packet"; +import type { PacketArgs, PacketDefinition, PacketListener } from "../shared/packet"; /** Where a packet received by the client comes from. */ export type ClientPacketInfo = { channel: Channel }; @@ -37,12 +37,13 @@ export interface ClientPacketsApi { read(packet: PacketDefinition): (ClientPacketInfo & { data: T })[]; /** * Send a packet to the server, on the definition's channel unless another - * `channel` is given. + * `channel` is given. `data` is left out for a packet without payload: + * `send(Play)`. * * @throws When the channel's port was not configured, or `data` cannot be * encoded as JSON. */ - send(packet: PacketDefinition, data: NoInfer, options?: ChannelOptions): void; + send(packet: PacketDefinition, ...args: PacketArgs>): void; } /** diff --git a/modules/network/src/server/index.ts b/modules/network/src/server/index.ts index 5b21bee6..7c365e1a 100644 --- a/modules/network/src/server/index.ts +++ b/modules/network/src/server/index.ts @@ -4,6 +4,7 @@ export type { NetworkConfig } from "../shared/config.network"; export { Channel, type ChannelOptions } from "../shared/channels"; export { NetworkData, type NetworkPayload } from "../shared/network-data"; export { + type PacketArgs, type PacketDefinition, type PacketListener, type PacketOptions, diff --git a/modules/network/src/server/network-server-context.type.ts b/modules/network/src/server/network-server-context.type.ts index 7a29072a..446e1194 100644 --- a/modules/network/src/server/network-server-context.type.ts +++ b/modules/network/src/server/network-server-context.type.ts @@ -1,6 +1,6 @@ import type { Channel, ChannelOptions } from "../shared/channels"; import type { NetworkData, NetworkPayload } from "../shared/network-data"; -import type { PacketDefinition, PacketListener } from "../shared/packet"; +import type { PacketArgs, PacketDefinition, PacketListener } from "../shared/packet"; import type { ClientId, ClientInfo, ClientsApi } from "./client-registry"; /** Where a packet received by the server comes from. */ @@ -38,12 +38,13 @@ export interface ServerPacketsApi { read(packet: PacketDefinition): (ServerPacketInfo & { data: T })[]; /** * Send a packet to every client reachable on the definition's channel, - * unless another `channel` is given. + * unless another `channel` is given. `data` is left out for a packet without + * payload: `sendToAll(Start)`. * * @throws When the channel's port was not configured, or `data` cannot be * encoded as JSON. */ - sendToAll(packet: PacketDefinition, data: NoInfer, options?: ChannelOptions): void; + sendToAll(packet: PacketDefinition, ...args: PacketArgs>): void; /** * Send a packet to one client, on the definition's channel unless another * `channel` is given. @@ -54,8 +55,7 @@ export interface ServerPacketsApi { sendToClient( clientId: ClientId, packet: PacketDefinition, - data: NoInfer, - options?: ChannelOptions, + ...args: PacketArgs> ): void; } diff --git a/modules/network/src/server/server.network.library.ts b/modules/network/src/server/server.network.library.ts index 0a06cf40..6a1783ec 100644 --- a/modules/network/src/server/server.network.library.ts +++ b/modules/network/src/server/server.network.library.ts @@ -72,10 +72,14 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary private readonly _packets: ServerPacketsApi = { on: (packet, listener) => this._dispatcher.on(packet, listener), read: (packet) => this._dispatcher.read(packet), - sendToAll: (packet, data, { channel = packet.channel } = {}) => - this.sendBytesToAll(channel, encodeTypedPacket(packet, data)), - sendToClient: (clientId, packet, data, { channel = packet.channel } = {}) => - this.sendBytesToClient(clientId, channel, encodeTypedPacket(packet, data)), + sendToAll: (packet, ...[data, options]) => + this.sendBytesToAll(options?.channel ?? packet.channel, encodeTypedPacket(packet, data)), + sendToClient: (clientId, packet, ...[data, options]) => + this.sendBytesToClient( + clientId, + options?.channel ?? packet.channel, + encodeTypedPacket(packet, data), + ), }; /** Client sessions shared by the TCP and UDP servers. */ diff --git a/modules/network/src/shared/packet.ts b/modules/network/src/shared/packet.ts index 3dabc9f1..81412395 100644 --- a/modules/network/src/shared/packet.ts +++ b/modules/network/src/shared/packet.ts @@ -1,6 +1,6 @@ import type { Context } from "@nanoforge-dev/common"; -import { type Channel, DEFAULT_CHANNEL } from "./channels"; +import { type Channel, type ChannelOptions, DEFAULT_CHANNEL } from "./channels"; import { NetworkData } from "./network-data"; const encoder = new TextEncoder(); @@ -61,7 +61,7 @@ export interface PacketDefinition { * export const Move = definePacket<{ id: number; x: number; y: number }>("move", { * channel: Channel.UnreliableOrdered, * }); - * export const Play = definePacket("play"); // payload: null + * export const Play = definePacket("play"); // no payload: `send(Play)` * ``` * * @param name - Name identifying the type on the wire, unique in the game. @@ -69,7 +69,7 @@ export interface PacketDefinition { * @returns The packet definition, used by `packets.on`, `read` and `send`. * @throws When the name is empty or longer than 255 UTF-8 bytes. */ -export const definePacket = ( +export const definePacket = ( name: string, { channel = DEFAULT_CHANNEL, validate }: PacketOptions = {}, ): PacketDefinition => { @@ -98,14 +98,27 @@ export const encodePacket = (name: string | null, payload: Uint8Array): Uint8Arr }; /** - * Encode a typed packet: its payload as JSON, prefixed with its type name. + * Arguments of the typed send methods after the definition: the payload, which + * can be left out when it accepts `undefined` (a packet without payload is + * `void`), then the options. + * + * @typeParam T - Payload type of the packet. + */ +export type PacketArgs = undefined extends T + ? [data?: T, options?: ChannelOptions] + : [data: T, options?: ChannelOptions]; + +/** + * Encode a typed packet: its payload as JSON, or nothing for `undefined`, + * prefixed with its type name. * * @throws `TypeError` When the payload cannot be encoded as JSON. */ -export const encodeTypedPacket = ( - packet: PacketDefinition, - data: T, +export const encodeTypedPacket = ( + packet: PacketDefinition, + data: unknown, ): Uint8Array => { + if (data === undefined) return encodePacket(packet.name, new Uint8Array()); const json: string | undefined = JSON.stringify(data); if (json === undefined) throw new TypeError(`Cannot send a ${typeof data} as packet "${packet.name}"`); @@ -222,7 +235,7 @@ export class PacketDispatcher { for (const { name, data, info } of packets) { let value: unknown; try { - value = JSON.parse(decoder.decode(data)); + value = data.length === 0 ? undefined : JSON.parse(decoder.decode(data)); } catch (error) { console.warn(`Network: packet "${name}" is not valid JSON, dropped`, { cause: error }); continue; diff --git a/modules/network/test/client/client.network.library.spec.ts b/modules/network/test/client/client.network.library.spec.ts index f67d16e0..a7f26d10 100644 --- a/modules/network/test/client/client.network.library.spec.ts +++ b/modules/network/test/client/client.network.library.spec.ts @@ -399,13 +399,13 @@ describe("NetworkClientLibrary", () => { lib.packets.on(Play, (play, info) => calls.push([play, info])); receive(encodeTypedPacket(Move, { x: 1 }), Channel.UnreliableUnordered); - receive(encodeTypedPacket(Play, null), Channel.ReliableOrdered); + receive(encodeTypedPacket(Play, undefined), Channel.ReliableOrdered); tcpSocket().onmessage({ data: new Uint8Array([0, 0, 7]).buffer }); expect(calls).toStrictEqual([]); await lib.__run(ctx); expect(calls).toStrictEqual([ - [null, { channel: Channel.ReliableOrdered, ctx }], + [undefined, { channel: Channel.ReliableOrdered, ctx }], [{ x: 1 }, { channel: Channel.UnreliableUnordered, ctx }], ]); expect(lib.packets.read(Move)).toStrictEqual([ @@ -421,7 +421,7 @@ describe("NetworkClientLibrary", () => { const listener = vi.fn(); lib.packets.on(Play, listener); - receive(encodeTypedPacket(Play, null), Channel.ReliableOrdered); + receive(encodeTypedPacket(Play, undefined), Channel.ReliableOrdered); expect(lib.getReceivedPackets()).toStrictEqual([]); await lib.__run(ctx); expect(listener).toHaveBeenCalledOnce(); @@ -447,12 +447,12 @@ describe("NetworkClientLibrary", () => { await lib.__init(makeInitContext(tcpOnly)); await lib.connect(); const listener = vi.fn(() => { - receive(encodeTypedPacket(Play, null), Channel.ReliableOrdered); + receive(encodeTypedPacket(Play, undefined), Channel.ReliableOrdered); lib.getReceivedPackets(); }); lib.packets.on(Play, listener); - receive(encodeTypedPacket(Play, null), Channel.ReliableOrdered); + receive(encodeTypedPacket(Play, undefined), Channel.ReliableOrdered); await lib.__run(ctx); await lib.__run(ctx); expect(listener).toHaveBeenCalledTimes(2); @@ -464,12 +464,12 @@ describe("NetworkClientLibrary", () => { await lib.connect(); lib.packets.send(Move, { x: 1 }); - lib.packets.send(Play, null, { channel: Channel.ReliableUnordered }); + lib.packets.send(Play, undefined, { channel: Channel.ReliableUnordered }); expect(dataChannel("unreliable-unordered").send).toHaveBeenCalledWith( encodeTypedPacket(Move, { x: 1 }), ); expect(tcpSocket().send).toHaveBeenCalledWith( - new Uint8Array([1, ...encodeTypedPacket(Play, null)]), + new Uint8Array([1, ...encodeTypedPacket(Play, undefined)]), ); }); diff --git a/modules/network/test/shared/packet.spec-d.ts b/modules/network/test/shared/packet.spec-d.ts index fe98a315..5dafdfe0 100644 --- a/modules/network/test/shared/packet.spec-d.ts +++ b/modules/network/test/shared/packet.spec-d.ts @@ -2,7 +2,7 @@ import { describe, it } from "vitest"; // From the type files: both entries augment `Context.network`, so they cannot share a program. import type { ClientPacketsApi } from "../../src/client/network-client-context.type"; -import { definePacket } from "../../src/index"; +import { Channel, definePacket } from "../../src/index"; import type { ServerPacketsApi } from "../../src/server/network-server-context.type"; declare const client: ClientPacketsApi; @@ -14,7 +14,10 @@ const Play = definePacket("play"); describe("packets (type-level)", () => { it("accepts the payload of the definition", () => { client.send(Move, { x: 1 }); - client.send(Play, null); + client.send(Play); + client.send(Play, undefined, { channel: Channel.UnreliableOrdered }); + server.sendToClient("client-0", Play); + client.on(Play, () => {}); server.sendToAll(Move, { x: 1 }); server.sendToClient("client-0", Move, { x: 1 }); client.on(Move, (move) => move.x.toFixed()); @@ -27,8 +30,12 @@ describe("packets (type-level)", () => { client.send(Move, { x: "1" }); // @ts-expect-error x must be a number server.sendToAll(Move, { x: "1" }); - // @ts-expect-error Play carries null + // @ts-expect-error Play carries no payload server.sendToClient("client-0", Play, { x: 1 }); + // @ts-expect-error Move needs its payload + client.send(Move); + // @ts-expect-error Move needs its payload + server.sendToAll(Move); // @ts-expect-error the listener must take the definition's payload client.on(Move, (move: { y: number }) => move.y); }); diff --git a/modules/network/test/shared/packet.spec.ts b/modules/network/test/shared/packet.spec.ts index fd1456fd..612a2cc2 100644 --- a/modules/network/test/shared/packet.spec.ts +++ b/modules/network/test/shared/packet.spec.ts @@ -69,8 +69,15 @@ describe("packet header", () => { expect(decodePacket(new Uint8Array([4, 1, 2]))).toBeNull(); }); + it("should send a packet without payload as its name only", () => { + const packet = encodeTypedPacket(Play, undefined); + + expect(packet).toStrictEqual(new Uint8Array([4, ...utf8("play")])); + expect(decodePacket(packet)).toStrictEqual({ name: "play", data: new Uint8Array() }); + }); + it("should refuse a payload JSON cannot encode", () => { - expect(() => encodeTypedPacket(Play as never, undefined as never)).toThrow(TypeError); + expect(() => encodeTypedPacket(Play, () => {})).toThrow(TypeError); }); it("should sort raw and typed packets, dropping broken ones", () => { @@ -78,13 +85,17 @@ describe("packet header", () => { const typed: TypedPacket[] = []; sortPackets( - [encodePacket(null, new Uint8Array([1])), new Uint8Array(), encodeTypedPacket(Play, null)], + [ + encodePacket(null, new Uint8Array([1])), + new Uint8Array(), + encodeTypedPacket(Play, undefined), + ], info, raw, typed, ); expect(raw).toStrictEqual([new NetworkData(new Uint8Array([1]))]); - expect(typed).toStrictEqual([{ name: "play", data: utf8("null"), info }]); + expect(typed).toStrictEqual([{ name: "play", data: new Uint8Array(), info }]); }); }); @@ -116,6 +127,15 @@ describe("PacketDispatcher", () => { ]); }); + it("should hand undefined for a packet without payload", () => { + const dispatcher = new PacketDispatcher(); + const listener = vi.fn(); + dispatcher.on(Play, listener); + + dispatcher.dispatch([{ name: "play", data: new Uint8Array(), info }], ctx); + expect(listener).toHaveBeenCalledWith(undefined, { ...info, ctx }); + }); + it("should stop calling a removed listener", () => { const dispatcher = new PacketDispatcher(); const listener = vi.fn(); From 32b7fc8a835a25d4c56907128169a61bd0977e4f Mon Sep 17 00:00:00 2001 From: Leo Date: Sat, 10 Oct 2026 02:01:03 +0200 Subject: [PATCH 3/3] fix(network): move connect in init, openning all for chans --- docs/docs/network/network-client-api.mdx | 27 +--- docs/docs/network/network-client.mdx | 3 - example/pong-network/apps/client/src/main.ts | 5 +- modules/network/README.md | 11 +- .../src/client/client.network.library.ts | 126 +++++++----------- modules/network/src/client/index.ts | 2 +- .../src/client/network-client-context.type.ts | 7 +- modules/network/src/shared/channels.ts | 23 +--- .../client/client.network.library.spec.ts | 90 ++++--------- 9 files changed, 91 insertions(+), 203 deletions(-) diff --git a/docs/docs/network/network-client-api.mdx b/docs/docs/network/network-client-api.mdx index b8b9d7d7..eaff3a77 100644 --- a/docs/docs/network/network-client-api.mdx +++ b/docs/docs/network/network-client-api.mdx @@ -14,35 +14,20 @@ description: "API reference for the client networking package" Reliable channels need `SERVER_TCP_PORT` and unreliable ones `SERVER_UDP_PORT`. Using a channel whose port is not configured throws. +The client connects during `__init`, so its channels are open once the game runs. +TCP connects first, then UDP with the TCP session token, so both share one client +id. `__init` rejects when a transport fails, or when its channels are not open +within 10 seconds. + ```ts import { Channel } from "@nanoforge-dev/network/client"; -await ctx.network.connect({ channels: [Channel.ReliableOrdered, Channel.UnreliableOrdered] }); - ctx.network.sendData({ type: "join" }); // Channel.ReliableOrdered ctx.network.sendData({ x: 1, y: 2 }, { channel: Channel.UnreliableOrdered }); const messages = ctx.network.getReceivedPackets().map((packet) => packet.json()); // every channel ``` -## connect(options?) - -Connect to the server and resolve once the channels are open. Only the -transports the channels need are opened: TCP for reliable channels, UDP for -unreliable ones. TCP connects first, then UDP with the TCP session token, so both -share one client id. Calling it again for channels that are already open resolves -right away. - -Option `channels`: `Channel[]`, the channels to open and wait for. Every -configured channel by default. - -Option `timeout`: `number`, milliseconds before giving up. `10000` by default. - -Rejects when a channel's port is not configured, when a transport fails to -connect, or when the channels are not open before the timeout. - -Returns: `Promise` - ## sendData(data, options?) Send a payload to the server. @@ -140,7 +125,7 @@ Returns: `boolean` ## clientId Client id assigned by the server, shared by TCP and UDP. `undefined` until -`connect` has been welcomed by the server. It is only kept in memory: a page +the server has welcomed the client. It is only kept in memory: a page reload gets a new id. Type: `string | undefined` diff --git a/docs/docs/network/network-client.mdx b/docs/docs/network/network-client.mdx index 48991520..38926f8d 100644 --- a/docs/docs/network/network-client.mdx +++ b/docs/docs/network/network-client.mdx @@ -23,9 +23,6 @@ four [channels](packet-framing), picked with the `channel` option: ```javascript import { Channel } from "@nanoforge-dev/network/client"; -// Open the channels the game uses and wait until they are ready -await network.connect({ channels: [Channel.ReliableOrdered, Channel.UnreliableOrdered] }); - // Send a value as JSON, on Channel.ReliableOrdered network.sendData({ hello: "world" }); diff --git a/example/pong-network/apps/client/src/main.ts b/example/pong-network/apps/client/src/main.ts index 2b4ef183..3f71c251 100644 --- a/example/pong-network/apps/client/src/main.ts +++ b/example/pong-network/apps/client/src/main.ts @@ -3,7 +3,7 @@ import { NanoforgeFactory } from "@nanoforge-dev/core"; import { EcsLibrary } from "@nanoforge-dev/ecs/client"; import { Circle, Graphics2DLibrary, Layer, Rect } from "@nanoforge-dev/graphics-2d"; import { InputEnum, InputLibrary } from "@nanoforge-dev/input"; -import { Channel, NetworkClientLibrary } from "@nanoforge-dev/network/client"; +import { NetworkClientLibrary } from "@nanoforge-dev/network/client"; import { AssignId, Play } from "../../../shared/protocol"; import { @@ -95,9 +95,6 @@ export const main = async (options: ClientRunOptions): Promise => { registry.addSystem(controlPlayer); registry.addSystem(draw); - await network.connect({ - channels: [Channel.ReliableOrdered, Channel.UnreliableOrdered, Channel.UnreliableUnordered], - }); network.packets.send(Play); await app.run(); diff --git a/modules/network/README.md b/modules/network/README.md index 6c9a46e2..358bb16d 100644 --- a/modules/network/README.md +++ b/modules/network/README.md @@ -164,9 +164,6 @@ import { Channel, NetworkClientLibrary } from "@nanoforge-dev/network/client"; app.use(new NetworkClientLibrary()); -// Once, at game start: open the channels the game uses and wait for them. -await ctx.network.connect({ channels: [Channel.ReliableOrdered, Channel.UnreliableOrdered] }); - // Every frame: ctx.network.sendData({ type: "join" }); // Channel.ReliableOrdered ctx.network.sendData({ x: 1, y: 2 }, { channel: Channel.UnreliableOrdered }); @@ -189,9 +186,9 @@ ctx.network.sendToAll(snapshot, { channel: Channel.UnreliableOrdered }); ctx.network.sendToClient(clientId, { type: "event" }); // Channel.ReliableOrdered ``` -`connect` only opens the transports its channels need, and rejects when a -transport fails or the channels are not open within `timeout` milliseconds -(10 seconds by default). The server starts listening during `__init`. +The client connects during `__init` and its channels are open before the game +starts: `__init` rejects when a transport fails or the channels are not open +within 10 seconds. The server starts listening during `__init`. ### Typed packets @@ -239,7 +236,7 @@ ctx.network.clients.onDisconnect((info) => console.log(`${info.id} left`)); const info = ctx.network.getClientInfo(clientId); ``` -On the client, `ctx.network.clientId` holds the id once `connect` has been welcomed. A page reload +On the client, `ctx.network.clientId` holds the id once the server has welcomed the client. A page reload starts a new session with a new id. Using a channel whose port wasn't configured on that side throws. diff --git a/modules/network/src/client/client.network.library.ts b/modules/network/src/client/client.network.library.ts index ed9e33ca..71aea663 100644 --- a/modules/network/src/client/client.network.library.ts +++ b/modules/network/src/client/client.network.library.ts @@ -7,7 +7,6 @@ import { CONNECT_TIMEOUT_MS, type Channel, type ChannelOptions, - type ConnectOptions, DEFAULT_CHANNEL, isReliableChannel, } from "../shared/channels"; @@ -34,7 +33,7 @@ import type { import { TCPClient } from "./tcp.client.network"; import { UDPClient } from "./udp.client.network"; -/** Delay between two checks of the channels while `connect` waits. */ +/** Delay between two checks of the channels while `__init` waits for them. */ const CONNECT_POLL_MS = 50; /** @@ -42,10 +41,10 @@ const CONNECT_POLL_MS = 50; * * @remarks * Reads network configuration from the environment via `ClientConfigNetwork` - * during `__init`. Call `connect` to open the connections: TCP (WebSocket) - * carries the reliable channels and UDP (WebRTC data channels) the unreliable - * ones. TCP connects first; once welcomed, UDP presents the session token so - * both transports share one client id. + * and opens the connections during `__init`, so the channels are open before + * the game runs: TCP (WebSocket) carries the reliable channels and UDP (WebRTC + * data channels) the unreliable ones. TCP connects first; once welcomed, UDP + * presents the session token so both transports share one client id. * * Configuration (via environment variables): * - `SERVER_ADDRESS` — hostname or IP of the server (required). @@ -56,7 +55,6 @@ const CONNECT_POLL_MS = 50; * * @example * ```ts - * await ctx.network.connect({ channels: [Channel.ReliableOrdered, Channel.UnreliableOrdered] }); * ctx.network.sendData({ type: "join" }); // Channel.ReliableOrdered * ctx.network.sendData({ x: 1, y: 2 }, { channel: Channel.UnreliableOrdered }); * const messages = ctx.network.getReceivedPackets().map((packet) => packet.json()); // every channel @@ -68,8 +66,6 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary private _tcp?: TCPClient; private _udp?: UDPClient; - /** Pending or settled connection of each transport, so it is only opened once. */ - private readonly _connecting = new Map>(); private readonly _session: ClientSession = {}; private readonly _dispatcher = new PacketDispatcher(); /** Typed packets drained from the transports, dispatched by the next `__run`. */ @@ -125,60 +121,8 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary this._session, ); } - } - /** - * Connect to the server and wait until `channels` are open. - * - * @remarks - * Only opens the transports the channels need: TCP for reliable channels, - * UDP for unreliable ones. Calling it again for already open channels - * resolves right away. - * - * @param options - `channels` to wait for (every configured channel by - * default) and a `timeout` in milliseconds (10 seconds by default). - * @throws When a channel's port was not configured, a transport fails to - * connect, or the channels are not open before the timeout. - */ - public async connect({ - channels = this.configuredChannels(), - timeout = CONNECT_TIMEOUT_MS, - }: ConnectOptions = {}): Promise { - if (!this._tcp && !this._udp) { - throw new Error("NetworkClientLibrary: connect() was called before __init"); - } - const deadline = Date.now() + timeout; - const reliable = channels.find(isReliableChannel); - const unreliable = channels.find((channel) => !isReliableChannel(channel)); - - // TCP first, so UDP joins its session with the token from the TCP welcome. - const transports: [string, TCPClient | UDPClient][] = []; - if (reliable !== undefined) transports.push(["TCP", this.requireTcp(reliable)]); - if (unreliable !== undefined) transports.push(["UDP", this.requireUdp(unreliable)]); - - const failures: unknown[] = []; - for (const [name, transport] of transports) { - try { - await this.connectTransport(name, transport); - } catch (error) { - failures.push(error); - } - } - if (failures.length > 0) { - throw new Error("NetworkClientLibrary: could not connect to the server", { - cause: failures.length === 1 ? failures[0] : failures, - }); - } - - while (!channels.every((channel) => this.isConnected({ channel }))) { - if (Date.now() >= deadline) { - const pending = channels.filter((channel) => !this.isConnected({ channel })); - throw new Error( - `NetworkClientLibrary: channels ${pending.join(", ")} not open after ${timeout}ms`, - ); - } - await new Promise((resolve) => setTimeout(resolve, CONNECT_POLL_MS)); - } + await this.openChannels(); } /** @@ -259,7 +203,6 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary // eslint-disable-next-line @typescript-eslint/no-this-alias const library = this; return { - connect: (options) => library.connect(options), sendData: (data, options) => library.sendData(data, options), getReceivedPackets: (options) => library.getReceivedPackets(options), isConnected: (options) => library.isConnected(options), @@ -272,6 +215,50 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary }; } + /** + * Connect every configured transport and wait until its channels are open. + * + * @remarks + * Channels cannot be opened once the game runs, and connection errors are + * best caught at start, so this runs during `__init`. + * + * @throws When a transport fails to connect, or the channels are not open + * within `CONNECT_TIMEOUT_MS`. + */ + private async openChannels(): Promise { + const deadline = Date.now() + CONNECT_TIMEOUT_MS; + const channels = this.configuredChannels(); + + // TCP first, so UDP joins its session with the token from the TCP welcome. + const transports: [string, TCPClient | UDPClient][] = []; + if (this._tcp) transports.push(["TCP", this._tcp]); + if (this._udp) transports.push(["UDP", this._udp]); + + const failures: unknown[] = []; + for (const [name, transport] of transports) { + try { + await transport.connect(); + } catch (error) { + failures.push(new Error(`${name} connection failed`, { cause: error })); + } + } + if (failures.length > 0) { + throw new Error("NetworkClientLibrary: could not connect to the server", { + cause: failures.length === 1 ? failures[0] : failures, + }); + } + + while (!channels.every((channel) => this.isConnected({ channel }))) { + if (Date.now() >= deadline) { + const pending = channels.filter((channel) => !this.isConnected({ channel })); + throw new Error( + `NetworkClientLibrary: channels ${pending.join(", ")} not open after ${CONNECT_TIMEOUT_MS}ms`, + ); + } + await new Promise((resolve) => setTimeout(resolve, CONNECT_POLL_MS)); + } + } + private configuredChannels(): Channel[] { return CHANNELS.filter((channel) => isReliableChannel(channel) ? this._tcp !== undefined : this._udp !== undefined, @@ -301,17 +288,4 @@ export class NetworkClientLibrary extends Library implements EditorAwareLibrary if (!this._udp) throw new Error(`Channel ${channel} needs SERVER_UDP_PORT to be set`); return this._udp; } - - private connectTransport(name: string, transport: TCPClient | UDPClient): Promise { - let connecting = this._connecting.get(transport); - if (!connecting) { - connecting = transport.connect().catch((error: unknown) => { - // Forget the failure so a later `connect` can retry. - this._connecting.delete(transport); - throw new Error(`${name} connection failed`, { cause: error }); - }); - this._connecting.set(transport, connecting); - } - return connecting; - } } diff --git a/modules/network/src/client/index.ts b/modules/network/src/client/index.ts index e4f7f13e..9ad6f534 100644 --- a/modules/network/src/client/index.ts +++ b/modules/network/src/client/index.ts @@ -1,7 +1,7 @@ import "./context-augmentation"; export type { NetworkConfig } from "../shared/config.network"; -export { Channel, type ChannelOptions, type ConnectOptions } from "../shared/channels"; +export { Channel, type ChannelOptions } from "../shared/channels"; export { NetworkData, type NetworkPayload } from "../shared/network-data"; export { type PacketArgs, diff --git a/modules/network/src/client/network-client-context.type.ts b/modules/network/src/client/network-client-context.type.ts index 9d210d3f..68b660ea 100644 --- a/modules/network/src/client/network-client-context.type.ts +++ b/modules/network/src/client/network-client-context.type.ts @@ -1,4 +1,4 @@ -import type { Channel, ChannelOptions, ConnectOptions } from "../shared/channels"; +import type { Channel, ChannelOptions } from "../shared/channels"; import type { NetworkData, NetworkPayload } from "../shared/network-data"; import type { PacketArgs, PacketDefinition, PacketListener } from "../shared/packet"; @@ -59,11 +59,6 @@ export interface NetworkClientContextApi { readonly packets: ClientPacketsApi; /** Client id assigned by the server, shared by TCP and UDP. `undefined` until welcomed. */ readonly clientId: string | undefined; - /** - * Connect to the server and resolve once `channels` (every configured - * channel by default) are open. Rejects on failure or after `timeout` ms. - */ - connect(options?: ConnectOptions): Promise; /** * Send bytes, a string or a JSON value to the server, on * `Channel.ReliableOrdered` unless another `channel` is given. diff --git a/modules/network/src/shared/channels.ts b/modules/network/src/shared/channels.ts index fe8dffd6..df91d086 100644 --- a/modules/network/src/shared/channels.ts +++ b/modules/network/src/shared/channels.ts @@ -52,30 +52,9 @@ export type ChannelOptions = { channel?: Channel; }; -/** How long `connect` waits for its channels by default. */ +/** How long the client waits for its channels to open during `__init`. */ export const CONNECT_TIMEOUT_MS = 10_000; -/** - * Options of the client's `connect`. - * - * @example - * ```ts - * await ctx.network.connect({ - * channels: [Channel.ReliableOrdered, Channel.UnreliableUnordered], - * }); - * ``` - */ -export type ConnectOptions = { - /** Channels to open and wait for, every configured channel by default. */ - channels?: Channel[]; - /** - * Give up after this many milliseconds. - * - * @default 10000 - */ - timeout?: number; -}; - /** * Tell whether a channel is carried by the TCP WebSocket. * diff --git a/modules/network/test/client/client.network.library.spec.ts b/modules/network/test/client/client.network.library.spec.ts index a7f26d10..5aa9b1d8 100644 --- a/modules/network/test/client/client.network.library.spec.ts +++ b/modules/network/test/client/client.network.library.spec.ts @@ -83,6 +83,7 @@ describe("NetworkClientLibrary", () => { }); afterEach(() => { + vi.useRealTimers(); vi.unstubAllGlobals(); vi.restoreAllMocks(); }); @@ -108,7 +109,6 @@ describe("NetworkClientLibrary", () => { }); const lib = new NetworkClientLibrary(); await lib.__init(ctx); - await lib.connect(); expect(() => lib.sendData(new Uint8Array([1]))).not.toThrow(); expect(() => lib.sendData(new Uint8Array([1]), { channel: Channel.UnreliableOrdered }), @@ -123,7 +123,6 @@ describe("NetworkClientLibrary", () => { }); const lib = new NetworkClientLibrary(); await lib.__init(ctx); - await lib.connect(); expect(RTCPeerConnection).toHaveBeenCalledWith({ iceServers: [{ urls: "stun:stun.example.com:3478" }], }); @@ -136,7 +135,6 @@ describe("NetworkClientLibrary", () => { }); const lib = new NetworkClientLibrary(); await lib.__init(ctx); - await lib.connect(); expect(() => lib.sendData(new Uint8Array([1]), { channel: Channel.UnreliableUnordered }), ).not.toThrow(); @@ -153,7 +151,6 @@ describe("NetworkClientLibrary", () => { }); const lib = new NetworkClientLibrary(); await lib.__init(ctx); - await lib.connect(); expect(() => lib.sendData(new Uint8Array([1]))).not.toThrow(); expect(() => lib.sendData(new Uint8Array([1]), { channel: Channel.UnreliableUnordered }), @@ -168,7 +165,6 @@ describe("NetworkClientLibrary", () => { }); const lib = new NetworkClientLibrary(); await lib.__init(ctx); - await lib.connect(); expect(vi.mocked(WebSocket).mock.calls.map(([url]) => url)).toStrictEqual([ "ws://127.0.0.1:8080", @@ -185,9 +181,8 @@ describe("NetworkClientLibrary", () => { SERVER_ADDRESS: "127.0.0.1", }); const lib = new NetworkClientLibrary(); - await lib.__init(ctx); - await expect(lib.connect()).rejects.toThrow( + await expect(lib.__init(ctx)).rejects.toThrow( "NetworkClientLibrary: could not connect to the server", ); expect(vi.mocked(WebSocket).mock.calls[1]?.[0]).toBe("ws://127.0.0.1:8081"); @@ -201,69 +196,52 @@ describe("NetworkClientLibrary", () => { }); }); - describe("connect", () => { - it("should not open any connection during __init", async () => { - await new NetworkClientLibrary().__init(makeInitContext(both)); - expect(WebSocket).not.toHaveBeenCalled(); + describe("connection", () => { + it("should open only the configured transports during __init", async () => { + await new NetworkClientLibrary().__init(makeInitContext(udpOnly)); + + expect(vi.mocked(WebSocket).mock.calls.map(([url]) => url)).toStrictEqual([ + "ws://127.0.0.1:8081", + ]); }); - it("should only open the transports the channels need", async () => { + it("should open all four channels by default when both ports are configured", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(both)); - await lib.connect({ channels: [Channel.UnreliableUnordered] }); - expect(vi.mocked(WebSocket).mock.calls.map(([url]) => url)).toStrictEqual([ - "ws://127.0.0.1:8081", + expect(Object.values(Channel).map((channel) => lib.isConnected({ channel }))).toStrictEqual([ + true, + true, + true, + true, ]); }); it("should wait until every channel is open", async () => { dataChannelState = "connecting"; const lib = new NetworkClientLibrary(); - await lib.__init(makeInitContext(udpOnly)); - let connected = false; - const connecting = lib - .connect({ channels: [Channel.UnreliableOrdered] }) - .then(() => (connected = true)); + let initialized = false; + const init = lib.__init(makeInitContext(udpOnly)).then(() => (initialized = true)); await new Promise((resolve) => setTimeout(resolve, 100)); - expect(connected).toBe(false); + expect(initialized).toBe(false); dataChannel("unreliable-ordered").readyState = "open"; - await connecting; - expect(connected).toBe(true); + dataChannel("unreliable-unordered").readyState = "open"; + await init; + expect(initialized).toBe(true); }); it("should reject when the channels are not open in time", async () => { dataChannelState = "connecting"; - const lib = new NetworkClientLibrary(); - await lib.__init(makeInitContext(udpOnly)); + vi.useFakeTimers(); - await expect(lib.connect({ timeout: 100 })).rejects.toThrow( - "channels unreliable-ordered, unreliable-unordered not open after 100ms", + const init = new NetworkClientLibrary().__init(makeInitContext(udpOnly)); + const rejection = expect(init).rejects.toThrow( + "channels unreliable-ordered, unreliable-unordered not open after 10000ms", ); - }); - - it("should not reconnect a transport that is already open", async () => { - const lib = new NetworkClientLibrary(); - await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); - await lib.connect({ channels: [Channel.ReliableUnordered] }); - - expect(WebSocket).toHaveBeenCalledTimes(1); - }); - - it("should reject a channel whose port is not configured", async () => { - const lib = new NetworkClientLibrary(); - await lib.__init(makeInitContext(tcpOnly)); - - await expect(lib.connect({ channels: [Channel.UnreliableOrdered] })).rejects.toThrow( - "Channel unreliable-ordered needs SERVER_UDP_PORT to be set", - ); - }); - - it("should reject when called before __init", async () => { - await expect(new NetworkClientLibrary().connect()).rejects.toThrow("before __init"); + await vi.advanceTimersByTimeAsync(10_000); + await rejection; }); }); @@ -271,7 +249,6 @@ describe("NetworkClientLibrary", () => { it("should send on ReliableOrdered when no channel is given", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); lib.sendData(new Uint8Array([1])); expect(tcpSocket().send).toHaveBeenCalledWith(new Uint8Array([0, 0, 1])); @@ -280,7 +257,6 @@ describe("NetworkClientLibrary", () => { it("should send each reliable channel over the same TCP socket", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); lib.sendData(new Uint8Array([1]), { channel: Channel.ReliableOrdered }); lib.sendData(new Uint8Array([2]), { channel: Channel.ReliableUnordered }); @@ -294,7 +270,6 @@ describe("NetworkClientLibrary", () => { it("should send strings as UTF-8 and other values as JSON", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(both)); - await lib.connect(); lib.sendData({ type: "play" }); lib.sendData("hi", { channel: Channel.UnreliableUnordered }); @@ -310,7 +285,6 @@ describe("NetworkClientLibrary", () => { it("should open one data channel per unreliable channel", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(udpOnly)); - await lib.connect(); const peerConnection = vi.mocked(RTCPeerConnection).mock.instances[0] as any; expect( @@ -321,7 +295,6 @@ describe("NetworkClientLibrary", () => { it("should send unreliable channels on their own data channel", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(udpOnly)); - await lib.connect(); lib.sendData(new Uint8Array([3]), { channel: Channel.UnreliableUnordered }); expect(dataChannel("unreliable-unordered").send).toHaveBeenCalledWith(new Uint8Array([0, 3])); @@ -331,7 +304,6 @@ describe("NetworkClientLibrary", () => { it("should read one channel, or every channel with reliable ones first", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(both)); - await lib.connect(); dataChannel("unreliable-unordered").onmessage({ data: new Uint8Array([0, 9]).buffer }); tcpSocket().onmessage({ data: new Uint8Array([1, 0, 2]).buffer }); @@ -350,7 +322,6 @@ describe("NetworkClientLibrary", () => { it("should return packets that read back as JSON", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); tcpSocket().onmessage({ data: new Uint8Array([0, 0, ...utf8('{"type":"move"}')]).buffer }); @@ -362,7 +333,6 @@ describe("NetworkClientLibrary", () => { it("should report connected only once every configured channel is", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(both)); - await lib.connect(); tcpSocket().readyState = 1; dataChannel("unreliable-ordered").readyState = "connecting"; @@ -393,7 +363,6 @@ describe("NetworkClientLibrary", () => { it("should dispatch typed packets in __run, reliable ones first", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(both)); - await lib.connect(); const calls: unknown[] = []; lib.packets.on(Move, (move, info) => calls.push([move, info])); lib.packets.on(Play, (play, info) => calls.push([play, info])); @@ -417,7 +386,6 @@ describe("NetworkClientLibrary", () => { it("should keep typed packets out of getReceivedPackets until __run", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); const listener = vi.fn(); lib.packets.on(Play, listener); @@ -430,7 +398,6 @@ describe("NetworkClientLibrary", () => { it("should drop sendData packets left unread for a whole tick", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); tcpSocket().onmessage({ data: new Uint8Array([0, 0, 1]).buffer }); await lib.__run(ctx); @@ -445,7 +412,6 @@ describe("NetworkClientLibrary", () => { it("should not lose packets drained by a listener", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); const listener = vi.fn(() => { receive(encodeTypedPacket(Play, undefined), Channel.ReliableOrdered); lib.getReceivedPackets(); @@ -461,7 +427,6 @@ describe("NetworkClientLibrary", () => { it("should send on the definition's channel unless another is given", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(both)); - await lib.connect(); lib.packets.send(Move, { x: 1 }); lib.packets.send(Play, undefined, { channel: Channel.ReliableUnordered }); @@ -482,7 +447,6 @@ describe("NetworkClientLibrary", () => { it("delegates to the library", async () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); const api = lib.expose(); api.sendData(new Uint8Array([5]), { channel: Channel.ReliableUnordered });