Distribute stream on an other stream

rx-distib.js

console.clear();

const items$ = Rx.Observable
  .interval(100)
  .map(x => `item-${x}`)
  .take(100)
  .share();

const cc$ = Rx.Observable
  .interval(0)
  .take(2)
  .scan(count => count + 1, 0)
  .share();

const zip$ = Rx.Observable
  .zip(
    items$, 
    items$.scan(count => count + 1, 0)
  );

Rx.Observable
  .combineLatest(zip$, cc$, (z, cc) => {
    return [z[0], z[1], cc];
  })
  //.subscribe(x => console.log(x));
  .groupBy(([i, c, cc]) => c % cc)
  .flatMap(x => x)
  .subscribe(x => console.log(x));

console.clear();

document.body.innerHTML += "<p style='width: 400px' id='div'></p>"
let div = document.querySelector('#div');

function rxDistribute(items$, clients$) {
  const cc$ = clients$
    .scan(count => count + 1, 0)
    .share();

  const zip$ = Rx.Observable.zip(
    items$, 
    items$.scan(count => count + 1, 0)
  );

  return Rx.Observable
    .combineLatest(zip$, cc$, ([item, index], count) => 
        [index, index, count]
    )
    .groupBy(
      ([i, c, cc]) => c % cc,
      ([item, index, count]) => [item, index % count]
    )
    .flatMap(x => x)
    .flatMap(([item, client]) => Rx.Observable.combineLatest(
      Rx.Observable.of(item),
      clients$.elementAt(client),
      (item, clientItem) => [item, clientItem]
    ));
}

const items$ = Rx.Observable
  .interval(100)
  .map(x => `item-${x}`)
  .take(16)
  .share();

const clients$ = Rx.Observable
  .interval(500)
  .map(client => ({client}))
  .take(5);

rxDistribute(items$, clients$)
    .filter(([_, c]) => c.client == 3)
    .scan(count => count + 1, 0)
    .subscribe(x => console.log(x));