rxjs가 있는 사용 가능한 Buffer를 생성하는 방법 5

rxjs가 있는 사용 가능한 Buffer를 생성하는 방법 5

나는 내가 생각하기에 사용 가능한 버퍼로 만들려고 노력하고 있다.

나는 이것을 위해 그들의 코드를 누군가 공유하게 했지만, 어떻게 그것을 사용자 정의 작업으로 바꿀 수 있는지 이해할 수 없다. (형식 없이/ 단지 ES6만).

const attach = Rx.Observable.timer(0 * 1000, 8 * 1000).mapTo('@');
const detach = Rx.Observable.timer(4 * 1000, 8 * 1000).mapTo('#');

const input = Rx.Observable.interval(1* 1000);
const pauser = attach.mapTo(true).merge(detach.mapTo(false));

input
  .publish(_input => _input
    .combineLatest(pauser, (v, b) => b)
    .filter(e => e)
    .publish(_switch => _input.bufferWhen(() => _switch.take(1)))
  )
  .flatMap(e => Rx.Observable.from(e))
  .concatMap(e => Rx.Observable.empty().delay(150).startWith(e))

내가 그렇게 할 수 있도록 누가 도와 줄 수 있을까?input.pausableBuffer(pauser)(그리고 startsWith를 정의할 수도 있다.)

다음과 같이 프로토타입에 추가할 수 있다.

var pausableBuffer = function(pauser) {
  return this.publish(_input => _input
    .combineLatest(pauser, (v, b) => b)
    .filter(e => e)
    .publish(_switch => _input.bufferWhen(() => _switch.take(1)))
  )
  .flatMap(e => Rx.Observable.from(e));
}

Rx.Observable.prototype.pausableBuffer = pausableBuffer;

한 가지 기억해야 할 것은 이것이 일시 중지된 상태에서 시작된다는 것이다.활성 상태에서 시작하려면.startWith(true)로pauser.

var pausableBuffer = function(pauser) {
  return this.publish(_input => _input
    .combineLatest(pauser.startWith(true), (v, b) => b)
    .filter(e => e)
    .publish(_switch => _input.bufferWhen(() => _switch.take(1)))
  )
  .flatMap(e => Rx.Observable.from(e));
}

Rx.Observable.prototype.pausableBuffer = pausableBuffer;

업데이트 2019: RxJs 6 스타일:

var pausableBuffer = function(pauser) {
  return (source) => source.pipe(publish(_input => 
  combineLatest(_input, pauser.pipe(startWith(true))).pipe(
    map(([inp, pa]) => pa),
    filter(pa => pa),
    publish(_switch => _input.pipe(bufferWhen(() => _switch.pipe(take(1)))))
  )),
    mergeMap(e => from(e))
  );
}

데모

참조URL: https://stackoverflow.com/questions/40557715/how-can-i-create-a-pausablebuffer-w-rxjs-5