RXJS：限制时间加上最后一个值

Question

我有一个场景，其中许多事件可以在短时间内发送到流。我想要一个操作符，它是debounceTime和throttleTime的混合体。

下面的演示可以用来说明我想要什么，https://stackblitz.com/edit/rxjs6-demo-jxbght?file=index.ts.我希望订阅服务器获取第一个发出的事件，然后等待x毫秒。如果在等待时间内发出更多事件，则应在等待时间后将最后一个事件发送给订阅服务器。等待时间应该在每个新发出的事件上重置，就像去盎司一样。

如果在1秒内单击按钮3次，则应打印1和3。如果在1秒内仅单击1次，则应仅打印4次。如果再次单击3次，则应打印5和7。

这不适用于debounceTime，因为它不会给我第一个事件，也不适用于throttleTime，因为它不会给我等待时间结束后最后发出的值。

有没有关于如何实施的建议？

更新

在Martins回答的帮助下，我创建了一个自定义运算符。不确定它是否100%正确工作，或者是否有更好的方法来做，但它似乎做了我想要的。

import { Observable, empty } from 'rxjs';
import { exhaustMap, timeoutWith, debounceTime, take, startWith } from 'rxjs/operators';

export function takeFirstThenDebounceTime(waitTime) {
    return function takeFirstThenDebounceTimeImplementation(source) {
        return Observable.create(subscriber => {
            const subscription = source.
                pipe(
                    exhaustMap(val => source.pipe(
                        timeoutWith(waitTime, empty()),
                        debounceTime(waitTime),
                        take(1),
                        startWith(val)
                    )),
                )
                .subscribe(value => {
                    subscriber.next(value);
                },
                    err => subscriber.error(err),
                    () => subscriber.complete());

            return subscription;
        });
    }
}

Answer 1

此运算符应该为您提供第一个值、受限制的流和最后一个值：

js prettyprint-override">export function throunceTime<T>(duration: number): MonoTypeOperatorFunction<T> {
  return (source: Observable<T>) =>
    merge(source.pipe(throttleTime(duration)), source.pipe(debounceTime(duration)))
      .pipe(throttleTime(0, undefined, { leading: true, trailing: false }));
}

Answer 2

在RxJS 6中，有一些额外的选项用于限制时间运算符，这些运算符现在没有文档化，您可以在持续时间的开始和结束时使其发出。所以也许这能帮到你。

https://github.com/ReactiveX/rxjs/blob/master/src/internal/operators/throttleTime.ts#L55

https://github.com/ReactiveX/rxjs/blob/master/src/internal/operators/throttle.ts#L12

但是，由于要在每次发射时重置超时，因此会更加复杂。这应该是随机发射的简化示例，但我想知道是否有更简单的方法：

const shared = source.pipe(shareReplay(1))

shared
  .pipe(
    exhaustMap(val => shared.pipe(
      timeout(1000),
      catchError(() => empty()),
      debounceTime(1000),
      take(1),
      startWith(val),
    ))
  )
  .subscribe(v => console.log(v))

此演示在175ms间隙后衰减。我希望这对你有意义。

演示：https://stackblitz.com/edit/rxjs6-demo-ztppwy?file=index.ts