rxjs - BehaviorSubject 'grouped' -

i'm getting started rxjs see if can replace manual data streams. 1 thing i'm trying port situation whereby last value in stream remembered, future observers 'current' value subsequent ones. seems fulfilled behaviorsubject.

however, need group of entities. example, might have data represents message user:

{ userid: 1, message: "hello!" } 

and want behaviorsubject-like object that'll store last message users. can rxjs out-of-the-box, or need myself? (if so, pointers appreciated).

edit: after more thought, perhaps seems logical having 'incoming' subject, observer updates map, , function can call initialises observable map values, , merges incoming stream...?

i use rxjs redux-like state setup. have behaviorsubject holds current state, , every time event/action fired current state gets passed through functions produce new state, subject subscribed to.

here's simplified version of use:

export default class flux {    constructor(state) {     //all resources saved here disposal (for example, when hot loading)     //this flux dispatcher     this.dispatcher = new rx.subject(),     // behaviorsuject constructed initial state     this.state = new rx.behaviorsubject(state),   }    addstore(store, initialstate, feature = store.feature) {       this.dispatcher         .share()         // "reduction" happens. store reducer          // takes existing state , returns new state         .flatmap(({action, payload}) =>             store(this.state.getvalue(), action, payload))         .startwith(initialstate || {})         .subscribe(this.state)     );      return this;   }    addactions(actions: rx.subject) {     // actions fed dispatcher     this.resources.push(actions.share().subscribe(this.dispatcher));     return this;   }  } 

i create global flux object manages state. then, every "feature" or "page" or whatever wish add actions , stores. makes managing state easy, , things time-travel given rx.