switch break statements are important and easy to forget.. mutate subscriptions broken but not far from fixed

Author: RepComm

dst/client.js

@@ -84,6 +84,12 @@ export class Client {
  });
  return;
  }
+ else if (json.response.type === "sub-inst") {
+ this.walkSubscribers(json.response.topic, undefined, (_cb) => {
+ _cb(json.response.id, undefined, true);
+ });
+ return;
+ }
  const { resolve, reject } = this.responseResolvers.get(json.id);
  if (resolve) {
  //if we did, json.response is our answer and we stop listening
@@ -124,15 +130,18 @@ export class Client {
  return res;
  }
  subscribe(topic, cb) {
- let cfg = topic;
+ let cfg = undefined;
  if (typeof (topic) === "string") {
- // cfg.onlyDeliverDeltas = false;
+ cfg = {
+ topic
+ };
  }
  else {
  cfg = topic;
  topic = cfg.topic;
  }
  this.addSubscriber(topic, cfg.id, cb);
+ console.log("sending sub to server", cfg);
  return this.sendMessage("sub", cfg);
  }
  unsubscribe(topic) {

dst/server.js

@@ -2,7 +2,6 @@ import { createServer } from "http";
 import serveHandler from "serve-handler";
 import { server as WebSocketServer } from "websocket";
 import { watchFile } from "fs";
-import { resolve as pathResolve } from "path";
 function originIsAllowed(origin) {
  return true;
 }
@@ -16,8 +15,6 @@ async function main() {
  return mod.auth;
  }
  let authFunc = await loadAuthFunc();
- const currentDirectory = pathResolve('.');
- console.log("Current dir (aka . ) is", currentDirectory);
  watchFile("./dst/auth.js", {
  persistent: true,
  interval: 4000
@@ -31,7 +28,7 @@ async function main() {
  }
  });
  httpServer = createServer((req, res) => {
- console.log(req.url);
+ // console.log(req.url);
  if (req.url.startsWith("/api")) {
  res.writeHead(200, "success", {
  "Content-Type": "application/json"
@@ -83,19 +80,26 @@ async function main() {
  const storage = subStorageGetOrCreate(topic);
  if (id !== undefined) {
  idSubsListGetOrCreate(storage, id).add(ws);
- console.log("Added sub", topic, id, ws.remoteAddress);
+ console.log(`[sub] ${topic}:${id} -> client`);
  }
  else {
  storage.topicSubscribers.add(ws);
- console.log("Added sub", topic, ws.remoteAddress);
+ console.log(`[sub] ${topic} -> client`);
  }
  }
  function walkSubscribers(topic, id = undefined, cb) {
  const storage = subStorageGetOrCreate(topic);
- let list = id === undefined ?
- storage.topicSubscribers :
- idSubsListGetOrCreate(storage, id);
- for (const ws of list) {
+ //call ID specific listeners first if present
+ if (id !== undefined) {
+ const idSubs = storage.idSubScribers.get(id);
+ if (idSubs) {
+ for (const ws of idSubs) {
+ cb(ws);
+ }
+ }
+ }
+ //call topic subscribers always if present
+ for (const ws of storage.topicSubscribers) {
  cb(ws);
  }
  }
@@ -137,23 +141,25 @@ async function main() {
  res.error = `Invalid auth to create schema`;
  }
  else {
- console.log("schema-set", topic);
+ console.log(`[schema] created "${topic}"`);
  schemas.set(topic, {
  shape,
  instances: new Map()
  });
  }
  }
  break;
- case "schema-get": {
- let { topic } = req.msg;
- const schema = schemas.get(topic);
- if (!schema) {
- res.error = "no schema for topic";
- break;
+ case "schema-get":
+ {
+ let { topic } = req.msg;
+ const schema = schemas.get(topic);
+ if (!schema) {
+ res.error = "no schema for topic";
+ break;
+ }
+ res.response.shape = schema.shape;
  }
- res.response.shape = schema.shape;
- }
+ break;
  case "instance":
  {
  const topic = req.msg.topic;
@@ -165,7 +171,24 @@ async function main() {
  const instanceId = Math.floor(Math.random() * Number.MAX_SAFE_INTEGER).toString();
  res.response.id = instanceId;
  storage.instances.set(instanceId, {});
- // console.log("instance", instanceId);
+ console.log("instance req", req);
+ /**create a message to send to topic subscribers
+ * to let them know a new instance exists
+ */
+ const subInstRes = {
+ response: {
+ type: "sub-inst",
+ topic,
+ id: instanceId
+ },
+ id: -1
+ };
+ const subInstResStr = JSON.stringify(subInstRes);
+ console.log(`[schema] instanced "${topic}:${instanceId}"`);
+ walkSubscribers(topic, undefined, (ws) => {
+ console.log("Notifying", ws.remoteAddress, "of instance", instanceId);
+ ws.send(subInstResStr);
+ });
  }
  break;
  case "mut":
@@ -210,11 +233,9 @@ async function main() {
  },
  id: -1
  };
- // console.log("received mutate", change);
  const subResStr = JSON.stringify(subRes);
  walkSubscribers(topic, id, (ws) => {
  ws.send(subResStr);
- // console.log("Send mut to", subResStr, ws.remoteAddress);
  });
  }
  break;
@@ -228,17 +249,19 @@ async function main() {
  res.error = "unsub is not impl yet";
  break;
  case "list":
- const topic = req.msg.topic;
- const storage = schemas.get(topic);
- if (!storage) {
- res.error = `schema for topic was not found`;
- break;
+ {
+ const topic = req.msg.topic;
+ const storage = schemas.get(topic);
+ if (!storage) {
+ res.error = `schema for topic was not found`;
+ break;
+ }
+ const list = {};
+ storage.instances.forEach((v, k) => {
+ list[k] = v;
+ });
+ res.response.list = list;
  }
- const list = {};
- storage.instances.forEach((v, k) => {
- list[k] = v;
- });
- res.response.list = list;
  break;
  }
  let str = JSON.stringify(res);

dst/test.js

@@ -24,8 +24,7 @@ async function main() {
  //wait for connection
  await client.connect();
  //wait for authentication
- const a = await client.authenticate({ apiKey }); //not impl yet
- console.log("Authenticated", a);
+ await client.authenticate({ apiKey }); //not impl yet
  //create a storage for players, will be owned by our client
  if (!await client.hasSchema("players")) {
  await client.createSchema("players", {
@@ -45,22 +44,31 @@ async function main() {
  // //upload our initial player data
  await client.mutate("players", localId, {
  name: prompt("Enter player name", "testbot"),
- x: 1,
- y: 2
+ x: 0.5,
+ y: 0.5
  });
  const players = new Map();
  const insts = await client.listInstances("players");
  const { list } = insts.response;
+ function addPlayer(id, data) {
+ players.set(id, data);
+ }
  for (const id in list) {
- console.log("Player found by id", id);
  const p = list[id];
- players.set(id, p);
- client.subscribe({ topic: "players", id }, (pid, change) => {
- const original = players.get(id);
- Object.assign(original, change);
- // console.log("Player", id, "updated", change);
- });
+ addPlayer(id, p);
+ console.log("Player identified in list", id);
  }
+ await client.subscribe("players", (pid, change, isNewInstance) => {
+ console.log("sub msg");
+ if (isNewInstance) {
+ console.log("Player instanced", pid);
+ addPlayer(pid, { x: 0.5, y: 0.5, name: "" });
+ }
+ else {
+ const original = players.get(pid);
+ Object.assign(original, change);
+ }
+ });
  const mouseMoveDebounce = {
  timeLast: 0,
  timeWait: 50
@@ -72,7 +80,6 @@ async function main() {
  x: x / window.innerWidth,
  y: y / window.innerHeight
  });
- // console.log("Sending mutate");
  };
  window.addEventListener("mousemove", (evt) => {
  handlePointerMove(evt.clientX, evt.clientY);

src/auth.ts

@@ -1,6 +1,5 @@
-import { MsgReq } from "./common";
-
 
+import type { MsgReq } from "./common";
 
 export async function auth (msg: MsgReq<any>): Promise<string> {
  return Math.floor(Math.random() * Number.MAX_SAFE_INTEGER).toString();

src/client.ts

@@ -14,7 +14,7 @@ export interface ResolveReject<T> {
 }
 
 export interface SubCb<T> {
- (id: string, change: T): void;
+ (id: string, change?: T, isNewInstance?: true): void;
 }
 export type TopicId = string;
 export type InstanceId = string;
@@ -134,6 +134,15 @@ export class Client {
  }
  );
  return;
+ } else if (json.response.type === "sub-inst") {
+ this.walkSubscribers(
+ json.response.topic,
+ undefined,
+ (_cb)=>{
+ _cb(json.response.id, undefined, true);
+ }
+ );
+ return;
  }
 
  const {resolve, reject} = this.responseResolvers.get(json.id);
@@ -179,14 +188,17 @@ export class Client {
  return res;
  }
  subscribe<InstanceType> (topic: string|SubConfig, cb: SubCb<InstanceType>) {
- let cfg = topic as SubConfig;
+ let cfg = undefined;
  if (typeof(topic) === "string") {
- // cfg.onlyDeliverDeltas = false;
+ cfg = {
+ topic
+ };
  } else {
  cfg = topic;
  topic = cfg.topic as string;
  }
  this.addSubscriber(topic, cfg.id, cb as any);
+ console.log("sending sub to server", cfg);
  return this.sendMessage("sub", cfg);
  }
  unsubscribe (topic: string) {