Publish modified array to Observable

Question

The task

Suppose we implement an Angular service and need to publish an Observable<number[]> to the world:

numbers: Observable<number[]>;

We want subscribers to:

1. receive the last value upon subscription
2. receive the whole modified array every time we change and publish it

Since #1, internally the Observable<number[]> should be "at least" a BehaviorSubject<number[]>.

But what about #2? Let's suppose we need to implement a method publishNumbersChange() which is called whenever we need to change and publish the changed array:

private publishNumbersChange() {
    // Get current numbers array
    ...        

    changeArray();

    // Now publish changed numbers array
    ...
}

The question

What is the RxJS 5 pattern to implement the task of publishing modified array based on its previous items?

Since I'm asking it mainly because currently I'm doing Angular stuff, here is the second part of the question:
What code does Angular (and the like frameworks based on RxJS) use when they provide an Observable which type parameter is an array to subsequently publish updated array?
Do they just keep a copy of the currently published array separately?

Some thoughts

It seems that storing the underlying array separately, so we always have access to it, is the simplest thing. But at the same time it does not look like a RxJS way (need to have a state outside a RxJS stream).

On the other hand, we could do something like the following:

private publishNumbersChange() {
    // To get the latest value from the stream, we have to subscribe
    const subscription: Subscription = this.numbers.subscribe((numbers: number[]) => {
        // We got the last value in stream in numbers argument. Now make changes to the array
        changeArray();

        // And push it back as a new value to the stream
        this.numbers.next(numbers);
    });

    // Also we have to unsubscribe
    subscription.unsubscribe();
}

I see here at least one issue (not counting the complexity\reusability): "race condition" between executing subscription callback and unsubscribing. Looking at that code you can't tell for sure whether the callback would be actually executed. So it doesn't look as a proper way of doing this either.

Answer 1

It sounds like the operator you may be looking for is scan.

let arraySubject = new BehaviorSubject([]);
let array$ = arraySubject.scan((fullArray, newValue) => fullArray.concat([newValue]), [])

Scan accumulates values over time in an observable stream, and each item in the stream gets the last emitted value and the current value as parameters. executes a function on them and then emits the result. the above example takes a new value and appends it to your full array, the second parameter initializes it to an empty array.

This is clearly kind of restricting though since it only does ONE thing, which may not be robust enough. in this case you need to get clever:

let arraySubject = new BehaviorSubject([]);
let array$ = arraySubject.scan((fullArray, {modifier, payload}) => modifier(fullArray, payload), []);

Now you're passing in an "action" which has a modifier function, which defines how you want to modify the full array, and a payload of any additional data the modifier might need to go into the modifier function along with the full array

so you might do:

let modifier = (full, item) => full.splice(full.indexOf(item), 1);
arraySubject.next({modifier, payload: itemToRemove});

which removes the item you sent through. You can extend this pattern to literally any array modification.

A "gotcha" with scan though is that subscribers only get the accumulated value from the TIME THEY SUBSCRIBED. So, this will happen:

let arraySubject = new BehaviorSubject([]);
let array$ = arraySubject.scan((fullArray, {modifier, payload}) => modifier(fullArray, payload), []);
let subscriber1 = array$.subscribe();
//subscriber1 gets []
let modifier = (full, val) => full.concat([val]);
arraySubject.next({modifier, payload:1});
//subscriber1 gets [1]
arraySubject.next({modifier, payload:2});
//subscriber1 gets [1,2]
let subscriber2 = array$.subscribe();
//subscriber2 gets [2]
arraySubject.next({modifier, payload:3});
//subscriber1 gets [1,2,3]
//subscriber2 gets [2,3]

See what happened there? the only thing stored in the behaviorsubject was the second event, not the full array, scan is storing the full array, so the second subscriber only gets the second action since it wasn't subscribed during the 1st action. So you need a persistent subscriber pattern:

let arraySubject = BehaviorSubject([]);
let arrayModifierSubject = new Subject();
arrayModifierSubject.scan((fullArray, {modifier, payload}) => modifier(fullArray, payload), []).subscribe(arraySubject);

and you modify by calling next on arrayModifierSubject:

let modifier = (full, val) => full.concat([val]);
arrayModifierSubject.next({modifier, payload: 1});

and your subscribers get the array from the array source:

subscriber1 = arraySubject.subscribe();

In this set up, all array modifications go through the modifier subject who in turns broadcasts it to the behaviorsubject who stores the full array for future subscribers and broadcasts it to current subscribers. The behaviorsubject (the store subject) is persistently subscribed to the modifier subject (the action subject), and is the ONLY subscriber to the action subject, so the full array is never lost as the entire history of actions is always maintained.

some sample usages (with the above set up):

// insert 1 at end
let modifier = (full, value) => full.concat([value]);
arrayModifierSubject.next({modifier, payload: 1});

// insert 1 at start
let modifier = (full, value) => [value].concat(full);
arrayModifierSubject.next({modifier, payload: 1});

// remove 1
let modifier = (full, value) => full.splice(full.indexOf(value),1);
arrayModifierSubject.next({modifier, payload: 1});

// change all instances of 1 to 2
let modifier = (full, value) => full.map(v => (v === value.target) ? value.newValue : v);
arrayModifierSubject.next({modifier, payload: {target: 1, newValue: 2}});

you can wrap any of these functions in a "publishNumbersChange" function. How you exactly implement this depends on your needs, you can make functions like:

insertNumber(numberToInsert:number) => {
   let modifier = (full, val) => full.concat([val]);
   publishNumbersChange(modifier, numberToInsert);
}

publishNumbersChange(modifier, payload) => {
   arrayModifierSubject.next({modifier, payload});
}

or you can declare an interface and make classes and use that:

publishNumbersChange({modifier, payload}) => {
   arrayModifierSubject.next({modifier, payload});
}

interface NumberArrayModifier {
    modifier: (full: number[], payload:any) => number[];
    payload: any;
}

class InsertNumber implements NumberArrayModifier {
    modifier = (full: number[], payload: number): number[] => full.concat([payload]);
    payload: number;
    constructor(numberToInsert:number) {
        this.payload = numberToInsert;
    }
}

publishNumbersChange(new InsertNumber(1));

And you can also extend similar functionality to any array modification. One last protip: lodash is a huge help with defining your modifiers in this type of system

so, how might this look in an angular service context?

This is a very simple implementation that isn't highly reusable, but other implementations could be:

const INIT_STATE = [];
@Injectable()
export class NumberArrayService {
    private numberArraySource = new BehaviorSubject(INIT_STATE);
    private numberArrayModifierSource = new Subject();
    numberArray$ = this.numberArraySource.asObservable();

    constructor() {
        this.numberArrayModifierSource.scan((fullArray, {modifier, payload?}) => modifier(fullArray, payload), INIT_STATE).subscribe(this.numberArraySource);
    }

    private publishNumberChange(modifier, payload?) {
        this.numberArrayModifierSource.next({modifier, payload});
    }

    insertNumber(numberToInsert) {
        let modifier = (full, val) => full.concat([val]);
        this.publishNumberChange(modifier, numberToInsert);
    }

    removeNumber(numberToRemove) {
        let modifier = (full, val) => full.splice(full.indexOf(val),1);
        this.publishNumberChange(modifier, numberToRemove);
    }

    sort() {
        let modifier = (full, val) => full.sort();
        this.publishNumberChange(modifier);
    }

    reset() {
        let modifier = (full, val) => INIT_STATE;
        this.publishNumberChange(modifier);
    }
}

Usage here is simple, subscribers just subscribe to numberArray$ and modify the array by calling functions. You use this simple pattern to extend functionality however you like. This controls access to your number array and makes sure it is always modified in ways defined by the api and your state and your subject are always one in the same.

OK but how is this made generic/reusable?

export interface Modifier<T> {
    modifier: (state: T, payload:any) => T;
    payload?: any;
}

export class StoreSubject<T> {
    private storeSource: BehaviorSubject<T>;
    private modifierSource: Subject<Modifier<T>>;
    store$: Observable<T>;

    publish(modifier: Modifier<T>): void {
        this.modifierSource.next(modifier);
    }

    constructor(init_state:T) {
        this.storeSource = new BehaviorSubject<T>(init_state);
        this.modifierSource = new Subject<Modifier<T>>();
        this.modifierSource.scan((acc:T, modifier:Modifier<T>) => modifier.modifier(acc, modifier.payload), init_state).subscribe(this.storeSource);
        this.store$ = this.storeSource.asObservable();
    }
}

and your service becomes:

const INIT_STATE = [];
@Injectable()
export class NumberArrayService {
    private numberArraySource = new StoreSubject<number[]>(INIT_STATE);
    numberArray$ = this.numberArraySource.store$;

    constructor() {
    }

    insertNumber(numberToInsert: number) {
        let modifier = (full, val) => full.concat([val]);
        this.numberArraySource.publish({modifier, payload: numberToInsert});
    }

    removeNumber(numberToRemove: number) {
        let modifier = (full, val) => full.splice(full.indexOf(val),1);
        this.numberArraySource.publish({modifier, payload: numberToRemove});
    }

    sort() {
        let modifier = (full, val) => full.sort();
        this.numberArraySource.publish({modifier});
    }

    reset() {
        let modifier = (full, val) => INIT_STATE;
        this.numberArraySource.publish({modifier});
    }
}