Replace the stub loopback server with a local, but multi-client server.

src/Game/Main.ts

@@ -7,7 +7,7 @@ import { Location, Polygon, PolygonComponent, RenderBounds } from "../Ecs/Compon
 import { Create } from "../Ecs/Data";
 import { RunRenderBounds } from "../Ecs/Renderers";
 import { LockstepClient } from "../Net/LockstepClient";
-import { Loopback } from "../Net/LoopbackServer";
+import { LoopbackServer } from "../Net/LoopbackServer";
 import { Data, Engine, PlayerControl } from "./GameComponents";
 import { Buttons } from "./Input";
 
@@ -19,7 +19,9 @@ export class Main extends LockstepClient<KeyName[], KeyName[][], Data> {
         super(new Engine());
         keys.setHandler(this.buttons);
 
-        this.connect(Loopback);
+        const server = new LoopbackServer<KeyName[], Data>();
+        this.connect(server.socket);
+        server.resetState({});
 
         pipe(
             this.renderFrames,

src/Net/LoopbackServer.ts

@@ -1,28 +1,57 @@
 import { Callbag } from "callbag";
+import pipe from "callbag-pipe";
+import share from "callbag-share";
 
-import { ClientMessage, MessageTypes, ServerMessage } from "./LockstepClient";
+import { INPUT_FREQUENCY } from "../Ecs/Lockstep";
+import { catchTalkback, defer, interval, makeSubject, map, merge } from "../Utilities/Callbag";
+import { ClientMessage, MessageTypes, Server, ServerMessage } from "./LockstepClient";
 
 type Client<LocalInput, State> = Callbag<ServerMessage<LocalInput[], State>, ClientMessage<LocalInput, State>>;
 
-/** Stub loopback server that handles a single client, for schemes where GlobalInput = LocalInput[] */
+/** Stub loopback server that handles multiple clients, for schemes where GlobalInput = LocalInput[] */
-export function Loopback<LocalInput, State>(start: number, data?: Client<LocalInput, State> | ClientMessage<LocalInput, State>) {
+export class LoopbackServer<LocalInput, State> {
-    if(start != 0) return;
 
-    const sink = data as Client<LocalInput, State>;
+    private nextClientId = 0;
+    private inputBuffer: LocalInput[] = [];
 
-    sink(0, (type: number, data?: Client<LocalInput, State> | ClientMessage<LocalInput, State>) => {
+    private heartbeat = pipe(
-        if(type == 1) {
+        interval(INPUT_FREQUENCY),
-            // message from client; just reflect for now
+        map(() => ({
-            const message = data as ClientMessage<LocalInput, State>;
-            switch(message.t) {
-                case MessageTypes.INPUT:
-                    sink(1, {
             t: MessageTypes.INPUT,
-                        i: [message.i],
+            i: this.inputBuffer.slice()
-                    });
+        } as ServerMessage<LocalInput[], State>)),
+    );
+    private broadcast = makeSubject<ServerMessage<LocalInput[], State>>();
+
+    private serverFeed = share(merge(this.heartbeat, this.broadcast));
+
+    public readonly socket = defer(() => {
+        const playerNumber = this.nextClientId++;
+        return pipe(
+            this.serverFeed,
+            catchTalkback<ServerMessage<LocalInput[], State>, ClientMessage<LocalInput, State>>(message => {
+                switch (message.t) {
+                    case MessageTypes.INPUT:
+                        if (playerNumber >= 0) {
+                            this.inputBuffer[playerNumber] = message.i;
+                        }
                         break;
                 }
+            }),
+            map(message => {
+                if (message.t === MessageTypes.SET_STATE && playerNumber >= 0) {
+                    return {
+                        ...message,
+                        u: playerNumber,
+                    };
+                } else {
+                    return message;
                 }
+            })
+        );
     });
-    sink(1, {t: MessageTypes.SET_STATE, u: 0, s: {}});
+
-};
+    public resetState(newState: Partial<State>) {
+        this.broadcast(1, { t: MessageTypes.SET_STATE, u: -1, s: newState });
+    }
+}