Simple TypeScript Message Bus (supporting Angular DI)

MessageService.ts

import { Injectable } from '@angular/core';
import { Observable, Subject } from 'rxjs/Rx';
import { filter, map } from 'rxjs/operators';

export interface IClassConstructor<T> {
	new (...args: any[]): T;
}

export interface IMessage {
	channel: Function;
	content: any;
}

/** simple class-based publish/subscribe message bus to provide decoupled communication of events, actions, commands */
@Injectable()
export class MessageService {
	private message = new Subject<IMessage>();

	public publish<T>(messageInstance: T): void {	// Flux/Redux: 'dispatch'
		const channel = messageInstance.constructor;
		this.message.next({'channel': channel, 'content': messageInstance}); // Redux: {'type': string, 'payload': any }
	}

	public subscribe<T>(messageClass: IClassConstructor<T>): Observable<T> {
		const channel: IClassConstructor<T> = messageClass;
		return this.message.pipe(
			filter((message: IMessage) => {
				return message.channel === channel;
			}),
			map((message: IMessage) => {
				return message.content;
			})
		);
	}

}

MessageServiceClasses.ts

// Just some sample message classes

/** encapsulation class for messages sent to the MessageService message bus to provide channel type and TypeScript typing */
export class UserNotificationMessage  {
    constructor(
        public message: string,
        public notificationType?: NotificationType,
        public options?: any,
        public notifier?: any
    ) { }
}
export type NotificationType = 'success' | 'info' | 'warning' | 'danger' | 'error';

/** base class for all DomainModelService UPDATE operations */
abstract class DomainModelUpdateRequestMessage  {
    constructor(
        public id: string,
        public domainModelObject: domain.IDomainModel,
        public newValue?: any,
        public oldValue?: any,
    ) { }
}

/** encapsulation class for messages sent to the MessageService message bus to provide TypeScript typing */
export class ProcedureUpdateNameOnClientRequestMessage extends DomainModelUpdateRequestMessage {
    public domainModelObject: domain.IProcedure;
}

publish.ts

import * as message from './MessageServiceClasses';

// assume MessageService is injected
this.MessageService.publish(new message.ProcedureUpdateNameOnClientRequestMessage(this.id, this, newValue, this._name));

subscribe.ts

import * as message from './MessageServiceClasses';

// assume MessageService is injected & some callback handlers defined to pass in
this.MessageService.subscribe(message.ProcedureUpdateNameOnClientRequestMessage)
            .subscribe(this.updateNameOnClientRequestHandle.bind(this), this._genericErrorHandler.bind(this))