javascript - Using RxJs to handle namespaced events -

i have question best practice implements namespacing behavior in rx. in current implementation, i'm using event namespacing , propagation in tree. namespace n1:n2:n3. n1 root namespace, n2 first child , n3 first child of n2. tree following one.

# tree result n1 └─ n2    └─ n3 

events sent n1:n2 triggered on n2 on n1 (if n2 doesn't stop propagation). in same way, if event sent n1:n2:n3 triggered on n3 on n2 on n1 (we don't take care stopping propagation here).

i'm looking best practice use rxjs implement behavior (i'm working on back-end side). tried do:

const eventemitter = require('events'); // node 6.9.1 const rx = require('rxjs'); // v4.0 class eventbus extends eventemitter {    emit(event) {       event = event.trim();       if(!this._isvalid(event)) {          return;       }       let args = [].splice.call(arguments, 1);       const parts = event.split(eventbus .glue);       for(let = parts.length; > 0; i--) {          let next = false;          let _args = _.clonedeep(args);          let eventname = parts.join(eventbus .glue);          _args.unshift(eventname);          next = eventemitter.prototype.emit.apply(this, _args);          if(!next) {             break;          }          parts.pop();       }    } } eventbus .glue = ':'; const mediator = new eventbus(); // yes, i'm trying implement mediator pattern old way of thinking new (and better) way of coding :). const myfilter = function(args) { /* ... */ }; // ... const n1 = rx.observable.fromevent(mediator, 'n1', myfilter); const n1n2 = rx.observable.fromevent(mediator, 'n1:n2', myfilter); const n1n2n3 = rx.observable.fromevent(mediator, 'n1:n2:n3', myfilter); // ... 

that become tedious stream namespace number increases.

stream encapsulation using flatmap, or filtering may works before digging code, i'm asking expert (since i'm pure newbie in field) guide me way of thinking concern / implementation.

so have recommendation, backed experiences, way of doing efficiently , smartly rxjs ?

is ins't implemented "natively" on rxjs ?

thanks reading !

regards