RXJS: 스로틀시간 + 마지막 값

RXJS: 스로틀시간 + 마지막 값

나는 짧은 시간 안에 많은 이벤트를 스트림에 보낼 수 있는 시나리오를 가지고 있다.나는 일종의 혼용된 연산자를 갖고 싶다.debounceTime그리고throttleTime.

다음의 데모들은 내가 무엇을 갖고 싶은지 설명하는데 사용될 수 있다. https://stackblitz.com/edit/rxjs6-demo-jxbght?file=index.ts.구독자가 첫 번째 이벤트를 받은 다음 x ms를 기다리길 바란다.대기 시간 중에 더 많은 이벤트가 방출된 경우, 마지막 이벤트는 대기 시간 이후 가입자에게 전송되어야 한다.대기 시간은 디바운스가 하는 것과 마찬가지로 각각의 새로운 방출 이벤트에서 재설정되어야 한다.

1초 이내에 버튼을 3번 클릭하면 1과 3이 인쇄된다.그런 다음 1초 이내에 1번만 클릭하면 4번만 인쇄된다.그런 다음 다시 3번 클릭하면 5번과 7번이 인쇄된다.

이 작업은 다음에서 수행되지 않음debounceTime첫 번째 사건도 안 되고 잘 안 되니까throttleTime대기 시간이 끝난 후에 마지막으로 내보낸 값을 알려주지 않겠어?

이를 구현하는 방법에 대한 제안이 있으십니까?

갱신하다

나는 마틴스의 대답으로 맞춤 제작자를 만들었다.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;
        });
    }
}

RxJS 6에는 다음과 같은 몇 가지 추가 옵션이 있다.throttleTime현재 기록되지 않은 연산자 및 시작과 종료 시 모두 방출할 수 있는 위치.그래서 아마 이게 너에게 도움이 될 거야.

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

이 연산자는 첫 번째 값, 조절 스트림 및 마지막 값을 제공해야 한다.

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 }));
}

참조URL: https://stackoverflow.com/questions/52038067/rxjs-throttletime-plus-last-value