Наблюдаемый горячий диапазон RxJs

Question

Я пытаюсь создать наблюдаемую горячую зону. Это означает, что когда у меня есть наблюдатель, наблюдающий наблюдаемое после определенного времени ожидания, он не должен получать значения, которые уже были опубликованы. Я создал следующую программу:

import Rx from "rxjs/Rx";

var x =   Rx.Observable.range(1,10).share()


x.subscribe(x => {
    print('1: ' + x);
  });

setTimeout(() => {
  x.subscribe(x => {
    print('2: ' + x);
  });
}, 1000);

  function print(x) {
    const element = document.createElement('div');
    element.innerText = x;
    document.body.appendChild(element)
  }

Я ожидаю, что эта программа напечатает от 1 до 10, а затем вторая наблюдаемая ничего не печатает, поскольку значения от 1 до 10 создаются в течение первой секунды. Ожидаемый результат показан ниже.

1: 1
1: 2
..
1:10

Однако я вижу, что он также печатает все значения. Несмотря на то, что я поставил оператор share () позади него. Результат показан ниже.

1: 1
..
1: 10
2: 1
..
2: 10

Может кто-нибудь объяснить мне это?

Answer 1

share возвращает наблюдаемое, которое подсчитывается для подписок. Когда счетчик ссылок меняется с нуля на единицу, общая наблюдаемая подписывается на источник - в вашем случае на наблюдаемую range. И когда счетчик ссылок возвращается к нулю, он отписывается от источника.

Ключевым моментом в вашем фрагменте является то, что range генерирует свои значения синхронно, а затем завершается. И завершение приводит к отмене подписки для общей наблюдаемой, и это приводит к тому, что счетчик ссылок возвращается к нулю, что приводит к отказу от общей наблюдаемой от ее источника.

Если вы замените share на publish, вы должны увидеть ожидаемое поведение:

var x =   Rx.Observable.range(1,10).publish();
x.subscribe(x => print('1: ' + x));
x.connect();

publish возвращает ConnectableObservable, который не является счетчиком ссылок, и предоставляет метод connect, который можно вызывать для явного подключения - то есть подписки - к источнику.