RxJS - 缓冲唯一值并在 x 秒内没有发出其他值时发出？答案

【问题标题】：RxJS - Buffer unique values and emit when no other values have been emitted for x seconds?RxJS - 缓冲唯一值并在 x 秒内没有发出其他值时发出？
【发布时间】：2018-05-08 09:58:38
【问题描述】：

用户在表单中键入值，并且每次用户编辑特定字段时都会发出一个事件，该值是他们编辑的字段。

例如，用户在描述字段中键入 3 次，然后在名称字段中键入两次，看起来像

"描述" => "描述" => "描述" => "名称" => "名称" => ...

我想缓冲 unique 值并在用户停止输入 x 秒时将它们作为数组发出。一个值可能会再次出现在不同的缓冲区窗口中。

本质上，这是为了跟踪当用户停止输入并与服务器通信以保存编辑的值时更新了哪些字段。

到目前为止，我有这个每 3000 毫秒发出一次，而且它不会在缓冲时防止重复，而是我们之后对数组进行“重复数据删除”。

   this.update$
      .bufferTime(3000)
      .filter(buffer => buffer.length > 0)
      .map(buffer => [...new Set(buffer)])
      .subscribe(x => console.log(x));

所以它应该一直监听直到发出一个值，然后缓冲 unique 值直到在 x 秒内不再发出任何值，然后发出缓冲区并重复。怎样才能做到这一点？

【问题讨论】：

您询问是否可以在没有自定义运算符的情况下完成此操作，但您没有在代码中使用它。您能详细说明您的问题是什么吗？
@Yoshi 我通常不确定如何执行此操作，我提供的代码解决了类似的问题，但不是同一个问题。我已经删除了自定义运算符位，希望现在问题更清楚了。
我还是有点迷茫，你能解释一下为什么/你的解决方案不够用吗？它似乎已经做到了你所描述的！？
@Yoshi 无论发生什么，我的解决方案都将每 3 秒发出一次缓冲区（我想推回发射，并且只发射直到 x 秒内不再发射任何值）。此外，我的缓冲区不是唯一的，我必须手动完成并删除重复数据。请参阅下面的答案，希望它更清楚。

标签： javascript rxjs rxjs5 reactive-extensions-js

【解决方案1】：

这可能是一个替代版本：

const { Observable } = Rx;

const log = (prefix) => (...args) => { console.log(prefix, ...args); };

const inputs = document.querySelectorAll('input');

const updates$ = Observable
  .fromEvent(inputs, 'input')
  .pluck('target', 'id')
;

// wait x ms after last update
const flush$ = updates$.debounceTime(3000);

const buffered$ = updates$
  // use `distinct` without `keySelector`, but reset with flush$
  .distinct(null, flush$)
  
  // flush the buffer using flush$ as `notifier`
  .buffer(flush$)
;

buffered$.subscribe(log('buffered$ =>'));

<script src="https://unpkg.com/@reactivex/rxjs@^5/dist/global/Rx.min.js"></script>

<div><input type="text" placeholder="foo" id="input.foo"></div>
<div><input type="text" placeholder="bar" id="input.bar"></div>
<div><input type="text" placeholder="baz" id="input.baz"></div>

【讨论】：

我真的很喜欢这个，它让我不必定义自定义运算符。谢谢！

【解决方案2】：

也许我的问题不够清楚，无论如何我已经设法解决了它：（以防它帮助别人）

为了让缓冲区仅在流静默 3 秒时发出，我会在每次用户输入内容（update$ 上发出的事件）时启动一个新计时器，并使用 switchMap取消上一个。

this.update$
  .buffer(this.update$.switchMap(x => Observable.timer(3000)))
  .filter(buffer => buffer.length > 0)
  .map(buffer => [...new Set(buffer)])
  .subscribe(console.log);

然后为了让缓冲区本身是唯一的而不是手动去重，我必须创建一个自定义运算符uniqueBuffer。

this.update$
  .uniqueBuffer(this.update$.switchMap(x => Observable.timer(3000)))
  .filter(buffer => buffer.length > 0)
  .subscribe(console.log);

function uniqueBuffer(emitObservable) {
  return Observable.create(subscriber => {
    const source = this;
    const uniqueBuffer = new Set();

    const subscription = source.subscribe(value => {
      uniqueBuffer.add(value);
    });

    emitObservable.subscribe(emit => {
      subscriber.next([...uniqueBuffer]);
      uniqueBuffer.clear();
    })
  })
}

Observable.prototype.uniqueBuffer = uniqueBuffer;

【讨论】：