NestJS Websockets Messages and RxJS Subjects.

There may come a time in your NestJS career where you find that you want to be able to inject a websocket server into a service so that you can allow the service to be in charge of emitting events back to users dynamically. While this can be done, I'd like to show a different approach using RxJS that I find cleaner and easier to manage.

What is RxJS

You may have seen this four letter library in the NestJS docs or in some front end code and wondered what it is. Without diving too deep into it, RxJS is a reactive library that can be used for working with asynchronous code in an extensible way. Think of it as a super-powered callback library.

You may be wondering “why we'd be wanting to use anything that might be related to callbacks in modern JavaScript?” The truth of the matter is the library is much more than just callbacks, it's also able to handle multiple values returned through a callback-like method, which is where this starts to come in handy.

Unlike promises, Observables, RxJS's asynchronous construct, are not fired off until they are subscribed to. Other than that, Observables have a very familiar API when compared to JavaScript's Array.prototype, along with RxJS specific ones that have more utility depending on your needs. There are many crash courses and lots of information about RxJS out there, so we won't go into great detail here, but hopefully this gives you a primer if you're just now learning about RxJS.

Think of it as a super-powered callback library. You may be wondering why we'd be wanting to use anything that might be callback in modern JavaScript, but the truth of the matter is it's more than just callbacks, it's also able to handle multiple values returned through the callback-like method, which is where this starts to come in handy, and unlike promises, Observables, RxJS's asynchronous construct, are not fired until they are subscribed to.

So how can we use RxJS with WebSockets?

According to NestJS's docs, the @WebSocketServer() decorator can be used to inject the WebSocket Server instance in a gateway class. Because this can only be used in a gateway class, the only way to get the websocket server instance into a service class would be to pass it to a setter of the service. This is doable, but not a very developer friendly way to write everything.

As RxJS is already a dependency of NestJS, it would make sense to use something that can make the developer experience more manageable in the long run.

So instead of writing a setter for the websocket server, we can create a Subject from RxJS that has the type of { name: string, data: unknown }. You could use any instead of unknown, but that's generally less secure.

@Injectable()
export class WebsocketService {
  private subject = new Subject<{ name: string, data: unknown }>();
}

Subjects are a useful construct here because they can be turned into an Observable that can later be subscribed to, but like an array can have data pushed onto them. So next we'll create a method for adding an event to the subject.

addEvent(eventName: string, eventData: unknown): void {
  this.subject.next({ name: eventName, data: eventData });
}

With this, so long as our WebsocketService is exported from the WebsocketModule which we will make soon, then we can push events on to the Subject for the actual Websocket Gateway to make use of and emit later. Now let's add a method for our gateway to get this Subject, as we initially made it private.

getEventSubject$(): Observable<{ name: string, data: unknown }> {
  return this.subject.asObservable();
}

The $ at the end of the method name is just a convention to say we are getting back an observable, feel free to not use this convention if you do not like it. Now, in our WebsocketGateway, let's implement the afterInit lifecycle hook so that we can hook into this Observable.

@WebSocketGateway()
export class WebsocketGateway implements OnGatewayInit, OnApplicationShutdown {
  private eventSubscription: Subscription;
  constructor(private readonly service: WebsocketService) {}
  afterInit(server: ServerType): void {
    this.eventSubscription = this.service.getEventSubject$.subscribe({
      next: (event) => server.emit(event.name, event.data),
      error: (err) => server.emit('exception', err),
    })
  }
  onApplicationShutdown() {
    this.eventSubscription.unsubscribe();
  }
}

If necessary, you can also add in filtering to the observable and the event data to ensure that you only emit events to the correct client, as Observables have a similar filter API as Array.prototype. You can also map the data for serialization, or debounce to ensure values aren't emitted too quickly. As we mentioned before, there's a lot that can be done with RxJS.

Finally, let's create the WebsocketModule that will have the providers set up correctly.

@Module({
  providers: [WebsocketService, WebsocketGateway],
  exports: [WebsocketService],
})
export class WebsocketModule {}

The exports: [WebsocketService] is very important here. Because providers are singleton through modules, and because we are making use of the Subject inside of the WebsocketService, we need to ensure we only have one instance of providers: [WebsocketService] in the entire application, otherwise we will end up creating multiple Subject instances and our server.emit won't ever be invoked.

Conclusion

So there you have it, the building blocks for emitting an unknown number of dynamic events from NestJS. We used RxJS Subjects to create an Observable that our websocket server could subscribe to, with data that allowed for dynamic events to be sent back on to the client on behalf of the server, without needing to pass the websocket server instance to the service where the Subject was created, while still allowing other services within the application to add events to emit. From here, you can expand on the example to add in filtering, throttling, or anything else you could want with your websocket server.

To view a working example of this, you can clone this repository and play around with it locally. Extra steps can be found in the README of the repo.