An updated MessageBus Service for Angular

I am beginning to learn Angular (not AngularJS / old Angular, but the new hotness), and came across the need to communicate a simple status from one component to another. In most other component-based UI programming paradigms, the thing to do here is to use a Message Bus service to communicate between components. The advantage is low coupling; components don’t need to know about each other, they just need to know about this thing called a Message Bus.

I did some digging, and came across this old blog post about how to build a message bus service in Angular. Unfortunately, copying the code into my IDE gave me an error about a missing filter method…  Alas. After fiddling for a while, here’s what I came up with instead:

import { Injectable } from '@angular/core';

import {Observable, Subject, Subscription} from 'rxjs';

import {filter} from 'rxjs/operators';



interface Message {

  channel: string;
  data: any;

}



@Injectable({
  providedIn: 'root'
})

export class MessagingService {



  private message$: Subject<Message>;



  constructor() {

    this.message$ = new Subject<Message>();

  }



  public publish<T>(message: T): void {

    const channel = (<any>message.constructor).name;

    this.message$.next({ channel: channel, data: message });

  }



  public subscribe<T>(messageType: { new(...args: any[]): T }, subscriber: (value: T) => void): Subscription {

    const channel = (<any>messageType).name;

    const filtered = this.message$.subscribe(m => {

      if (m.channel === channel) {

        subscriber(m.data);

      }

    });

    return filtered;

  }

}


Now I recognize that there are quite a few issues with this implementation. Not the least of which is that each subscriber gets every message, and then has to do string comparison on a payload field to determine if the message is one that they are interested in. After reworking it, I came up with a much more CPU-efficient implementation:

import { Injectable } from '@angular/core';
import {Subject, Subscription} from 'rxjs';

@Injectable({
  providedIn: 'root'
})
export class MessagingService {

  private subjects$: Map<string, Subject<any>>;

  constructor() {
    this.subjects$ = new Map<string, Subject<any>>();
  }

  private getOrCreateChannel(channelName) {
    let channel = this.subjects$.get(channelName);
    if (channel === undefined) {
      channel = new Subject<any>();
      this.subjects$.set(channelName, channel);
    }
    return channel;
  }

  public publish<T>(message: T): void {
    const channelName = (<any>message.constructor).name;
    const channel = this.getOrCreateChannel(channelName);
    channel.next(message);
  }

  /**
   * A more efficient publish method, for cases where a publisher repeatedly produces values on the same topic
   * @param messageType the type of message that we expect tp publish
   */
  public getPublisher<T>(messageType: { new(...args: any[]): T }): (T) => void {
    return this.getOrCreateChannel((<any>messageType).name).next;
  }

  public subscribe<T>(messageType: { new(...args: any[]): T }, subscriber: (value: T) => void): Subscription {
    const channelName = (<any>messageType).name;
    const channel = this.getOrCreateChannel(channelName);
    return channel.subscribe(subscriber);
  }
}