DEV Community: NestJS

NestJS Production Checklist: 17 Checks Before You Deploy

Kamil Mysliwiec — Wed, 30 Sep 2026 08:55:09 +0000

Locally, a Nest app runs as one process, restarts when you save, talks to a database with no other clients, and gets requests straight from your browser. In production it runs as several replicas behind a load balancer, gets stopped on every deploy, shares its database with other processes, and depends on APIs that sometimes don't answer. A lot of defaults that are fine for the first setup are wrong for the second.

Startup and deploys

1. Validate config at boot, and keep secrets out of the image

A missing or malformed environment variable should stop the process before it accepts a request. Otherwise the app boots, passes its health check, takes traffic, and fails the first request that reads the value.

@nestjs/config accepts any Standard Schema (Zod, Valibot, ArkType) as validationSchema:

import { z } from 'zod';

@Module({
  imports: [
    ConfigModule.forRoot({
      isGlobal: true,
      validationSchema: z.object({
        NODE_ENV: z.enum(['development', 'test', 'production']),
        DATABASE_URL: z.url(),
        STRIPE_SECRET_KEY: z.string().min(1),
        PORT: z.coerce.number().default(3000),
      }),
    }),
  ],
})
export class AppModule {}

Read required values with config.getOrThrow('STRIPE_SECRET_KEY') rather than get(). During a rolling deploy, a version that fails on boot never becomes ready and the previous one keeps serving. That's the failure mode you want.

While you're here, check where the values come from. Secrets belong in the platform's secret store (Kubernetes Secrets, AWS Secrets Manager, Vault, your PaaS's config), injected at runtime. Not in a .env file copied into the image, and not as a Docker ARG/ENV: both end up in the image layers, readable by anyone who can pull the image. A .dockerignore with .env* in it is a cheap guard.

And don't log the config object. console.log(config) at startup is a common debugging step that stays in, and it ships every credential to your log pipeline. If you need to see what loaded, log the keys, not the values.

2. Handle SIGTERM, and get the container right

Deploys, scale-downs and node drains all send SIGTERM. Nest doesn't listen for it unless you ask, so by default onModuleDestroy and onApplicationShutdown never run: pools stay open, queue workers are killed mid-job, in-flight requests are dropped.

const app = await NestFactory.create(AppModule);
app.enableShutdownHooks();
await app.listen(process.env.PORT ?? 3000);

With hooks enabled, a signal runs: the HTTP adapter is marked as closing, onModuleDestroy, beforeApplicationShutdown, the HTTP server, gateways and microservices close, then onApplicationShutdown. Most integrations clean up in that last one. @nestjs/bullmq, for example, closes its workers there, and worker.close() waits for the jobs they're running.

Two ways the signal gets lost in containers:

Something between the runtime and Node. CMD npm run start:prod, or the shell form CMD node dist/main.js, puts npm or a shell in front of Node, and neither reliably forwards signals. Use the exec form: CMD ["node", "dist/main.js"].
Node running as PID 1. The kernel doesn't apply default signal behaviour to PID 1. enableShutdownHooks installs its own handler, so the hooks run, but Nest then exits by removing that handler and re-sending the signal to its own process. As PID 1, that second signal is ignored, and the process only exits if the event loop happens to be empty. Run with an init process (docker run --init, or tini as the entrypoint), or use app.enableShutdownHooks(undefined, { useProcessExit: true }) so Nest calls process.exit().

The rest of the image is worth a look while you're in the Dockerfile. Build in one stage and run in another, so compilers, dev dependencies and your source tree don't ship; set NODE_ENV=production; and don't run as root, since the official Node images already include a node user:

FROM node:24-slim AS build
WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build && npm prune --omit=dev

FROM node:24-slim
ENV NODE_ENV=production
WORKDIR /app
COPY --from=build /app/package.json ./
COPY --from=build /app/node_modules ./node_modules
COPY --from=build /app/dist ./dist
USER node
CMD ["node", "--enable-source-maps", "dist/main.js"]

(package.json is copied on purpose: the Nest 12 starter is ESM, and Node reads "type": "module" from it.)

Then check memory. If the container has a memory limit and V8's heap limit is higher than what the container can actually give it, the kernel kills the process before V8 gets a chance to complain: exit code 137, no JavaScript error, no log line, just a restart. Depending on the Node version and how the limit is set, the default heap size doesn't necessarily follow the container's limit, so set it explicitly to leave room for everything that isn't heap (buffers, native memory, the stack):

env:
  - name: NODE_OPTIONS
    value: "--max-old-space-size=768" # ~75% of a 1 GiB limit

Now a leak or an oversized payload ends in a JavaScript heap out of memory error with a stack trace in your logs, instead of a silent kill. Alert on restarts either way, and on OOMKilled specifically.

3. Drain before you close

When Kubernetes terminates a pod, two things start at the same time: the pod is removed from the Service's endpoints, and the kubelet stops the container (preStop hook, then SIGTERM). Endpoint removal takes a few seconds to reach every load balancer and kube-proxy. A server that closes immediately on SIGTERM refuses the requests routed to it during that window.

The usual fix is a short delay before shutdown starts:

lifecycle:
  preStop:
    exec:
      command: ["sleep", "5"]

Nest 12 also has return503OnClosing: true on NestFactory.create(). It answers new requests with a 503 and Connection: close while in-flight ones finish. That's useful once traffic has drained, but during the propagation window those 503s reach clients, so use it together with the delay, not instead of it.

Then check the time budget. Kubernetes sends SIGKILL after terminationGracePeriodSeconds (default 30), and the preStop time counts against it. A queue job that runs longer than what's left gets killed, BullMQ marks it stalled, and another worker runs it again from the start. Either raise the grace period or keep jobs short, and make them idempotent regardless, because a crash does the same thing.

4. Scheduled jobs run on every replica

@Cron(), @Interval() and @Timeout() from @nestjs/schedule run in-process. Three replicas means the nightly invoice export runs three times. It's one of the most common production bugs in Nest apps, because it can't happen on a single dev instance.

@nestjs/locks handles this with decorators:

@Cron('0 2 * * *')
@OnOneInstance({ key: 'invoices:export' })
async exportInvoices() {
  // runs on exactly one instance; the others skip the tick
}

@Cron('*/5 * * * *')
@WithoutOverlapping()
async reconcile() {
  // skips a tick while the previous run is still going
}

@OnOneInstance() holds a lease that the owning instance renews every ttl / 3 (30 s TTL by default). If that instance dies, another one takes over once the TTL runs out. @WithoutOverlapping() only holds the lock for the length of a run, which covers the other common bug: a 5-minute job that sometimes takes 7.

The lock store must be shared, and it's yours to provide: you register a LockStore backed by Redis or Postgres (the docs include both). The in-memory store only excludes other callers in the same process, so LocksModule refuses to use it in production unless you set allowInMemoryStorage. Missed ticks aren't replayed, so write jobs that process current state ("export everything not yet exported") rather than "yesterday's rows".

For code that has to be safe if a lock is lost mid-run (a long GC pause, a network partition), each acquisition comes with a signal that aborts on lease loss and a monotonically increasing fencingToken. Pass the signal to your I/O, and use the token to reject writes from a holder whose lock was taken over.

Queues have their own version of this. BullMQ doesn't retry unless you set attempts, a job whose worker dies or blocks the event loop past its lock gets marked stalled and run again, and completed and failed jobs stay in Redis until something removes them. The minimum is to set attempts, backoff and removeOnComplete/removeOnFail explicitly in defaultJobOptions, and make every job safe to run twice.

The edge

5. Make authentication deny-by-default

The usual pattern is @UseGuards(AuthGuard) on each controller. It fails open: the one controller where someone forgets the decorator is public, and nothing tells you. Invert it. Require authentication globally and opt routes out explicitly.

@nestjs/authentication works that way out of the box. AuthenticationModule registers a global guard, so every route requires a signed-in user and anything else gets a 401, and you mark the exceptions:

@Public()
@Controller('auth')
export class AuthController {
  // sign-in, sign-up, password reset
}

It covers sessions, JWT bearer tokens, OpenID Connect, magic links and API keys, and @CurrentUser() gives handlers the resolved user.

If you keep your own guard, get the same property by registering it with APP_GUARD and letting a metadata decorator opt out:

export const IS_PUBLIC = 'isPublic';
export const Public = () => SetMetadata(IS_PUBLIC, true);

@Injectable()
export class AuthGuard implements CanActivate {
  constructor(private readonly reflector: Reflector) {}

  canActivate(context: ExecutionContext) {
    const isPublic = this.reflector.getAllAndOverride<boolean>(IS_PUBLIC, [
      context.getHandler(),
      context.getClass(),
    ]);
    if (isPublic) return true;
    // verify the token or session here
  }
}

// app.module.ts
providers: [{ provide: APP_GUARD, useClass: AuthGuard }],

Either way, a review now has one thing to check: every @Public() in the diff.

6. Security headers, CORS, rate limits, and the proxy in front

Nest now sets the standard security headers itself, with the same defaults as Helmet 8 (CSP, HSTS, X-Content-Type-Options, X-Frame-Options and the rest):

const app = await NestFactory.create(AppModule);
app.useSecurityHeaders();

It has to be called before app.init()/app.listen(), and only once; otherwise it throws. Individual headers are configurable, e.g. app.useSecurityHeaders({ contentSecurityPolicy: { directives: { ... } } }).

CORS: app.enableCors() with no arguments allows every origin. For an API used by one frontend with cookies, list the origins:

app.enableCors({ origin: ['https://app.example.com'], credentials: true });

Rate-limit at least the expensive or brute-forceable endpoints (login, password reset, anything that sends email or SMS):

ThrottlerModule.forRoot([{ ttl: 60_000, limit: 100 }]),
// ...
providers: [{ provide: APP_GUARD, useClass: ThrottlerGuard }],

Counters are per process unless you configure shared storage, so with N replicas the effective limit is N times higher.

Rate limiting also depends on knowing who the client is, and behind a load balancer every request comes from the load balancer's address. Until Express trusts X-Forwarded-*, req.ip is that address, all clients share one limit, and req.protocol is http even for HTTPS clients, which breaks secure cookies and audit logs too:

const app = await NestFactory.create<NestExpressApplication>(AppModule);
app.set('trust proxy', 1); // number of proxies in front of the app

Set the actual hop count. true trusts whatever the client sends in X-Forwarded-For, which lets any client choose its own IP.

Input and output

7. Validate every input

With class-validator classes:

app.useGlobalPipes(
  new ValidationPipe({
    whitelist: true,
    forbidNonWhitelisted: true,
    transform: true,
  }),
);

whitelist strips properties that aren't declared on the DTO. That's what stops { "email": "...", "role": "admin" } from reaching a repository.save(dto). forbidNonWhitelisted rejects them with a 400 instead, which also surfaces client typos.

With Zod, Valibot or ArkType, use StandardSchemaValidationPipe and attach the schema to the parameter:

app.useGlobalPipes(new StandardSchemaValidationPipe());

export const createUserSchema = z.object({
  email: z.email(),
  password: z.string().min(8),
});
export type CreateUserDto = z.infer<typeof createUserSchema>;

@Post()
create(@Body({ schema: createUserSchema }) dto: CreateUserDto) {
  return this.users.create(dto);
}

Here unknown keys are the schema's business, not the pipe's. z.object() strips them by default; use z.strictObject() if you want them rejected. Either way, the handler receives the schema's output, so coercions and defaults are applied.

One thing to know about useGlobalPipes(): it's applied to the app instance, so an e2e test that builds its own app from AppModule doesn't get it. Either repeat the call in the test setup, or register the pipe as an APP_PIPE provider so the module brings it along.

Validation is also where you bound list endpoints. A limit or take query parameter without a maximum lets one request load every row in the table into memory; so does a list endpoint with no default. Put the cap in the DTO or schema, and give every list query a default:

export const listOrdersQuery = z.object({
  limit: z.coerce.number().int().min(1).max(100).default(20),
  cursor: z.string().optional(),
});

(@Max(100) plus a default value does the same with class-validator.) The same applies to anything that fans out per item: a bulk endpoint that accepts 10,000 ids is a list endpoint too.

The body size limit is already reasonable: Express's JSON parser stops at 100 KB. The mistake is raising it globally for one upload endpoint. app.useBodyParser('json', { limit: '50mb' }) applies to every route, and parsing a 50 MB JSON body blocks the event loop for every other request on that process. Stream large payloads on the route that needs them, or upload them directly to object storage.

8. Control what goes out

Returning ORM entities from controllers leaks every column, including the ones added after the endpoint was written. Map to response DTOs, or use @Exclude() with a global ClassSerializerInterceptor. The interceptor only applies to class instances, so plain objects (raw query results, { ...user, extra }) are serialized as-is.

Don't expose Swagger publicly in production. SwaggerModule.setup() publishes a complete map of the API. Skip it when NODE_ENV === 'production', or put it behind auth.

9. Make writes safe to repeat

Clients retry. A mobile app on a flaky connection, a user who double-clicks, a proxy or SDK that retries on a timeout: any of them can send the same POST /orders/:id/pay twice, and the second one charges the card again. Your handler can't tell a retry from a new request unless the client says so.

The standard contract is an Idempotency-Key header: the client generates a key per logical operation and sends it with every attempt, and the server runs the operation once and replays the stored response for repeats. @nestjs/idempotency implements it:

IdempotencyModule.forRoot({
  // keys are per user, so two clients can't collide or read each other's results
  scope: (req: { user?: User }) => req.user?.id,
}),

@Post(':id/pay')
@Idempotent({ required: true })
pay(@Param('id') id: string, @Body() dto: PayOrderDto, @CurrentUser() user: User) {
  return this.orders.pay(id, user, dto);
}

A repeat with the same key gets the stored response (with Idempotent-Replayed: true) and the handler doesn't run. A duplicate that arrives while the first is still running gets a 409 with Retry-After. The same key with a different body is a 422, which catches client bugs that reuse keys. 5xx responses aren't stored, so a request that failed on your side can be retried and will run again. required: true rejects requests without a key, which is what you want on payment-like endpoints.

The records need a shared store (Postgres or Redis; the docs have both), for the same reason as the lock store in item 4: in memory, a retry that lands on another replica, or arrives after a deploy, runs the handler again. The module refuses to start in production without one. Stored responses are kept for 24 hours by default (ttl), and the whole response body is stored, so keep that in mind for endpoints that return large payloads.

This is different from @Retry({ idempotent: true }) in item 12, which lets the server re-run a handler within one request. @Idempotent() deduplicates the client's retries across requests. If you use both, register IdempotencyModule before ResilienceModule in imports, because import order sets the order of their global interceptors.

Dependencies

10. Run current versions of Node, Nest and your dependencies

Old versions are a production risk even when nothing is visibly broken: security fixes, memory leak fixes and bug fixes only land in supported release lines, and the longer you wait, the bigger and riskier the eventual upgrade.

Node: run an LTS release that's still supported, Active or Maintenance. Odd-numbered majors never become LTS, and an end-of-life line gets no security patches at all: Node 20 reached end-of-life in April 2026, so anything still on it is unpatched. Pin the major in the base image (node:24-slim) and rebuild regularly so patch releases actually reach production, rather than building once and running that image for a year.
Nest: stay on the current major, and keep every @nestjs/* package on matching versions. Mismatched @nestjs/common and @nestjs/core versions cause confusing DI and metadata errors. Several items in this list (useSecurityHeaders(), StandardSchemaValidationPipe, return503OnClosing) only exist in recent releases.
Everything else: commit the lockfile, install with npm ci in CI and in the image, and let Renovate or Dependabot open small update PRs continuously. Twenty small upgrades with passing tests are much cheaper than one big one after two years. Run npm audit --omit=dev in CI to catch known vulnerabilities in what actually ships.

11. Time out every outbound call

A dependency that stops responding is worse than one that errors. Every request waiting on it holds memory, a socket and often a database connection, and under load you run out of those before the dependency recovers. Few clients time out by default:

fetch waits up to 5 minutes for response headers. Pass signal: AbortSignal.timeout(5_000).
For HTTP APIs, use @nestjs/http-client, the Promise-based replacement for the axios-based HttpModule. Configure a timeout per client:

  HttpClientModule.register({
    name: 'payments',
    baseUrl: 'https://api.payments.example.com',
    timeout: '5s',
  }),

The timeout applies per attempt, and retries are on by default for idempotent methods (GET, HEAD, OPTIONS, PUT, DELETE: 3 attempts with exponential backoff, and on timeouts too). So a 5 s timeout can mean 15 s or more before the call fails. Make sure that fits the caller's budget, or tune retry.

Database queries: set a statement timeout. With pg, that's statement_timeout in the pool config, whatever ORM sits on top.

12. Plan for dependencies that are down

Timeouts cap one call. When a dependency is down for minutes, you also want to stop calling it for a while, limit how much of your capacity it can occupy, and serve something reasonable in the meantime. @nestjs/resilience provides that as decorators and presets:

ResilienceModule.forRoot({
  presets: {
    carrier: {
      timeout: '2s',
      retry: { attempts: 2 },
      circuitBreaker: { failureRateThreshold: 50, minimumCalls: 10, openDuration: '30s' },
    },
  },
}),

@Get('quotes')
@Resilience('carrier')
@Fallback('flatRateQuotes', { handleIf: (error) => error instanceof CircuitOpenError })
getQuotes(@Query('orderId') orderId: string, @Signal() signal: AbortSignal) {
  return this.shipping.getQuotes(orderId, signal);
}

When the breaker is open, calls fail immediately with a 503 and Retry-After instead of waiting 2 s each, and the fallback answers. @Bulkhead() caps concurrent calls, so a slow dependency can't tie up every request on the process. Pass the injected signal down to your I/O so a timed-out attempt actually stops. Inside services (jobs, cron, anything that isn't a handler), use ResilienceService.preset('carrier').execute(...). Two details: @Retry() only applies to safe methods unless the handler is also @Idempotent() (item 9) and the retry is marked idempotent: true, and retries multiply across layers. A resilience retry around an @nestjs/http-client call that retries itself makes 3 x 3 = 9 attempts, so decide which layer owns retries.

13. Run migrations as a deploy step

Every ORM has a way to sync the schema directly from your model: TypeORM's synchronize: true, drizzle-kit push, prisma db push. They're convenient in development and unsafe against production data, because they apply whatever diff they compute, and a renamed field can look like a drop plus an add.

In production, apply reviewed migrations (TypeORM migrations, drizzle-kit migrate, prisma migrate deploy) as a separate step before the new version starts: a Kubernetes Job, a release phase, a CI step. Don't run them from app startup (migrationsRun: true and equivalents) when you have several replicas, because they all start at once and race.

Also check that migrations are backward compatible with the version still running. During a rolling deploy, old and new code share the new schema. Dropping or renaming a column the old version still reads breaks it for the length of the rollout, so split those into expand and contract steps across two deploys.

The same rule is what makes rollback possible. The goal is that at any moment you can redeploy the previous image and it works against the current schema, because down-migrations under incident pressure are rarely tested and often lossy. If a deploy can't be rolled back that way, it should be a deliberate, reviewed exception, not something you discover during an incident.

14. Check the connection pool math

Every replica has its own pool. Five replicas with a pool of 20 is 100 connections. That's Postgres's default max_connections, and you haven't counted the migration job, the workers, an admin tool, or the deploy window where old and new pods run at the same time.

Set the pool size explicitly (for pg, the max option of the pool, however your ORM passes it through). Multiply by the maximum replica count during a deploy, and check it fits. When it no longer fits, add a pooler like PgBouncer rather than raising max_connections.

15. Watch for request-scoped chains

Scope.REQUEST propagates up the dependency graph. Anything that injects a request-scoped provider becomes request-scoped, and so does anything that injects that, up to the controllers. A request-scoped logger near the bottom of the graph can mean a dozen objects constructed per request across a module, without any code saying so.

Check for it before launch. If you need per-request data rather than per-request instances, AsyncLocalStorage (or nestjs-cls) provides it without rebuilding the graph. If you need per-tenant instances, durable providers reuse them per tenant instead of creating them per request.

Errors and visibility

16. Errors: statuses, logs, source maps

Keep the default exception handling semantics: real statuses for HttpException, an opaque 500 plus a log entry for everything else. If you customise the response shape, extend BaseExceptionFilter and delegate to it. A filter written from scratch drops Nest's ExceptionsHandler log line, and unhandled errors stop appearing in logs.
Errors outside handlers. An unhandled promise rejection crashes the process (Node's default since v15). That's the correct behaviour. Make sure the error is logged with its stack before exit, and that repeated restarts alert someone.
Source maps. The starter already emits them. Start Node with --enable-source-maps (or NODE_OPTIONS=--enable-source-maps) so stack traces point at .ts files and lines.

17. Logs, health checks, monitoring

Logs: JSON, production log levels, structured params, and a trace id on every line. How to monitor a NestJS app covers each.

Health checks: separate liveness from readiness. Liveness asks whether this process is stuck, and it shouldn't check dependencies: if the database goes down and liveness fails, Kubernetes restarts every pod, which doesn't help and adds a reconnect storm when the database comes back. Readiness asks whether the pod should receive traffic, and that's where the @nestjs/terminus database check goes.

Monitoring: the items above prevent known failure modes. For the rest, you need per-route latency and error rates, traces that show which of your methods and queries took the time, errors grouped into defects with an alert on new ones, and visibility into jobs and scheduled runs.

For Nest specifically, that's what we built NestJS Observe for. It instruments through Nest's DI container, so spans are your own providers, and setup is a module and one option:

// app.module.ts
export const { ObserveModule, ObserveInstrument } = createObserveModule();

@Module({
  imports: [
    ObserveModule.forRoot({
      appKey: process.env.OBSERVE_APP_KEY!,
      appSecret: process.env.OBSERVE_APP_SECRET!,
      serviceId: 'orders-api',
      // nothing is ignored by default; keep probes out of the latency numbers
      http: { ignore: [/^\/health(?:\/|\?|$)/] },
    }),
  ],
})
export class AppModule {}

// main.ts
const app = await NestFactory.create(AppModule, { instrument: ObserveInstrument });

That gives you requests, jobs, cron runs and messages with p95 and error rate, traces with the SQL under each method, unhandled errors grouped with the failing source line and an email on new ones, and trace ids on ConsoleLogger output. The free tier covers 300,000 events a month.

Summary

Config is validated at boot, required values use getOrThrow, secrets come from a secret store, and the config object is never logged.
enableShutdownHooks() is on, Node receives SIGTERM directly, and there's an init process; the image is multi-stage, runs as non-root, and the heap limit fits the container's memory limit.
Pods drain before closing, and the grace period covers the longest job.
Scheduled jobs are guarded with @OnOneInstance()/@WithoutOverlapping() on a shared lock store, and every job is idempotent.
Authentication is global, and every @Public() is deliberate.
useSecurityHeaders() is on, CORS lists its origins, sensitive endpoints are rate-limited, and trust proxy matches the number of proxies.
A global ValidationPipe (with whitelist) or StandardSchemaValidationPipe is registered, list endpoints have a maximum page size, and the body limit is unchanged.
Responses are DTOs or serialized classes; Swagger isn't public.
Endpoints that must not run twice require an Idempotency-Key, backed by a shared store.
Node is on a supported LTS, @nestjs/* packages are current and aligned, and dependency updates arrive continuously.
Every outbound call and query has a timeout, and retry budgets add up.
Unreliable dependencies have a circuit breaker, a bulkhead or a fallback, and retries are owned by one layer.
Migrations run as a separate deploy step and are backward compatible, so the previous image can always be redeployed.
Pool size times peak replica count fits in the database.
No unintended request-scoped chains.
Error handling keeps statuses and logs; stack traces resolve to source.
Logs are structured, liveness doesn't depend on the database, and the service is monitored.

If you'd rather see the monitoring side of this (item 17) in action than read about it, here's a walkthrough of setting up NestJS Observe: NestJS Observe: Zero-Config Observability for NestJS on YouTube.

How to monitor a NestJS app

Kamil Mysliwiec — Mon, 21 Sep 2026 13:07:01 +0000

How to monitor a NestJS app

When something breaks in production, the first question is usually why it happened. Monitoring gives you a much better answer than a collection of isolated logs, so here's a practical way to think about the signals that help.

Many applications start with logs and a health check, while a broader monitoring plan waits for a quieter week. That's a reasonable place to begin. Logs are much more useful when you already know what to look for, though, and during an incident that context is often the first thing you need.

So let's go through what's actually worth having, roughly in the order I'd add it.

Health checks

This is a useful first layer, and Nest has @nestjs/terminus for it:

@Controller("health")
export class HealthController {
  constructor(
    private health: HealthCheckService,
    private db: TypeOrmHealthIndicator,
  ) {}

  @Get()
  @HealthCheck()
  check() {
    return this.health.check([() => this.db.pingCheck("database")]);
  }
}

Point your load balancer at /health and it knows when an instance is dead. Keep in mind that's all it knows. An app can answer /health in 2 ms while every real request takes 9 seconds, and as far as the load balancer is concerned everything is fine.

It is worth excluding this endpoint from the other instrumentation below. A probe every few seconds creates thousands of identical requests a day and can pull latency numbers toward a route that does very little real work.

Logs

The built-in Logger is often enough during development. In production, a few ConsoleLogger options make a real difference:

const levels = (process.env.LOG_LEVELS ?? "fatal,error,warn,log").split(
  ",",
) as LogLevel[];

const logger = new ConsoleLogger({
  json: true,
  flattenParams: true,
  logLevels: levels,
});

const app = await NestFactory.create(AppModule, { logger });

json: true gives your log system one object per line instead of coloured text it has to parse back apart.

Levels are a contract

logLevels decides what gets written at all, but the levels only help if everyone on the team uses them the same way. The split I'd use:

fatal: the process can't continue. Bad config at startup, a lost connection that won't come back.
error: an operation failed and somebody should look at it. If nobody would act on it, it isn't an error.
warn: something went wrong and was handled. A retry, a fallback, a degraded response. One of these is noise; a hundred in a minute is an early warning.
log: business events you'll want during an incident. Order placed, payment captured, job finished.
debug and verbose: off in production. They bury the lines you're looking for, and you pay to ship and store every one of them.

Structured params

Since Nest 12, plain objects passed after the message are treated as data attached to the entry, not printed as extra messages. With flattenParams, their keys land at the top level of the JSON line:

this.logger.warn("Payment declined", { orderId, provider: "stripe", attempt });
// {"level":"warn",...,"message":"Payment declined","context":"PaymentsService","orderId":"ord_91f2","provider":"stripe","attempt":2}

Now your log system can filter on orderId and group by provider instead of you grepping text. A few rules make this pay off:

Constant message, variables in params. "Payment declined" written 400 times is a pattern you can count and alert on. 400 different strings with the order id baked into each one are not.
Objects, never extra strings. this.logger.log('Order shipped', order.id) doesn't do what it looks like. A trailing string is how Nest passes a context, so on a context-less logger the id replaces the context, and on a new Logger(OrdersService.name) it's printed as a second, separate log line. { orderId: order.id } is always safe.
Errors keep their stack. Params go before the stack: this.logger.error('Charge failed', { orderId }, err.stack) produces one line with orderId and a stack field.
Framework fields win. A param called message, level or timestamp is dropped silently rather than overwriting the real one. Name it reason or status.
Ids, not objects. { user } ships the whole entity, five levels deep, including whatever PII is on it. { userId: user.id } is what you'll actually filter on.

Trace ids

What the built-in logger can't give you is a trace id on every line. Without one, finding the dozen entries for a failed request among thousands of lines written that minute is guesswork, and filtering by orderId only helps if every line in the request happened to include it.

A hand-rolled request id (a middleware plus AsyncLocalStorage) gets you part of the way, but it only covers HTTP. Queue jobs, cron runs and microservice handlers never pass through that middleware, and it isn't the id of anything else you can look at: not a trace, not an error report.

ConsoleLogger has no idea which operation it's logging for, so something that does has to add the id. That's one of the things @nestjs/observe does. It hooks into ConsoleLogger, and every line written during a traced operation (a request, a queue job, a cron run, a microservice message) gets the id of that trace:

{
  "level": "warn",
  "pid": 4242,
  "timestamp": 1789992138000,
  "message": "Payment declined",
  "context": "PaymentsService",
  "orderId": "ord_91f2",
  "provider": "stripe",
  "attempt": 2,
  "traceId": "4bf92f3577b34da6a3ce929d0e0e4736"
}

The id is read when the line is written, so it's the right one even when a hundred requests interleave across awaits. It's the same id as the trace in the dashboard, so even if your logs stay in your own pipeline, you can go from a slow request to its log lines and back. If you log through pino or winston instead, the hook doesn't apply; TracerService.currentTraceId() gives you the id to add yourself, for example from pino's mixin.

Metrics

Requests per second, error rate, latency, memory, event loop delay. Cheap to collect, and they're what you put alerts on.

Two choices matter here. First, average latency can hide one request in twenty taking four seconds, so p95 is usually a more useful view of the slower experience. Second, break the data down by route. A global number changes when your traffic mix changes, even if the code has not become slower.

Metrics tell you that something changed. They never tell you why.

Traces

Traces are often what turns a broad symptom into a short list of likely causes, although they do take more setup than logs and metrics.

A trace is a single request broken into the things that happened during it: the guard, the controller method, the service it called, the three queries that service ran, the call to Stripe. Each one with a start time and a duration. Instead of "it's probably the database" you get a picture, and the picture is usually surprising.

The traditional setup can be substantial: the OpenTelemetry SDK, auto-instrumentation packages, an exporter, a collector, and a backend to store the data. Depending on the existing infrastructure, it can take a few days before the first useful trace is available, which is difficult to prioritize alongside product work.

If you do go shopping for a tracing tool, here's what I'd look for in a Nest app specifically. Spans named after your code (OrdersService.create, not middleware - <anonymous>). Self time and not only total time, because a controller that awaits a slow repository didn't do anything wrong and you don't want it at the top of your list. Database queries and outbound HTTP calls as their own spans, with the statement visible. And traces that survive a queue, so the request that enqueues a job and the job itself are one story.

Error tracking

Logs contain errors, but they do not automatically tell you that this TypeError has happened 3,140 times since the last deploy, that those occurrences represent one bug, or that the bug is new.

That's what error tracking is for: group occurrences into actual defects, remember when each one first showed up and in which release, and tell you when a new one appears. It should also know the difference between errors you threw on purpose (a NotFoundException is your API doing its job) and the ones you didn't see coming. If those two end up in the same bucket, your error rate is permanently at 4% and everybody learns to ignore it.

Queues

If you use BullMQ or Bull, half your app runs somewhere no HTTP dashboard can see. For each queue I want to know how long jobs wait before a worker picks them up, how often they fail and on which attempt, and which request enqueued them. I wrote a separate post about this because it deserves one.

Doing all of this without losing a week

You can assemble everything above from separate tools. Terminus, a log shipper, Prometheus and Grafana, an OpenTelemetry pipeline, an error tracker, something for queues. Lots of teams run exactly that and it works. It's also five or six systems to keep alive.

This is the reason we built NestJS Observe. We're the framework team, so we could do something nobody else can: instrument the app from the inside, using Nest's own hooks, instead of wrapping a generic Node.js agent around it. The whole setup is this:

npm install @nestjs/observe

and with this package installed, the rest is configuration:

// app.module.ts
import { createObserveModule } from "@nestjs/observe";

export const { ObserveModule, ObserveInstrument } = createObserveModule();

@Module({
  imports: [
    ObserveModule.forRoot({
      appKey: process.env.OBSERVE_APP_KEY,
      appSecret: process.env.OBSERVE_APP_SECRET,
      serviceId: "orders-api",
      http: {
        // keep the load balancer probe out of your percentiles
        ignore: [/^\/health(?:\?|$)/],
      },
    }),
  ],
})
export class AppModule {}

// main.ts
const app = await NestFactory.create(AppModule, {
  instrument: ObserveInstrument,
});

And that's it. No spans to write, nothing to preload, no collector to run. From that point you get requests, GraphQL operations, microservice messages, WebSocket messages, queue jobs and cron jobs, per route, with throughput, error rate and p95. Every request has a trace where the spans are your own classes and methods. SQL queries show up under the method that ran them (through pg, mysql2 and mongodb, so TypeORM, Drizzle, MikroORM and Mongoose just work), and so do outbound HTTP calls. We only record the shape of a statement, never the values. Unhandled errors come with the failing source line, get grouped into defects, and you get an email when a new one shows up. Jobs carry their wait time and attempts, and they share a trace with the request that enqueued them. And every line your ConsoleLogger writes during any of that carries its trace id.

It needs @nestjs/core 11.1 or newer and works with both Express and Fastify. The free tier is 300,000 events a month, which is plenty to find out if it's useful to you. Trace ids in your logs are on every plan. Log forwarding (the lines themselves shipped to Observe and placed on the trace's timeline, with redaction on by default) and alert rules are on the paid plans.

Two caveats. Keep your Terminus health check, because your orchestrator still needs an endpoint to hit. And if your company runs five languages and has standardised on OpenTelemetry, stick with that. A shared standard across your whole stack is worth more than a shortcut for one framework.

If I were starting from zero today

Health check first, excluded from everything else. Then JSON logs with production log levels and structured params. Then tracing and error tracking together, since they come from the same instrumentation and an error without its trace is half a story, and the same instrumentation puts a trace id on every log line. Alerts on error rate and p95 once you have a week of data to set thresholds against. Queue monitoring the day you add a queue.

The goal is not a collection of attractive dashboards. It is a shorter path from a production symptom to the line of code or dependency that needs attention.

You can click around a live project without signing up here: the demo dashboard.

Setting Up Sessions with NestJS, Passport, and Redis

Jackie McDoniel — Tue, 24 Aug 2021 23:51:24 +0000

Jay is a member of the NestJS core team, primarily helping out the community on Discord and Github and contributing to various parts of the framework.

If you're here, you're either one of the avid readers of my, just stumbling about dev.to looking for something interesting to read, or you're searching for how to implement sessions with Passport and NestJS. Nest's own docs do a pretty good job of showing how to set up the use of JWTs with Passport, but are lacking when it comes to how to use sessions. Maybe you want to use a session store because of supporting some legacy software. Maybe it's because JWTs bring too much complexity with scope. Maybe it's because you're looking for an easier way to set up refresh tokens. Whatever the case, this article is going to be for you.

Pre-requisites

I'm going to be using NestJS (it's in the title, so I hope that's obvious) and I'm going to be making use of Guards so if you don't know what those are, I highly suggest reading up on them first. Don't worry, I'll wait.

I'm also going to be not using an HTTP client like Postman or Insomnia, but using cURL instead. I lke living in the terminal as much as I can, as it gives me immediate feedback between my terminals. Feel free to use whichever you prefer, but the code snippets will be curls.

And speaking of immediate feedback, I'm also going to be using tmux, which is a terminal multiplexer, to allow me to run multiple terminals at a time within the same window and logical grouping. This allows me to keep a single terminal window up and view my server logs, docker-compose instance and/or logs, and make curls without having to alt-tab to change views. Very handy, and very customizable.

Lastly, I'll be using docker and a docker-compose file to run a Redis instance for the session storage and to allow for running a redis-cli to be able to query the redis instance ran by Docker.

All of the code will be available to follow along with and run here. Just note that to run it after you clone and run the install for the repo, you'll need to cd blog-posts/nestjs-passport-sessions and then run nest start --watch yourself. Just a side effect of how the repo is set up for my dev.to blogs.

Following along from scratch

If you're following along with the code that's pre-built, feel free to skip over this.

To set up a similar project from scratch, you'll need to first set up a Nest project, which is easiest through the Nest CLI

nest new session-authentication

Choose your package manager of choice, and then install the follow dependencies

pnpm i @nestjs/passport passport passport-local express-session redis connect-redis bcrypt

And the following peer dependencies

pnpm i -D @types/passport-local @types/express-session @types/connect-redis @types/bcrypt @types/redis

npm and yarn work fine as well, I just like pnpm as a package manager

Now you should be okay to follow along with the rest of the code, building as we go.

NestJS and Passport

The AuthGuard()

Like most @nestjs/ packages, the @nestjs/passport package is mostly a thin wrapper around passport, but Nest does do some cool things with the passport package that I think are worth mentioning. First, the AuthGuard mixin. At first glance, this mixin may look a little intimidating, but let's take it chunk by chunk.

export const AuthGuard: (type?: string | string[]) => Type<IAuthGuard> = memoize(createAuthGuard);

Ignoring the memoize call, this createAuthGuard is where the magic of class creation happens. We end up passing the type, if applicable, to the createAuthGuard method and will eventually pass that back to the @UseGuards(). Everything from here on, unless mentioned otherwise, will be a part of the createAuthGuard method.

class MixinAuthGuard<TUser = any> implements CanActivate {
  constructor(@Optional() protected readonly options?: AuthModuleOptions) {
    this.options = this.options || {};
    if (!type && !this.options.defaultStrategy) {
      new Logger('AuthGuard').error(NO_STRATEGY_ERROR);
    }
  }
...

The constructor allows for an optional injection of AuthModuleOptions. This is what is passed to PassportModule.register(). This just allows Nest to figure out if the defaultStrategy is used or the named one passed to AuthGuard.

  async canActivate(context: ExecutionContext): Promise<boolean> {
    const options = {
      ...defaultOptions,
      ...this.options,
      ...await this.getAuthenticateOptions(context)
    };
    const [request, response] = [
      this.getRequest(context),
      this.getResponse(context)
    ];
    const passportFn = createPassportContext(request, response);
    const user = await passportFn(
      type || this.options.defaultStrategy,
      options,
      (err, user, info, status) =>
        this.handleRequest(err, user, info, context, status)
    );
    request[options.property || defaultOptions.property] = user;
    return true;
  }

This here reads through decently well, we have custom methods for getting the authentication options (defaults to returning undefined), getting the request and response objects (defaults to context.switchToHttp().getRequest()/getResponse()), and then this createPassportContext method that is called and then it's return is immediately called with the strategy name and options. Then, we set req.user to the return of passportFn and return true to let the request continue. The next code block is not a part of the mixin or MixinAuthGuard class.

const createPassportContext = (request, response) => (type, options, callback: Function) =>
  new Promise<void>((resolve, reject) =>
    passport.authenticate(type, options, (err, user, info, status) => {
      try {
        request.authInfo = info;
        return resolve(callback(err, user, info, status));
      } catch (err) {
        reject(err);
      }
    })(request, response, err => (err ? reject(err) : resolve())),
  );

Here's where some magic may be seen to happen: Nest ends up calling passport.authenticate for us, so that we don't have to call it ourselves. In doing so, it wraps passport in a promise, so that we can manage the callback properly, and provides it's own handler to the authenticate function. This entire method is actually creating a different callback function so that we can end up calling this.handleRequest with the err, user, info, and status returned by passport. This can take a bit of time to understand, and isn't necessarily needed, but it's usually good to know what some of the code under the hood is doing.

  handleRequest(err, user, info, context, status): TUser {
    if (err || !user) {
      throw err || new UnauthorizedException();
    }
    return user;
  }

This is pretty straightforward, but it's useful to know this method is here. As mentioned in Nest's docs if you need to do any debugging about why the request is failing, here is a good place to do it. Generally just adding the line console.log({ err, user, info, context, status }) is enough, and will help you figure out pretty much anything going wrong within the passport part of the request.

There's two other classes I want to talk about before getting to the implementation, but I promise it'll be worth it!

The PassportStrategy()

So the next mixin we have to look at is the PassportStrategy mixin. This is how we end up actually registering our strategy class's validate method to passport's verify callback. This mixin does a little bit more in terms of some advance JS techniques, so again, lets take this chunk by chunk.

export function PassportStrategy<T extends Type<any> = any>(
  Strategy: T,
  name?: string | undefined
): {
  new (...args): InstanceType<T>;
} {
  abstract class MixinStrategy extends Strategy {

This part is pretty straightforward, we're just passing the passport strategy class and an optional renaming of the strategy to the mixin.

constructor(...args: any[]) {
  const callback = async (...params: any[]) => {
    const done = params[params.length - 1];
    try {
      const validateResult = await this.validate(...params);
      if (Array.isArray(validateResult)) {
        done(null, ...validateResult);
      } else {
        done(null, validateResult);
      }
    } catch (err) {
      done(err, null);
    }
  };

This is the first half of the constructor. You'll probably notice right of the bat that we don;'t call super, at least not yet. This is because we're setting up the callback to be passed to passport later. So what's happening here is we're setting up a function that's going to be calling this.validate and getting the result from it. If that result happens to be an array, we spread the array (passport will use the first value), otherwise we'll end up calling the done callback with just the result. If there happens to be an error, in good ole callback style, it'll be passed as the first value to the done method.

  super(...args, callback);
  const passportInstance = this.getPassportInstance();
  if (name) {
    passportInstance.use(name, this as any);
  } else {
    passportInstance.use(this as any);
  }
}

Now we end up calling super, and in doing so, we overwrite the original verify with the new callback we just created. This sets up the entire passport Strategy class that we're going to use for the strategy's name. Now all that's left to do is tell passport about it, by calling passportInstance.use(this) (or passing the custom name as the first argument).

If any of that went a little too deep, don't worry. It's something you can come back to if you really want, but isn't necessary for the rest of ths article.

PassportSerializer

Finally, an actual class! This is the most straightforward and the last bit of passport I'll talk about before getting into the implementation of sessions. This class usually won't be used in Nest applications _unless you are using sessions, and we're about to see why.

So passport has the notion of serializing and deserializing a user. Serializing a user is just taking the user's information and compressing it/ making it as minimal as possible. In many cases, this is just using the ID of the user. Deserializing a user is the opposite, taking an ID and hydrating an entire user out of it. This usually means a call to a database, but it's not necessary if you don't want to worry about it. Now, Nest has a PassportSerializer class like so:

export abstract class PassportSerializer {
  abstract serializeUser(user: any, done: Function);
  abstract deserializeUser(payload: any, done: Function);

  constructor() {
    const passportInstance = this.getPassportInstance();
    passportInstance.serializeUser((user, done) => this.serializeUser(user, done));
    passportInstance.deserializeUser((payload, done) => this.deserializeUser(payload, done));
  }

  getPassportInstance() {
    return passport;
  }
}

You should only ever have one class extending the PassportSerializer, and what it should do is set up the general serialization and deserialization of the user for the session storage. The user passed to serializeUser is usually the same value as req.user, and the payload passed to deserializeUser is the value passed as the second parameter to the done of serializeUser. This will make a bit more sens when it is seen in code.

Break Time

Okay, that was a lot of information about NestJS and Passport all at once, and some pretty complex code to go through. Take a break here if you need to. Get some coffee, stretch your legs, go play that mobile game you've been wanting to. Whatever you want to do, or continue on with the article if you want.

Running Redis Locally

You can either install and run redis locally on your machine, or you can use a docker-compose.yml file to run redis inside a container. The following compose fle is what I used while working on this article

# docker-compose.yml

version: '3'
services:
  redis:
    image: redis:latest
    ports:
      - '6379:6379'
  rcli:
    image: redis:latest
    links:
      - redis
    command: redis-cli -h redis

And then to run redis, I just used docker compose up redis -d. When I needed to run the redis CLI, I used docker compose run rcli to connect to the redis instance via the docker network.

Setting Up the Middleware

Now on to the middleware we're going to be using: for setting up sessions and a way to store them, I'm going to be using express-session, and connect-redis for the session and session store, and redis as the redis client for connect-redis. I'm also going to be setting up our middleware via a Nest middleware instead of using app.use in the bootstrap so that when we do e2e testing, the middleware is already set up (that's out of the scope of this article). I've also got redis set up as a custom provider using the following code

// src/redis/redis.module.ts

import { Module } from '@nestjs/common';
import * as Redis from 'redis';

import { REDIS } from './redis.constants';

@Module({
  providers: [
    {
      provide: REDIS,
      useValue: Redis.createClient({ port: 6379, host: 'localhost' }),
    },
  ],
  exports: [REDIS],
})
export class RedisModule {}

// src/redis/redis.constants.ts

export const REDIS = Symbol('AUTH:REDIS');

which allows for us to use @Inject(REDIS) to inject the redis client. Now we can configure our middleware like so:

// src/app.module.ts

import { Inject, Logger, MiddlewareConsumer, Module, NestModule } from '@nestjs/common';
import * as RedisStore from 'connect-redis';
import * as session from 'express-session';
import * as passport from 'passport';
import { RedisClient } from 'redis';

import { AppController } from './app.controller';
import { AppService } from './app.service';
import { AuthModule } from './auth';
import { REDIS, RedisModule } from './redis';

@Module({
  imports: [AuthModule, RedisModule],
  providers: [AppService, Logger],
  controllers: [AppController],
})
export class AppModule implements NestModule {
  constructor(@Inject(REDIS) private readonly redis: RedisClient) {}
  configure(consumer: MiddlewareConsumer) {
    consumer
      .apply(
        session({
          store: new (RedisStore(session))({ client: this.redis, logErrors: true }),
          saveUninitialized: false,
          secret: 'sup3rs3cr3t',
          resave: false,
          cookie: {
            sameSite: true,
            httpOnly: false,
            maxAge: 60000,
          },
        }),
        passport.initialize(),
        passport.session(),
      )
      .forRoutes('*');
  }
}

and have passport ready to use sessions. There's two important things to note here:

passport.initialize() must be called before passport.session().
session() must be called before passport.initialize()

With this now out of the way, let's move on to our auth module.

The AuthModule

To start off, let's define our User as the following

// src/auth/models/user.interface.ts

export interface User {
  id: number;
  firstName: string;
  lastName: string;
  email: string;
  password: string;
  role: string;
}

And then have RegisterUserDto and LoginUserDto as

// src/auth/models/register-user.dto.ts

export class RegisterUserDto {
  firstName: string;
  lastName: string;
  email: string;
  password: string;
  confirmationPassword: string;
  role = 'user';
}

and

// src/auth/models/login-user.dto.ts

export class LoginUserDto {
  email: string;
  password: string;
}

Now we'll set up our LocalStrategy as

// src/auth/local.strategy.ts

import { Injectable } from '@nestjs/common';
import { PassportStrategy } from '@nestjs/passport';
import { Strategy } from 'passport-local';
import { AuthService } from './auth.service';

@Injectable()
export class LocalStrategy extends PassportStrategy(Strategy) {
  constructor(private readonly authService: AuthService) {
    super({
      usernameField: 'email',
    });
  }

  async validate(email: string, password: string) {
    return this.authService.validateUser({ email, password });
  }
}

Notice here we're passing usernameField: 'email' to super. This is because in our RegisterUserDto and LoginUserDto we're using the email field and not username which is passport's default. You can change the passwordField too, but I had no reason to do that for this article. Now we'll make our AuthService,

// src/auth/auth.service.ts

import { BadRequestException, Injectable, UnauthorizedException } from '@nestjs/common';
import { compare, hash } from 'bcrypt';

import { LoginUserDto, RegisterUserDto } from './models';
import { User } from './models/user.interface';

@Injectable()
export class AuthService {
  private users: User[] = [
    {
      id: 1,
      firstName: 'Joe',
      lastName: 'Foo',
      email: 'joefoo@test.com',
      // Passw0rd!
      password: '$2b$12$s50omJrK/N3yCM6ynZYmNeen9WERDIVTncywePc75.Ul8.9PUk0LK',
      role: 'admin',
    },
    {
      id: 2,
      firstName: 'Jen',
      lastName: 'Bar',
      email: 'jenbar@test.com',
      // P4ssword!
      password: '$2b$12$FHUV7sHexgNoBbP8HsD4Su/CeiWbuX/JCo8l2nlY1yCo2LcR3SjmC',
      role: 'user',
    },
  ];

  async validateUser(user: LoginUserDto) {
    const foundUser = this.users.find(u => u.email === user.email);
    if (!user || !(await compare(user.password, foundUser.password))) {
      throw new UnauthorizedException('Incorrect username or password');
    }
    const { password: _password, ...retUser } = foundUser;
    return retUser;
  }

  async registerUser(user: RegisterUserDto): Promise<Omit<User, 'password'>> {
    const existingUser = this.users.find(u => u.email === user.email);
    if (existingUser) {
      throw new BadRequestException('User remail must be unique');
    }
    if (user.password !== user.confirmationPassword) {
      throw new BadRequestException('Password and Confirmation Password must match');
    }
    const { confirmationPassword: _, ...newUser } = user;
    this.users.push({
      ...newUser,
      password: await hash(user.password, 12),
      id: this.users.length + 1,
    });
    return {
      id: this.users.length,
      firstName: user.firstName,
      lastName: user.lastName,
      email: user.email,
      role: user.role,
    };
  }

  findById(id: number): Omit<User, 'password'> {
    const { password: _, ...user } = this.users.find(u => u.id === id);
    if (!user) {
      throw new BadRequestException(`No user found with id ${id}`);
    }
    return user;
  }
}

our controller

// src/auth/auth.controller.ts

import { Body, Controller, Post, Req, UseGuards } from '@nestjs/common';

import { LocalGuard } from '../local.guard';
import { AuthService } from './auth.service';
import { LoginUserDto, RegisterUserDto } from './models';

@Controller('auth')
export class AuthController {
  constructor(private readonly authService: AuthService) {}

  @Post('register')
  registerUser(@Body() user: RegisterUserDto) {
    return this.authService.registerUser(user);
  }

  @UseGuards(LocalGuard)
  @Post('login')
  loginUser(@Req() req, @Body() user: LoginUserDto) {
    return req.session;
  }
}

and our serializer

// src/auth/serialization.provider.ts

import { Injectable } from '@nestjs/common';
import { PassportSerializer } from '@nestjs/passport';

import { AuthService } from './auth.service';
import { User } from './models/user.interface';

@Injectable()
export class AuthSerializer extends PassportSerializer {
  constructor(private readonly authService: AuthService) {
    super();
  }
  serializeUser(user: User, done: (err: Error, user: { id: number; role: string }) => void) {
    done(null, { id: user.id, role: user.role });
  }

  deserializeUser(payload: { id: number; role: string }, done: (err: Error, user: Omit<User, 'password'>) => void) {
    const user = this.authService.findById(payload.id);
    done(null, user);
  }
}

along with our module

// src/auth/auth.module.ts

import { Module } from '@nestjs/common';
import { PassportModule } from '@nestjs/passport';

import { AuthController } from './auth.controller';
import { AuthService } from './auth.service';
import { LocalStrategy } from './local.strategy';
import { AuthSerializer } from './serialization.provider';

@Module({
  imports: [
    PassportModule.register({
      session: true,
    }),
  ],
  providers: [AuthService, LocalStrategy, AuthSerializer],
  controllers: [AuthController],
})
export class AuthModule {}

All we need to do for the AuthSerializer is to add it to the providers array. Nest will instantiate it, which will end up calling passport.serializeUser and passport.deserializeUser for us (told you going over that would be useful).

The Guards

So now let's get to our guards, as you'll notice up in the AuthController we're not using AuthGuard('local'), but LocalGuard. The reason for this is because we need to end up calling super.logIn(request), which the AuthGuard has, but doesn't make use of by default. This just ends up calling request.login(user, (err) => done(err ? err : null, null)) for us, which is how the user serialization happens. This is what kicks off the session. I'll repeat that because it's super important. super.logIn(request) is how the user gets a session. To make use of this method, we can set up the LocalGuard as below

// src/local.guard.ts

import { ExecutionContext, Injectable } from '@nestjs/common';
import { AuthGuard } from '@nestjs/passport';

@Injectable()
export class LocalGuard extends AuthGuard('local') {
  async canActivate(context: ExecutionContext): Promise<boolean> {
    const result = (await super.canActivate(context)) as boolean;
    await super.logIn(context.switchToHttp().getRequest());
    return result;
  }
}

We have another guard as well, the LoggedInGuard. This guards ends up just calling request.isAuthenticated() which is a method that passport ends up adding to the request object when sessions are in use. We can use this instead of having to have the user pass us the username and password every request, because there will be a cookie with the user's session id on it.

// src/logged-in.guard.ts

import { CanActivate, ExecutionContext, Injectable } from '@nestjs/common';


@Injectable()
export class LoggedInGuard implements CanActivate {
  canActivate(context: ExecutionContext) {
    return context.switchToHttp().getRequest().isAuthenticated();
  }
}

And now we have one other guard for checking if a user is an admin or not.

// src/admin.guard.ts

import { ExecutionContext, Injectable } from '@nestjs/common';

import { LoggedInGuard } from './logged-in.guard';

@Injectable()
export class AdminGuard extends LoggedInGuard {
  canActivate(context: ExecutionContext): boolean {
    const req = context.switchToHttp().getRequest();
    return super.canActivate(context) && req.session.passport.user.role === 'admin';
  }
}

This guard extends our usual LoggedInGuard and checks for the user's role, which is saved in the redis session, via the AuthSerializer we created earlier.

A couple of extra classes

There's a few other classes that I'm making use of. It'll be easiest to view them in the GitHub repo, but I'll add them here if you just want to copy paste:

// src/app.controller.ts

import { Controller, Get, UseGuards } from '@nestjs/common';
import { AdminGuard } from './admin.guard';

import { AppService } from './app.service';
import { LoggedInGuard } from './logged-in.guard';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get()
  publicRoute() {
    return this.appService.getPublicMessage();
  }

  @UseGuards(LoggedInGuard)
  @Get('protected')
  guardedRoute() {
    return this.appService.getPrivateMessage();
  }

  @UseGuards(AdminGuard)
  @Get('admin')
  getAdminMessage() {
    return this.appService.getAdminMessage();
  }
}

// src/app.service.ts

import { Injectable } from '@nestjs/common';

@Injectable()
export class AppService {
  getPublicMessage(): string {
    return 'This message is public to all!';
  }

  getPrivateMessage(): string {
    return 'You can only see this if you are authenticated';
  }

  getAdminMessage(): string {
    return 'You can only see this if you are an admin';
  }
}

// src/main.ts

import { Logger } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';

import { AppModule } from './app.module';

const bootstrap = async () => {
  const app = await NestFactory.create(AppModule);
  const logger = app.get(Logger);
  await app.listen(3000);
  logger.log(`Application listening at ${await app.getUrl()}`);
};

bootstrap();

Testing out the flow

So now, we can run everything all together and test out the flow. First things first, make sure the Redis instance is running. Without that, the server won't start. Once it's running, run nest start --watch to start the server in dev mode which will recompile and restart on file change. Now it's time to send some curls.

Testing Existing Users

So let's start off with some existing user test. We'll try to log in as Joe Foo.

curl http://localhost:3000/auth/login -d 'email=joefoo@test.com&password=Passw0rd!' -c cookie.joe.txt

If you aren't familiar with curl, the -d make the request a POST, and sends the data as application/x-www-form-urlencoded which Nest accepts by default. The -c tells curl that it should start the cookie engine and save the cookies to a file. If all goes well, you should get a response like

{"cookie":{"originalMaxAge":60000,"expires":"2021-08-16T05:30:51.621Z","httpOnly":false,"path":"/","sameSite":true},"passport":{"user":1}}

Now we can send a request to /protected and get our protected response back

curl http://localhost:3000/protected -b cookie.joe.txt

With -b we are telling curl to use the cookies found in this file.

Now let's check the registration:

curl http://localhost:3000/auth/register -c cookie.new.txt -d 'email=new.email@test.com&password=password&confirmationPassword=password&firstName=New&lastName=Test'

You'll notice that no session was created for the new user, which means they still need to log in. Now let's send that login request

curl http://localhost:3000/auth/login -c cookie.new.txt -d 'email=new.email@test.com&password=password'

And check that we did indeed create a session

curl http://localhost:3000/protected -b cookie.new.txt`

And just like that, we've implemented a session login with NestJS, Redis, and Passport.

To view the session IDs in redis, you can connect the redis-cli to the running instance and run KEYS * to get all of the set keys. By default connect-redis uses sess: as a session key prefix.

Conclusion

Phew, okay, that was definitely a longer article than I had anticipated with a much deeper focus on Nest's integration with Passport, but hopefully it helps paint a picture of how everything ties together. With the above, it should be possible to integrate sessions with any kind of login, basic, local, OAuth2.0, so long as the user object remains the same.

One last thing to note, when using sessions, cookies are a must. The client must be able to work with cookies, otherwise the session will essentially be lost on each request.

If you have any questions, feel free to leave a comment or find me on the NestJS Discord Server

Providing Providers to Dynamic NestJS Modules

Jackie McDoniel — Tue, 27 Jul 2021 17:18:14 +0000

Jay is a member of the NestJS core team, primarily helping out the community on Discord and Github and contributing to various parts of the framework.

If you've been working with NestJS for a while, you've probably heard of dynamic modules. If you're rather new to them, the docs do a pretty good job explaining them, and there's an awesome article by John about them as well that I would highly encourage reading.

I'm going to skip over the basics of dynamic modules, as the above links do a great job of explaining the concepts around them, and I'm going to be jumping into an advanced concept of providing a provider to a dynamic module. Let's unpack that sentence for a moment: we are wanting to call a dynamic module's registration method and provide a service to use instead of the default service the dynamic module already has.

The Use Case

Let's go with the general use case of having a general AuthModule with an AuthService that injects USER_SERVICE, to allow for swapping between database types (mongo, typeorm, neo4j, raw sql, etc). In doing this, we'd be able to publish the authentication module and let anyone make use of the package, while providing their own USER_SERVICE so long as it adheres to the interface we've defined.

The Setup

For this, I'm going to be using a package called @golevelup/nestjs-modules to help with the creation of the dynamic module. Instead of having to set up the entire forRoot and forRootAsync methods, we can extend a mixin and let the package take care of the setup for us. everything in this article will work without the package, I just like using it for the sake of simplicity. So, lets dive into setting up our AuthModule to be a dynamic module. First we need to create our injection token for the options

// src/auth.constants.ts

export const AUTH_OPTIONS = Symbol('AUTH_OPTIONS');
export const AUTH_SECRET = Symbol('AUTH_SECRET');
export const USER_SERVICE = Symbol('USER_SERVICE');

For now, you can ignore the AUTH_SECRET and USER_SERVICE symbols, but we'll need it here in a moment. The next is to set up the AuthModule's options interface

// src/auth.interface.ts

import { UserService } from './user-service.interface';

export interface AuthModuleOptions {
  secret: string;
  userService: UserService;
}

And the UserService interface defined as such:

// src/user-service.interface.ts

interface User {
  id: string;
  name: string;
  email: string;
}

export interface UserService {
  find: (id: string) => User;
  insert: (user: Exclude<User, 'id'>) => User;
}

Now, to make our AuthModule we can simply use the createConfigurableDynamicModule method like so:

// src/auth.module.ts

import { createConfigurableDynamicRootModule } from '@golevelup/nestjs-modules';
import { Module } from '@nestjs/common';

import { AUTH_OPTIONS } from './auth.constants';
import { AuthModuleOptions } from './auth.interface';
import { AuthService } from './auth.service';

@Module({
  providers: [AuthService],
})
export class AuthModule extends createConfigurableDynamicRootModule<AuthModule, AuthModuleOptions>(AUTH_OPTIONS) {}

And just like that, the module now has a forRoot, a forRootAsync, and a externallyConfigured static method that can all be taken advantage of (for more on the externallyConfigured method, take a look at the package's docs).

The Solution

So now, how do we ensure that users can pass in a UserService of their own, and how does our AuthService make use of it? Well, let's say that we have the following AuthService

// src/auth.service.ts

import { Injectable, Inject } from '@nestjs/common';
import { sign, verify } from 'jsonwebtoken';

import { AUTH_SECRET, USER_SERVICE } from './auth.constants';
import { UserService } from './user-service.interface';

@Injectable()
export class AuthService {
  constructor(
    @Inject(AUTH_SECRET) private readonly secret: string,
    @Inject(USER_SERVICE) private readonly userService: UserService,
  ) {}

  findUser(id: string) {
    return this.userService.find(id);
  }

  signToken(payload: Record<string, any>) {
    return sign(payload, this.secret);
  }

  verifyToken(token: string) {
    return verify(token, this.secret);
  }
}

We have it set up to inject two providers, AUTH_SECRET and USER_SERVICE (told you they'd be needed). So now all we need to do is provide these injection tokens. But how do we do that with a dynamic module? Well, taking the module from above, we can pass in a second parameter to the createConfigurableDynamicModule method and set up providers that should exist inside the module like so

// src/auth.module.with-providers.ts

import { createConfigurableDynamicRootModule } from '@golevelup/nestjs-modules';
import { Module } from '@nestjs/common';

import { AUTH_OPTIONS, AUTH_SECRET, USER_SERVICE } from './auth.constants';
import { AuthModuleOptions } from './auth.interface';
import { AuthService } from './auth.service';

@Module({
  providers: [AuthService],
})
export class AuthModule extends createConfigurableDynamicRootModule<AuthModule, AuthModuleOptions>(AUTH_OPTIONS, {
  providers: [
    {
      provide: AUTH_SECRET,
      inject: [AUTH_OPTIONS],
      useFactory: (options: AuthModuleOptions) => options.secret,
    },
    {
      provide: USER_SERVICE,
      inject: [AUTH_OPTIONS],
      useFactory: (options: AuthModuleOptions) => options.userService,
    },
  ],
}) {}

Using this approach, we are able to make use of the options passed in at the forRoot/forRootAsync level, while still allowing for injection of separate providers in the AuthService. Making use of this AuthModule would look something like the below:

// src/app.module.ts

import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';

import { AuthModule } from './auth';
import { UserModule, UserService } from './user';

@Module({
  imports: [
    AuthModule.forRootAsync({
      imports: [ConfigModule, UserModule],
      inject: [ConfigService, UserService],
      useFactory: (config: ConfigService, userService: UserService) => {
        return {
          secret: config.get('AUTH_SECRET_VALUE'),
          userService,
        };
      },
    }),
  ],
})
export class AppModule {}

This approach can work in many different contexts, and allows for a nice separation of all the options into smaller, more injectable sub-sets of options, which is what I do in my OgmaModule.

Introducing nest-commander

Jackie McDoniel — Thu, 08 Apr 2021 20:23:02 +0000

Jay is a member of the NestJS core team, primarily helping out the community on Discord and Github and contributing to various parts of the framework.

What's up Nestlings! So you've been developing web servers with NestJS for a while now, and you're in love with the DI context that it brings. You're enthralled by the class first approach, and you have "nest-itis" where you just want everything to work with Nest because, well, it just feels good. But now, after building some web servers and application, you realize you need to be able to run some one-off methods, or maybe you want to provide a full CLI package for one of your projects. You could write these commands using shell scripts (and I wish you the best if so), or you could use a CLI package like yargs or commander. Those are all fine options, but none of them are using Nest, which is sad for those of you with "nest-itis". So what do you do? You go to Google, type in "nest commander package" and search through the results, finding nest-commander. And that's where you see the light.

What sets nest-commander apart from other Nest CLI packages

`nestjs-command`

uses yargs as the underlying CLI engine
uses parameter decorators for command
has a CommandModuleTest built into the package

`nestjs-console`

uses commander for the underlying CLI engine
has a ConsoleService to create commands or can use decorator
no immediate testing integration

`nest-commander`

uses commander for the underlying CLI engine
uses a decorator on a class for a command and a decorator on class methods for command options
has a separate testing package to not bloat the final CLI package size
has an InquirerService to integrate inquirer into your CLI application
has a CommandFactory to run similar to NestFactory for a familiar DX

How to use it

Okay, so we've talked about what's different, but let's see how we can actually write a command using nest-commander.

Let's say we want to create a CLI that takes in a name, and an age and outputs a greeting. Our CLI will have the inputs of -n personName and -a age. In commander itself, this would look something like

const program = new Command();
program
  .option('-n <personName>')
  .option('-a <age>');
program.parse(process.argv);
const options = program.options();
options.age = Number.parseInt(options.age, 10);
if (options.age < 13) {
  console.log(`Hello ${options.personName}, you're still rather young!`);
} else if (12 < options.age && options.age < 50) {
  console.log(`Hello ${options.personName}, you're in the prime of your life!`);
} else {
  console.log(`Hello ${options.personName}, getting up there in age, huh? Well, you're only as young as you feel!`);
}

This works out well, and it pretty easy to run, but as your program grows it may be difficult to keep all of the logic clean and separated. Plus, in some cases you may need to re-instantiate services that Nest already manages for you. So enter, the @Command() decorator and the CommandRunner interface.

All nest-commander commands implement the CommandRunner interface, which says that every @Command() will have an async run(inputs: string[], options?: Record<string, any>): Promise<void> method. inputs are values that are passed directly to the command, as defined by the arguments property of the @Command() decorator. options are the options passed for the command that correlate back to each @Option() decorator. The above command could be written with nest-commander like so

@Command({ name: 'sayHello', options: { isDefault: true }})
export class SayHelloCommand implements CommandRunner {
  async run(inputs: string[], options: { name: string, age: number }): Promise<void> {
    if (options.age < 13) {
      console.log(`Hello ${options.personName}, you're still rather young!`);
    } else if (12 < options.age && options.age < 50) {
      console.log(`Hello ${options.personName}, you're in the prime of your life!`);
    } else {
      console.log(`Hello ${options.personName}, getting up there in age, huh? Well, you're only as young as you feel!`);
    }
  }

  @Option({ flags: '-n <personName>' })
  parseName(val: string) {
    return val;
  }

  @Option({ flags: '-a <age>' })
  parseAge(val: string) {
    return Number.parseInt(val, 10);
  }
}

Now all we need to do is add the SayHelloCommand to the Nest application and make use of CommandFactory in our main.ts.

// src/say-hello.module.ts
@Module({
  providers: [SayHelloCommand],
})
export class SayHelloModule {}

// src/main.ts
import { CommandFactory } from 'nest-commander';
import { SayHelloModule } from './say-hello.module';

async function bootstrap() {
  await CommandFactory.run(SayHelloModule);
}
bootstrap();

And there you have it, the command is fully operational. If you end up forgetting to pass in an option, commander will inform you the call is invalid. To run the command from here, compile your code as you normally would (either using nest build or tsc) and then run node dist/main.

Now, this is all fine and dandy, but the real magic, as mentioned before, is that all of Nest's DI context still works! So long as you are using singleton or transient providers, there's no limitation to what the CommandFactory can manage.

InquirerService

So now what? You've got this fancy CLI application and it runs awesome, but what about when you want to get user input during runtime, not just when starting the application. Well, that's where the InquirerService comes in. The first thing that needs to happen is a class with @QuestionSet() needs to be created. This will be the class that holds the questions for the named set. The name is important as it will be used in the InquirerService later. Say that we want to get the name and age at runtime or at start time, first we need to change the options to optional by changing from chevrons to brackets (i.e. <personName> to [personName]). Next, we need to create our question set

@QuestionSet({ name: 'personInfo' })
export class PersonInfoQuestions {
  @Question({
    type: 'input',
    name: 'personName',
    message: 'What is your name?'
  })
  parseName(val: string) {
    return val;
  }

  @Question({
    type: 'input',
    name: 'age',
    message: 'How old are you?'
  })
  parseAge(val: string) {
    return Number.parseInt(val, 10);
  }
}

Now in the SayHelloCommand we need to add in the InquirerService and ask for the information.

@Command({ name: 'sayHello', options: { isDefault: true })
export class SayHelloCommand implements CommandRunner {
  constructor(private readonly inquirerService: InquirerService) {}

  async run(inputs: string[], options: { personName?: string, age?: number }): Promise<void> {
    options = await this.inquirerService.ask('personInfo', options);
    if (options.age < 13) {
      console.log(`Hello ${options.personName}, you're still rather young!`);
    } else if (12 < options.age && options.age < 50) {
      console.log(`Hello ${options.personName}, you're in the prime of your life!`);
    } else {
      console.log(`Hello ${options.personName}, getting up there in age, huh? Well, you're only as young as you feel!`);
    }
  }
...

The rest of the class follows as above. Now we can pass in the options commander already found, and inquirer will skip over asking for them again, allowing for the a great UX by not having to duplicate their information (now if only resume services were so nice). Now in SayHelloModule we add in the PersonInfoQuestions to the providers and everything else just works ™️

@Module({
  providers: [
    SayHelloCommand,
    PersonInfoQuestions,
  ]
})
export class SayHelloModule {}

And just like that, we've now created a command line application using nest-commander, allowing for users to pass the info in via flags or using prompts and asking for it at runtime.

For more information on the project you can check the repo here. There's also a testing package to help with testing both the commander input and the inquirer input. Feel free to raise any issues or use the #nest-commander channel on the official NestJS Discord

Using the NestJS CLI GraphQL Plugin for E2E Testing

Jackie McDoniel — Wed, 09 Sep 2020 19:28:24 +0000

Jay is a member of the NestJS core team, primarily helping out the community on Discord and Github and contributing to various parts of the framework.

If you've been using NestJS for GraphQL, you probably know about the sweet @nestjs/cli plugin to auto-annotate all of your DTOs with GraphQL metadata, meaning you can cut down on code and make your development cycle faster (and let's be honest, who doesn't want that?). So you've written your code, using that awesome plugin, your unit tests are passing, you fire up your e2e tests and everything fails. GraphQL errors left and right about missing metadata, jest complaining about failures, and everything right in the world is now all so very wrong. Why did this happen, and how can we fix it?

Why did this happen?

To understand the why of this, we need to figure out what all Nest's build command does and how it works from a very high level. Below is a short list of what the build command is responsible for

compile code from ts to js
map paths properly for us (if using path aliases)
annotate DTOs with swagger decorators (if applied)
annotates DTOs with GraphQL decorators (if applied)
set up an automatic watcher to recompile on change (if using start --watch)

Now all of this looks really sweet compared to the standard Typescript compiler. In fact, Nest uses the Typescript compiler under the hood by default (though this can be swapped out for Webpack if you choose to do so). What's important to note here though, is that this is more than what tsc does by default. And because of all this, when we try to run our e2e tests through ts-jest, the default tsc compiler just isn't enough.

Okay, so how do I fix it?

The way I see it, there's two main options:

1) Create a ts-jest compiler plugin that can be swapped out with typescript
2) Create a process for building your test files specifically for e2e testing

While the first option sounds great, there's a couple of problems with it. It's specific to ts-jest so if anyone wanted to use something like mocha a whole new plugin would need to be written, and it's difficult to write, because Nest's build command is a wrap around of tsc, not an entirely new compiler in the first place.

So we go with option two and write a build process for e2e tests.

The Build Process

The first thing that we will need now is a new tsconfig, so that we can get the build to include both our source code and our test code. So, taking the standard setup from a nest new project, we can make a tsconfig.test.json that looks like this

{
  "extends": "./tsconfig.json",
  "compilerOptions": {
    "outDir": "dist-test"
  },
  "include": ["src", "test"]
}

This will test nest build to include both the src directory and the test dir, and use a different output directory than our normal build process.

hint: you should probably add the dist-test dir to your .gitignore so it isn't checked into version control

Now we need to update our jest-e2e.json that Nest provides for us. We need to remove the transform property, and change the regex used for finding the tests. The new jest-e2e.json should look something like this:

{
  "moduleFileExtensions": ["js", "json", "ts"],
  "rootDir": ".",
  "testEnvironment": "node",
  "testRegex": ".e2e-spec.js$"
}

Now the last thing to do is to create a build script and a test script to build the entire source code with tests and move them all to a single directory for tests. We can make use of pre scripts here and make two new scripts in the package.json that look like this

"pretest:e2e": "nest build -p tsconfig.test.json && cp ./test/jest-e2e.json ./dist-test/test/",
"test:e2e": "jest --config ./dist-test/test/jest-e2e.json"

With these commands, when we run npm run test:e2e or yarn test:e2e, we are telling nest to build the project using the new tsconfig.test.json we created, move the jest-e2e.json to the ./dist-test/test/ directory, to act similarly to where it already lives in the ./test/ dir (Typescript won't move non ts files by default), and then for jest (not ts-jest) to run the tests based on the config file.

To see a running version of this you can view my sample repository here.

If you have questions or comments let me know. And if you want to discuss more about NestJS in general feel free to check out our Discord.

Part 5: Completing the Client Component

John Biundo — Wed, 04 Mar 2020 14:37:21 +0000

John is a member of the NestJS core team

Introduction

This is Part 5 of a six-part series. If you landed here from Google, you may want to start with Part 1.

In this article, we build the final iteration of the client component of the Faye Custom Transporter.

Reminder: Many of the concepts and terminology here are introduced and explained in this article series. That series serves as a good foundation for understanding the more advanced concepts covered in this series.

Get the Code

All of the code in these articles is available here. As always, these tutorials work best if you follow along with the code. The README covers all the steps you need to get the repository, build the apps, and follow along. It's easy! I strongly encourage you to do so. Note that each article has a corresponding branch in the repository. For example, this article (Part 5), has a corresponding branch called part5. Read on, or get more details here on using the repository.

Git checkout the current version

For this article, you should git checkout the branch part5. You can get more information about the git repository branches here.

Build the Apps for This Part

For each article in the series, we introduce some new components (and sometimes entirely new projects). For convenience, at the start of each article, you should run the following command from the top-level directory (where you cloned the repo): *

$ # from root directory of project (e.g., transporter-tutorial, or whatever you chose as the root)
$ sh build.sh

*This is a shell script that runs on Linux-like systems. You may have to make an adjustment for non-Linux systems. Note that the script is simply a convenience that runs npm install && npm run buildinside each top-level directory, so you can always fall back to that technique if you have trouble with the script.

Overview

I ended the last article with a discussion of the shortcomings of our Take 1 (first iteration) implementation of the Faye Custom Transporter client component. In this article, we'll address those shortcomings. Since this chapter is considerably more complex than the previous ones, we'll spend a fair amount of time clarifying the issues and discussing the strategy and approach before we get into the code. My intention is to clarify the concepts you'll need to address in any custom transporter, rather than focusing entirely on the code for the Faye-flavored Custom Transporter. Hopefully we're creating re-usable conceptual building blocks, and creating a big picture you can rely on in building any custom transporter!

Understanding the Multiplexing Challenge

The big problem we found with our first implementation is that it's a little naïve about handling multiple simultaneous requests. Let's make sure we understand why. The first thing to think through is the premise of multiplexing. What we're essentially doing is sharing a single pair of channels (per pattern) for sending and receiving unrelated requests. Let's focus on the behavior of the response channel, as it seems like the potentially problematic one (after all, the request channel is just sort of "fire and forget").

We uncovered the problem in our "race test" at the end of the last chapter. The triggering scenario for our problem, in basic terms, is: we sent two requests, where the second request finished before the first one. Our shared response channel just isn't prepared to handle this. It's well-behaved in the context of one-at-a-time requests: it subscribes and waits for a response, then unsubscribes. But in the face of multiple overlapping requests, its behavior is clearly incorrect. Let's try to describe why, as this will guide us to a solution.

The problem is this: we can have only a single active Faye subscription handler¹ for each pattern no matter how many messages we send on that pattern (this is by definition, since our design is committed to a single response channel). Each request for a given pattern causes us to overwrite the previous Faye subscription handler for that pattern when we re-issue the fayeClient.subscribe(...) call. Moreover, that active one, with its built-in Observer, notifies every subscriber for that pattern.

¹The generic term "subscription handler" is unfortunately overloaded in our discussion. In the last chapter we introduced our nifty observable subscriber function. Here we're talking about a Faye subscription handler. These are two separate concepts, and we need to try to be very careful with our language. Especially since we have been, and will continue to be, building objects that create a tight relationship between these two! I'll do my best to keep the terminology straight, and you please do too!

Strategy for Solving the Multiplexing Challenge

Let's start working on a strategy to solve this challenge. Given the complexity of the problem, it's helpful to approach a solution starting from a somewhat abstract level, and successively refine our understanding and implementation. We'll walk through the details of handling this in our Faye client component, but some of the details will vary from one transport library (e.g., broker) to the next because each has its own message protocol and API.

If you understand the concepts, there is a certain amount of boilerplate material that you can adapt to a particular transport library. Inevitably, if you need to build a new transporter client, you'll have to work through a few knotholes. In the mean time, pay more attention to the concepts than to understanding every line of code.

The Observable Part

We know ClientProxy#send() returns an Observable. The job of that Observable is to consume incoming Faye messages (boxes in the diagram below) and emit them as a stream (circles in the diagram below) so that our user-land subscriber can deal with them as a nice, well-behaved RxJS Observable.

Figure 1: Convert Messages to Stream

One valuable lesson we learned in the last chapter is that we can use the Observable/observer pattern to essentially "wrap our Faye client inside an Observable". This was our observable subscriber function pattern — implemented in the handleRequest() method (review the source code here) — which builds the observable subscriber function (the bit of code that actually emits values through our Observable). We're going to build on this lesson, and take it to the next level, to deal with the complexities of the multiplexing challenge.

Handling Multiple Requests: The Union of Observables and Correlation Ids

The problem we have to solve is how to associate a unique observable subscriber function with each Observable (i.e., with each send() request that returns that Observable as a response). Furthermore, we need to do this while having only a single active Faye subscription handler for each pattern. This is going to require a little higher order programming. Stick with me — this is the hardest part of the tutorial, but we can get through it.

Let's propose defining our problem as follows: we are binding the observable subscriber function logic to our Faye subscription handler too soon/too statically. Our solution needs to do late/dynamic binding of the observable subscriber function logic. To be precise, it needs to delay binding the the observable subscriber function into the Faye subscription handler until a request is made so it can associate a unique one to each request. This is tough, because there is only a single Faye subscription handler for each pattern. What to do? We have to make a little leap here.

Conceptually, what we're going to do is the following. We'll make our single observable subscriber function logic dynamic by extracting a chunk of what was previously static code and instead have it produced dynamically by a factory function. We'll invoke the factory function for each send() request. The code we extract and produce from the factory will be the code that actually translates inbound messages to Observer emits. We'll call that code the response emitter and the factory that produces it the response emitter factory. We'll see the factory in a moment.

We'll store a unique instance of the response emitter function each time we create an Observable. Later, when we get a response, we'll retrieve that stored function and plug it back into the single static observable subscriber function, giving us a unique observable subscriber function for each request.

Let's get started. The response emitter factory function is actually provided by the framework (it's called createObserver, but we're going to stick with our response emitter factory label in this article):

// from https://github.com/nestjs/nest/blob/master/packages/microservices/client/client-proxy.ts#L82
  protected createObserver<T>(
    observer: Observer<T>,
  ): (packet: WritePacket) => void {
    return ({ err, response, isDisposed }: WritePacket) => {
      if (err) {
        return observer.error(err);
      } else if (response !== undefined && isDisposed) {
        observer.next(response);
        return observer.complete();
      } else if (isDisposed) {
        return observer.complete();
      }
      observer.next(response);
    };
  }

We've gotten quite meta here! Let's take a quick timeout to clarify all our terms!

Faye subscription handler: A callback registered with the Faye client library that is invoked whenever a message with a particular topic is received.

Observable subscriber function: A callback defined during Observable creation to convert application events (e.g., inbound Faye messages) to Observer callbacks. For example, this is the mechanism that enables us to call the user-land handler with the value emitted by our Observer every time an inbound message is received.

Response emitter factory: A higher order function, provided by the framework, that returns a response emitter (defined below).

Response emitter: A function that delivers the dynamic part of our observable subscriber function. It handles the actual Observer emit calls to produce the Observable stream. We will have a unique instance of the response emitter for each Observable, overcoming the multiplexing problem.

Let's head toward implementation. The approach we'll take to implement this goes something like this:

At the time a user-land send() call is made, we'll call the response emitter factory to create a unique instance of the response emitter; we'll also generate a unique identifier* (id field) for the request, and store an association between the unique id and the response emitter in a map.

*At last, we see the genesis of the id field we've been carrying along with our messages all this time!
We'll subscribe to the response channel.
When a response message is received, we'll extract the id from the message, and use it to look up the response emitter. We'll then call this unique instance of the response emitter from within the single shared observable subscriber function, and use this to emit the result. Since the only Observable associated with this response emitter is the originator of the request, that's the only one that receives the emitted result.
To tie these things together, we need to pass the id all the way through the request/response flow. The requesting code generates the id property, it's copied (by the server side code) to the corresponding response message, and finally retrieved again on the corresponding inbound message. Then we use it as described in step 3 above.

The pattern we described in step 3 above is often referred to as using correlation ids (for example, here's a tutorial showing how a similar concept can be used for writing native RabbitMQ client apps).

Other Bookkeeping

One related issue we kind of glossed over is managing the Faye (response channel) subscription process intelligently. We discussed the problem that we are essentially overwriting the Faye subscription handler each time we handle a new request for a given pattern. So an additional consequence of our design goal of sharing a single inbound response channel per pattern is that we we must have only a single active subscription for the response channel, per pattern.

When all active requests for a given pattern/topic have completed, we unsubscribe. When a new request comes in, we subscribe again, and leave the subscription open until the channel quiesces (i.e., there are no more inflight requests) again. We'll need to add some bookkeeping to improve our current behavior (of simply overwriting the handler with each new request on a pattern).

Connection Management

One thing we'll find, as we integrate our code with the framework, is that we need to adhere to its expectations for how we provide a client library connection (i.e., the connection to the broker). Since we're packaging up a bunch of stuff inside observable subscriber functions, the framework expects us to provide access to the connection in a particular (RxJS friendly) way. For the most part, we can just utilize some boilerplate to handle this. There's only a small bit that is specific to a client library. This is all packaged up in a connect() method that we must implement in our ClientFaye class.

Odds and Ends

We'll also see sprinkled in some functions that deal with other loose ends:

we should return channel names using a function, rather than hard coding them with the _ack and _res modifiers to avoid scattering magic strings throughout the code
since user-land patterns can be arbitrary objects (we use strings in these articles, but Nest allows constructs like client.send({cmd: 'get-customers'}, {})), we'll use a built-in inherited function to "normalize" (essentially flatten) the pattern for internal usage

Implementation

With this strategy in mind, here's the outline of how we'll implement it.

Note that there's a fair amount of interaction back and forth between our ClientFaye subclass and the ClientProxy superclass it extends, so pay close attention to that, and I'll try to give clear signposts to guide you.

Unlike our Take 1 version, where we created a new standalone class, we're now going to start by extending the ClientProxy class. This is, of course, how we "plug in" to the framework.
We're going to let the framework take over the implementation of send() for us (at least the shell of the implementation), rather than creating a concrete implementation ourselves, as we did in Part 4. The superclass send() method is now the entry point for processing a user-land send() call, and dictates how we plug in our pieces without disrupting the overall machinery.
The superclass send() method calls upon a concrete implementation of publish(), which is where we'll implement the strategy we've been discussing, dealing with our response emitter factory construct, unique identifiers, and so forth, as described above.
We're going to implement a method to unsubscribe from a Faye topic.
We're going to handle events (ClientProxy#emit()) by providing a concrete implementation of dispatchEvent().
We're going to beef up connection management, as we mentioned above, to enable the framework to efficiently share a Faye client library connection across multiple calls.
We'll take care of a few other minor details.

We have our shopping list, so let's get started!

Take 2 Code Review

The Superclass `send()` Method

Let's begin at a logical place, the send() method. As mentioned, we'll now let the superclass handle this for us, and plug in our code appropriately. Let's take a look:

// From the ClientProxy superclass.  See source code here:
// https://github.com/nestjs/nest/blob/master/packages/microservices/client/client-proxy.ts#L82
  public send<TResult = any, TInput = any>(
    pattern: any,
    data: TInput,
  ): Observable<TResult> {
    if (isNil(pattern) || isNil(data)) {
      return _throw(new InvalidMessageException());
    }
    return defer(async () => this.connect()).pipe(
      mergeMap(
        () =>
          new Observable((observer: Observer<TResult>) => {
            const callback = this.createObserver(observer);
            return this.publish({ pattern, data }, callback);
          }),
      ),
    );
  }

Stripping this method to the bone, here's what it's doing:

Reusing a connection if it exists, otherwise obtaining a new one. Note that since it depends on client library particulars to deal with obtaining a connection, it depends on the connect() method — which we are responsible for implementing and will get to soon.
Creating and returning the Observable that is effectively the "container" within which we'll implement the strategy we discussed above.

Let's look at that Observable creation step for a moment. The first line const callback = this.createObserver(observer) is the call to the response emitter factory, wherein we create our request-specific response emitter.

The second line calls publish(), which is an abstract method on the superclass that we must implement. This is where we'll implement the custom details of our strategy. Let's tackle that next. But first, let's note that send() calls publish() with two parameters:

the arguments from the user-land client.send() call (e.g., if the user wrote client.send('/get-customers', {}), the first argument would contain {pattern: '/get-customers', data: {}}).
the newly minted response emitter

The `ClientFaye#publish()` Method

This is where it all comes together. Let's look at the whole method, then we'll break it down. For now, I suggest you read the comments and lightly scan the code below. We'll dive into the code in a moment.

// nestjs-faye-transporter/src/requestor/clients/faye-client.ts
protected publish(
    partialPacket: ReadPacket,
    callback: (packet: WritePacket) => any,
  ): Function {
    try {
      // prepare the outbound packet, and do other setup steps
      const packet = this.assignPacketId(partialPacket);
      const pattern = this.normalizePattern(partialPacket.pattern);
      const serializedPacket = this.serializer.serialize(packet);
      const responseChannel = this.getResPatternName(pattern);

      // start the Faye subscription handler "bookkeeping"
      let subscriptionsCount =
        this.subscriptionsCount.get(responseChannel) || 0;

      // build the call to publish the request; in addition to making the Faye API
      // `publish()` call, when this function is called, it:
      // * updates the bookkeeping associated with the Faye subscription handler
      //   count
      // * stashes our *response emitter* (called `callback` below) in the map
      const publishRequest = () => {
        subscriptionsCount = this.subscriptionsCount.get(responseChannel) || 0;
        this.subscriptionsCount.set(responseChannel, subscriptionsCount + 1);
        this.routingMap.set(packet.id, callback);
        this.fayeClient.publish(
          this.getAckPatternName(pattern),
          serializedPacket,
        );
      };

      // build the Faye subscription handler
      // this function retrieves the *response emitter*
      // and binds it to the rest of the Faye subscription handler logic
      const subscriptionHandler = this.createSubscriptionHandler(packet);

      // perform Faye `subscribe()` first (if needed), then Faye `publish()`
      if (subscriptionsCount <= 0) {
        const subscription = this.fayeClient.subscribe(
          responseChannel,
          subscriptionHandler,
        );
        subscription.then(() => publishRequest());
      } else {
        publishRequest();
      }

      // remember, this is an *observable subscriber function*, so it returns
      // the unsubscribe function.
      return () => {
        this.unsubscribeFromChannel(responseChannel);
        this.routingMap.delete(packet.id);
      };
    } catch (err) {
      callback({ err });
    }
  }

With the time we put in on the strategy discussion, this code should hopefully make sense, at least at a high level. Let's first dispense with a couple of the minor details so they don't distract, then we can focus on the core functionality.

At the top of the method (remember, this method is called synchronously when a user-land request is made), we prepare the outbound packet (the request message to be published), including assigning the packet id and serializing the packet.
Response channel subscription management should be straightforward to follow. We essentially keep a counter of response channel subscriptions per pattern. We (logically speaking) do a count++ when publishing a request on that channel (pattern), and a count-- when receiving a response from the response channel. This let's us decide whether or not we need to subscribe to the response channel before publishing a request (solving our "we can only have one active subscription per pattern at a time" issue).

The `createSubscriptionHandler` Method: Binding the Response Emitter

Finally, let's talk about the call to createSubscriptionHandler() — first at a high level, and then the details. First, let's recognize that this a factory that returns the actual Faye subscription handler that gets bound to the Faye subscribe() call on the response channel (i.e., <message-pattern>_res). In it, we do the late binding of the observable subscriber function. We do the late binding by looking up the response emitter by id, matching it with the request Observable, and calling it with the destructured inbound message fields. Here's the code:

  public createSubscriptionHandler(packet: ReadPacket & PacketId): Function {
    return (rawPacket: unknown) => {
      const message = this.deserializer.deserialize(rawPacket);
      if (
        message.id &&
        message.id !== (rawPacket as ReadPacket & PacketId).id
      ) {
        return undefined;
      }
      const { err, response, isDisposed, id } = message;

      const callback = this.routingMap.get(id);
      if (!callback) {
        return undefined;
      }

      if (isDisposed || err) {
        return callback({
          err,
          response,
          isDisposed: true,
        });
      }
      callback({
        err,
        response,
      });
    };
  }

Let's explore a little further. The function returned by this factory, as we said, is the Faye subscription handler. As such, it has only one argument (as dictated by the Faye API) — an actual Faye inbound message. Here are the steps the returned function performs:

Deserialize the packet.
Destructure the message so we can deal with its constituent values: err, response, isDisposed and id
Use the map to lookup the correct response emitter
Discard any messages which are not destined for this client¹
Finally, return the actual observable subscriber function. Note: we automatically add isDisposed: true if there's an error, to force the closure of the Observable.

¹This is the code fragment:
if (!callback) {
return undefined;
}

We take up discussion of this snippet in its own section immediately below.

With this, we've completed the process of converting to a late binding of the observable subscriber function! We have a couple more details to work through next.

Discarding Messages for Other Clients

Consider that in a distributed microservice-based architecture, we may have multiple clients (e.g., instances of nestHttpApp or other requestor apps), connected via the broker, to the same responder (e.g., nestMicroservice). In such a configuration, each client may issue the same requests (i.e., utilize the same message pattern). When that happens, we'll have multiple clients communicating using the same Faye topic. This is, of course, completely natural for Faye (and any message broker). However, since we may have multiple Faye subscribers on the same pattern/topic (e.g., '/get-customers'), each client (each subscriber) will be notified with any response message that matches the response topic. Only those that come from the originating client will have a matching id and response emitter. We can surmise that non-matching responses are properly destined to be handled by other client instances. We can simply ignore these, returning undefined to our Faye subscription handler, which is essentially a no-op.

Loose Ends

Back in our massive discussion of the publish() method, we briefly mentioned unsubscribing our Faye subscription handler for a particular message pattern after the response channel quiesces.

Let's talk about Observable life-cycle hooks briefly. While it's possible for us to unsubscribe from an Observable in user-land (a subscribe() call on an Observable returns a function to unsubscribe), that's rarely needed. The RxJS Observer takes care of this automatically for us when the final event in a stream is emitted, and has a hook for any further cleanup we need to do. In Nest transporter-land, this life-cycle event happens when we receive the last message in a stream, with the isDisposed property set to true.

From the Observable unsubscribe hook, we can run any cleanup code we need upon stream completion, such as determining whether to unsubscribe our Faye subscription handler. To do this, we call our custom cleanup code from that unsubscribe hook. Remember, this is the function returned from the observable subscriber function, so we can expect to find it back in our publish() method. Let's take a look. Note the function returned at the bottom of the publish() method body. That's our unsubscribe hook. From it, we can do any cleanup, as shown below:

// nestjs-faye-transporter/src/requestor/clients/faye-client.ts
      ...
      return () => {
        this.unsubscribeFromChannel(responseChannel);
        this.routingMap.delete(packet.id);
      };
      ...

We do two important things at the termination of each message stream:

We decide whether to unsubscribe for messages on that topic on the Faye response channel
We delete the relevant response emitter function from the map

Deciding whether to unsubscribe requires us to revisit the subscription management topic. You should be able to connect the dots by looking at the unsubscribeFromChannel() method shown below, and the discussion we had earlier about subscription bookkeeping. Basically, we unsubscribe when the count of in-flight requests for a given pattern/topic reaches 0.

  protected unsubscribeFromChannel(channel: string) {
    const subscriptionCount = this.subscriptionsCount.get(channel);
    this.subscriptionsCount.set(channel, subscriptionCount - 1);

    if (subscriptionCount - 1 <= 0) {
      this.fayeClient.unsubscribe(channel);
    }
  }

Connection Management

The framework follows a specific protocol for accessing the broker client library connection object. This protocol deals with connection establishment and connection errors as Observable events. Rather than review this code in detail, we can safely say that it is mostly boilerplate. Let's call out the parts that are specific to Faye.

We need to implement a connect() method, which is tied to a class member called connection. Let's take a look:

  public async connect(): Promise<any> {
    if (this.fayeClient) {
      return this.connection;
    }
    const { url, serializer, deserializer, ...options } = this.options;
    this.fayeClient = new faye.Client(url, options);
    this.fayeClient.connect();
    this.connection = await this.connect$(
      this.fayeClient,
      ERROR_EVENT,
      CONNECT_EVENT,
    )
      .pipe(share())
      .toPromise();
    this.handleError(this.fayeClient);
    return this.connection;
  }

We populate this.connection by first generating a live connection using the the library-specific protocol. For Faye, this requires instantiating a Client object, then calling connect().

Once we have a connection, we run this through the framework by calling this.connect$() as shown above. This method takes the connection object (this.fayeClient), and constants defining the events the client library defines for "success" and "failure". In the case of Faye, we define them in the file nestjs-faye-transporter/src/constants.ts, and import those constants to this (the faye-client.ts) file.

In this manner, the framework uses a connection if it exists, or creates one if needed, and handles the connection lifecycle events in a uniform and efficient way.

The call to this.handleError() is optional, and let's us emit a log message in the event the connection fails.

As mentioned, most of this is boilerplate; if you're integrating another broker, you can mostly just plug in the client library-specific calls where indicated for the Faye library calls above.

Event Handling

The final feature to implement is event handling — handling user-land client.emit(...) calls. As we found on the server side in Part 3, this is far simpler than dealing with request/response style messaging because we are dealing with one-way messages: we simply send a message over the outbound (request) channel, and we're done. No subscribing to responses necessary.

Because this is straightforward and follows a predictable pattern, the framework handles most of this for us in the ClientProxy#emit method. Here's what that code looks like:

// from https://github.com/nestjs/nest/blob/master/packages/microservices/client/client-proxy.ts
  public emit<TResult = any, TInput = any>(
    pattern: any,
    data: TInput,
  ): Observable<TResult> {
    if (isNil(pattern) || isNil(data)) {
      return _throw(new InvalidMessageException());
    }
    const source = defer(async () => this.connect()).pipe(
      mergeMap(() => this.dispatchEvent({ pattern, data })),
      publish(),
    );
    (source as ConnectableObservable<TResult>).connect();
    return source;
  }

Here, once again, the framework is handling connection management for us, and all we really need to do is provide a concrete implementation for dispatchEvent(). The superclass defines an abstract dispatchEvent method as follows:

// from https://github.com/nestjs/nest/blob/master/packages/microservices/client/client-proxy.ts
protected abstract dispatchEvent<T = any>(packet: ReadPacket): Promise<T>;

So our implementation simply needs to accept a request packet and publish the appropriate message using the Faye client API. We need to wrap that in a promise and return the promise to match the signature above.

Here's how we implement the dispatchEvent() method in our Faye client component:

  protected dispatchEvent(packet: ReadPacket): Promise<any> {
    const pattern = this.normalizePattern(packet.pattern);
    const serializedPacket = this.serializer.serialize(packet);

    return new Promise((resolve, reject) =>
      this.fayeClient.publish(pattern, serializedPacket),
    );
  }

The logic should be easy to follow. We simply extract the pattern, normalize it, serialize the outbound packet (no need for an id on this one), and then return a promise that resolves to the results of the fayeClient.publish() call. The framework handles the rest (which is really very little, other than efficient connection management).

Acceptance Testing

At this point, we should have a complete implementation of the Faye Custom Transporter! 🍺 🍺 🍺!

To test it, you'll want to start up the nestHttpApp and nestMicroservice, and run any of the routes. See these notes for more help in running the tests.

Conclusion

We've come a long way! Thanks for hanging in there so long! I hope you found this series useful. Please hit me up in the comments with any questions!

For those of you braving the task of writing your own Custom Transporter, please share your experiences in the comments.

For those just interested in a deeper understanding of this key aspect of Nest Microservices, I hope this was helpful, and feel free to share your thoughts or questions in the comments!

What's Next

I have a sixth article planned, in which I'll compare several of Nest's built-in transporter implementations to provide further insight into the nuances of each implementation. I'll gauge from the responses to this series whether that's something people would find useful, so please comment below to encourage me to write that final article! 😄

Feel free to ask questions, make comments or suggestions, or just say hello in the comments below. And join us at Discord for more happy discussions about NestJS. I post there as Y Prospect.

Part 4: Basic Client Component

John Biundo — Wed, 04 Mar 2020 14:06:40 +0000

John is a member of the NestJS core team

Introduction

This is Part 4 of a six-part series. If you landed here from Google, you may want to start with Part 1.

In this article, we build the first iteration of the client component of the Faye Custom Transporter.

Get the Code

All of the code in these articles is available here. As always, these tutorials work best if you follow along with the code. The README covers all the steps you need to get the repository, build the apps, and follow along. It's easy! I strongly encourage you to do so. Note that each article has a corresponding branch in the repository. For example, this article (Part 4), has a corresponding branch called part4. Read on, or get more details here on using the repository.

Git checkout the current version

For this article, you should git checkout the branch part4. You can get more information about the git repository branches here.

Build the Apps for This Part

$ # from root directory of project (e.g., transporter-tutorial, or whatever you chose as the root)
$ sh build.sh

Overview

It's time to turn our attention to the client component. But let's start by recalling that we spent a lot of effort ensuring that the server component can return an observable stream of results to a requestor. Let's have a quick visual to cement that idea, since we'll start right out with consuming that stream on the client side.

Figure 1: Remote Observable Stream

What's Next

In this animated sequence, our nestMicroservice responder app responds to a request message (ClientProxy#send) by emitting a stream of three results (circles). We saw in the last article how our transporter server component converts that stream to a sequence of 3 messages (boxes) to the broker, sending those messages on the response channel. Our first meaningful task with the client component will be to consume those broker messages and convert them back into an Observable stream (circles) in the ClientProxy layer so that our user-land controllers and services can use them as regular RxJS observables.

In this iteration (Take 1), we'll take a similar approach to our first iteration of the server. We'll get a basic functioning client working so we can focus on its responsibilities and the main code path. Later, we'll add more robustness. As with the server side, we'll ignore events (the ClientProxy#emit path) for now.

First Iteration (Take 1) of the Client Component

Let's get started. First, let's look at how someone would use the client component of our Faye Custom Transporter. Open the app.controller.ts file for our nestHttpApp.

// nestHttpApp/src/app.controller.ts
import {
  Controller,
  Get,
  Logger,
  Param,
  Response,
  Post,
  Body,
} from '@nestjs/common';

import {
  ClientFaye,
  InboundResponseIdentityDeserializer,
  OutboundMessageIdentitySerializer,
} from '@faye-tut/nestjs-faye-transporter';


import { tap, reduce, filter } from 'rxjs/operators';

@Controller()
export class AppController {
  logger = new Logger('AppController');
  client: ClientFaye;

  constructor() {
    this.client = new ClientFaye({
      url: 'http://localhost:8000/faye',
      serializer: new OutboundMessageIdentitySerializer(),
      deserializer: new InboundResponseIdentityDeserializer(),
    });
  }

We simply instantiate the client in the class constructor. We could also use constructor injection. Note: one thing we can not currently do is use the @Client() decorator. That's not really a significant limitation, as the decorator is merely a convenience, and is not the preferred way of instantiating a ClientProxy.

The JSON options object we pass to ClientFaye should be quite similar to the options object we support on the server side, as it is either configuring the Faye connection, or passing some generic transporter options, like serializer.

Once we instantiate a client like this, we can use it just like a ClientProxy object to make requests and emit events. Let's take a look at one such request. Further down in the app.controller.ts file, notice this call. Here, we make a call to client.send() and — since, in this case, we want to manipulate the entire stream of results (what we do with those results isn't important here, we're just showing that we are wanting to access the stream) — we pipe that stream through the RxJS tap() operator.

// nestHttpApp/src/app.controller.ts
@Get('jobs-stream1/:duration')
  stream(@Param('duration') duration) {
    return this.client.send('/jobs-stream1', duration).pipe(
      // do notification
      tap(step => {
        this.notify(step);
      }),
    );
  }

Take 1 Requirements

The request above is the one represented in the animation at the top of the article. Feel free to take a look at the server-side code (nestMicroservice/src/app.controller.ts) to get familiar with what's happening here. You can run this now by issuing an HTTP request to the nestHttpClient app as shown below. Assuming you've got your multi-terminal setup going, you'll be able to follow the full flow of this request and response through all of the components.

$ # HTTPie command to request `/jobs-stream1` with base duration of 1
$ http get localhost:3000/jobs-stream1/1

You can also read the deep dive on Observables for a thorough treatment of the elegance of streaming server side observable results back to the client. For now, all we need to know is what we see in the animation above: when we make this request, we get a stream of results in the form of a sequence of messages.

Let's define our acceptance criteria for Take 1 as follows. Our ClientFaye object should let us send the '/jobs-stream1' request, and handle the response, converting the message stream back into an Observable.

In part5 we'll complete the implementation and have a fully functioning Faye Custom Transporter — both client and server side.

Take 1 Strategy

Before diving into the code, let's review our overall strategy. At a top level, it's pretty simple. For each user-land request (e.g., the this.client.send(...) call in the app.controller.ts file shown above), we want to issue the remote request and wait for the results, converting the response message(s) into an Observable stream.

Issuing the request and waiting for the results using the broker API is familiar territory. We'll just implement our very familiar STRPTQ (subscribe-to-the-response-then-publish-the-request) pattern to accomplish this.

How do we convert the broker message stream into an Observable stream? We can use the Observable/observer pattern to do this. While I won't go into a lot of detail (you can read more, including simple examples, here), the basic idea is simple, and is the primary use case for Observables. The easiest way to see this is through the code review below.

Take 1 Code Review

Let's dive into the code. Open the file nestjs-faye-transporter/src/requestor/clients/faye-client.ts.

You might notice that in this iteration, we're not actually extending any framework class. We'll need to do that in the next iteration, but it makes our code smaller and easier to focus on the core logic if we omit that in this step and just define a new class.

Class Definition

Let's start with the class members and constructor:

// nestjs-faye-transporter/src/requestor/clients/faye-client.s
export class ClientFaye {
  private readonly serializer;
  private readonly deserializer;
  protected fayeClient;

  constructor(protected readonly options) {
    this.fayeClient = this.connect();
    this.serializer = options.serializer || new IdentitySerializer();
    this.deserializer =
      options.deserializer || new IncomingResponseDeserializer();
  }

We should be familiar with the serializer and deserializer properties, and how they're initialized, from the server side. The fayeClient holds our connection to the Faye broker and is the object we'll use to make Faye client API calls. The constructor just sets this stuff up. It defaults to using the built-in (essentially no-op) serializer and deserializer if the user doesn't provide one in the options.

Implementing `send()`

With that covered, we can look at the API our class exposes. First, recall that the user-land send() call we need to support is:

// nestHttpApp/src/app.controller.ts
@Get('jobs-stream1/:duration')
  stream(@Param('duration') duration) {
    return this.client.send('/jobs-stream1', duration).pipe(
      // do notification
      tap(step => {
        this.notify(step);
      }),
    );
  }

We expose this feature, of course, with the send() method right at the top of the faye-client.ts file, as shown here:

// nestjs-faye-transporter/src/requestor/clients/faye-client.ts
  public send(pattern, data): Observable<any> {
    return new Observable(observer => {
      return this.handleRequest({ pattern, data }, observer);
    });
  }

The `send()` Method Returns an Observable

As required, send() returns an Observable. We need to delve into Observable-land here for a while. This can be confusing/frustrating if you're not familiar with the topic, but by the time you're finished here, I think the 💡 will go on.

In the send() method body, we construct an Observable using the normal Observable creation pattern¹ and return it to user-land. Applying the normal Observable creation pattern in our context means we pass the constructor a function that accesses the Faye message stream and emits the values through our Observable. Traditionally, this callback is called the observable subscriber function.

Simple Observables (the ones you'll find in a lot of Medium articles 😉) often define this observable subscriber function body right in place. For example:

return new Observable(observer => {
  observer.next('hello');
  observer.next('world');
  observer.complete();
});

But we already know that ours is going to be a little complex because the source of the data we are going to emit comes from the Faye broker. Specifically, it comes from the callback we're going to register in the Faye client API subscribe() call which is invoked each time a message comes in on the response channel. That code — which listens through Faye for inbound messages — is the code that needs to run our observer.next() (or equivalent) function call.

In these more complex scenarios, to keep the code clean, we delegate this sophisticated observer subscriber function to a specialized function. That's what we're doing in the code above by calling handleRequest().

For the moment, you need to know two things:

handleRequest() is our observable subscriber function
handleRequest() is going to make a Faye client API call to subscribe to the response topic, causing it to register to receive future inbound requests. As it receives them, it's going to emit the results in an Observable stream, analogous to our observer.next('hello') statement above. We'll see that code in a minute, but let's keep it abstract for another minute.

This can all seem kind of strange if you're not familiar with the Observable pattern¹. Let's see if it helps to think about it this way. We're setting up callbacks at several levels in a sort of cascade. At the "innermost" level, we have the Faye subscription handler callback that will run when we get an inbound message on the response channel. When this handler runs, we want to in turn trigger a callback through the Observer mechanism to a higher-level callback — the user-land response handler (the code that looks like pipe(tap(step) => {...}) in the app.controller.ts file, shown above).

Here are the connections you need to make:

inner callback (broker) called when inbound message received...
this triggers calling the observable subscription function...
which in turn calls back to our user-land client.send(...) subscription handler

That's what the send() method is setting up with the Observable it creates. One more thing about Observables to keep in mind. What we just described effectively "sets up the cascade of callbacks", but these functions are cold in the sense that all we've done is register callbacks. The act of subscribing to the Observable kicks things in motion. The act of subscribing is the event that triggers the observable subscriber function to run; when you think about that, you realize that "Aha! So this in turn causes the Faye API subscribe() call inside handleRequest() to run", and we see that this is how we start the stream flowing and turn the process from cold to hot.

Another thing to know is that if the user-land client.send() call doesn't explicitly subscribe, but instead just returns the result, Nest automatically subscribes to the result. As a ClientProxy developer, you don't have to worry about this. But as a quick aside, note that this behavior is useful as a default, and works great in most situations.

Now if you've clicked ahead to the next question and you're wondering how does Nest return a potential stream of results over HTTP? The answer is that it returns the last result in the stream. Again, in many cases this is fine. In our user-land client.send(...) call (shown again below), however, we want to access the values in the stream, thus we add a pipe() operator that allows us to operate on the stream itself, once it's subscribed:

// nestHttpApp/src/app.controller.ts
@Get('jobs-stream1/:duration')
  stream(@Param('duration') duration) {
    return this.client.send('/jobs-stream1', duration).pipe(
      // do notification
      tap(step => {
        this.notify(step);
      }),
    );
  }

If this still isn't completely intuitive, I feel your pain, but I advise you to just let it marinate in the background and proceed. Trust me, you can still follow most of this even if this still feels strange. If you really must scratch that itch, see the footnote below.

¹I'm assuming familiarity with this pattern. If you're an RxJS junky, this should be easy to follow. If you're not, I'll try not to wave my hands too much. I too struggled with this concept for a while, so I've created a small side excursion here to help with this concept. This should take you no more than 5-10 minutes to work through, and might be a helpful pre-requisite if some of this section has you a little boggled.

So let's dig into that requestHandler() method.

The Request Handler

The handleRequest() method is where we marry up our Observable pattern with our request/response handling pattern. It's easiest if we decompose this and look at a version that abstracts out the Observable code first. Then we'll see how it all wires together.

In the first pseudocode-ish version below, we can focus directly on the request/response handling pattern. You'll quickly recognize this as our old friend the STRPTQ pattern. (As a refresher, feel free to review how we implemented this functionality in our native Faye customerApp way back in Episode 1. Here's a link to that code. Or just open up customerApp/src/customer-app.ts).

With that in mind, the pseudocode-ish version of handleRequest() below should be pretty clear. I'll make a few explanatory comments below the code sample.

  public handleRequest(partialPacket, observer): Function {
    const packet = Object.assign(partialPacket, { id: uuid() });
    const serializedPacket = this.serializer.serialize(packet);

    const requestChannel = `${packet.pattern}_ack`;
    const responseChannel = `${packet.pattern}_res`;

    const subscriptionHandler = ... /* we'll fill this in shortly!  This function
      will provide the code we want to plug in as our *Faye subscription handler*.
      Guess what?  This is where we'll wire in our Observable :)
    */

    // Execute the familiar subscribe-to-response-then-publish-the-request pattern
    const subscription = this.fayeClient.subscribe(
      responseChannel,
      subscriptionHandler,
    );

    subscription.then(() => {
      this.fayeClient.publish(requestChannel, serializedPacket);
    });

    /*
      Handle any other Observable requirements
    */
  }

A few other notes from the code:

The partialPacket parameter has most of the elements of our outbound request message. It's been invoked from send() with an object like:
```
{
  pattern: '/jobs-stream1/',
  data: 1
}
```
We complete the formation of our outbound packet by using uuid() to add that pesky id parameter (that we still haven't talked about — I promise to address this in Part 5!).
After running our serializer, the request packet is ready to send.
In Faye, the subscribe() call returns a promise that resolves when the Faye server returns an acknowledgement.
Upon that acknowledgement (promise resolution), we publish the serialized packet we built earlier.

Now let's deal with marrying that up with our Observable. Here's the final version of handleRequest().

  public handleRequest(partialPacket, observer): Function {
    const packet = Object.assign(partialPacket, { id: uuid() });
    const serializedPacket = this.serializer.serialize(packet);

    const requestChannel = `${packet.pattern}_ack`;
    const responseChannel = `${packet.pattern}_res`;

    const subscriptionHandler = rawPacket => {
      const message = this.deserializer.deserialize(rawPacket);
      const { err, response, isDisposed } = message;
      if (err) {
        return observer.error(err);
      } else if (response !== undefined && isDisposed) {
        observer.next(response);
        return observer.complete();
      } else if (isDisposed) {
        return observer.complete();
      }
      observer.next(response);
    };

    const subscription = this.fayeClient.subscribe(
      responseChannel,
      subscriptionHandler,
    );

    subscription.then(() => {
      this.fayeClient.publish(requestChannel, serializedPacket);
    });

    return () => {
      this.fayeClient.unsubscribe(responseChannel);
    };
  }

Let's focus in on the newly added part:

    const subscriptionHandler = rawPacket => {
      const message = this.deserializer.deserialize(rawPacket);
      const { err, response, isDisposed } = message;
      if (err) {
        return observer.error(err);
      } else if (response !== undefined && isDisposed) {
        observer.next(response);
        return observer.complete();
      } else if (isDisposed) {
        return observer.complete();
      }
      observer.next(response);
    };

Let's review the purpose of this chunk of code. This is (in terms of the concepts we started out with), our inner callback — the function that will be passed in our Faye client API subscribe() call to handle inbound messages. As we handle each message in this function, this code converts the message back into a value in the Observable stream. We do that by taking the following steps for each inbound message received from Faye:

Deserialize it and destructure it, leaving us with an error, response and isDisposed value for each inbound message
Based on the contents of the message, determine if the inbound message stream has errored or is complete
Emit an Observable value, based on step 2

As in any normal observable subscriber function, we handle these cases with the observer.error(), observer.next() and observer.complete() calls. If this confuses you, remember that all of this plugs into the Observable we created in our send() method. You might take a minute to grok all of this, and if it's still a little fuzzy, you can visit my mini Observable factory tutorial here.

Handling Unsubscribe

Any observable subscriber function must return a way to unsubscribe from the Observable. This is part of the Observable creation pattern. The unsubscribe method can be called by the user; more often, it is called automatically when the Observable stream ends (e.g., when the final message is returned from our remote responder).

The last bit of code in handleRequest() fulfills this contract for our observable subscriber function:

    return () => {
      this.fayeClient.unsubscribe(responseChannel);
    };

The net effect of this is that when the stream completes (e.g., when the inbound message has isDisposed set to true), we unsubscribe from the response channel.

The Rest of the Code

The only other thing left is the connection management to the Faye broker. That's handled in the connect() method, which for now we're calling in the constructor. This should be pretty understandable. The only slightly funky thing is that we use destructuring to extract out the url and the options object from the constructor options so we can easily format our call to fayeClient.connect().

Connection management will become more important and consequently more sophisticated in our next iteration.

  public connect() {
    if (this.fayeClient) {
      return this.fayeClient;
    }
    const { url, serializer, deserializer, ...options } = this.options;
    this.fayeClient = new faye.Client(url, options);
    this.fayeClient.connect();
    return this.fayeClient;
  }

Acceptance Testing

We're ready to test our code. In previous articles, I gave a lot of suggestions about managing your multiple terminals. Going forward, I'll simply indicate the steps you should take, and assume you've got a working environment with the appropriate logical terminals set up. Here are the steps to run our test:

Make sure our Faye server is up and running in one terminal.
Make sure the nestMicroservice app is running in one terminal.
Make sure the nestHttpApp app is running in one terminal.

Now, issue a request (expressed below as an HTTPie request, but use your favorite HTTP testing tool):

$ # HTTPie request to get the jobs stream with a base duration of 1
$ http get localhost:3000/jobs-stream1/1

If all is working, you should get a coordinated set of logs in each of the terminal windows indicating that we have, indeed, successfully subscribed to the results of our remote request. You'll see our RxJS pipe come into play, making simulated calls to our notifier as each step in the job completes (as each message in the stream arrives). 🍺!

Limitations of Take 1

We knowingly deferred a few things to Take 2 to keep focused on the core requirements and implementation of a client. We'll address them in Part 5. But there's one big issue lurking. Any guesses on what that might be? I've definitely foreshadowed this along the way (hint: repeatedly deferring the discussion of that pesky message id).

OK, let's get right to it. Ask yourself this: since our client is embedded in an HTTP app that is itself responding to multiple incoming requests, many of which in turn are triggering these client.send(...) remote requests, how does the transporter client keep track of all of the outstanding remote requests and route them back to the correct originating (HTTP) request handler (i.e., the code with the original client.send() call?

Let's try it out. You'll need one extra terminal window for this to enable launching two simultaneous requests. You'll notice that in this branch, there is an extra route in nestHttpApp. That route makes a new request ('/race'), which has a corresponding @MessagePattern() in nestMicroservice. The point of these new features is to enable us to launch multiple requests that overlap in their execution. Feel free to take a look at the code (these new routes/handlers are at the bottom of their respective Controller files), but all you really need to know is this: the race route in our nestHttpApp takes two parameters:

a requestId
a delay period (in seconds) specifying how long the server should wait before responding

For example, try making the following request (HTTPie format):

$ http get localhost:3000/race/1/10

This will return the following response after 10 seconds (cid in the response corresponds to the first route parameter, and delay in the response corresponds to the second parameter (expressed in milliseconds)):

{
    "cid": "1",
    "customers": [
        {
            "id": 1,
            "name": "fake"
        }
    ],
    "delay": 10000
}

With this in place, we can run two overlapping requests. Let's go ahead and run the following two requests. You'll need to do this in two separate terminals, and you should launch the second request within 5 seconds of launching the first (to achieve the desired overlap):

$ # run this in terminal 1, and be prepared to run the second one in terminal 2 a few seconds later
$ http get localhost:3000/race/1/10
$ # run this one in terminal 2, within a couple of seconds of running the top 1
$ http get localhost:3000/race/2/5

The results may (well, maybe not with all the foreshadowing 😉) surprise you. Both requests return the following result:

{
  "cid": "2",
  "customers": [
      {
          "id": 1,
          "name": "fake"
      }
  ],
  "delay": 5000
}

This is obviously incorrect for the first request. The issue demonstrates what I'll refer to as the multiplexing challenge. That is, we are multiplexing our requests for each pattern through a single pair of channels. When we issue request #1, it subscribes to a response on the '/race_res' channel. Request #2 does the same thing, and since request #2 finishes first, the request #1 subscription gets request #2's result (the wrong answer!) on the response channel.

Also, there's another nagging question... why do both subscribers get the result? Hmmm... clearly something isn't working quite right here. We'll have to dig in and work on this in our next (and final!) revision!

What's Next

With these issues in mind, we're ready to build a final version of the Faye Custom Transporter client component, completing our development project. In Part 5 we cover:

Solving the multiplexing issue
Client library connection management
Event handling on the client side

Feel free to ask questions, make comments or suggestions, or just say hello in the comments below. And join us at Discord for more happy discussions about NestJS. I post there as Y Prospect.

Part 3: Completing the Server Component

John Biundo — Mon, 02 Mar 2020 18:02:41 +0000

John is a member of the NestJS core team

Introduction

This is Part 3 of a six-part series. If you landed here from Google, you may want to start with Part 1.

In this article, we build the final iteration of the server component of our Faye Custom Transporter.

Get the Code

All of the code in these articles is available here. As always, these tutorials work best if you follow along with the code. The README covers all the steps you need to get the repository, build the apps, and follow along. It's easy! I strongly encourage you to do so. Note that each article has a corresponding branch in the repository. For example, this article (Part 3), has a corresponding branch called part3. Read on, or get more details here on using the repository.

Git checkout the current version

For this article, you should git checkout the branch part3. You can get more information about the git repository branches here.

Build the Apps for This Part

$ # from root directory of project (e.g., transporter-tutorial, or whatever you chose as the root)
$ sh build.sh

Overview

I ended the last article with a discussion of the shortcomings of our Take 1 (first iteration) implementation of the Faye Custom Transporter server component. In this article, we'll address those shortcomings. Most of this work is fairly straightforward, but I want to take a moment to discuss the "Observable issue" a bit further.

You could actually ignore this issue entirely. The server component we built in the last chapter is functional, and we'll clean up a few loose ends here to make it typesafe, etc. However, by ignoring proper handling of Observables, you'd be leaving out a feature that you may not miss until you really need it. Plus, let me hit you with the full commercial:

It's super fun to understand how it works, and if you're not an RxJS whiz, don't worry, that won't stop you (and you might even pick up a few nifty tricks about RxJS along the way)
It builds a deeper appreciation and understanding of the elegance of the NestJS framework
It's really, really easy to implement, once you get on the happy path
The feature provides some amazing benefits, virtually for free
Your transporter really should be plug-and-play compatible with other transporters to be a good Nest citizen (Nestizen? 😃)

I could go on and on, but instead, I've encapsulated all the concrete examples and detailed analysis here to try to keep this tutorial on track. Suffice to say, I strongly encourage you to follow that link and take that little side trip to both understand why this is an important step — and because you might even get inspired to use Nest microservices in some new and interesting ways.

With all that said, let's jump in!

Addressing Observables

After that build-up, you might think we're about to embark on some form of brain surgery (maybe I should say "🚀 science" in keeping with the trekkie theme? 😉). Fortunately, that's not really the case. Let's start by considering the requirements.

At the highest level, we need to consider the possibility that the user-land handler returns a stream (Observable that emits multiple values). Our earlier version won't handle that (as we saw at the end of the last article). The problem lies with our response message publishing logic being unaware of a response stream. It simply awaits a single response and publishes. Let's take a quick look at the part of the code that publishes to refresh ourselves:

// nestjs-faye-transporter/src/responder/transporters/server-faye.ts
// on branch part2
public getMessageHandler(pattern: string, handler: Function): Function {
  return async message => {
    const inboundPacket = this.deserializer.deserialize(message);
    const response = await handler(inboundPacket.data);
    const outboundRawPacket = {
      err: null,
      response,
      isDisposed: true,
      id: (message as any).id,
    };
    const outboundPacket = this.serializer.serialize(outboundRawPacket);
    this.fayeClient.publish(`${pattern}_res`, outboundPacket);
  };
}

We could try to handle this streaming ourselves, but it's a tad tricky bit of code, and as such, a great candidate for the framework to handle for us. Which it does! As always with frameworks, if we want to delegate a responsibility, we have to invert our thinking a bit. Rather than publishing directly, we instead need to build a publish function and hand it to the framework

Now, here's the new, improved, Observable-aware version. This is what you'll see now (on the current branch, part3) when you open the nestjs-faye-transporter/src/responder/transporters/server-faye.ts file:

//nestjs-faye-transporter/src/responder/transporters/server-faye.ts
// on branch part3
public getMessageHandler(pattern: string, handler: Function): Function {
  return async (message: ReadPacket) => {
    const inboundPacket = this.deserializer.deserialize(message);
    const fayeCtx = new FayeContext([pattern]);

    const response$ = this.transformToObservable(
      await handler(inboundPacket.data, fayeCtx),
    ) as Observable<any>;

    const publish = (response: any) => {
      Object.assign(response, { id: (message as any).id });
      const outgoingResponse = this.serializer.serialize(response);
      return this.fayeClient.publish(`${pattern}_res`, outgoingResponse);
    };

    response$ && this.send(response$, publish);
  };
}

Let's discuss the differences. For now, ignore the fayeCtx object — we'll cover that below.

The reorganized code does two main things:

It converts our handler response to an Observable
It builds a publish function that it can delegate to the framework (which will then use it to publish a stream if that's what the user-land handler returns)

For the first item, we call a built-in (inherited from Server) utility to convert the response to an Observable.

const response$ = this.transformToObservable(
  await handler(inboundPacket.data, {}),
) as Observable<any>;

This step is pretty intuitive, and the utility is correspondingly pretty simple. Feel free to take a moment to review the transformToObservable() method's source code.

Building the publish function is also fairly straightforward.

const publish = (response: any) => {
  Object.assign(response, { id: (message as any).id });
  const outgoingResponse = this.serializer.serialize(response);
  return this.fayeClient.publish(`${pattern}_res`, outgoingResponse);
};

This code is very similar to the way we published in the Take 1 version (from branch part2), but now packaged as a function we can hand off to the framework. Study it for a moment and make sure you understand it — it should be pretty straightforward. The Object.assign(...) code's job is to copy the id field from the inbound request to the outbound response. We did that in the earlier version as well, but there we were defining the response object in place, so the technique was a little different.

Note: We still haven't discussed why we need that id. For now, just trust the process 😃. Don't worry, we'll cover this in Part 5.

Finally, we delegate the publishing step to the framework with the final line:

response$ && this.send(response$, publish);

The send() method is inherited from Server; it takes care of the details of dealing with streams.

The significance of the previous sentence should not be overlooked. To fully appreciate both why this is so useful and how the framework makes this as easy as delegating a publish() function, you really should read this deep dive.

Anyway, let's discuss what the framework is doing for us. As mentioned, send() is inherited from Server. See here for the full source code of the Server. Let's reproduce it below to get a feel for it.

Note: we don't need to fully understand how this method works, and I won't belabor it here. It's fair to treat it as a black box, as long as we understand how to use it. We'll make sure we do!

// from https://github.com/nestjs/nest/blob/master/packages/microservices/server/server.ts
  public send(
    stream$: Observable<any>,
    respond: (data: WritePacket) => void,
  ): Subscription {
    let dataBuffer: WritePacket[] = null;
    const scheduleOnNextTick = (data: WritePacket) => {
      if (!dataBuffer) {
        dataBuffer = [data];
        process.nextTick(() => {
          dataBuffer.forEach(buffer => respond(buffer));
          dataBuffer = null;
        });
      } else if (!data.isDisposed) {
        dataBuffer = dataBuffer.concat(data);
      } else {
        dataBuffer[dataBuffer.length - 1].isDisposed = data.isDisposed;
      }
    };
    return stream$
      .pipe(
        catchError((err: any) => {
          scheduleOnNextTick({ err, response: null });
          return empty;
        }),
        finalize(() => scheduleOnNextTick({ isDisposed: true })),
      )
      .subscribe((response: any) =>
        scheduleOnNextTick({ err: null, response }),
      );
  }

While at first a little obscure, this really isn't that hard to understand if you work through it, though it does require some understanding of observables. Essentially, we are subscribing to the response coming back from our user-land handler. If the subscription produces multiple values (a stream), we buffer them (dataBuffer), then publish each value using the publish() function we built a moment ago (inside this method the publish() function is accessed as the respond parameter). When the stream completes, we signify that by setting isDisposed to true on the final emitted response.

Handle Events

Thus far we've skipped handling events — those user-written handlers that are decorated with @EventPattern(...). Let's take care of those now. They turn out to be a lot easier than request/response type messages precisely because they don't require any response. Take a look at the bindHandlers() method of server-faye.ts. The comments should pretty much explain the simple change we've made here to handle events.

// nestjs-faye-transporter/src/responder/transporters/server-faye.ts
public bindHandlers() {
  /**
   * messageHandlers is populated by the Framework (on the `Server` superclass)
   *
   * It's a map of `pattern` -> `handler` key/value pairs
   * `handler` is the handler function in the user's controller class, decorated
   * by `@MessageHandler()` or `@EventHandler`, along with an additional boolean
   * property indicating its Nest pattern type: event or message (i.e.,
   * request/response)
   */
  this.messageHandlers.forEach((handler, pattern) => {
    // In this version (`part3`) we add the handler for events
    if (handler.isEventHandler) {
      // The only thing we need to do in the Faye subscription callback for
      // an event, since it doesn't return any data to the caller, is read
      // and decode the request, pass the inbound payload to the user-land
      // handler, and await its completion.  There's no response handling,
      // hence we don't need all the complexity of `getMessageHandler()`
      this.fayeClient.subscribe(pattern, async (rawPacket: ReadPacket) => {
        const fayeCtx = new FayeContext([pattern]);
        const packet = this.parsePacket(rawPacket);
        const message = this.deserializer.deserialize(packet, {
          channel: pattern,
        });
        await handler(message.data, fayeCtx);
      });
    } else {
      this.fayeClient.subscribe(
        `${pattern}_ack`,
        this.getMessageHandler(pattern, handler),
      );
    }
  });
}

Two additional (unrelated) changes that show up on this branch are:

The use of the FayeContext object. We'll cover that in the next section.
Passing { channel: pattern } to the deserialize() call. We'll cover that later in the article.

Subscription Context

Consider what happens when we subscribe to a topic via the broker API. With Faye, it's straightforward. We register a Faye subscription handler on a topic, and when a message matching that topic arrives, our handler is called with the message payload containing only what the publisher sent.

For some brokers, the subscription handler can receive additional context about the message — sometimes in the callback parameters, and sometimes in the message payload. For example, with NATS, the message payload packs the actual content sent by the publisher in a top-level data property, and adds two additional top-level properties with context metadata: one describing the topic the caller was subscribed to, and one containing the optional reply topic. We refer to this metadata as Context, and Nest allows you to pass this information through to the user-land handler (method decorated by @MessagePattern() or @EventPattern()).

Since Faye simply doesn't provide any such metadata, we'll demonstrate this behavior by passing the channel in this context object. This is not particularly useful in this case, as we're not actually providing any new information, but with Faye it's the best we can do to demonstrate the concept of the Context object.

Creating the context object

For any given broker, the technique for creating the context object may vary, based on the discussion above. In the end, you'll need to extract any context information (from the message payload, if context is encoded there, or from the subscribe() callback parameters) and pass it to the user-land handlers. We use a custom object to pass context. Open the file nestjs-faye-transporter/src/responder/ctx-host/faye-context.ts. It's a pretty simple object that extends BaseRpcContext.

// nestjs-faye-transporter/src/responder/ctx-host/faye-context.ts
import { BaseRpcContext } from '@nestjs/microservices/ctx-host/base-rpc.context';

type FayeContextArgs = [string];

export class FayeContext extends BaseRpcContext<FayeContextArgs> {
  constructor(args: FayeContextArgs) {
    super(args);
  }

  /**
   * Returns the name of the Faye channel.
   */
  getChannel() {
    return this.args[0];
  }
}

Essentially, it keeps an array of context values — in our case just strings, though you can specify a more complex type if you need to. For example, MQTT uses:

// from https://github.com/nestjs/nest/blob/master/packages/microservices/ctx-host/mqtt.context.ts
type MqttContextArgs = [string, Record<string, any>];

export class MqttContext extends BaseRpcContext<MqttContextArgs> {
  constructor(args: MqttContextArgs) {
    super(args);
  }

For each context value type (e.g., "channel", in our case, used for carrying the channel/pattern name), you supply a getter function. Remember, this is a simple data structure as there are generally very few context values, so your getter is just pulling out a particular row of the array.

Passing Context to User-land Handlers

With this in place, we then need to simply find the right places to:

Instantiate a FayeContext object with the correct values
Pass it to the user-land handlers

We already encountered this behavior in the previous sections. Let's look again at the getMessageHandler() method, focusing on the relevant snippet from that method:

  return async (message: ReadPacket) => {
    const inboundPacket = this.deserializer.deserialize(message);
    const fayeCtx = new FayeContext([pattern]);

    const response$ = this.transformToObservable(
      await handler(inboundPacket.data, fayeCtx),
    ) as Observable<any>;

It should be clear how we're using an instance of FayeContext here. We instantiate it with the relevant context, and then pass the instance to our user-land handler method. The code above takes care of context for request/response type handlers. We also need to handle this appropriately for events. We saw where that happened in the last section, in the bindHandlers() method. Here's the relevant snippet:

    if (handler.isEventHandler) {
      // The only thing we need to do in the Faye subscription callback for
      // an event, since it doesn't return any data to the caller, is read
      // and decode the request, pass the inbound payload to the user-land
      // handler, and await its completion.  There's no response handling,
      // hence we don't need all the complexity of `getMessageHandler()`
      this.fayeClient.subscribe(pattern, async (rawPacket: ReadPacket) => {
        const fayeCtx = new FayeContext([pattern]);
        const packet = this.parsePacket(rawPacket);
        const message = this.deserializer.deserialize(packet);
        await handler(message.data, fayeCtx);
      });

Remember, where and how you gather the context for any particular broker may vary, but the concept remains the same.

Testing Context

In this branch (part3), we provided a new project, nestHttpApp. This is a simple Nest HTTP app that includes a few routes for testing our nestMicroservice over the Faye transporter. It's worth mentioning that this branch also includes a full final version of the ClientFaye subclass of ClientProxy — this is necessary for our nestHttpApp to instantiate a client to be able to run ClientProxy#send() and ClientProxy#emit() calls. We'll actually build that ClientFaye component from scratch in the next two articles, but it's sitting there in this branch to help you test. With this in place, you can see the context by following these steps:

Start up the nestHttpApp

If you ran the build.sh script as described at the beginning of this chapter, all of this project's dependencies have been installed. If not, simply open a terminal in the top level nestHttpApp directory and run npm install. Then, start the application with npm run start:dev.
Issue an HTTP request like:

GET /customers

Note: You can do this in any number of ways — via a browser, browsing to localhost:3000/customers, via something like HTTPie - my favorite, or with a desktop program like Postman. See here for more help on running these requests.

Keep an eye on the nestMicroservice app's console log. You should see a line like the following, demonstrating that we've successfully passed the FayeContext data to the handler:

[AppController] Faye Context: {"args":["/get-customers"]}"

Deserialization Options

The transporter main class we inherit from (Server) has hooks for calling custom serialization and deserialization methods to encode/decode broker messages. I cover this topic extensively in this article. The main use case is for translating between non-Nest (i.e., external) message formats and the internal Nest microservice transporter message protocol. One thing we need to do to fully enable custom deserialization (i.e., let users hook their own custom deserializer class) with our custom transporter is fully implement the call to the Deserializer#deserialize() method. Here's its interface:

export interface Deserializer<TInput = any, TOutput = any> {
  deserialize(value: TInput, options?: Record<string, any>): TOutput;
}

This means we can pass in some context via a second argument that the user can access as the options parameter to utilize in their custom deserialization code. Usually this context is just the channel (pattern), but it can also include other contextual information. For example, the NATS transporter passes the following information (note the replyTo property):

const message = this.deserializer.deserialize(rawMessage, { channel, replyTo });

In our case, with Faye, we'll stick to the usual pattern and pass the channel. This is done in bindHandlers(), where we added the deserialization context as shown below:

    const message = this.deserializer.deserialize(packet, {
      channel: pattern,
    });

With this in place, any user-written deserializer will have access to the channel name to make decisions about transforming inbound messages. I encourage you to read this article for a deeper understanding of deserialization to help inform what you should pass as deserialization options.

Type Safety

In this branch, we added types to several calls to help ensure type safety when end users use the custom transporter. You can examine the code and explore how we use the following newly imported interfaces:

// nestjs-faye-transporter/src/responder/transporters/server-faye.ts
import { FayeClient } from '../../external/faye-client.interface';
import { FayeOptions } from '../../interfaces/faye-options.interface';

Let's take a quick moment to examine the FayeOptions interface:

// nestjs-faye-transporter/src/interfaces/faye-options.interface.ts
import { Serializer, Deserializer } from '@nestjs/microservices';

export interface FayeOptions {
  /**
   * faye server mount point (e.g., http://localhost:8000/faye)
   */
  url?: string;
  /**
   * time in seconds to wait before assuming server is dead and attempting reconnect
   */
  timeout?: number;
  /**
   * time in seconds before attempting a resend a message when network error detected
   */
  retry?: number;
  /**
   * connections to server routed via proxy
   */
  proxy?: string;
  /**
   * per-transport endpoint objects; e.g., endpoints: { sebsocket: 'http://ws.example.com./'}
   */
  endpoints?: any;
  /**
   * backoff scheduler: see https://faye.jcoglan.com/browser/dispatch.html
   */
  // tslint:disable-next-line: ban-types
  scheduler?: Function;
  /**
   * instance of a class implementing the serialize method
   */
  serializer?: Serializer;
  /**
   * instance of a class implementing the deserialize method
   */
  deserializer?: Deserializer;
}

These options (modulo serializer and deserializer, which are consumed by the ServerFaye class directly) are passed through to the Faye client library in createFayeClient() to customize the Faye connection:

  public createFayeClient(): FayeClient {
    // pull out url, and strip serializer and deserializer properties
    // from options so we conform to the `faye.Client()` interface
    const { url, serializer, deserializer, ...options } = this.options;
    return new faye.Client(url, options);
  }

The other benefit of this interface is that it provides intellisense for users constructing the options object in the call to NestFactory.createMicroservice(...) in the main.ts file.

Figure 1: Editor Intellisense

Error Handling

What should we do if the server loses its connection to the broker (e.g., the Faye broker goes offline)? This is a good question that probably deserves deeper discussion than what we'll cover here. The Faye client library appears to be fairly robust at re-establishing a connection, including buffering failed attempts, so we are going to simply defer to it, and log an error message to the console whenever a connection goes down. To handle this, we simply add a listener for a disconnect event and log the message:

  public handleError(stream: any) {
    stream.on(ERROR_EVENT, (err: any) => {
      this.logger.error('Faye Server offline!');
    });
  }

Checkpoint

We covered a lot of ground in this chapter. Because we provided the nestHttpApp and a full implementation of the Faye flavored ClientProxy, you now have a fully working Faye Custom Transporter. You can feel free to experiment at large with the nestHttpApp and nestMicroservice. Try issuing HTTP requests to the nestHttpApp like:

GET /customers

and

POST /customer // <-- with a payload

See here for more help on running these and similar tests..

As mentioned earlier in this article, there's also a series of routes related to exploring how microservices handle promises and observables when they are returned from a request. You can read more here.

What's Next

In Part 4), we turn our attention to the "client" side of the transporter. We'll build the Faye-flavored ClientProxy from scratch. Like the server construction we've done, we'll break this into two parts so we can work through the main concepts first, then handle some edge cases and details in the subsequent part.

Feel free to ask questions, make comments or suggestions, or just say hello in the comments below. And join us at Discord for more happy discussions about NestJS. I post there as Y Prospect.

Part 2: Basic Server Component

John Biundo — Mon, 02 Mar 2020 17:40:05 +0000

John is a member of the NestJS core team

Introduction

This is Part 2 of a six-part series. If you landed here from Google, you may want to start with Part 1.

In this article, we build the first iteration of the server component of our Faye Custom Transporter. We'll then test it with a simple Nest responder (microservice) app.

Get the Code

All of the code in these articles is available here. As always, these tutorials work best if you follow along with the code. The README covers all the steps you need to get the repository, build the apps, and follow along. It's easy! I strongly encourage you to do so. Note that each article has a corresponding branch in the repository. For example, this article (Part 2), has a corresponding branch called part2. Read on, or get more details here on using the repository.

Git checkout the current version

For this article, you should git checkout the branch part2. You can get more information about the git repository branches here.

Build the Apps for This Part

$ # from root directory of project (e.g., transporter-tutorial, or whatever you chose as the root)
$ sh build.sh

Organizing the Code

Even though the scope of this tutorial is relatively small — at the end of the day we'll just be creating a few main classes — we're going to be accumulating a number of assets along the way to test and exercise our code. It's time to think a little bit about code organization. First, let's take a look at what we'll end up with:

Our Faye Custom Transporter is going to consist of some interfaces and constants, some serializers/deserializers (discussed extensively in the previous article series), and the main components: a client component (class) providing the ClientProxy subclass we can use in any Nest requestor apps and a server component (class) we can use in any Nest responder apps.
The native apps we wrote in the previous article that let us explore the Faye client API and give us some test drivers.
A pair of regular Nest apps that will host and exercise our transporter code as we build it: a Nest requestor that is a simple Nest HTTP app called nestHttpApp and a Nest responder that is a classic Nest microservice, called nestMicroservice. (Note: see the previous article series for more on this terminology, but think of a Nest requestor as an app that makes remote requests over some non-HTTP transport (like Faye) and a Nest responder as a Nest application that listens for inbound requests over that transport).

Where to keep all of these assets?

The main thing to think about now is where to keep the Faye Custom Transporter components. Since we'll want to easily re-use them across different Nest apps, the logical place is an NPM package. To that end, when we work on any of those components, they'll go into that package.

The Custom Transporter Package

At this point, you should be on the part2 branch of the repo that you cloned (read more here for details). Once you've checked out that branch, you'll notice a new folder called nestjs-faye-transporter. This is organized as an NPM package (you can learn more about creating NestJS-friendly NPM packages in my article Publishing NestJS Packages with npm, including how the tsconfig.json file, package.json scripts and properties, and file organization all interact to create a reusable NPM package).

Here's a run-down on the contents of the nestjs-faye-transporter package, and how we'll use each:

The scripts (package.json) let you build (npm run build or npm run build:watch) and, if you want, publish (npm publish) the package.
Our workflow, in dev, will be to build the package and then use npm link to install it in our Nest apps. More on this soon.
The code is organized in a directory structure as follows:
- Code common to our client and server is in top level files or folders, like src/external (holds things like interfaces to the Faye client library which are external to the Nest environment), src/interfaces (holds interfaces used within the Nest constructs), and src/constants.
- All code defining the client is in the src/requestor folder. Ultimately, this will consist of a sub-class of ClientProxy from the @nestjs/microservices package — the class which provides methods like client.send(...) and client.emit(...)) — and a variety of related classes.
- All code defining the server is in the src/responder folder. Ultimately, this will consist of a sub-class of Server from the @nestjs/microservices package — the class which implements a transporter's server-side strategy — and a variety of related classes.

First Iteration (Take 1) of the Server Component

Let's get started building the "server" side of the equation — the part of the custom transporter that you'll use to help build Nest microservices (AKA Nest responders).

Using a Custom Transporter Strategy

Before we start writing transporter code, let's first take a look at how we're going to use a custom transporter. To see this in action, let's look at the nestMicroservice app (this app has been added in this branch). The code for it is in the top-level nestMicroservice folder.

Instantiating the Custom Transporter

Open the src/main.ts file. Notice the structure of the createMicroservice call:

// nestMicroservices/src/main.ts
const app = await NestFactory.createMicroservice(AppModule, {
  strategy: new ServerFaye({
    url: 'http://localhost:8000/faye',
    serializer: new OutboundResponseIdentitySerializer(),
    deserializer: new InboundMessageIdentityDeserializer(),
  }),
});

This is very similar to the structure used for any built-in Nest transporter. For example, with MQTT it looks like:

const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.MQTT,
  options: {
    host: 'localhost',
    port: 1883,
    serializer: new OutboundResponseIdentitySerializer(),
    deserializer: new InboundMessageIdentityDeserializer(),
  },
});

It should be clear what we're doing: instead of passing the normal transporter options object with transport and options properties, we pass a single property, strategy, whose value is an instance of our custom transporter class. This class takes a single options object in its constructor, pretty much mirroring the options you pass to the options property with the built-in transporters.

Using the Custom Transporter

Let's briefly review how our nestMicroservice app works. Its only real job is to take in a '/get-customers' message and return a list of customers. It performs the exact same function as the customerApp we built in Part 1. The action is in the nestMicroservices/src/app.controller.ts file:

// nestMicroservices/src/app.controller.ts
@Controller()
export class AppController {
  logger = new Logger('AppController');

  /**
   * Register a message handler for '/get-customers' requests
   */
  @MessagePattern('/get-customers')
  async getCustomers(data: any): Promise<any> {
    const customers =
      data && data.customerId
        ? customerList.filter(cust => cust.id === parseInt(data.customerId, 10))
        : customerList;
    return { customers };
  }
}

All it really does is check for the existence of a customerId in the inbound message, and if it exists, uses it to filter the customer list it returns.

Take 1 Requirements

Alright, we're finally ready to look at our ServerFaye class — the one we'll instantiate and pass as the value of that strategy property above. We'll take this in a couple of steps.

This iteration (Take 1) is going to be the absolute bare-bones class needed to implement the server component of the Faye transporter. We're going to keep our first version basic so that we can focus on the core flow. Our first cut will:

minimize error handling
not yet utilize some of the nice features of the framework that make our code really robust
not handle events (e.g., inbound messages coming from client.emit(...))
not really be type safe (we omit a bunch of typing to declutter the code)

To state it in terms of requirements: the goal is to respond to a well-formed inbound request (i.e., a request from a request-response style message). We'll test this requirement by replacing our native customerService responder app from the last article with our nestMicroservices app running our new Faye Custom Transporter, and sending it the same '/get-customers' request from our native customerApp.

In Part 3, we'll complete the implementation and have a fully functioning Faye Custom Transporter (server component). At that point, you'll also have all of the concepts in place to write your own custom transporter server component, as well as the ability to look inside the built-in transporters (like the MQTT transporter server) and understand what's going on. That will prepare you for even more adventures, like customizing the Nest built-in transporters to add features— the subject of my next NestJS microservice tutorial (already underway, and coming very soon 💥)!

Take 1 Code Review

Let's dive into the code. Open the file nestjs-faye-transporter/src/responder/transporters/server-faye.ts.

An important concept is that this class extends Server (view this parent class here: @nestjs/microservices/server/server.ts). We won't walk through all the details of the Server superclass, but the main things to know are:

1) Our ServerFaye class inherits some properties and methods from the built-in Server class
2) After being instantiated in the main.ts file, the ServerFaye instance is essentially managed for us as part of the Nest lifecycle. This means that when the application bootstraps, and our custom ServerFaye class is instantiated, its inherited properties are populated with data, and the framework calls the entry point of our class. That entry point is the listen() method.

So let's dig into the ServerFaye class, starting with the listen() entry point. Its job is to make a connection to the broker, and then run start(), which is where the fun begins.

The interesting thing in start() is the call to this.bindHandlers(). Take a look at that method:

// nestjs-faye-transporter/src/responder/transporters/server-faye.ts
  public bindHandlers() {
    /**
     * messageHandlers is populated by the Framework (on the `Server` superclass)
     *
     * It's a map of `pattern` -> `handler` key/value pairs
     * `handler` is the handler function in the user's controller class, decorated
     * by `@MessageHandler()` or `@EventHandler`, along with an additional boolean
     * property indicating its Nest pattern type: event or message (i.e.,
     * request/response)
     */
    this.messageHandlers.forEach((handler, pattern) => {
      // only handling `@MessagePattern()`s for now
      if (!handler.isEventHandler) {
        this.fayeClient.subscribe(
          `${pattern}_ack`,
          this.getMessageHandler(pattern, handler),
        );
      }
    });
  }

  public getMessageHandler(pattern: string, handler: Function): Function {
    return async message => {
      const inboundPacket = this.deserializer.deserialize(message);
      const response = await handler(inboundPacket.data);
      const outboundRawPacket = {
        err: null,
        response,
        isDisposed: true,
        id: (message as any).id,
      };
      const outboundPacket = this.serializer.serialize(outboundRawPacket);
      this.fayeClient.publish(`${pattern}_res`, outboundPacket);
    };
  }

The comments should help with understanding what's happening, but at a high level, the strategy should be pretty clear:

Iterate over all of our "message handlers"*. These are the user-land Controller methods decorated with @MessagePattern(); a list of them is made available to us (in this.messageHandlers) by the framework, which discovers them using introspection with the Reflect API during the bootstrap process.
For each one, subscribe to the inbound channel (the _ack form of the topic). Remember, a Faye client's subscribe() call registers a Faye subscription handler to be invoked whenever the Faye client library receives an inbound message matching this topic. So this is the step where we map patterns to handlers. In short, this is our "router".
The Faye subscription handler in step 2, when invoked, runs the actual user supplied handler (the user-land handler method decorated with something like @MessagePattern('/get-customer')), and returns the result it gets from that method. When we say "returns", we mean publishes a reply on the outbound channel (the _res form of the topic).
Along the way, we run our deserializer on the inbound message, our serializer on the outbound response, and we package up the data produced by the user's pattern handler in an appropriately shaped standard Nest transporter message object.

*For Take 1, we are omitting event handlers (methods decorated with @EventPattern(...)). We'll handle these in Take 2 of our server component, in the next article.

The only slightly tricky part is the call to getMessageHandler(). This is just a higher-order function that is returning us the actual Faye subscription handler function.

If all this looks somewhat familiar, it's because we're basically following the same STRPTQ (subscribe-to-the-response-then-publish-the-request) pattern we used in the native customerService app in the previous article (Part 1).

Acceptance Testing

We should be ready to test out our code. We need to do a tiny bit of setup first. This is going to run best if you have four separate terminals open so you can watch things unfold across the whole process as the various pieces communicate.

Now's a good time to mention a couple of things about the development setup:

I run this kind of stuff on an Ubuntu machine. There's nothing platform-specific anywhere in the Nest-related code, but, I find running builds, starting and stopping processes, running multiple terminal sessions, etc., to be much smoother on Linux. You can run this wherever you want, but if you're not on Linux, you may have to make slight adjustments to your package.json, or other actual build steps.
Since you kind of need to run multiple terminal sessions to see the full impact, I strongly recommend Tmux. I mentioned this briefly at the end of the last article, so you now know I feel strongly about it 😃. You can instead start multiple terminal programs, or use a tabbed console program if you prefer, but if you want what I think is the best DX for this kind of work, checkout Tmux. I covered some detailed Tmux recommendations in the last article series.

In the following steps, I'll reference these (logical) terminals as:

Terminal 1: run the Faye broker here
Terminal 2: run live builds (npm build:watch) of the transporter server code we're working on here
Terminal 3: run the "requestor code" here. This is usually the customerApp; in the future, we'll also interact with nestHttpApp (making it the requestor) using HTTPie commands from the OS prompt (you can also use something like Postman or curl to issue HTTP requests, of course). We can use one terminal for this since we don't typically run both the customerApp and the nestHttpApp at the same time
Terminal 4: run the nestMicroservice Nest responder application here (this is the plain old Nest microservice app that will be using our new Faye Custom Transporter)

Primary Acceptance Test

Going back to our requirements, our main acceptance test is simple: send a well-formed message from our native customerApp (acting as a requestor) to our nestMicroservice (acting as a responder), and get back a proper response.

Figure 1: Faye Transporter Acceptance Test

Using `npm link` in Our Development Environment

Before we can run the test, we need to cover one more preliminary. Ask yourself this: how does our nestMicroservice app know how to find the code in our nestjs-faye-transporter package? The answer is pretty simple (though it belies the awesomeness of NPM!).

Note: If you ran the build.sh script at the beginning of this article, you probably do not need to run the npm link commands below. Doing so again is harmless, and it's useful to understand what's happening so I include the steps below. Also, sometimes things just get flakey with the links, and you need to re-run them. I'd actually suggest you go ahead and do so even if you did already run the build.sh script, just to see the behavior. These steps are prety quick anyway.

We're going to use the npm link command (read details here, but the following is all you need to know for this tutorial). There are two simple steps:

In terminal 2, make sure you're in the folder nestjs-faye-transporter (our NPM package folder). Then run npm link at the OS level. You should see output like this (I use nvm; if you don't your output will look slightly different):

$ # from directory nestjs-faye-transporter
$ npm link

@faye-tut/nestjs-faye-transporter@1.0.1 prepare /home/john/code/nest-micro/nestjs-faye/nestjs-faye-transporter
npm run build

@faye-tut/nestjs-faye-transporter@1.0.1 build /home/john/code/nest-micro/nestjs-faye/nestjs-faye-transporter
tsc

 (... some lines omitted ...)

/home/john/.nvm/versions/node/v10.18.1/lib/node_modules/@faye-tut/nestjs-faye-transporter ->
/home/john/code/nest-micro/nestjs-faye/nestjs-faye-transporter

In terminal 4, make sure you're in the folder nestMicroservice. Then run npm link @faye-tut/nestjs-faye-transporter at the OS level. This command references the NPM package name (found in nestjs-faye-transporter/package.json) for our custom transporter. You should see output like this:
```
$ # from directory nestMicroservice
$ npm link @faye-tut/nestjs-faye-transporter
/home/john/code/nest-micro/nestjs-faye/nestMicroservice/node_modules/@faye-tut/nestjs-faye-transporter ->
/home/john/code/nest-micro/nestjs-faye/nestjs-faye-transporter
```

At this point, our nestMicroservice testing app is live-linked to the custom transporter code in the @faye-tut/nestjs-faye-transporter NPM package we're working on. Any changes we make to that code will trigger a rebuild, and because we linked, those changes are immediately visible. Very cool stuff there.

Running the Test

We're ready to rock and roll 🎵!

In terminal 1, make sure the Faye broker is running. First make sure you're in the faye-server directory, then run npm run start. You should see something like this:

$ # from directory faye-server
$ npm run start
simple-faye-server@1.0.0 start /home/john/code/nest-micro/nestjs-faye/faye-server
node server.js

listening on http://localhost:8000/faye
========================================

In terminal 4, start the nestMicroservice. First make sure you're in the nestMicroservice directory, then run npm run start:dev. You should see the usual Nest startup logs, followed by the message Microservice is listening....

In terminal 3, we'll run the native customerApp to make the request. Earlier, we used this app to send requests to the native customerService app. Now we're sending those exact same messages (via the Faye broker, of course) to the nestMicroservice. Make sure you're in the customerApp directory, then...

... keep your eyes on all three terminal windows so you don't miss the magic, and...

# in terminal 3 window...
# in folder customerApp
npm run get-customers

If all went well, you should see a flurry of log messages. I strongly encourage you to take the time to look at them and make sure you can follow the flow of what's happening. In terminal 3, we should get a nice response that looks like this:

$ npm run get-customers

> faye-customer-app@1.0.0 get-customers /home/john/code/nest-micro/nestjs-faye/customerApp
> node dist/customer-app get

Faye customer app starts...
===========================
<== Sending 'get-customers' request with payload:
{"pattern":"/get-customers","data":{},"id":"415fdcef-88e8-42e7-a464-ebdb3b10ff57"}

==> Receiving 'get-customers' reply:
{
  "customers": [
    {
      "id": 1,
      "name": "nestjs.com"
    }
  ]
}

Terminals 1 and 4 show the corresponding message flow in the Faye broker and nestMicroservice app respectively. They all should weave together nicely to let you trace the entire saga through the layers of the system. By the way, if you're wondering where the message level logging in terminal 4 (nestMicroservice) comes from, take a look at the serializer/deserializer implementations in nestjs-faye-transporter/src/responder/serializers/outbound-response-identity-serializer.ts and nestjs-faye-transporter/src/responder/deserializers/inbound-message-identity-deserializer.ts respectively. These are our so-called "identity" serializer/deserializer from the previous article series, which are very helpful for visualizing message flow.

Hooray! We're done right? 🍺 🍺 🍺?

Not so fast... did you forget this is a six-part series 😃? Read on to see where we've still got work to do.

Understanding the Limitations of Take 1

We already know we have to clean up a few things, like adding event handling (e.g., handlers decorated with @EventPattern(...)), adding TypeScript types, and plugging in more cleanly to the framework. But the biggest limitation of our Take 1 implementation is its (lack of) handling of RxJS Observables. To fully appreciate this, we'll have to take a bit deeper dive into the overall flow and handling of requests through the Nest system.

We'll explore this in greater detail in the next article, but let's start with a picture. The following animation shows the path a hypothetical inbound HTTP request would take through our application. Inner boxes with a red background are part of the Nest infrastructure. User supplied code lives in the user-land space with a white background. Controllers are in light blue, and "Services" are in yellow.

Figure 2: Nest Request Handling

An inbound HTTP request kicks off the following sequence of events. Bolded words represent Nest system responsibilities. Underlined words represent user code. There's probably nothing terribly surprising going on here, but let's just briefly walk through it.

The request is routed to a route handling method (like getCustomers in our controller).
The route handling method can either make a remote request directly, or call a service that makes a remote request.
The remote request is handled by the broker client library and sent to the broker.
The remote request is received by the broker client library in the nestMicroservice app.
The request is routed to the correct handler (e.g., a method decorated with @MessagePattern('/get-customers')) based on matching the pattern in the request with the pattern in the method decorator.
The request handling method may make a call to other services (which in turn, could make their own internal calls or remote calls). Let's say it does make such a call, to the custService.getCustomers() method, and that method has a signature like this:
```
getCustomers(id: integer): Observable<Customer>
```

Once the getCustomers method returns, we start the return trip, where things get more interesting. This "more interesting" part is mainly because Nest is very Observable-aware.

Figure 3: Nest Response Handling

In this sequence, I'll introduce a part of the transporter infrastructure responsible for what I'm informally calling "Marshalling" (there's no such official term or single component inside Nest that does Marshalling). Conceptually, it's the part(s) of the system that handle(s) dealing with transferring Observable streams over the network.

In Part 3 we'll go through a few use cases for why Observables are so cool, why they're a perfect fit in this flow, AND how they're actually really easy to use. I know this diagram doesn't make it seem that way, but hey, we're building our own transporter (my inner trekky can't help but giggle over that 🚀). The beauty of it is that once we handle this case properly — and the framework will make this easy as we'll see in the next chapter — everything we might want to do with Observables (and their potential is just, well... mind bending) just works.

Here's the walk through of the return trip flow:

Our handler in nestMicroservice (perhaps directly, or perhaps by calling a service) returns the result. In terms of our example, it's returning a list of customers.
Now, if the response is a "plain" value (JavaScript primitive, array or object), there's not much work to be done other than send it back to the broker. But, if the response is an Observable stream, Nest steps in and makes this extremely easy for everything downstream to work with. That's where marshalling comes in.
Once the response is prepared from nestMicroservice, it's delivered to the Faye broker via the broker client library.
The broker takes care of publishing the response.
On the nestHttpApp side, the broker client library (which has previously subscribed to the broker on the response channel — though we haven't coded that part yet), receives the response message.
The ClientProxy class (again, we haven't built this yet) takes care of routing (and marshalling, if it's dealing with a stream of responses).
This routes the response to the originating service, then back to the originating controller.
Finally, the Nest HttpAdapter software again, if needed, marshalls responses to a suitable form for HTTP transport. For example, if the response is an Observable or a promise, it converts it to an appropriate form for return over HTTP.

So what exactly is the issue?

As you might expect from the above, things work fine if our handlers return plain objects. For example, we currently return an object in our nestMicroservices getCustomers() handler (last line below):

// nestMicroservice/src/app.controller.ts
@MessagePattern('/get-customers')
async getCustomers(data: any): Promise<any> {
  const customers =
    data && data.customerId
      ? customerList.filter(cust => cust.id === parseInt(data.customerId, 10))
      : customerList;
  return { customers };
}

But what happens if our handler returns an Observable? The framework enables this for all built-in transporters, so we should handle it too. Let's test this really quickly. Replace that last line in app.controller.ts with:

return of({customers});

You'll also have to add the following line to the top of the file:

import { of } from 'rxjs';

This construct uses the RxJS of operator to convert our getCustomers() method handler response to an Observable — a stream (containing only a single value in our case, but still, a stream) of values.

If you make this change, then re-issue the /get-customers message (run npm run get-customers in terminal 3), you'll get a rather ugly failure in the nestMicroservice window. This is our fault! If you think about it, we aren't doing any marshalling anywhere in our code. So we aren't handling this case, which, again, is expected of any Nest microservice transporter.

What's Next

With these issues in mind, we're ready to step up our game and make the Faye Custom Transporter server component much more robust. We'll tackle that in the next article. In Part 3 we cover:

A little side expedition on how and why you should care about the "Observables issue" we just uncovered
Addressing that issue
Handling events (e.g., @EventPattern(...) decorated methods) in our responder app
Adding TypeScript types
A few other minor items to clean up our app and make it production worthy

Feel free to ask questions, make comments or suggestions, or just say hello in the comments below. And join us at Discord for more happy discussions about NestJS. I post there as Y Prospect.

Part 1: Introduction and Setup

John Biundo — Mon, 02 Mar 2020 17:18:40 +0000

John is a member of the NestJS core team

Introduction

This article series covers the topic of building a custom transporter for the NestJS microservices subsystem. Is this article right for you? It is if you fit into one of the following two audiences:

You're trying to build or modify a NestJS microservices transporter. That one was probably pretty obvious 😄.
You're a NestJS application developer (someone who writes code in what I'll often refer to as "user-land"), and want a deeper understanding of how NestJS microservices work. Candidly, this is where I came from, and a main motivator for doing the research to write this article series. Even if you never intend to build your own transporter, you can still benefit from this deeper understanding, as I have done. Along the way, I'll try to share insights on how you can become a better user of Nest microservices.

Why Custom Transporters?

Let's start with the Why? question. The Nest microservices subsystem provides a communications layer abstraction that makes it easy for applications (both Nest and non-Nest) to communicate over a wide variety of what are called transporters. This provides several key benefits, covered fully in the previous article series. Nest comes with a variety of built-in transporters, including NATS, RabbitMQ, Kafka, and others. But what if you want to build your own transporter — say for ZeroMQ, Google Cloud Pub/Sub, Amazon Kinesis or Apache ActiveMQ? If that's your desire, you've landed on the right article! Or, if you just want to understand more about how this key part of the Nest infrastructure works, read on!

A related question is "How can I extend the capabilities of existing Nest transporters?" For example, maybe you'd like to use MQTT's QoS feature with the MQTT transporter. That's a topic I'll cover in another upcoming series (already under development 💥), but this series provides a baseline for addressing that requirement.

Article Series Overview

Part 1 (this article): Sets the background for the entire series. We'll begin by examining the simple Faye message broker to set the stage for building a custom Nest transporter that sits on top of it. By the time we're done with Part 5, we'll have a fully functional Faye Transporter!
Part 2: Tackles building an initial version of the server side of our custom transporter. The result is a working version that should familiarize you with the main concepts needed to understand how to build the server component of any custom transporter. It defers implementation of a few important details to the next article so as to keep us focused on the main path.
Part 3: Cleans up some loose ends, and dives deep on a few of the more advanced and impressive features of the framework, and shows you how to enable these features in your custom transporter.
Part 4: Switches to the client side. As with the server side, this article walks through a basic functional implementation for Faye, deferring a few of the more complex features to the next article.
Part 5: Completes the client-side journey, resulting in a finished, fully functional Custom Transporter for Faye.
Part 6: (Coming soon!) Surveys a few of the built-in NestJS transporters and compares their implementation to the Faye implementation to shed light on some broker-specific nuanced implementation details.

Let's get started!

Get the Code

All of the code in these articles is available here. As always, these tutorials work best if you follow along with the code. The README covers all the steps you need to get the repository, build the apps, and follow along. It's easy! I strongly encourage you to do so. Note that each article has a corresponding branch in the repository. For example, this article (Part 1), has a corresponding branch called part1. Read on, or get more details here on using the repository.

Git checkout the current version

For this article, you should git checkout the branch part1. You can get more information about the git repository branches here.

Build the Apps for This Part

$ # from root directory of project (e.g., transporter-tutorial, or whatever you chose as the root)
$ sh build.sh

Overview for Part 1

For this case study, we'll build a custom transporter to work with the Faye message broker. Faye is a simple OSS JavaScript publish/subscribe message broker that runs nicely on Node.js. Getting it up and running and understanding its API are simple, making it a convenient target. At the same time, it provides all the features Nest needs to build a transporter.

Faye uses a very simple — almost canonical — publish/subscribe protocol, as depicted in Figure 1 below.

Figure 1: Faye Message Protocol

Building Basic Faye Client Apps

As we did in the NestJS Microservices in Action series, we'll start by building simple native requestor and responder apps. These are TypeScript apps that exercise the Faye client API directly. This will help us make sure we understand the Faye client API, and give us a convenient test bench.

As covered in the previous article series, one of the challenges the Nest microservices package deals with is to layer a request/response message style on top of publish/subscribe semantics. In other words, Nest requestors need to be able to run code like this, which issues a request and returns a response from a remote microservice:

  @Get('customers')
  getCustomers(): Observable<any> {
    this.logger.log('client#send -> topic: "get-customers"');
    // the following request returns a response as an Observable
    return this.client.send('/get-customers', {});
  }

Supporting Request/Response

Since Faye (as well as most brokers) only supports publish/subscribe, we need to follow a simple recipe to provide request/response semantics. Here's a recap of the generic pattern for how this is often done.

Let's say component A wishes to "get customers" from component B, which has access to a customer DB. Component A can publish a '/get-customers' message, and (assuming it has subscribed to that topic) component B receives it, queries the customer DB for a list of customers, and sends a response message which is delivered back to A. Handling the response message is where the magic happens. In order for B to respond to A, they both must do a few things, agreed upon by convention:

A chooses a response topic (sometimes called a reply subject)
A subscribes to the response topic
A publishes a request, passing the response topic as part of the message
B publishes a response, using the response topic as the topic of its message

💡 We've seen this subscribe-to-the-response-then-send-the-request pattern a few times now, and we'll see it again. It's a vital pattern, and worth permanently etching into your synapses. To that end, let's invent a simple mnemonic. That way we can mentally conjure this pattern quickly whenever we need it. I'll refer to it as STRPTQ. The letters come from "Subscribe To the Response, then Publish The ReQuest". I use Q to make it easier to remember reQuest (vs. Response), in the proper order.

As we'll see in developing our Faye transporter, Nest makes this a bit easier by automatically choosing the response topic name for you. In fact, Nest builds two topics from each pattern you declare. For example, assume you define a Nest microservice responder message pattern like this:

@MessagePattern('/get-customers')

(Note: Nest uses the term channel as a generic internal name for topics (also called subjects and channels by some brokers), and to help disambiguate them from the user-land concept of a pattern, from which they are derived.)

Internally, Nest builds two channels from the above pattern:

'/get-customers_ack' - this is the physical channel name we'll use in the Faye transporter to publish/subscribe to request messages on the topic '/get-customers'
'/get-customers_res' - this is the physical channel name we'll use in the Faye transporter to publish/subscribe to response messages resulting from '/get-customers' requests

Let's build those native apps we mentioned. Note: later we'll want to use these native apps to test our Nest transporter, so to make them interact seamlessly with Nest, we'll utilize the Nest channel names described above. We'll start with the responder app.

Native Responder App (customerService)

This is the app that responds to inbound messages (requests) and generates responses. Assuming you're following along on the part1 branch, open the file customerService/src/service.ts.

Here's the implementation of the getCustomers Faye subscription handler. This is the code we register to run when our app receives an inbound message on the '/get-customers_ack' channel (see the subscribe() call lower in the file):

// customerService/src/service.ts
function getCustomers(packet): void {
  const message = parsePacket(packet);
  console.log(
    `\n========== <<< 'get-customers' message >>> ==========\n${JSON.stringify(
      message,
    )}\n=============================================\n`,
  );

  // filter customers list if there's a `customerId` param
  const customers =
    message.data && message.data.customerId
      ? customerList.filter(
          cust => cust.id === parseInt(message.data.customerId, 10),
        )
      : customerList;

  client.publish('/get-customers_res', getPayload({ customers }, message.id));
}

The logic should be easy to understand. Refer to the full listing and make sure you see how we are using the '/get-customers_ack' and '/get-customers_res' channels here. We'll be seeing a lot of that pattern, so make sure it makes sense to you.

One detail to point out is the call to getPayload(). Let's discuss that. As mentioned, we're building these native apps to conform to Nest protocols. This includes using the Nest-standard channel names, as well as matching the internal Nest message format (message format is covered extensively in the NestJS Microservices in Action series). Because we take this approach, the apps provide a convenient test bench for the Nest transporter Server and ClientProxy we're building in this article series. The message format requirement means we need to wrap responses in a standard object we can depict like this:

   {
     /**
      *  Request error message, if any
      */
     err: any,
     /**
      *  The response message payload (return value of a
      *  message pattern handler)
      */
     response: any
     /**
      *  Status of an Observable response. `true` once the final
      *  stream value has been retrieved.
      */
     isDisposed: boolean
     /**
      *  Unique identifier, corresponding to the id field received
      *  from the initial request
      */
     id: string
   }

So getPayload() is a helper that wraps payloads in this standard Nest message structure. If you're paying close attention, you'll notice that getPayload() (not shown here — look at the service.ts source code to see it) copies the message.id field from the inbound request to the outbound message. We haven't discussed these id fields yet, but they'll become very important when we dive into the "client side" of our transporter starting in Part 5.

Native Requestor App (customerApp)

This is the user-facing app that makes outbound requests. The requestor is in customerApp/src/customer-app.ts. Below is the implementation of the getCustomers() function from that app.

There's a little bit of boilerplate, but the main logic should be easy to see. It implements the requestor side of that STRPTQ request/response protocol we walked through a couple of minutes ago.

// customerApp/src/customer-app.ts
async function getCustomers(customerId) {
  // build Nest-shaped message
  const payload = getPayload('/get-customers', { customerId }, uuid());

  return new Promise((resolve, reject) => {
    // subscribe to the response message
    const subscription = client.subscribe('/get-customers_res', result => {
      console.log(
        `==> Receiving 'get-customers' reply: \n${JSON.stringify(
          result.response,
          null,
          2,
        )}\n`,
      );
    });

    // once response is subscribed, publish the request
    subscription.then(() => {
      console.log(
        `<== Sending 'get-customers' request with payload:\n${JSON.stringify(
          payload,
        )}\n`,
      );
      const pub = client.publish('/get-customers_ack', payload);
      pub.then(() => {
        // wait .5 second to ensure subscription handler executes
        // then unsubscribe and resolve
        setTimeout(() => {
          subscription.cancel();
          resolve();
        }, 500);
      });
    });
  });
}

We first subscribe to '/get-customers_res' (the response channel). We pass it a simple Faye subscription handler that prints out the result when a message on this channel comes back. This output becomes the "final response" to an end-user command (request) that's been issued from the command line (received by way of the basic command line processing logic at the bottom of the file).

Then we publish the request on '/get-customers_ack'. Note that the call to subscribe() returned a subscription which is a Promise that resolves once the Faye server acknowledges that the subscribe operation has completed.

Because we're running this as a standalone Node.js command line program, and thus want to exit after sending a request (which let's us use it as a command processor to conveniently send requests), we wait 500ms after publishing the request, then cancel the subscription and exit. Cancelling the subscription is important hygiene — if we don't, subscriptions hang around (though Faye will eventually detect them and clean them up). This is a bit brute force, and we'll want to do more sophisticated subscription management in our transporter client, but suffices for now.

Finally, you'll notice that much like the customerService responder app, we have a helper getPayload() utility to wrap requests in the Nest message protocol. You'll also notice that we are generating the id field — the same id field we are handling on our responder app above. For now, just think of the id field as something we need to have in order to speak native Nest message protocol. Later, we'll dive into this in more detail.

For completeness, let's take a very quick look at how we run the Faye server. Though you can treat it as a black box, it's worth a moment to look at how it runs, and appreciate its native node.js-ness and simplicity. Open up faye-server/server.js. The code is shown below. You can read more about running the Faye server here, but we're basically just starting it up, and for instrumentation purposes, listening for various events that we can log to the console to make it easy to trace the handshaking and message exchange. You shouldn't have to mess with this at all, but if you do, it's pretty basic stuff.

// faye-server/server.js
const http = require('http');
const faye = require('faye');

const mountPath = '/faye';
const port = 8000;

const server = http.createServer();
const bayeux = new faye.NodeAdapter({ mount: mountPath, timeout: 45 });

bayeux.attach(server);
server.listen(port, () => {
  console.log(
    `listening on http://localhost:${port}${mountPath}\n========================================`,
  );
});

bayeux.on('handshake', clientId =>
  console.log('^^ client connect (#', clientId.substring(0, 4), ')'),
);

bayeux.on('disconnect', clientId =>
  console.log('vv client disconnect (#', clientId.substring(0, 4), ')'),
);

bayeux.on('publish', (clientId, channel, data) => {
  console.log(
    `<== New message from ${clientId.substring(0, 4)} on channel ${channel}`,
    `\n    ** Payload: ${JSON.stringify(data)}`,
  );
});

bayeux.on('subscribe', (clientId, channel) => {
  console.log(
    `++ New subscription from ${clientId.substring(0, 4)} on ${JSON.stringify(
      channel,
    )}`,
  );
});

bayeux.on('unsubscribe', (clientId, channel) => {
  console.log(
    `-- Unsubscribe by ${clientId.substring(0, 4)} on ${JSON.stringify(
      channel,
    )}`,
  );
});

Testing the Apps

We've now completed our requestor and responder, and can test them out. While I only reviewed the '/get-customers' message flow above, the code in this branch also implements the '/add-customer' message. Take a minute to run the code now by following these simple instructions. Here's what it looks like. In this video loop, the Faye broker is running in the top pane, the customerService app is running in the middle pane, and the customerApp app is running in the lower pane (and if you're curious, this is running in the tmux virtual terminal; you can get my setup here if you're interested).

Screen Capture: Native App Demo

What's Next

In Part 2, we'll write and test a basic version of the Faye transporter's server component. This is the component that runs in the context of a microservice listener to enable apps to function as Nest responders (these terms and concepts are discussed extensively in the NestJS Microservices in Action series). We'll test this server component by using it in a simple Nest responder (microservice) app. That Nest microservice app will replace the native customerApp we just built, and we'll make sure it responds to the exact same customer-related messages.

Feel free to ask questions, make comments or suggestions, or just say hello in the comments below. And join us at Discord for more happy discussions about NestJS. I post there as Y Prospect.

Part 3: Techniques for Application Integration with NestJS and NATS

John Biundo — Wed, 12 Feb 2020 14:25:30 +0000

John is a member of the NestJS core team

Part 3: Solving the Message Incompatibility Problem

In Part 2 of this series, we described the challenges of integrating Nest apps with non-Nest apps using NATS as an intermediary.

In this article, we explore solutions to that problem.

Get the Code

Reminder: all the code for these articles is available here, with complete instructions here.

Serialization and Deserialization

Serialization and deserialization are the steps Nest uses to translate between message formats. In other words, Nest receives inbound messages (from the broker), deserializes them (converts arbitrary message formats into Nest message formats), and then processes them. For outbound messages, the process is reversed: the last step before sending a message to the broker is to (optionally) serialize it (convert from Nest message format to some other format).

The nifty part is that while Nest does all this automatically, it also provides hooks where you can easily plug in your own serializers and deserializers to handle arbitrary message formats. Serializers and deserializers are simply classes that implement a defined interface.

Since requestors and responders perform their own unique roles (each with their own context) and each are both message senders and message receivers, there are four hooks where serialization/deserialization take place. These are examined in the recipes below. If there's one section of this article series that you should bookmark for future reference, this is it. Each recipe describes exactly where you should look to handle a serialization/deserialization task, including the full context you need to understand why and how to do it.

Serialization/Deserialization Hooks and Recipes

Let's examine the two different contexts (requestors and responders) in which we need to serialize/deserialize. For each, we'll examine handling inbound and outbound messages.

Nest as Requestor

Nest requestors live in the context of a ClientProxy instance. It follows that we customize serialization/deserialization behavior by modifying the ClientProxy behavior. For example:



// app.controller.ts
@Client({
  transport: Transport.NATS,
  options: {
    url: 'nats://localhost:4222',
    serializer: new OutboundRequestSerializer(),
    deserializer: new InboundResponseDeserializer(),
  }
})
client: ClientProxy;

In the code fragment above, OutboundRequestSerializer is the class that answers the question "how do does my Nest requestor format outgoing requests so that an external app can process them?", while InboundResponseDeserializer is the class that addresses "how does my requestor translate incoming external responses so that Nest can process them?".

Note: you can choose any name for these classes, but trust me, it's highly recommended that you choose a naming convention similar to the ones above to keep things straight!

Nest as Responder

Nest responders live in the context of a microservice instance. It follows that we customize serialization/deserialization behavior by modifying the microservice instance behavior. For example:



// main.ts
async function bootstrap() {
  const app = await NestFactory.createMicroservice(AppModule, {
    transport: Transport.NATS,
    options: {
      url: 'nats://localhost:4222',
      deserializer: new InboundMessageDeserializer(),
      serializer: new OutboundResponseSerializer(),
    },
  });
  app.listen(() => console.log('Microservice is listening...'));
}

So InboundMessageDeserializer is the class that answers the question "how does my responder translate incoming external messages so that Nest can process them?", while OutboundResponseSerializer is the class that handles "how does my responder format responses so that the external app can understand them?".

This feature results in the following diagrams — ones that remove all those ugly red X's in the last diagram!

Here's the diagram to keep in mind when writing serializers/deserializers for Nest responders.

Figure 1: Nest Responder (De)serialization

In the above diagram, the Nest responder serializer and deserializer are configured in the NestFactory.creatMicroservice() call.

And here's the corresponding diagram for Nest requestors.

Figure 2: Nest Requestor (De)serialization

In the above diagram, the Nest requestor serializer and deserializer are configured in the ClientProxy configuration (decorator or injectable, depending on which method you use).

Implementing Serializers and Deserializers

Before we start writing serializer/deserializer classes, let's take a look at their interfaces.

Serializer Interface

A serializer class needs to implement the serialize() method, which takes a single argument — the outbound payload to be serialized — and returns the serialized payload.



export interface Serializer<TInput = any, TOutput = any> {
  serialize(value: TInput): TOutput;
}

We'll now implement an identity serializer for our Nest responder as an exercise. An identity serializer simply returns the same message it receives. In addition to exercising the interface, we can use it to "spy on" the otherwise silent serialization function that Nest always performs on our behalf.

Responder Identity Serializer Implementation

Here, we'll be working in the nestMicroservice project (i.e., nest-nats-sample/nestMicroservice directory). Take a look at the src/common/serializers/outbound-response-identity.serializer.ts file:



// src/common/serializers/outbount-response-identity.serializer.ts
import { Serializer, OutgoingResponse } from '@nestjs/microservices';
import { Logger } from '@nestjs/common';

export class OutboundResponseIdentitySerializer implements Serializer {
  private readonly logger = new Logger('OutboundResponseIdentitySerializer');
  serialize(value: any): OutgoingResponse {
    this.logger.debug(
      `-->> Serializing outbound response: \n${JSON.stringify(value)}`
    );
    return value;
  }
}

Now simply plug this serializer into the main.ts file for the Nest responder app. If you're following along with the github repository, the main.ts file will have all of this code (and more that we'll get to soon) in place, but commented out. Simply uncomment the line shown below.



// src/main.ts
const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.NATS,
  options: {
    queue: 'customers',
    url: 'nats://localhost:4222',
    /**
      * Use the "Identity" (de)serializers for observing messages for
      * nest-only deployment.
      */
    serializer: new OutboundResponseIdentitySerializer(),
  },
});

If you now send some requests to the nestMicroservice app from the nestHttpApp (here's how), you'll see logging information that looks something like this, showing you the internal layout of the outbound message:



[Nest] 8786   - 02/04/2020, 10:19:04 AM   [OutboundResponseIdentitySerializer] -->> Serializing outbound response:
{"err":null,"response":[],"isDisposed":true,"id":"41da72e8-09d0-4720-9404-bd0977b034a0"}

Deserializer Interface

Now let's take a look at the required interface for a deserializer.



export interface Deserializer<TInput = any, TOutput = any> {
  deserialize(value: TInput, options?: Record<string, any>): TOutput;
}

A deserializer class needs to implement the deserialize() method, which takes a two arguments — the payload to be deserialized and an optional options object — and returns the deserialized payload.

The options object contains metadata about the incoming message. For NATS, the object contains the value of the replyTo message property, if one exists.

Let's implement a deserializer. To start with, we'll build an "identity deserializer" that just logs out the contents of any incoming message without doing any transformation on it.

Responder Identity Deserializer Implementation

We're still working on the Nest responder — the nestMicroservice project (i.e., nest-nats-sample/nestMicroservice directory). Take a look at the src/common/deserializers/inbound-message-identity.deserializer.ts file:



// src/common/serializers/inbound-message-identity.deserializer.ts
import { ConsumerDeserializer, IncomingRequest } from '@nestjs/microservices';
import { Logger } from '@nestjs/common';

export class InboundMessageIdentityDeserializer
  implements ConsumerDeserializer {
  private readonly logger = new Logger('InboundMessageIdentityDeserializer');

  deserialize(value: any, options?: Record<string, any>): IncomingRequest {
    this.logger.verbose(
      `<<-- deserializing inbound message:\n${JSON.stringify(
        value
      )}\n\twith options: ${JSON.stringify(options)}`
    );
    return value;
  }
}

Like our identity serializer, we can quickly plug in an identity deserializer to spy on the incoming messages from our requestor. Again, if you're following along using the github repository, simply comment out the lines so the active (de)serializers match those shown below.



// src/main.ts
const app = await NestFactory.createMicroservice(AppModule, {
  transport: Transport.NATS,
  options: {
    queue: 'customers',
    url: 'nats://localhost:4222',
    /**
      * Use the "Identity" (de)serializers for observing messages for
      * nest-only deployment.
      */
    serializer: new OutboundResponseIdentitySerializer(),
    deserializer: new InboundMessageIdentityDeserializer(),
  },
});

If you again send some requests to the nestMicroservice app from the nestHttpApp (like this), you'll see logging information that looks something like this, showing you the layout of the inbound message:



[Nest] 8786   - 02/04/2020, 10:19:04 AM   [InboundMessageIdentityDeserializer] <<-- deserializing inbound message:
{"pattern":"get-customers","data":{},"id":"41da72e8-09d0-4720-9404-bd0977b034a0"}
        with options: {"channel":"get-customers","replyTo":"_INBOX.IDML09N4JB6W0MF4P6GBPB.IDML09N4JB6W0MF4P6GCX2"}

Notice the value for the replyTo field in the logging output. You can probably guess what that is. This is a detail that's going to come in handy down the line, so file that away for future reference.

Implementing the Integration Use Cases

We've nearly arrived at our final destination. With the understanding we've gained, we can specify what we need to complete our integration use cases from Figure 1 in Part 1.

Here are the requirements:

The nestHttpApp, in its role as a requestor, must implement:
- A) an outbound message external serializer that translates a Nest formatted request into a request understood by our external service. For example:
  - From Nest format
    - topic: 'get-customers'
    - reply topic: '_INBOX.XVRL...'
    - payload: {pattern: 'get-customers', data: {}, id: 'abc...'}
  - To external format
    - topic: 'get-customers'
    - reply topic: '_INBOX.XVRL...'
    - payload: {}
- B) an inbound response external deserializer that translates an external response into a format understood by Nest. For example:
  - From external format
    - topic: '_INBOX.XVRL...'
    - payload: {customers: [{id: 1, name: 'nestjs.com'}]}
  - To Nest format
    - topic:'_INBOX.XVRL...'
    - payload: {err: undefined, response: {customers: [{id: 1, name: 'nestjs.com'}]}, isDisposed: true}
The nestMicroservice app, in its role as a responder, must implement:
- A) an inbound message external deserializer that translates an external request into a format understood by Nest. For example:
  - From external format
    - topic: 'get-customers'
    - reply: '_INBOX.XVRL...'
    - payload: {}
  - To Nest format
    - topic 'get-customers'
    - reply topic: '_INBOX.XVRL...'
    - payload: {pattern: 'get-customers', data: {}, id: 'abc...'}
- B) an outbound response external serializer that translates a Nest formatted response into a response understood by our external service. For example:
  - From Nest format
    - topic: '_INBOX.XVRL...'
    - payload: {err: undefined, response: {customers: [{id: 1, name: 'nestjs.com'}]}, isDisposed: true}
  - To external format
    - topic: '_INBOX.XVRL...'
    - payload: {customers: [{id: 1, name: 'nestjs.com'}]}

With that in mind, let's get busy! The bolded descriptions above inform the names of our classes.

For requirement 1-A, we have src/common/serializers/outbound-message-external.serializer.ts in our nestHttpApp project. Here's the code. The comments explain its intent.



// src/common/serializers/outbound-message-external.serializer.ts
import { Logger } from '@nestjs/common';
import { Serializer } from '@nestjs/microservices';

export class OutboundMessageExternalSerializer implements Serializer {
  private readonly logger = new Logger('OutboundMessageExternalSerializer');
  serialize(value: any) {
    this.logger.debug(
      `-->> Serializing outbound message: \n${JSON.stringify(value)}`,
    );

    /**
     * Here, we are merely "unpacking" the request payload from the Nest
     * message structure and returning it as a "plain" top-level object.
     */
    return value.data;
  }
}

For requirement 1-B, we have src/common/deserializers/inbound-response-external.deserializer.ts in our nestHttpApp project. Here's the code. The comments explain its intent.



// src/common/deserializers/inbound-response-external.deserializer.ts
import { WritePacket, Deserializer } from '@nestjs/microservices';
import { Logger } from '@nestjs/common';

export class InboundResponseExternalDeserializer implements Deserializer {
  private readonly logger = new Logger('InboundResponseExternalDeserializer');

  deserialize(value: any): WritePacket {
    this.logger.verbose(
      `<<-- deserializing inbound response:\n${JSON.stringify(value)}`,
    );

    /**
     * Here, we wrap the external payload received in a standard Nest
     * response message.  Note that we have omitted the `id` field, as it
     * does not have any meaning from an external responder.  Because of this,
     * we have to also:
     *   1) implement the `Deserializer` interface instead of the
     *      `ProducerDeserializer` interface used in the identity deserializer
     *   2) return an object with the `WritePacket` interface, rather than
     *      the`IncomingResponse` interface used in the identity deserializer.
     */
    return {
      err: undefined,
      response: value,
      isDisposed: true,
    };
  }
}

For requirement 2-A, we have src/common/deserializers/inbound-message-external.deserializer.ts in our nestMicroservice project. Here's the code. The comments explain its intent.



// src/common/serializers/inbound-message-external.deserializer.ts
import { Logger } from '@nestjs/common';
import * as uuid from 'uuid/v4';
import { ConsumerDeserializer } from '@nestjs/microservices';

export class InboundMessageExternalDeserializer
  implements ConsumerDeserializer {
  private readonly logger = new Logger('InboundMessageExternalDeserializer');
  deserialize(value: any, options?: Record<string, any>) {
    this.logger.verbose(
      `<<-- deserializing inbound external message:\n${JSON.stringify(
        value,
      )}\n\twith options: ${JSON.stringify(options)}`,
    );

    /**
     * Here, we merely wrap our inbound message payload in the standard Nest
     * message structure.
     */
    return {
      pattern: undefined,
      data: value,
      id: uuid(),
    };
  }
}

For requirement 2-B, we have src/common/serializers/outbound-response-external.serializer.ts in our nestMicroservice project. Here's the code. The comments explain its intent.



// src/common/serializers/outbound-response-external.serializer.ts
import { Serializer, OutgoingResponse } from '@nestjs/microservices';
import { Logger } from '@nestjs/common';

export class OutboundResponseExternalSerializer implements Serializer {
  private readonly logger = new Logger('OutboundResponseExternalSerializer');
  serialize(value: any): OutgoingResponse {
    this.logger.debug(
      `-->> Serializing outbound response: \n${JSON.stringify(value)}`,
    );

    /**
     * Here, we are merely "unpacking" the response payload from the Nest
     * message structure, and returning it as a "plain" top-level object.
     */

    return value.response;
  }
}

Enabling the External (De)serializers

The last step is to plug in these (de)serializers in at the appropriate "hook points". We've seen this before. For the nestHttpApp, this is done wherever you configure the ClientProxy. In our case, for ease of code organization, we've done this with the @Client() decorator in the src/app.controller.ts file. To enable the external (de)serializers, update that file to look like this:



// src/app.controller.ts
  @Client({
    transport: Transport.NATS,
    options: {
      url: 'nats://localhost:4222',
      /**
       * Use the "Identity" (de)serializers for observing messages for
       * nest-only deployment.
       */
      // serializer: new OutboundMessageIdentitySerializer(),
      // deserializer: new InboundResponseIdentityDeserializer(),

      /**
       * Use the "External" (de)serializers for transforming messages to/from
       * (only) an external responder
       */
      serializer: new OutboundMessageExternalSerializer(),
      deserializer: new InboundResponseExternalDeserializer(),
    },
  })
  custClient: ClientProxy;

For the nestMicroservice, this is done in the src/main.ts file. To enable the external (de)serializers, update that file to look like this:



// src/main.ts
  const app = await NestFactory.createMicroservice(AppModule, {
    transport: Transport.NATS,
    options: {
      queue: 'customers',
      url: 'nats://localhost:4222',
      /**
       * Use the "Identity" (de)serializers for observing messages for
       * nest-only deployment.
       */
      // serializer: new OutboundResponseIdentitySerializer(),
      // deserializer: new InboundMessageIdentityDeserializer(),

      /**
       * Use the "External" (de)serializers for transforming messages to/from
       * an external requestor
       */
      serializer: new OutboundResponseExternalSerializer(),
      deserializer: new InboundMessageExternalDeserializer(),
    },
  });

At this point, you should be able to send and receive requests between any combination of apps. For example, from nestHttpApp to customerService, from customerApp to nestMicroservice, and all other combinations. I highly recommend you do so now, using the instructions from here and here to run all components at the same time*. Pay close attention to the log outputs which show how (de)serialization is happening at each step.

*We haven't described the use of queues, but when you start running the full mixed configuration (Nest and non-Nest components), one thing you'll notice is that requests will be randomly routed between the nestMicroservice app and the customerService app. This is because they are members of a NATS queue. If you want to send a request to a particular responder, you'll need to shut down the other one to guarantee the message is sent there. I'll discuss NATS queues in more detail the final article of this series.

Here's a nice surprise! We've magically handled Figure 1 Case D along the way. If you think carefully about it, this is both not a surprise, and has a somewhat strict limitation. The reason this works is that we're now serializing and deserializing to and from a canonical external format. So all requests look like they come from an external requestor, and all responses look like they come from an external responder. With this in place, our serializers and deserializers work in all combinations. We've created a "lingua franca" message protocol. The limitation with this approach is perhaps a subtle one, and is something we'll address later (see bullet point #2 in What's Next? below).

Conclusion 🚀

We've covered a lot of ground. Hopefully you've got both a conceptual framework for how to think about integrating external applications with Nest apps using tools like Nest microservices and the NATS message broker, and some boilerplate code you can work with. Some of these concepts also translate directly to other Nest microservice transporters, but as you can imagine, the details are different. Holler in the comments if you'd like me to cover other transporters in this fashion in another series, and I'll see if I can do so.

What's next? ❓

There are a few remaining subtle topics to cover, which I'll do in the next installment of this series. As a teaser, these include:

Can we run both the nestMicroservice app and the external customerService app at the same time, and load balance requests across them? Spoiler: yes! In fact, we already did so, as mentioned above. This is simple to do with NATS distributed queues, which we'll cover briefly in the next article.
We covered the case where we're running our modified nestHttpApp in this hybrid (Nest and non-Nest) environment. What happens if we have a mixture of Nest apps that are unmodified — that is, they are deployed apps that we don't want to touch to implement our "lingua franca" message protocol (i.e., they're running out-of-the-box standard de(serializers))? Can they be made to play nicely in this hybrid environment? Spoiler: yes!
What about events? We've covered what seems to be the trickier case here with request/response style messaging, but how about plain old events? In fact, you may have noticed we've built a feature for adding a customer (see the 'add-customer' messages and handlers sprinkled through our code). Go ahead and try them now. You'll see you have mixed results. Can we make events cooperate in this hybrid environment? Spoiler: yes!

Stay tuned for the answers to these and other exciting questions in our next installment! 😃

Feel free to ask questions, make comments or suggestions, or just say hello in the comments below. And join us at Discord for more happy discussions about NestJS. I post there as Y Prospect.