RxJS 并非所有订阅者都收到所有事件

Question

我正在做一个关于 RxJS 的练习。并且发生了一些非常奇怪的事情：

  typoStream.subscribe(x => console.log('wont get executed'));
  wordCompletedStream.subscribe(nextStream);
  typoStream.subscribe(x => console.log('will get executed'));

当应用程序运行时，第一个 console.log 不会打印，第二个会。
不管流是什么以及它们如何交互——这永远不应该发生，对吧？为什么当我订阅一个可观察对象时它很重要 - 它不应该向每个订阅者发送事件吗？

如果你想试试： http://embed.plnkr.co/xb8Yimo5RcYGPtgClYgY/

正确输入显示的单词，您可以看到 "error" 的效果。但它不会每次都发生 - 只有大部分时间。

这里是流：https://photos.app.goo.gl/Z4cpKzekAIuKzMF93

Answer 1

我玩过你发布的代码，关键修复是正确多播 checkWord observable。您可以像 wordStream 那样使用 .publish().refCount() 来执行此操作，或者您可以使用执行相同操作的快捷方法 .share().

  const checkStream = wordStream.combineLatest(inputStream).share();

之所以可行，是因为如果没有它，对 checkStream 的多个订阅或从它派生的任何流，例如 typoStream 和 wordCompletedStream 都会触发对wordStream observable（正确多播，因此不会发出新请求）和 inputStream observable，它将在输入上注册新的事件侦听器。

使用 .share() 运算符，对 checkStream 或派生的可观察对象进行了多少次订阅并不重要，只有第一个会触发对 inputStream 的订阅。

请注意，此修复后，typoStream 的两个订阅者都不会因输入正确的单词而触发。这是我对名为 typoStream 的可观察对象的期望。当输入不正确的字符时，两者都会触发。

Forked Plunkr here

或查看下面的代码片段：

(() => {

  // --- UI Stuff, NO NEED TO TOUCH THESE --- //
  const wordField = $('#TotDWord');
  const inputField = $('#TotDInput');
  // ----------------------------------------- //

  // A stream of the users string inputs
  const inputFieldStream = Rx.Observable.fromEvent(inputField, 'keyup')
    .map(x => x.target.value).distinctUntilChanged();

  // This stream is used to represent the users unput - we don't use the 
  // inputFieldStream directly because we want to manually send values aswell
  const inputStream = new Rx.Subject();


  // Feed the stream from the field into our inputStream
  inputFieldStream.subscribe(inputStream);

  // A stream that allows us to manually trigger that we need a new word
  const nextStream = new Rx.Subject();

  // When we want the next word we need to reset the users input
  nextStream.subscribe(() => {
    inputField.val('');
    inputStream.onNext('');
  });

  // This stream calls a server for a new random word every time the nextStream emits an event. We startWith a value to trigger the first word
  const wordStream = nextStream.startWith('')
    .flatMapLatest(getRandomWord)
    // publish & refCount cache the result - otherwise every .map on wordStream would cause a new HTTP request
    .publish().refCount();

  // When there is a new word, we display it
  wordStream.subscribe(word => {
    wordField.empty();
    wordField.append(word);
  });

  // Checkstream combines the latest word with the latest userinput. It emits an array, like this ['the word', 'the user input'];
  const checkStream = wordStream.combineLatest(inputStream).share();

  // Emits an event if the user input is not correct
  const typoStream = checkStream.filter(tuple => {
    const word = tuple[0];
    const input = tuple[1];
    return !word.startsWith(input);
  });

  // When there is a typo we need a new word
  typoStream.subscribe(nextStream);

  // Emits an event when the user has entered the entire word correctly
  const wordCompletedStream = checkStream.filter(tuple => {
    const word = tuple[0];
    const input = tuple[1];
    return word == input;
  });

  /**
   * THIS WILL (MOST OF THE TIME) NOT FIRE WHEN YOU COMPLETE A WORD
   */
  typoStream.subscribe(x => console.log('wont get executed'));

  // Whenever the word is completed, request a new word
  wordCompletedStream.subscribe(nextStream);

  /**
   * THIS WILL FIRE WHEN YOU COMPLETE A WORD
   */
  typoStream.subscribe(x => console.log('will get executed'));

  // Calls a server for a random word
  // returns a promise
  function getRandomWord() {
    return $.ajax({
      // Change the URL to cause a 404 error
      url: 'https://setgetgo.com/randomword/get.php'
    }).promise();
  }

})();

<script data-require="jquery" data-semver="3.1.1" src="https://ajax.googleapis.com/ajax/libs/jquery/3.1.1/jquery.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.6/rx.all.js"></script>

<div>
  <h1>Exercise: Typing of the Dead</h1>
  <div>
    Type the given word correctly and watch the console. Most of the time 1 of the 2 subscriptions on the typoStream will fire (when there should fire none).
  </div>
  <br />
  <div id="TotDWord"></div>
  <input type="text" name="" id="TotDInput" value="" /><span>Highscore: </span><span id="TotDHighscore"></span>
  <div id="TotDScore"></div>
</div>

<script>
  console.clear();
</script>

RxJS 并非所有订阅者都收到所有事件

RxJS not all Subscribers receive all events

javascript

rxjs

reactivex

reactive