If you're in need of an operator, scan might be the one you're searching for.

let subjectArray = new BehaviorSubject([]);
let array$ = subjectArray.scan((fullArr, newVal) => fullArr.concat([newVal]), [])

Scan function accumulates values over time within an observable stream, where each item in the stream receives the previously emitted value and the current value as inputs. It then applies a function to them and emits the result. For instance, the above example takes a new value and adds it to your full array, with the second parameter initializing the array as empty.

However, if you find this limitation restricting, you can get creative:

let subjectArray = new BehaviorSubject([]);
let array$ = subjectArray.scan((fullArr, {changeFunc, input}) => changeFunc(fullArr, input), []);

Now, by passing an "action" that contains a modifier function specifying how you want to alter the full array, along with any additional data required by the modifier function.

For example, you could do the following:

let changeFunc = (full, item) => full.splice(full.indexOf(item), 1);
subjectArray.next({changeFunc, input: itemToRemove});

This action removes the specified item. You can extend this approach to accommodate any array modifications.

One thing to note about scan is that subscribers only receive the accumulated value from the point they subscribed onwards. This results in:

let subjectArray = new BehaviorSubject([]);
let array$ = subjectArray.scan((fullArr, {changeFunc, input}) => changeFunc(fullArr, input), []);
let subscriber1 = array$.subscribe();
//subscriber1 gets []
let changeFunc = (full, val) => full.concat([val]);
subjectArray.next({changeFunc, input: 1});
//subscriber1 gets [1]
subjectArray.next({changeFunc, input: 2});
//subscriber1 gets [1,2]
let subscriber2 = array$.subscribe();
//subscriber2 gets [2]
subjectArray.next({changeFunc, input: 3});
//subscriber1 gets [1,2,3]
//subscriber2 gets [2,3]

In this scenario, the BehaviorSubject only stores the second event, not the entire array, while scan retains the complete array. As a result, the second subscriber only receives the second action because it wasn't subscribed during the first action. To address this issue, consider implementing a persistent subscriber pattern:

let subjectArray = BehaviorSubject([]);
let modifySubject = new Subject();
modifySubject.scan((fullArr, {changeFunc, input}) => changeFunc(fullArr, input), []).subscribe(subjectArray);

To make modifications, use next on modifySubject:

let changeFunc = (full, val) => full.concat([val]);
modifySubject.next({changeFunc, input: 1});

Your subscribers will obtain the array from the array source:

subscriber1 = subjectArray.subscribe();

In this setup, all array changes flow through the modifySubject, which broadcasts them to the BehaviorSubject for storage and distribution to subscribers. The BehaviorSubject remains persistently subscribed to the modifySubject and serves as the sole subscriber to ensure the entire history of actions is preserved.

Below are a few sample usages with the mentioned setup:

// insert 1 at the end
let changeFunc = (full, value) => full.concat([value]);
modifySubject.next({changeFunc, input: 1});

// insert 1 at the start
let changeFunc = (full, value) => [value].concat(full);
modifySubject.next({changeFunc, input: 1});

// remove 1
let changeFunc = (full, value) => full.splice(full.indexOf(value), 1);
modifySubject.next({changeFunc, input: 1});

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

You can encapsulate these functions in a "publishNumbersChange" method. The implementation varies based on your requirements—options include creating functions like:

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

publishNumbersChange(changeFunc, input) => {
   modifySubject.next({changeFunc, input});
}

Alternatively, you can define an interface, create classes, and leverage those:

publishNumbersChange({changeFunc, input}) => {
   modifySubject.next({changeFunc, input});
}

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

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

publishNumbersChange(new InsertNumber(1));

This approach can be extended to any array modification. A final tip: lodash proves invaluable when defining modifiers in such systems.

Considering an Angular service context, the following represents a simple yet somewhat non-reusable implementation:

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, {changeFunc, input?}) => changeFunc(fullArray, input), INIT_STATE).subscribe(this.numberArraySource);
    }

    private publishNumberChange(changeFunc, input?) {
        this.numberArrayModifierSource.next({changeFunc, input});
    }

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

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

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

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

Usage involves subscribing to numberArray$ and modifying the array by invoking functions. This straightforward pattern allows for easy extension to tailor functionality according to requirements. It safeguards access to your number array by ensuring modifications align with the API, state, and maintaining consistency between subjects.

How can this be made more generic/reusable?

export interface Modifier<T> {
    changeFunc: (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.changeFunc(acc, modifier.payload), init_state).subscribe(this.storeSource);
        this.store$ = this.storeSource.asObservable();
    }
}

The modified service would appear as follows:

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

    constructor() {
    }

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

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

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

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