1.1 The example app

Actually we will begin with our example application. A session is begun by POSTing a payload { "n": <n> } to /sum. The reply, and each subsequent reply until the last one, will be the sub-total along with a link to send the next number in that session. The number must be positive; if 0 (or less) is received the session is terminated.

Just read the code, the whole point is it’s supposed to be obvious:

class ExampleApp extends Handler {
  protected _sum = 0;

  async handle(request: RequestData) {
    let done = false;
    while (!done) {
      try {
        const { n } = request.parse<{ n: number }>();
        if (n <= 0) {
          done = true;
          request.replyOk({
            total: this._sum,
          });
        } else {
          this._sum += n;
          request = await this.suspend(
            (resumeId) =>
              void request.replyOk({
                subtotal: this._sum,
                resumeAt: `/_r/${resumeId}`,
              })
          );
        }
      } catch (err) {
        request = await this.suspend(
          (resumeId) =>
            void request.replyError(500, {
              error: JSON.stringify(err),
              subtotal: this._sum,
              resumeAt: `/_r/${resumeId}`,
            })
        );
      }
    }
  }
}

void httpRequests(8000).subscribe(
  router({
    "/sum": ExampleApp
  })
);

$ curl -XPOST -H'Content-Type: application/json' -d'{"n":3}' 'http://localhost:8000/sum'
{"subtotal":3,"resumeAt":"/_r/s-1"}
$ curl -XPOST -H'Content-Type: application/json' -d'{"n":2}' 'http://localhost:8000/_r/s-1'
{"subtotal":5,"resumeAt":"/_r/s-2"}
$ curl -XPOST -H'Content-Type: application/json' -d'{"n":1}' 'http://localhost:8000/_r/s-1'
{"error":"No continuation for s-1."}
$ curl -XPOST -H'Content-Type: application/json' -d'{"n":1}' 'http://localhost:8000/_r/s-2'
{"subtotal":6,"resumeAt":"/_r/s-3"}
$ curl -XPOST -H'Content-Type: application/json' -d'{"n":0}' 'http://localhost:8000/_r/s-3'
{"total":6}

This code has some desirable properties:

1. The majority of the code deals directly with the problem domain.
2. The algorithm is expressed in a straightforward, “direct” style like that of a one-off script, even though it spans multiple asynchronous requests. Legibility is the key to maintaining software with grace.
3. Behold: no RxJS or weird continuation stuff. Our efforts result in less homework for people just trying to get things done, not more.

suspend is our raison d’être. When called it will suspend the handler execution, register it somewhere with a unique identifier, and pass that identifier to a callback to be invoked before exiting. Thus the handler may suspend itself and reply with the information necessary to be resumed.

2.1 Will somebody please tell me what continuations are

The original paper begins assuming the reader already has some inkling of what continuations are.

The term continuation is used to refer to the state of the executing program and future work to be done.

Lisp languages, like Scheme, have support for assigning the “current continuation” to a variable that you can invoke as a function. It’s called call-with-current-continuation (or the shorter alternative call/cc). In some senses it’s the most fundamental control flow mechanism: the ability to pick up your current universe or timeline and store it to resume later.

We are going to build our own call/cc using a homemade clone of RxJS. This is for two reasons:

2.2 Double Negation and Inverting Control

An Observable inverts control. In the most basic setting, a function A -> B is simply applied to a suitable argument A to produce a concluding B. An Observable, instead, is an A-like thing which consumes functions A -> B as its arguments with a method called subscribe. The ultimate goal is the same: to annihilate an A with some A -> B.

The following is a stripped down imitation of the fundamental RxJS type which achieves this.

interface Observer<A, B = void> {
  next(value?: A): B;
}

class Observable<A> {
  constructor(
    private readonly _action: (c: Observer<A, unknown>) => (() => void) | void
  ) {}

  public subscribe<B = void>(
    observer: Observer<A, B> | ((a: A) => B)
  ): () => void {
    const unsubscribe =
      "function" === typeof observer
        ? this._action({
            next(v: A): B {
              return observer(v);
            },
          })
        : this._action(observer);
    return "function" === typeof unsubscribe
      ? unsubscribe
      : () => {
          return;
        };
  }
}

Borrowing a term from intuitionistic logic, for a given value A its negation is an implication which assumes A and concludes false (it literally negates A).Programming is an application of intuitionistic logic or I wouldn’t mention it.

A continuation is a negation in this sense: it represents future work to be performed, assuming some initial value. And since false implies true, a continuation can produce any result as long as it “doesn’t matter.”Consider how libraries like axios or fetch model remote network requests with callbacks (which return nothing).

Any function-like thing can be considered a negation in this way of course. RxJS has a special type, Observer<A>, which represents work to be done when its next method is called with a suitable argument of type A.

An Observable<A>, being a thing which negates things-which-negate-A, is a double negation.

interface Operator<A, B> {
  (s: Observable<A>): Observable<B>;
}

function keep<A>(promise: PromiseLike<A>): Observable<A> {
  let canceled = false;
  return new Observable((observer) => {
    void promise.then((value) => {
      if (!canceled) {
        setTimeout(() => void observer.next(value), 0);
      }
    });
    return () => {
      canceled = true;
    };
  });
}

function merge<A>(it: Iterable<Observable<A>>): Observable<A> {
  let unsubbers: (() => void)[] = [];
  return new Observable((observer) => {
    for (const upstream of it) {
      setTimeout(() => {
        const unsubscribe = upstream.subscribe(observer);
        unsubbers.push(unsubscribe);
      }, 0);
    }
    return () => {
      unsubbers.slice().forEach((unsubscribe) => void unsubscribe());
    };
  });
}

// Hang on to this one. We use it later to define `reset`.
function prune<A>(): Operator<A, A> {
  return (source) =>
    new Observable((observer) => {
      const unsubscribe = source.subscribe((value: A): unknown => {
        setTimeout(() => void unsubscribe(), 0);
        return observer.next(value);
      });
      return unsubscribe;
    });
}

const stream1 = merge([
  keep(
    new Promise((resolve) => void setTimeout(() => void resolve("C"), 100))
  ),
  keep(
    new Promise((resolve) => void setTimeout(() => void resolve("A"), 30))
  ),
  keep(
    new Promise((resolve) => void setTimeout(() => void resolve("B"), 50))
  ),
]);
const stream2 = prune()(stream1);
void stream2.subscribe((response) => {
  console.log("Received response:", response);
});
// prints "Received response: A"

2.3 Continuations in reactive terms

This is where the link between reactivity and continuations first became apparent to me. With a new function, reset, we can subscribe a Promise callback to an observable computation.

function reset<A>(observable: Observable<A>): Promise<A> {
  const pruned: Observable<A> = prune()(observable);
  return new Promise<A>((resolve) => void pruned.subscribe(resolve));
}

async function resetTest() {
  const result = await reset(stream1);
  console.log("Received response:", result);
}
resetTest();
// This ALSO prints "Received response: A".

The function resetTest is syntactic sugar for the following:

function resetTest_keto() {
  //                  +-- callback
  //                  V
  reset(stream1).then((result) => {
    console.log("Received response:", result);
  });
}

reset effectively subscribes the callback function noted above to an Observable; when combined with async/await syntax sugar what we end up with is, if you really use your imagination, a kind of call/cc for async functions.

Another example will help drive this point home. This next function uses two Observables in an unusual way to achieve non-linear control flow and (promise to) return the number 16.

async function sixteen(): Promise<number> {
  return reset(
    new Observable((kOuter) =>
      void new Observable((kInner) => {
        const eight = kInner.next(4);
        kOuter.next(kInner.next(eight) as number);
      }).subscribe((n) => {
        return (n as number) * 2;
      })
    )
  ) as number;
}
// My types could be improved - in the name of flexibility I am clearly
// throwing away critical and obvious information.

The inner Observable uses its continuation kInner twice, which is bound to an Observer which doubles its argument and returns it. Doubling 4 twice of course yields 16 which is then handed to the outer continuation kOuter.

Here is an important point: kOuter is bound to the resolve callback of a Promise and cannot be invoked more than once. Doing so will result in a runtime error. However, kInner may be used 0-or-more times as it is not bound to a Promise.

Observables may be used to implement complex control flow in ways Promises cannot due to their one-use restriction.

3.1 Suspensions and registries

A handful of definitions will clarify the meaning of Handler and its suspend method.

interface Suspension {
  (u: RequestData): unknown;
}

When a service Handler suspends execution its current continuation is registered somewhere with a unique ID so it can be resumed later. We model these suspensions as unary functions accepting RequestData arguments. This makes sense: execution will be resumed by a new request, after all.

type Identifier = string;

interface Registry {
  register(suspension: Suspension): Identifier;
  lookup(route: Identifier): Suspension | null;
  remove(route: Identifier): void;
}

We keep things simple and use strings as our unique identifiers. The interface for our suspension Registry is likewise straightforward: one may register, lookup, or remove suspended executions.

3.2 Handler

We have everything we need now to handle requests which may suspend themselves for later resumption by a new request.

Instead of having an ambient Registry in the global state like the original paper, we opt to inject it as a dependency to each Handler.

// eg, RouteParams<"foo"|"bar"> => { "foo": string, "bar": string }
type RouteParams<TParams extends string> = { [K in TParams]: string };

abstract class Handler<TParams extends string = never> {
  constructor(protected readonly suspensions: Registry) {}

  public abstract handle(
    request: RequestData,
    params: RouteParams<TParams>
  ): Promise<void>;

  protected suspend(fn: (id: Identifier) => void): Promise<RequestData> {
    return reset(
      new Observable<RequestData>((resolve) => {
        const resumeId = this.suspensions.register((x) => resolve.next(x));
        void fn(resumeId);
        return () => void this.suspensions.remove(resumeId);
      })
    );
  }
}

The original paper defines a function show which generates a unique suspension identifier on every request, whether it is needed or not. This code realizes show as two pieces: the handle method which initiates server sessions, and the suspend method which registers the current continuation on-demand. The end result simply means you don’t have to capture the suspension ID at the start of every service handler (but you can).

Regardless, the heart of the original essay and this paper is captured in the suspend method above.

3.3 Observing incoming requests

Of course, we may also use Observables the way they’re advertised: to process a stream, this one of HTTP requests. Node’s built-in http library exports a simple callback-based API for writing HTTP services, and inverting callbacks is RxJS’ bread and butter. The following code constructs an Observable of RequestData objects, each of which represents an incoming request in need of a reply.

function httpRequests(port: number): Observable<RequestData> {
  return new Observable((process) => {
    const server = createServer((request, response) => {
      const [route, queryString] = (request.url ?? "").split(/\?/);
      let data = "";
      request.on("data", (payload) => {
        data += payload;
      });
      request.on(
        "end",
        () =>
          void process.next(
            new RequestData(route, response, data, queryString ?? "")
          )
      );
    });
    server.listen(port, () => void console.log(`Listening on port ${port}`));
    return () => void server.close();
  });
}

3.4 Connecting everything with the router

For convenience here is the last part of the example application again:

void httpRequests(8000).subscribe(
  router({
    "/sum": ExampleApp
  })
);

The Observer we subscribe to our httpRequests Observable is the return value of the enigmatic router function. We’ll shootdefine it first and ask questions later.

type Routes = Record<Route, new (r: Registry) => Handler>;

function router(routes: Routes) {
  routes["_r/:id"] = class extends Handler<"id"> {
    async handle(request: RequestData, params: RouteParams<"id">) {
      const resume = this.suspensions.lookup(params.id);
      void (resume
        ? resume(request)
        : request.replyError(404, {
            error: `No continuation for ${params.id}.`
          }));
    }
  };
  return {
    /* This method defines an Observer. */
    async next(request: RequestData) {
      let handler, template;
      for (const [_template, _handler] of Object.entries(routes)) {
        if (matchesPath(_template, request.route)) {
          template = _template;
          handler = _handler;
          break;
        }
      }
      if (!handler || !template) {
        request.replyError(404, {
          error: `No handler found for route ${request.route}`
        });
        return;
      }
      try {
        await new handler(this).handle(
          request,
          pathToObject(template, request.route)
        );
      } catch (err) {.
        request.replyError(500, {
          error: err.toString()
        });
      }
      return;
    },
    /* These properties and methods define a Registry. */
    _store: {},
    _unique: 0,
    register(suspension: Suspension): Identifier {
      const id = `s-${++this._unique}`;
      this._store[id] = suspension;
      return id;
    },
    lookup(route: Identifier): Suspension | null {
      return this._store[route] ?? null;
    },
    remove(route: Identifier): void {
      delete this._store[route];
    }
  };
}

type Router = ReturnType<typeof router>;

Let’s review this.