javascript - 累积和重置流中的值

Question

我正在使用 RxJS 进行响应式编程，并偶然发现了一些我不确定如何解决的问题。

假设我们实现了一台自动售货机。您投入一枚硬币，选择一个商品，机器就会分发一个商品并找零。我们假设价格始终为 1 美分，因此插入 25 美分(25 美分)应返回 24 美分，依此类推。

“棘手”的部分是我希望能够处理用户插入 2 个硬币然后选择一个项目等情况。或者选择一个项目而不插入硬币。

将插入的硬币和选定的项目实现为流似乎很自然。然后，我们可以在这两个操作之间引入某种依赖关系 - 合并或压缩或组合最新的操作。

但是，我很快遇到了一个问题，我希望积累硬币直到分发元素，但不能再进一步。 AFAIU，这意味着我无法使用 sum 或 scan，因为无法在某个时刻“重置”以前的累积。

这是一个示例图:

coins: ---25---5-----10------------|->
acc:   ---25---30----40------------|->
items: ------------foo-----bar-----|->
combined: ---------30,foo--40,bar--|->
change:------------29------39------|->

以及相应的代码:

this.getCoinsStream()
  .scan(function(sum, current) { return sum + current })
  .combineLatest(this.getSelectedItemsStream())
  .subscribe(function(cents, item) {
    dispenseItem(item);
    dispenseChange(cents - 1);
  });

插入 25 和 5 美分，然后选择“foo”项目。积累硬币然后组合最新的将导致“foo”与“30”组合(这是正确的)，然后“bar”与“40”组合(这是不正确的；应该是“bar”和“10”)。

我浏览了all of the methods用于分组和过滤，但没有看到任何我可以使用的东西。

我可以使用的另一种解决方案是单独积累硬币。但这引入了流之外的状态，我真的很想避免这种情况:

var centsDeposited = 0;
this.getCoinsStream().subscribe(function(cents) {
  return centsDeposited += cents;
});

this.getSelectedItemsStream().subscribe(function(item) {
  dispenseItem(item);
  dispenseChange(centsDeposited - 1);
  centsDeposited = 0;
});

此外，这不允许使流相互依赖，例如等待硬币插入，直到所选操作可以返回项目。

我是否缺少已有的方法？实现此类目标的最佳方法是什么 - 累积值直到需要与另一个流合并的那一刻，但同时也要等待第一个流中至少有 1 个值，然后再将其与第二个流中的值合并？

Answer 1

您可以使用 scan/combineLatest 方法，然后使用 first 完成流，然后使用 repeat ，以便它“重新开始”流，但您的观察者不会看到它。

var coinStream = Rx.Observable.merge(
    Rx.Observable.fromEvent($('#add5'), 'click').map(5),
    Rx.Observable.fromEvent($('#add10'), 'click').map(10),
    Rx.Observable.fromEvent($('#add25'), 'click').map(25)
    );
var selectedStream = Rx.Observable.merge(
  Rx.Observable.fromEvent($('#coke'), 'click').map('Coke'),
  Rx.Observable.fromEvent($('#sprite'), 'click').map('sprite')
);

var $selection = $('#selection');
var $change = $('#change');

function dispense(selection) {
 $selection.text('Dispensed: ' + selection); 
 console.log("Dispensing Drink: " + selection);   
}

function dispenseChange(change) {
 $change.text('Dispensed change: ' + change); 
 console.log("Dispensing Change: " + change);   
}


var dispenser = coinStream.scan(function(acc, delta) { return acc + delta; }, 0)
                              .combineLatest(selectedStream, 
                                 function(coins, selection) {
                                   return {coins : coins, selection : selection};
                                 })

                              //Combine latest won't emit until both Observables have a value
                              //so you can safely get the first which will be the point that
                              //both Observables have emitted.
                              .first()

                              //First will complete the stream above so use repeat
                              //to resubscribe to the stream transparently
                              //You could also do this conditionally with while or doWhile
                              .repeat()

                              //If you only will subscribe once, then you won't need this but 
                              //here I am showing how to do it with two subscribers
                              .publish();

    
    //Dole out the change
    dispenser.pluck('coins')
             .map(function(c) { return c - 1;})
             .subscribe(dispenseChange);
    
    //Get the selection for dispensation
    dispenser.pluck('selection').subscribe(dispense);
    
    //Wire it up
    dispenser.connect();

<script src="https://ajax.googleapis.com/ajax/libs/jquery/2.1.1/jquery.min.js"></script>
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.0.6/rx.all.js"></script>
<button id="coke">Coke</button>
<button id="sprite">Sprite</button>
<button id="add5">5</button>
<button id="add10">10</button>
<button id="add25">25</button>
<div id="change"></div>
<div id="selection"></div>