Skip to content

Instantly share code, notes, and snippets.

@GFoley83
Last active June 2, 2020 12:55
Show Gist options
  • Star 8 You must be signed in to star a gist
  • Fork 5 You must be signed in to fork a gist
  • Save GFoley83/ac0004796972036b221ffbeb67aa60d9 to your computer and use it in GitHub Desktop.
Save GFoley83/ac0004796972036b221ffbeb67aa60d9 to your computer and use it in GitHub Desktop.
Angular 2 Message Bus / PubSub ex.
import { Injectable } from "@angular/core";
import { ReplaySubject, Observable } from "rxjs/Rx";
interface Message {
channel: string;
data: any;
}
@Injectable()
export class MessageBus {
public channelSubjects: { channel: string, subject: ReplaySubject<Message> }[];
constructor() {
this.channelSubjects = [];
}
private getTypeSub(channel: string) {
let existingChannelSubject = this.channelSubjects.find(x => x.channel === channel);
if (existingChannelSubject == null) {
existingChannelSubject = { channel: channel, subject: new ReplaySubject<Message>(1) };
this.channelSubjects.push(existingChannelSubject);
}
return existingChannelSubject;
}
public publish<T>(message: T): void {
const channel = Array.isArray(message) ? (<any>message[0].constructor).name + '[]' : (<any>message.constructor).name
const existingTypeSub = this.getTypeSub(channel);
existingTypeSub.subject.next({ channel: channel, data: message });
}
public listenFor<T>(messageType: { new (...args: any[]): T }): Observable<T>
public listenFor<T>(messageType: { new (...args: any[]): T }[]): Observable<T[]>
public listenFor<T>(messageType: { new (...args: any[]): T } | { new (...args: any[]): T }[]): Observable<T> | Observable<T[]> {
const channel = Array.isArray(messageType) ? (<any>messageType[0]).name + '[]' : (<any>messageType).name
const existingTypeSub = this.getTypeSub(channel);
return existingTypeSub.subject.map(m => m.data);
}
}
@GFoley83
Copy link
Author

GFoley83 commented Jan 23, 2017

Expanded on the example shown here: Message Bus pattern in Angular 2 and TypeScript.
Every type or type array that gets published or listened to, gets its own ReplaySubject with a buffer one. The insures that the most recent published value will always be returned for the requested type, even if you subscribe after the data has been published. This can be especially useful for async workflows or lazy-loaded modules.

Usage:

   
    // traditional type you use in your app e.g.
    class Todo {
        // some properties here
        . . .
    }

    // or create classes to represent certain events e.g.
    class TodoAdded {
        newTodo: Todo;
        // some other props here
        . . .
    }


    // Publish
    var myNewTodo = new Todo();
    var myNewTodoArray = [myNewTodo, <more Todos here>];

    this.msgBus.publish(myNewTodo);
    this.msgBus.publish(myNewTodoArray);

    var newTodoAddedEvent: TodoAdded = {
        newTodo: myNewTodo 
    }
    this.msgBus.publish(newTodoAddedEvent);
  
    // Listen

    // List for a new Todo
    this.todoSubscription = this.msgBus.listenFor(Todo)
      .subscribe(todo => { // strongly typed Todo
        console.log(`New Todo: ${todo}`);
      });

    // Listen for an array of Todos
    this.todosSubscription = this.msgBus.listenFor([Todo])
      .subscribe(todos => { // strongly typed Todo[]
        var numTodos = todos.length;
        console.log(`Got ${numTodos} todos`);
      });

    this.todoAdded = this.msgBus.listenFor(TodoAdded)
      .subscribe(todoAdded => { // strongly typed Todo
        console.log(`New Todo: ${todoAdded.newTodo}`);
      });

@gioragutt
Copy link

channelSubjects can be of type { [channel: string]: ReplaySubject<Message> } instead of an array.
What you're doing is making an array of key:value pairs, instead of using the built in map in javascript/typescript.

Also, why use ReplySubject instead of a plain Subject?

What's your opinion about my version of my message bus service?

@vlodko
Copy link

vlodko commented Feb 15, 2018

@gioragutt good question. Why would I need ReplaySubject here?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment