diff --git a/docs/docs/network/network-client-api.mdx b/docs/docs/network/network-client-api.mdx index 86542c7..eaff3a7 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. @@ -59,13 +44,75 @@ 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. 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. + +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`. Without a type, `T` is `void`: the packet +carries no payload. + +```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. @@ -78,15 +125,15 @@ 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` ## 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 310e3ef..38926f8 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" }); @@ -43,6 +40,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"); // 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 +}); +``` + +```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); // no payload to give +``` + +- 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 b4a7d8f..a0834c7 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,47 @@ 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. 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?) + +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 a1185c8..6350fc4 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 33a4cec..d3dce2a 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,27 @@ 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, + 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 + 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 +61,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 f7fcdc6..3f71c25 100644 --- a/example/pong-network/apps/client/src/main.ts +++ b/example/pong-network/apps/client/src/main.ts @@ -3,16 +3,18 @@ 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 { 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,16 +84,18 @@ 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); - // Wait for every channel the game uses, so no early move is missed. - await network.connect({ - channels: [Channel.ReliableOrdered, Channel.UnreliableOrdered, Channel.UnreliableUnordered], - }); - network.sendData({ type: "play" }); + network.packets.send(Play); 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 2fdcace..9ae87e9 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 9b8e599..62ec77e 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 dbcb5dd..0f2c2ac 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 28742f0..47eaf1a 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 0000000..555aa57 --- /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 df7fa4c..358bb16 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 | | -------------------------------------------------- | --------------------- | --------------------------------- | @@ -162,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 }); @@ -187,9 +186,39 @@ 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 + +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 @@ -207,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 53655a2..71aea66 100644 --- a/modules/network/src/client/client.network.library.ts +++ b/modules/network/src/client/client.network.library.ts @@ -7,19 +7,33 @@ import { CONNECT_TIMEOUT_MS, type Channel, type ChannelOptions, - type ConnectOptions, 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"; -/** 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; /** @@ -27,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). @@ -41,10 +55,10 @@ 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 + * ctx.network.packets.send(Input, { key: "up" }); // typed packet, see `definePacket` * ``` */ export class NetworkClientLibrary extends Library implements EditorAwareLibrary { @@ -52,9 +66,18 @@ 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`. */ + 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, options]) => + this.sendBytes(options?.channel ?? packet.channel, encodeTypedPacket(packet, data)), + }; /** Client id assigned by the server, `undefined` until transport is welcomed. */ public get clientId(): string | undefined { @@ -63,6 +86,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); @@ -88,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(); } /** @@ -152,17 +133,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 +152,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,26 +190,95 @@ 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; return { - connect: (options) => library.connect(options), 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; }, }; } + /** + * 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, ); } + 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; @@ -239,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 9e05745..9ad6f53 100644 --- a/modules/network/src/client/index.ts +++ b/modules/network/src/client/index.ts @@ -1,8 +1,19 @@ 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, + 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 f6e21a2..68b660e 100644 --- a/modules/network/src/client/network-client-context.type.ts +++ b/modules/network/src/client/network-client-context.type.ts @@ -1,5 +1,50 @@ -import type { 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"; + +/** 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. `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, ...args: PacketArgs>): void; +} /** * Public surface of `NetworkClientLibrary`, exposed on `Context.network`. @@ -10,13 +55,10 @@ 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; - /** - * 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. @@ -24,7 +66,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 34317cf..629936e 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 00f0e92..7c365e1 100644 --- a/modules/network/src/server/index.ts +++ b/modules/network/src/server/index.ts @@ -3,6 +3,13 @@ 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 PacketArgs, + type PacketDefinition, + type PacketListener, + type PacketOptions, + definePacket, +} from "../shared/packet"; export type { ClientId, ClientInfo, @@ -12,6 +19,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 6305026..446e119 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 { PacketArgs, 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. `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, ...args: PacketArgs>): 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, + ...args: PacketArgs> + ): 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 1b107d7..6a1783e 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,21 @@ 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, 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. */ public get clients(): ClientsApi { @@ -58,6 +89,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 +165,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 +182,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 +200,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 +264,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 +285,9 @@ export class NetworkServerLibrary extends Library implements EditorAwareLibrary get clients() { return library.clients; }, + get packets() { + return library.packets; + }, }; } @@ -249,6 +297,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 649243f..df91d08 100644 --- a/modules/network/src/shared/channels.ts +++ b/modules/network/src/shared/channels.ts @@ -44,37 +44,17 @@ 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; }; -/** 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/src/shared/packet.ts b/modules/network/src/shared/packet.ts new file mode 100644 index 0000000..8141239 --- /dev/null +++ b/modules/network/src/shared/packet.ts @@ -0,0 +1,288 @@ +import type { Context } from "@nanoforge-dev/common"; + +import { type Channel, type ChannelOptions, 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"); // no payload: `send(Play)` + * ``` + * + * @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; +}; + +/** + * 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: 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}"`); + 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 = data.length === 0 ? undefined : 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 c268737..5aa9b1d 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: () => {} }, @@ -82,6 +83,7 @@ describe("NetworkClientLibrary", () => { }); afterEach(() => { + vi.useRealTimers(); vi.unstubAllGlobals(); vi.restoreAllMocks(); }); @@ -107,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 }), @@ -122,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" }], }); @@ -135,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(); @@ -152,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 }), @@ -167,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", @@ -184,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"); @@ -200,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; }); }); @@ -270,44 +249,42 @@ 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, 1])); + expect(tcpSocket().send).toHaveBeenCalledWith(new Uint8Array([0, 0, 1])); }); 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 }); expect(tcpSocket().send.mock.calls).toStrictEqual([ - [new Uint8Array([0, 1])], - [new Uint8Array([1, 2])], + [new Uint8Array([0, 0, 1])], + [new Uint8Array([1, 0, 2])], ]); }); 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 }); 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 () => { const lib = new NetworkClientLibrary(); await lib.__init(makeInitContext(udpOnly)); - await lib.connect(); const peerConnection = vi.mocked(RTCPeerConnection).mock.instances[0] as any; expect( @@ -318,21 +295,19 @@ 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([3])); + expect(dataChannel("unreliable-unordered").send).toHaveBeenCalledWith(new Uint8Array([0, 3])); expect(dataChannel("unreliable-ordered").send).not.toHaveBeenCalled(); }); 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([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])], @@ -347,9 +322,8 @@ 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, ...utf8('{"type":"move"}')]).buffer }); + tcpSocket().onmessage({ data: new Uint8Array([0, 0, ...utf8('{"type":"move"}')]).buffer }); const [packet] = lib.getReceivedPackets(); expect(packet).toBeInstanceOf(NetworkData); @@ -359,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"; @@ -376,20 +349,114 @@ 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)); + 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, undefined), Channel.ReliableOrdered); + tcpSocket().onmessage({ data: new Uint8Array([0, 0, 7]).buffer }); + expect(calls).toStrictEqual([]); + + await lib.__run(ctx); + expect(calls).toStrictEqual([ + [undefined, { 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)); + const listener = vi.fn(); + lib.packets.on(Play, listener); + + receive(encodeTypedPacket(Play, undefined), 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)); + + 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)); + const listener = vi.fn(() => { + receive(encodeTypedPacket(Play, undefined), Channel.ReliableOrdered); + lib.getReceivedPackets(); + }); + lib.packets.on(Play, listener); + + receive(encodeTypedPacket(Play, undefined), 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)); + + lib.packets.send(Move, { x: 1 }); + 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, undefined)]), + ); + }); + + 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(); await lib.__init(makeInitContext(tcpOnly)); - await lib.connect(); 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 4bbb895..3d5e5ef 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 0000000..5dafdfe --- /dev/null +++ b/modules/network/test/shared/packet.spec-d.ts @@ -0,0 +1,42 @@ +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 { Channel, 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); + 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()); + 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 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 new file mode 100644 index 0000000..612a2cc --- /dev/null +++ b/modules/network/test/shared/packet.spec.ts @@ -0,0 +1,238 @@ +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 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, () => {})).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, undefined), + ], + info, + raw, + typed, + ); + expect(raw).toStrictEqual([new NetworkData(new Uint8Array([1]))]); + expect(typed).toStrictEqual([{ name: "play", data: new Uint8Array(), 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 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(); + 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 8270cab..c62afd5 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 7d13083..0fb6359 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)