reactive array

dispatch.js

import * as Rx from '@reactivex/rxjs'

function dispatch(key, dispatcher){
    let map = {};
    let newMap = {};
    let subscription = null;
    let extractId = (typeof key === 'string')? (v)=>v[key]: key;

    function createChannel(x){
        let subject = new Rx.BehaviorSubject(x);
        return {subject, observable: dispatcher(subject)};
    }
    
    return xs=>{
        xs = xs || [];
        let cs = xs.map((x)=>{
            let id = extractId(x);
            if(map[id]) return newMap[id] = map[id];
            else return newMap[id] = createChannel(x);
        });
        
        map = newMap;
        
        return Rx.Observable.create((obs)=>{
            if(subscription){
                subscription.unsubscribe();
            }
            cs.forEach((c, ic)=>c.subject.next(xs[ic]));
            subscription = Rx.Observable.zip.apply(null, cs.map((c)=>c.observable)).subscribe(obs);
        });
    };
}

let source = new Rx.BehaviorSubject();

function pretty(v){
    console.log("prett", v);
    return `value: ${v.value} id: ${v.id}`;
}

source.switchMap(dispatch('id', (s)=>s.map(pretty))).subscribe(v=>console.log(v));


source.next([
    {value: "a", id: 1},
    {value: "b", id: 2}
]);


source.next([
    {value: "a", id: 1},
    {value: "b", id: 2},
    {value: "c", id: 3}
]);