可观察：跳过（订阅者数量）

Question

我有一个 Observable，它在调用外部 api 的回调时发出。我想跳过（n）个排放，其中 n 是订阅 observable 的订阅者数量。

例如：订阅 2nd 的订阅者应该只收到第二个发射然后取消订阅。

skip 运算符不起作用，因为订阅数量可能会发生变化。

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 应该发出“2”，因为 subscription2 将在调用之前取消订阅。

控制台上的预期输出：

clear interval
subscription1 1
clear interval
subscription3 2
clear interval

Answer 1

跳过有效，您的第一个订阅 1 跳过 1 个值并取 1（跳过 0 得到 1） subscription3 跳过 3 个值 (0,1,2) 并取 1 (即 3)。 为什么应该是2？

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

当 Observable 源被创建并且初始值不再改变时执行一次。后面减少 toSkip 也没关系，源 3 是用 skip 3 值启动的。

还要记住，同一观察者的每个新订阅都会执行此代码

    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);

这意味着每次新订阅都会增加 toSkip。 例如，此代码还增加了 2 个单位的 ToSkip。

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

也 take(1) 自动完成收集并取消订阅所有触发你的取消订阅事件的订阅者。 由于它的动态特性，您可以使用过滤器而不是跳过，但是在可观察集合中使用具有数据状态的变量是不好的做法。 这不是企业解决方案：

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 有一个静态参数，但是在这种情况下我们应该使用动态变化的变量，所以需要改变操作符。

我们也不能将观察者创建中的函数命名为unsubscribe，因为它在完成后调用。我们无法跟踪我们取消订阅的次数，因此我们可以返回一个包装的方法来为我们执行此操作。

https://stackblitz.com/edit/rxjs-bpfcgm - 检查几个案例

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)