Running RxJS on KoaBuilding a core server
The Koa web framework provides a barebones middleware system that works with Promises on Node.js. It doesn’t have much built in, and it’s necessary to plug in additional middleware for routing and other considerations in the request lifecycle. If we prefer working with RxJS and Observables, and don’t mind writing our own middleware — whether for control or as a practical design exercise — it’s straightforward to put together a wrapper for Koa to run our own reactive middleware pipeline instead of native Koa middleware.
Prerequisites
Before getting started, we’ll need Node.js — I’m running 10.x for this project. Add the following scripts and dependencies in a package.json and install with npm install (or install via the NPM or Yarn CLI).
"scripts": {
"start": "ts-node src/index.ts"
}
"devDependencies": {
"@types/koa": "^2.0.48",
"ts-node": "^8.0.3",
"typescript": "^3.3.3333"
}
"dependencies": {
"koa": "^2.7.0",
"rxjs": "^6.4.0"
}
Create a tsconfig.json at the project root:
{
"compilerOptions": {
"alwaysStrict": true,
"strict": true,
"target": "es6",
"module": "commonjs",
"lib": ["es2015"],
"esModuleInterop": true
},
"include": ["./src/**/*"]
}
Note: All code will be assumed to be in src/ relative to the project root.
Router
A lightly opinionated API for defining routes consists of a predicate to match requests against, and a handler to operate on a request. Additionally, we’ll expose a handleRoute() helper for the server to consume.
// router.ts
import { Context, Request } from 'koa';
import { Observable, throwError } from 'rxjs';
export type Route = {
predicate: FilterPredicate;
handler: RouteHandler;
};
type FilterPredicate = (request: Request) => boolean;
type RouteHandler = (ctx: Context) => Observable<unknown>;
export const route = (predicate: FilterPredicate, handler: RouteHandler): Route => ({
predicate,
handler,
});
export const handleRoute = (ctx: Context, routes: Route[]) => {
const route = routes.find(route => route.predicate(ctx.request));
return route ? route.handler(ctx) : throwError(new Error());
};
Sample route
At this point, a route can be defined with a matcher for the desired URL. Let’s use /hello for this example. Writing to the response and then returning an Observable that immediately emits a single element will satisfy the router’s contract and allow the server to continue down its pipeline after processing each request.
// greetings/getGreeting.ts
import { of } from 'rxjs';
import { route } from '../router';
export const getGreeting = route(
request => request.url === '/hello',
ctx => {
const { response } = ctx;
response.body = 'hello, world';
return of(0);
},
);
Collecting routes
Collecting all our routes in one configuration file, we can export a single array to be consumed by our application.
// routes.ts
import { Route } from './router';
import { getGreeting } from './greetings/getGreeting';
export const routes: Route[] = [getGreeting];
Server
The server is the most complex component of this solution. It initializes a Koa application and handles the full request pipeline using one RxJS subscription per request, run within a single middleware.
// server.ts
import Koa, { Context } from 'koa';
import { Subject, of } from 'rxjs';
import { tap, catchError, mergeMap, mapTo, single } from 'rxjs/operators';
import { handleRoute, Route } from './router';
export type Props = {
routes: Route[];
preTap: (ctx: Context) => void;
postTap: (ctx: Context) => void;
catchTap: (error: any) => void;
};
export const server = ({ routes, preTap, postTap, catchTap }: Props) => {
const handleError = (error: any) => of(catchTap(error));
const listen = (...params: Parameters<typeof Koa.prototype.listen>) =>
new Koa()
.use(async (ctx, next) => {
// We'll emit the Koa ctx into this Observable:
const root$ = new Subject<Context>();
// This Observable will indicate that the pipeline has completed:
const done$ = new Subject<void>();
const subscription = root$
// Create a reactive pipeline to handle hooks and the main routing middleware:
.pipe(
tap(preTap),
// Wait for the route handler to emit its result, then re-emit the ctx for use by subsequent hooks:
mergeMap(ctx => handleRoute(ctx, routes).pipe(mapTo(ctx))),
single(),
tap(postTap),
catchError(handleError),
// When the pipeline is complete, complete the done$ Observable
tap(() => done$.complete()),
)
.subscribe();
// Send the request into the reactive pipeline; immediately complete the root$ Observable, as we only intend to
// handle one request per pipeline:
root$.next(ctx);
root$.complete();
// Wait for the reactive pipeline to complete:
await done$.toPromise();
// Clean up and then call Koa's next() callback to proceed with its middleware pipeline:
subscription.unsubscribe();
next();
})
.listen(...params);
return {
listen,
};
};
App root
Finally, to initialize our application, we can create a server, passing in routes and lifecycle hooks, and telling it where to listen. The listen methd is written to take the same parameters as Koa#listen, as it delegates to the Koa application instance.
// index.ts
import { server } from './server';
import { routes } from './routes';
server({
routes,
preTap: ({ request }) => console.log(request),
postTap: ({ response }) => console.log(response),
catchTap: error => console.error(error),
}).listen(3000);
If we have everything together and the start script as defined at the top of the article, with an npm run start, we should be able to see the running application at localhost:3000/hello.
Next steps
An additional capability to provide here would be extending the routing predicate functionality to extract route params and plumb these through to the handler. With request parsing for URL params and body content, standardized HTTP status handling, and additional middleware hooks, this solution would come together as a practical basis for a Node.js web application. For parts of these capabilities, lower-libraries exist outside the Koa middleware ecosystem that would be straightforward to plug into our reactive middleware pipeline.
Additionally, facading the Koa Context such that the full ctx object isn’t the primary API to work with at every stage of the pipeline would be good — although suitable for low-level middleware, route handlers could be provided a more opinionated interface to operate against.