why is the second item in these RxJS.Observable sequences getting skipped?

Question

I have the following RxJS code:

let items = Observable.of({
  val: "one",
  type: "string"
}, {
  val: "five",
  type: "string"
}, {
  val: 2,
  type: "integer"
}, {
  val: 10,
  type: "integer"
}, {
  val: 20,
  type: "integer"
});

items
  .groupBy(x => x.type)
  .subscribe(obs => {
    let head = obs.first();
    let tail = obs.skip(1);
    head
      .do(x => console.log('head == ', x))
      .concat(tail)
      .do(x => console.log('tail == ', x))
      .subscribe(x => console.log('sub == ', x));

  });

Here's the output:

head ==  Object {val: "one", type: "string"}
tail ==  Object {val: "one", type: "string"}
sub ==  Object {val: "one", type: "string"}

head ==  Object {val: 2, type: "integer"}
tail ==  Object {val: 2, type: "integer"}
sub ==  Object {val: 2, type: "integer"}
tail ==  Object {val: 20, type: "integer"}
sub ==  Object {val: 20, type: "integer"}

Why is the second item not getting emitted?

Here's a Plunk that demonstrates my problem.

Answer 1

It is a bit hard to explain what is happening here, but there is no magic so let's see if I can find the words (see thereafter). Following that lengthy explanation is a proposed workaround.

Let's say observer is obs => { let head = obs.first();....

• Observable.of(...) is a cold observable which will emit one by one the items... whenever an observer will subscribe (definition of cold observable).
• groupBy is also a cold observable, but it creates hot observables which are your type groups. That means that when a items comes in, it is associated to a group observable and immediately emitted to whatever observer were there (definition of hot observable). If no observer, then the value is lost.
• Your first subscribe(observer) leads to the following :
  • observer subscribes to groupBy which subscribes to Observable.of which hence emit the first item from items (val: "one", type: "string"). Let-s call it item_string_1.
  • groupBy creates the first group hot observable (call it obs_string), emits it to observer THEN emits the first item on that first group.
  • when obs_string is emitted, observer is executed (remember observers are executed every time their related observable emits). That means that item_string_1 will be emitted AFTER observer is executed.
  • First observer execution : Your subscribe in the observer leads to the following:
    • subscription of observer2 (x => console.log('sub == ', x)) to concat(tail)
    • concat(tail) subscribes immediately to head and DOES NOT subscribe to tail in this moment. It will subscribe to it only when head has completed. That's one of the key points.
    • head leads to subscription to first which leads to subscription to obs, which leads to... nothing as no value has been emitted by obs.
    • end of observer execution
  • then item_string_1 is emitted on obs_string. That leads to :
    • head emitting item_string_1, concat(tail) emitting item_string1 and all your console.log showing Object {val: "one", type: "string"}.
    • head completes
  • then item_string_2 is emitted on obs_string. That leads to :
    • concat(tail) subscribes to tail which subscribes to skip(1) which subscribes to obs (which is obs_string for the moment). That means that concat(tail) emits... nothing. It skips the first value of obs, first value counted from the moment it subscribed. That first value is item_string_2 which is the second value emitted. Conclusion : your second value is NEVER emitted.
  • then item_integer_1 is emitted. Same behaviour as item_string_1
  • then item_integer_2 is emitted. Same behaviour as item_string_2 (skipped)
  • then item_integer_3 is emitted. Now the skip(1) emits val: 20, type: "integer" which makes it way to the messages in console.log

Proposed workaround:

Not sure if I understand well, but if you want to do something specific depending on the index of the element, just use map(selector(value, index)) as follows :

items
  .groupBy(x => x.type)
  .subscribe(obs => obs.map(function(value, index){
                               return index == 0 ? process_first(value)
                                                 : process_others(value)
                            })
                       .subscribe(x => console.log('sub == ', x))