ReactiveX filtering on observable multiple times and merging

Question

I have a problem creating the following observable.
I want it to receive a predefined array of values
And I want to filter by different things, and be able to work with these as individual observables.
And then when it comes time to merge these filtered observables, I want to preserve the order from the original one

//Not sure the share is necessary, just thought it would tie it all together
const input$ = Observable.from([0,1,0,1]).share();
const ones$ = input$.filter(n => n == 1);
const zeroes$ = input$.filter(n => n == 0);

const zeroesChanged$ = zeroes$.mapTo(2);
const onesChanged$ = ones$.mapTo(3);
const allValues$ = Observable.merge(onesChanged$,zeroesChanged$);

 allValues$.subscribe(n => console.log(n));
//Outputs 3,3,2,2
//Expected output 3,2,3,2

EDIT: I am sorry I was not specific enough in my question. I am using a library called cycleJS, which separates sideeffects into drivers. So what I am doing in my cycle is this

export function socketCycle({ SOCKETIO }) {
  const serverConnect$ = SOCKETIO.get('connect').map(serverDidConnect);
  const serverDisconnect$ = SOCKETIO.get('disconnect').map(serverDidDisconnect);
  const serverFailedToConnect$ = SOCKETIO.get('connect_failed').map(serverFailedToConnect);
  return { ACTION: Observable.merge(serverConnect$, serverDisconnect$, serverFailedToConnect$) };
}

Now my problem arose when I wanted to write a test for it. I tried with the following which worked in the wrong matter(using jest)

const inputConnect$ = Observable.from(['connect', 'disconnect', 'connect', 'disconnect']).share();
const expectedOutput$ = Observable.from([
  serverDidConnect(),
  serverDidDisconnect(),
  serverDidConnect(),
  serverDidDisconnect(),
]);
const socketIOMock = {
  get: (evt) => {
    if (evt === 'connect') {
      return inputConnect$.filter(s => s === 'connect');
    } else if (evt === 'disconnect') {
      return inputConnect$.filter(s => s === 'disconnect');
    }
    return Observable.empty();
  },
};
const { ACTION } = socketCycle({ SOCKETIO: socketIOMock });
Observable.zip(ACTION, expectedOutput$).subscribe(
  ([output, expectedOutput]) => { expect(output).toEqual(expectedOutput); },
  (error) => { expect(true).toBe(false) },
  () => { done(); },
);

Maybe there is another way I can go about testing it?

Answer 1

When stream is partitioned, the timing guarantees between elements in different daughter streams is actually destroyed. In particular, even if connect events always come before disconnect events at the event source, the events of the connect Observable won't always come before their corresponding events items in the disconnect Observable. At normal timescales, this race condition probably quite rare but dangerous nonetheless, and this test shows the worst case.

The good news is that your function as shown is just a mapper, between events and results from handlers. If you can continue this model generally over event types, then you can even encode the mapping in a plain data structure, which benefits expressiveness:

const event_handlers = new Map({
 'connect': serverDidConnect,
 'disconnect': serverDidDisconnect,
 'connect_failed': serverFailedToConnect
});
const ACTION = input$.map(event_handlers.get.bind(event_handlers));

Caveat: if you were reducing over the daughter streams (or otherwise considering previous values, like with debounceTime), the refactor is not so straightforward, and would also depend on a new definition of "preserve order". Much of the time, it would still be feasible to reproduce with reduce + a more complicated accumulator.