rxjs - Rx - How to apply to stream distinctUntilChanged that allows emit duplicate item after some period?

Question

So, I have to implement some chain or perhaps some custom RxJava operator to an Observable, that will distinct items emitted from Observable, until they change, but only for some short period (like 1 sec), then duplicated item can be emitted again.

What I need is some nested combination of distinctUntilChanged, with throttle?

Main requirements are:

• different items have to be emitted without any delay
• the same item can not be emitted twice in given period

I couldn't find any operator that matches my requirement so, probably I'll need to write some custom Rx operator, but still I can't figure out how to start

标签： rxjsrx-javareactive-programming

Answer 1

您只需使用运算符即可完成此groupBy操作。由于每个都通过group$管道传输，throttleTime并且每个发射都通过这个group$Observable，因此它将忽略所有后续发射 1s：

source$
  .pipe(
    groupBy(item => item.whatever),
    mergeMap(group$ => group$.pipe(
      throttleTime(1000)
    )),
  )
  .subscribe(...);