loopback4-nats-connector

Overview

A LoopBack 4 extension that exposes NATS to LB4 applications as an injectable component. Decorate controller methods with @subscribe / @reply, inject a NatsPublisher, and configure one or more isolated named connections.

Installation

npm install loopback4-nats-connector

Basic usage (v1, single connection)

import {BootMixin} from '@loopback/boot';
import {RepositoryMixin} from '@loopback/repository';
import {RestApplication} from '@loopback/rest';
import {ServiceMixin} from '@loopback/service-proxy';
import {
  NatsConnectorComponent,
  NatsConnectorComponentBindings,
} from 'loopback4-nats-connector';

export class MyApp extends BootMixin(
  ServiceMixin(RepositoryMixin(RestApplication)),
) {
  constructor(options: ApplicationConfig = {}) {
    super(options);

    this.configure(NatsConnectorComponentBindings.COMPONENT).to({
      servers: ['nats://localhost:4222'],
      auth: {token: process.env.NATS_TOKEN!},
    });
    this.component(NatsConnectorComponent);

    this.controller(HelloController);
  }
}

import {inject} from '@loopback/core';
import {
  subscribe,
  reply,
  NatsPublisher,
  NatsConnectorComponentBindings as N,
  type SubscriptionContext,
} from 'loopback4-nats-connector';

export class HelloController {
  constructor(@inject(N.PUBLISHER) private publisher: NatsPublisher) {}

  @subscribe('hello.*')
  async greet(payload: {name: string}, ctx: SubscriptionContext) {
    await this.publisher.publish('audit', {
      greeted: payload.name,
      at: Date.now(),
    });
  }

  @reply('hello.health')
  async health() {
    return {ok: true};
  }
}

Default connection. Decorators with no connection option resolve via this rule: (1) default: '<name>' field on options, else (2) connection literally named 'default', else (3) boot fails. Flat shorthand auto-satisfies rule 2. For multi-connection with no obvious primary, set default: '<name>' or pass {connection: '<name>'} on every decorator.

Multiple isolated connections

A single application can talk to several NATS servers, each fully isolated (separate auth, codec, status emitter, subscription handles):

this.configure(NatsConnectorComponentBindings.COMPONENT).to({
  connections: {
    default: {
      servers: ['nats://internal-broker:4222'],
      auth: {token: process.env.INTERNAL_NATS_TOKEN!},
    },
    ingress: {
      servers: ['nats://public-broker:4222'],
      auth: {
        tls: {
          certFile: '/secrets/ingress.crt',
          keyFile: '/secrets/ingress.key',
        },
      },
    },
  },
});

Then route subscriptions to a specific connection:

@subscribe('public.>', {connection: 'ingress'})
async fanIn(payload: unknown, ctx) { /* … */ }

Inject a publisher for a named connection:

constructor(
  @inject(N.publisher('ingress')) private outbound: NatsPublisher,
  @inject(N.publisher('default')) private internal: NatsPublisher,
) {}

Per-tenant dynamic connections (v1.x)

Open / close NATS connections at runtime — one per tenant — and let shared handlers follow the fleet via connection: '*':

import {inject} from '@loopback/core';
import {
  NatsConnectionRegistry,
  NatsConnectorComponentBindings as N,
  subscribe,
  type SubscriptionContext,
} from 'loopback4-nats-connector';

class OnboardingService {
  constructor(@inject(N.REGISTRY) private nats: NatsConnectionRegistry) {}
  async onTenantCreated(t: {id: string; natsUrl: string; natsToken: string}) {
    await this.nats.add(t.id, {
      servers: [t.natsUrl],
      auth: {token: t.natsToken},
    });
  }
}

class FleetController {
  @subscribe('orders.created', {connection: '*'})
  async onOrder(payload: unknown, ctx: SubscriptionContext) {
    const tenantId = ctx.connection; // tenant ID = connection name
    /* ... */
  }
}

Dynamic connections & limitations

When subscriptions target a connection added via registry.add() after app.start(), the decorator must use connection: '*'.

SubscriptionBooter.start() calls resolveConnectionName() for every non-wildcard decorator. If the requested name is not in the static connections map and not already in the registry at boot time, it throws and the app never starts:

[nats] subscription on 'orders.created' targets unknown connection 'tenant1'

The connection: '*' path avoids this: decorators are stored as templates during start() and materialised per-connection each time registry.add(name, opts) fires.

Scenario	connection value	Works?
Static connection in config	'myconn'	✅
Dynamic, added before app.start()	'tenant1'	✅
Dynamic, added after app.start()	'tenant1'	❌ boot error
Dynamic, added after app.start()	'*'	✅

Reference implementation: examples/03-multi-tenant-dynamic/.

Examples

Runnable LB4 apps under examples/ — one per configuration (single connection, multi-connection static, multi-tenant dynamic, mTLS auth, NKEY auth, custom codec, queue groups, status events, JetStream consumer, KV repository, services API). Each example spawns its own nats-server and is the project's end-to-end test layer.

npm run examples         # build + run every example against spawned nats-server

Imports cheat sheet

Everything imports from the package root. Never import from nats directly — the extension re-exports the nats.js types/factories you need so consumers debug against nats.js docs without contract drift.

import {
  // component
  NatsConnectorComponent,
  NatsConnectorComponentBindings, // also exported as `N` alias in examples below
  // decorators (v1)
  subscribe,
  queueSubscribe,
  reply,
  // service
  NatsPublisher,
  // header factory + type (re-exported from nats.js)
  headers,
  type MsgHdrs,
  // raw connection type for escape hatch
  type NatsConnection,
  // handler context
  type SubscriptionContext,
  // dynamic registry (v1.x)
  NatsConnectionRegistry,
  // codec contract
  type Codec,
} from 'loopback4-nats-connector';

Troubleshooting

First-failure debugging path. Symptom on the left, where to look on the right.

Symptom	Look at
Boot rejects: cannot resolve N.connection('default')	No connection named default. Either rename one in connections, or pass {connection: '<name>'} on every decorator.
Boot rejects naming a decorator + connection	Decorator references unknown connection name. Fix the name in the @subscribe opts or add the connection to options.
Disconnects, reconnect noise, slow consumer	Subscribe to per-connection emitter at N.events('<name>') to monitor status events.
request() rejects with code: '503'	No responder for subject (nats.js behavior, surfaced unchanged). Check subscriber registered, broker reachable, subject typo.
Need a nats.js method missing from NatsPublisher (e.g. requestMany)	Inject raw NatsConnection at N.CONNECTION as an escape hatch.
headers() import not found	Re-exported from package root: import {headers} from 'loopback4-nats-connector'. Do not import from nats.
Multi-tenant: messages published in onboarding race window lost	Documented behavior on core NATS. Use a retry/queue mechanism before calling registry.add() to close the race window.
Codec decode fails on inbound	Raw bytes available on ctx.raw.
One connection failed at boot, others fine	Intentional. Subscriptions targeting the failed name raise loudly during booter pass.

Requirements

• Node.js >=22
• @loopback/core ^7.0.3 (peer dependency)
• A reachable NATS server (e.g. via the nats-server binary or a managed broker)

Development

See DEVELOPING.md for the contributor guide.

npm run build         # tsc → dist/
npm test              # rebuild → mocha → lint
npm run lint:fix      # eslint --fix + prettier --write

License

MIT © SourceFuse