broker.ts

import * as restate from "@restatedev/restate-sdk";

export type WorkItem = {
  item: number; // compute item factorial
};

export type WorkRequest = {
  awekable: string;
};

type BrokerState = {
  workers: WorkRequest[];
  work: WorkItem[];
};

export const broker = restate.object({
  name: "broker",
  handlers: {
    enqueueWork: async (
      ctx: restate.ObjectContext<BrokerState>,
      item: WorkItem
    ) => {
      const workers = (await ctx.get("workers")) ?? [];
      const worker = workers.shift();
      if (worker) {
        ctx.resolveAwakeable(worker.awekable, item);
        ctx.set("workers", workers);
        return;
      }
      const work = (await ctx.get("work")) ?? [];
      work.push(item);
      ctx.set("work", work);
    },

    requestWork: async (
      ctx: restate.ObjectContext<BrokerState>,
      worker: WorkRequest
    ) => {
      const work = (await ctx.get("work")) ?? [];
      const item = work.shift();
      if (item) {
        ctx.resolveAwakeable(worker.awekable, item);
        ctx.set("work", work);
        return;
      }
      const workers = (await ctx.get("workers")) ?? [];
      workers.push(worker);
      ctx.set("workers", workers);
    },
  },
});

export const queue = restate.service({
  name: "queue",
  handlers: {
    pull: (ctx: restate.Context) => {
      const { id, promise } = ctx.awakeable<WorkItem>();
      ctx.objectSendClient(broker, "1").requestWork({ awekable: id });
      return promise;
    },
  },
});

export type Broker = typeof broker;

restate.endpoint().bind(broker).bind(queue).listen();

To publish work:

...
 ctx.objectSendClient(broker, "1").enqueueWork( { ... });

To pull for work:

> curl localhost:8080/queue/pull
{"item":1337}⏎                                                                                                
> curl localhost:8080/queue/pull
{"item":1338}⏎                                                                                                
> curl localhost:8080/queue/pull
{"item":2000}⏎