rxjs - Rx-如何应用到允许某个时间段后发出重复项的stream uniqueUntilChanged？(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.

(因此，我必须对Observable实现某个链或某个自定义RxJava运算符，该操作将区分从Observable发出的项目，直到它们改变为止，但仅在短时间内（例如1秒），然后可以再次发出重复的项目。)

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

(我需要的是distinctUntilChanged的一些嵌套组合，与油门 ？)

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

(我找不到符合我要求的任何运算符，因此，可能需要编写一些自定义Rx运算符，但仍然不知道如何开始)

  ask by Patroy translate from so

Answer 1

You can do this with just groupBy operator.

(您可以仅使用groupBy运算符来执行此操作。)

Since each group$ is piped with throttleTime and each emission goes through this group$ Observable it will ignore all subsequent emission for 1s:

(由于每个group$都通过throttleTime传递，并且每个发射都通过该group$ Observable，它将忽略所有后续发射1s：)

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