No tarball for gmail

server/index.js

@@ -0,0 +1,38 @@
+import {default as createLogger, getLevel} from "../logger/index.js";
+import yargs from "yargs/yargs";
+
+import { startServer } from "../grpc/index.js";
+
+import pubsub from "./pubsub.js";
+
+const argv = yargs(process.argv.slice(2))
+  .option("port", {
+    alias: "p",
+    demandOption: true,
+    default: 50051,
+    describe: "Listen on this port for incoming client connections",
+    type: "number",
+  })
+  .option("bind", {
+    alias: "b",
+    demandOption: true,
+    default: "0.0.0.0",
+    describe: "Which IP to bind to",
+    type: "string",
+  })
+  .option("verbose", {
+    alias: "v",
+    default: 2,
+    describe: "Verbosity level, 0 to 6, 0=error, 1=warn, 2=info, 3+=debug",
+    type: "number",
+  })
+  .help()
+  .alias("help", "h").argv;
+
+const logger = createLogger({level: getLevel(argv.verbose), postfix: "server"}) 
+const address = `${argv.bind}:${argv.port}`;
+const server = startServer(address, pubsub({ logger }), logger);
+process.on('SIGINT', () => {
+  logger.info("Shutting down due to SIGINT")
+  server.forceShutdown()
+});

server/pubsub.js

@@ -0,0 +1,67 @@
+import _logger from "../logger/index.js";
+
+//Creates a consumer that automatically removes itsefl upon timing out
+function consumer(timeoutMs, cb, call, remove) {
+  const c = {
+    cb,
+  };
+  setTimeout(() => {
+    remove(c);
+    cb({
+      message: "Timedout",
+      status: 4, //Deadline exceeded
+    });
+  }, timeoutMs);
+  call.on("cancelled", () => remove(c));
+  return c;
+}
+
+async function next(topic) {
+  const consumer = topic.consumers.values().next().value;
+  const msg = topic.queue.shift();
+  if (consumer && msg) {
+    //Fire and forget. At-most-once semantics.
+    topic.consumers.delete(consumer);
+    consumer.cb(undefined, { message: msg });
+  } else {
+    if (msg) {
+      topic.queue.unshift(msg);
+    }
+  }
+}
+
+export default function create({ logger = _logger("impl") } = {}) {
+  const topics = {};
+
+  //If logging is above info, display the content of the queues every 5s for debuggin
+  if (logger.levels[logger.level] >= 4) {
+    setInterval(() => {
+      console.log(topics);
+    }, 5000);
+  }
+
+  function getTopic(topic) {
+    // Consumers are stored in a Map to allow O(1) deletion on timeout while
+    // allowing iteration in order of insertion
+    return (topics[topic] = topics[topic] || {
+      queue: [],
+      consumers: new Map(),
+    });
+  }
+
+  return {
+    publish: (call, cb) => {
+      const topic = getTopic(call.request.topic);
+      topic.queue.push(call.request.message);
+      cb(undefined, {});
+      next(topic);
+    },
+    consume: (call, cb) => {
+      const { topic: topicStr, timeoutMs } = call.request;
+      const topic = getTopic(topicStr);
+      const c = consumer(timeoutMs, cb, call, (c) => topic.consumers.delete(c));
+      topic.consumers.set(c, c);
+      next(topic);
+    },
+  };
+}

server/pubsub.test.js

@@ -0,0 +1,165 @@
+import test from "ava";
+import pubsub from "./pubsub.js";
+
+const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
+
+//Should mock those (TODO)
+function publishCall(topic, message) {
+  return {
+    request: {
+      topic,
+      message,
+    },
+  };
+}
+
+function consumeCall(topic, timeoutMs = 10000) {
+  return {
+    request: {
+      topic,
+      timeoutMs,
+    },
+    on: () => {},
+  };
+}
+
+test.beforeEach((t) => {
+  t.context.pubsub = pubsub();
+});
+
+
+test("Published messages are available to be consumed", async (t) => {
+  const { pubsub } = t.context;
+  t.plan(2);
+  await new Promise((resolve) => {
+    pubsub.publish(publishCall("topic", "message"), (error) => {
+      t.is(error, undefined);
+    });
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message");
+      resolve();
+    });
+  });
+});
+
+test("Published messages automatically create topics", async (t) => {
+  const { pubsub } = t.context;
+  t.plan(4);
+  await new Promise((resolve) => {
+    pubsub.publish(publishCall("topic", "message"), (error) => {
+      t.is(error, undefined);
+    });
+    pubsub.publish(publishCall("topic2", "message2"), (error) => {
+      t.is(error, undefined);
+    });
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message");
+    });
+    pubsub.consume(consumeCall("topic2"), (error, { message }) => {
+      t.is(message, "message2");
+      resolve();
+    });
+  });
+});
+
+test("Consumer waits for a published message", async (t) => {
+  const { pubsub } = t.context;
+  t.plan(2);
+  await new Promise((resolve) => {
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message");
+    });
+    pubsub.publish(publishCall("topic", "message"), (error) => {
+      t.is(error, undefined);
+      resolve();
+    });
+  });
+});
+
+test("Consumer timeout if no message published", async (t) => {
+  const { pubsub } = t.context;
+  t.plan(2);
+  await new Promise((resolve) => {
+    pubsub.consume(consumeCall("topic", 0), (error, nothing) => {
+      t.is(nothing, undefined);
+      t.deepEqual(error, { message: "Timedout", status: 4 });
+      resolve();
+    });
+  });
+});
+
+test("Multiple consumers wait for multiple messages, FIFO", async (t) => {
+  const { pubsub } = t.context;
+  t.plan(6);
+  await new Promise((resolve) => {
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message1");
+    });
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message2");
+    });
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message3");
+    });
+    pubsub.publish(publishCall("topic", "message1"), (error) => {
+      t.is(error, undefined);
+    });
+    pubsub.publish(publishCall("topic", "message2"), (error) => {
+      t.is(error, undefined);
+    });
+    pubsub.publish(publishCall("topic", "message3"), (error) => {
+      t.is(error, undefined);
+      resolve();
+    });
+  });
+});
+
+test("Multiple consumers, only consumers that have not timed out get a message, FIFO", async (t) => {
+  const { pubsub } = t.context;
+  t.plan(8);
+  await new Promise(async (resolve) => {
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message1");
+    });
+    pubsub.consume(consumeCall("topic", 0), (error, nothing) => {
+      t.is(nothing, undefined);
+      t.deepEqual(error, { message: "Timedout", status: 4 });
+    });
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message2");
+    });
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message3");
+    });
+    //Allow consumer to timeout
+    await sleep(1);
+    pubsub.publish(publishCall("topic", "message1"), (error) => {
+      t.is(error, undefined);
+    });
+    pubsub.publish(publishCall("topic", "message2"), (error) => {
+      t.is(error, undefined);
+    });
+    pubsub.publish(publishCall("topic", "message3"), (error) => {
+      t.is(error, undefined);
+      resolve();
+    });
+  });
+});
+
+test("Multiple consumers trying to read a message, only the first will succeed", async (t) => {
+  const { pubsub } = t.context;
+  t.plan(4);
+  await new Promise(async (resolve) => {
+    pubsub.publish(publishCall("topic", "message1"), (error) => {
+      t.is(error, undefined);
+    });
+    pubsub.consume(consumeCall("topic"), (error, { message }) => {
+      t.is(message, "message1");
+    });
+    pubsub.consume(consumeCall("topic", 100), (error, nothing) => {
+      t.is(nothing, undefined);
+      t.deepEqual(error, { message: "Timedout", status: 4 });
+      resolve()
+    });
+  });
+});