Service API

Inject PUBSUB_SERVICE for programmatic control.

import { Injectable, Inject, OnModuleInit } from '@nestjs/common';
import { PUBSUB_SERVICE, IPubSubService, IPubSubSubscription } from '@nestjs-redisx/pubsub';
import { OrderEvent } from './types';

@Injectable()
export class OrderFeed implements OnModuleInit {
  private subscription?: IPubSubSubscription;

  constructor(
    @Inject(PUBSUB_SERVICE)
    private readonly pubsub: IPubSubService,
  ) {}

  async onModuleInit(): Promise<void> {
    // Programmatic subscription with a typed handler.
    this.subscription = await this.pubsub.subscribe<OrderEvent>('order.updated', (message) => {
      console.log(`order ${message.data.orderId} -> ${message.data.status}`);
    });
  }

  async pause(): Promise<void> {
    // Releases the Redis subscription when this was the last handler.
    await this.subscription?.unsubscribe();
  }

  activeChannels(): string[] {
    return this.pubsub.getSubscriptions().channels;
  }
}

IPubSubService

• publish<T>(channel, data): Promise<number> — JSON-serializes and publishes; returns the number of subscribers that received the message. Throws PubSubPublishError on serialization or transport failure.
• subscribe<T>(channel, handler): Promise<IPubSubSubscription> — registers a handler; the Redis subscription is created for the first handler and shared by the rest. Throws PubSubSubscribeError when SUBSCRIBE fails (no phantom handler is left behind).
• psubscribe<T>(pattern, handler): Promise<IPubSubSubscription> — pattern subscription (*, ?, [..]); messages carry both pattern and the concrete channel.
• unsubscribeAll(): Promise<void> — removes every handler and releases all Redis subscriptions (called automatically on module destroy).
• getSubscriptions(): { channels, patterns } — non-mutating snapshot of logical names for monitoring/health.

Subscription handles

unsubscribe() on the returned handle removes that handler; when it was the last one for the channel/pattern, the underlying Redis subscription is released too. Calling it twice is safe.

A given handler function is registered at most once per channel: subscribing the same function to the same channel again is a no-op that returns a handle over the same registration (use separate closures if you need independent subscriptions). Concurrent subscribe/unsubscribe calls for the same channel are serialized internally, so the local handler registry can never desync from the Redis subscription state.

Delivery semantics

Pub/Sub is at-most-once and non-persistent: publish returns 0 when nobody is subscribed, and messages sent while an instance is disconnected are lost. Use Streams when you need durability.