Observable: skip(number-of-subscribers)

Question

I have an Observable that emits when the callback of an external api is invoked. I would like to skip(n) emissions where n is the amount of subscribers subscribed to the observable.

E.g.: Subscriber that subscribes 2nd should only receive second emission and then unsubscribe.

The skip operator does not work as the number of subscriptions may change.

https://stackblitz.com/edit/rxjs-qdnh9f

let toSkip = 0;
const source = () => {
  return Observable.create((observer) => {
    toSkip++;
    // External API callback
    const handler = (count) => () => {
      observer.next(count++);
    };
    const interval = setInterval(handler(1), 1000)
    const unsubscribe = () => {
      toSkip--;
      console.log('clear interval');
      clearInterval(interval)
    }
    observer.add(unsubscribe);
  }).pipe(
    skip(toSkip), 
    take(1)
  );
}


const subscription1 = source().subscribe(x => console.log('subscription1', x));
const subscription2 = source().subscribe(x => console.log('subscription2', x));
// subscription3 should emit "2" as subscription2 will unsubscribe never run
const subscription3 = source().subscribe(x => console.log('subscription3', x));

setTimeout(() => {
  subscription2.unsubscribe();
}, 500);

Subscription3 should emit "2" as subscription2 will unsubscribe before called.

Expected output on the console:

clear interval
subscription1 1
clear interval
subscription3 2
clear interval

Answer 1

Skip is working, Your first subscription1 skips 1 value and take 1 (0 is skipped 1 is got) subscription3 skips 3 value (0,1,2) and takes 1 ( that is 3). Why it should be 2?

.pipe(
skip(toSkip), 
take(1)

This section is executed once when Observable source is created and initial value is not changed anymore . And it does not matter that toSkip was decreased latter, source 3 was initiated with skip 3 value.

Also keeps in mind that each new subscription for the same observer executes this code

    toSkip++;
    // External API callback
    const handler = (count) => () => {
      observer.next(count++);
    };
    const interval = setInterval(handler(1), 1000)
    const unsubscribe = () => {
      toSkip--;
      console.log('clear interval');
      clearInterval(interval)
    }
    observer.add(unsubscribe);

It means toSkip will be increased for each new subscription. for example this code also increase ToSkip on 2 units.

var source = source();
const subscription1 = source.subscribe(x => console.log('subscription1', x));
const subscription1_1 = source.subscribe(x => console.log('subscription1_1', x));

Also take(1) automatically completes collection and unsubscribes all subscribers that triggers your unsubscribe event too. You could use filter instead skip due to its dynamic nature but using out variables with data state in observable collection is bad practice. that is not enterprise solution :

import { Observable } from 'rxjs'; 
import { map, skip, take, filter } from 'rxjs/operators';

let toSkip = 0;
const source = () => {
  let init;
  return Observable.create((observer) => {
    toSkip++;
    init = toSkip;
    // External API callback
    const handler = (count) => () => {

      observer.next(count++);
      console.log('count ' + count);
    };
    const interval = setInterval(handler(1), 1000)
    const unsubscribe = () => {    

      console.log(' clear interval ' + toSkip);
      clearInterval(interval)
    }
    observer.add(unsubscribe);
    console.log('skip ' + toSkip);
  }).pipe(
    filter((x) =>
    {
      console.log(x + ' - ' + toSkip);
       return x == init || x == toSkip
       }),
       take(1)
       );
}

const subscription1 = source().subscribe(x => {
   console.log('subscription1', x);   
   });

const subscription2 = source().subscribe(x => { 
  console.log('subscription2', x);

});
// subscription3 should emit "2" as subscription2 will unsubscribe never run
const subscription3 = source().subscribe(x => {
   console.log('subscription3', x)

});

setTimeout(() => {
   toSkip--;
  subscription2.unsubscribe();
}, 500)

Answer 2

Skip has a static argument, but in this situation we should use dynamically changed variable, so need to change operator.

Also we can't name function inside observer creation as unsubscribe because it calls after complete. We can't track how many times we unsubscribed so we can return a wrapped method which will do this for us.

https://stackblitz.com/edit/rxjs-bpfcgm - check there several cases

import { Observable } from 'rxjs'; 
import { map, take, filter } from 'rxjs/operators';


const getSource = function() {

  let inc = 0;
  let unsubscribed = 0;

  const source = () => {
    inc++;

    let created = inc;
    let handlerCount = 0;

    return Observable.create((observer) => {

      // External API callback
      const handler = (count) => () => {
        handlerCount++;
        observer.next(count++); // Emit any value here
      };
      const interval = setInterval(handler(1), 1000)
      const complete = () => {
        console.log('clear interval');
        clearInterval(interval)
      }
      return complete;
    }).pipe(
      filter(() => handlerCount >= created - unsubscribed), 
      take(1)
    );
  }

  const unsubscribe = o => {
    unsubscribed++;
    o.unsubscribe();
  }

  return [source, unsubscribe];
}

let [source, unsubscribe] = getSource();

const subscription1 = source().subscribe(x => console.log('subscription1', x));
const subscription2 = source().subscribe(x => console.log('subscription2', x));
// subscription3 should emit "2" as subscription2 will unsubscribe never run
const subscription3 = source().subscribe(x => console.log('subscription3', x));

setTimeout(() => {
  unsubscribe(subscription2);
}, 500)