Pattern for Observables that includes acknowledgem

问题:

I'm working on something that is recording data coming from a queue. It was easy enough to process the queue into an Observable so that I can have multiple endpoints in my code receiving the information in the queue.

Furthermore, I can be sure that the information arrives in order. That bit works nicely as well since the Observables ensure that. But, one tricky bit is that I don't want the Observer to be notified of the next thing until it has completed processing the previous thing. But the processing done by the Observer is asynchronous.

As a more concrete example that is probably simple enough to follow. Imagine my queue contains URLs. I'm exposing those as an Observable in my code. The I subscribe an Observer whose job is to fetch the URLs and write the content to disk (this is a contrived example, so don't take issue with these specifics). The important point is that fetching and saving are async. My problem is that I don't want the observer to be given the "next" URL from the Observable until they have completed the previous processing.

But the call to next on the Observer interface returns void. So there is no way for the Observer to communicate back to me that has actually completed the async task.

Any suggestions? I suspect there is probably some kind of operator that could be coded up that would basically withhold future values (queue them up in memory?) until it somehow knew the Observer was ready for it. But I was hoping something like that already existed following some established pattern.

回答1:

similar use case i ran into before

window.document.onkeydown=(e)=>{
  return false
}
let count=0;
let asyncTask=(name,time)=>{
  time=time || 2000
  return Rx.Observable.create(function(obs) {
      setTimeout(function() {
       count++
        obs.next('task:'+name+count);
           console.log('Task:',count ,'   ', time, 'task complete') 
        obs.complete();
      }, time);
    });
}

let subject=new Rx.Subject()
let queueExec$=new Rx.Subject()


Rx.Observable.fromEvent(btnA, 'click').subscribe(()=>{
 queueExec$.next(asyncTask('A',4000)) 
})

Rx.Observable.fromEvent(btnB, 'click').subscribe(()=>{
 queueExec$.next(asyncTask('B',4000)) 
})

Rx.Observable.fromEvent(btnC, 'click').subscribe(()=>{
 queueExec$.next(asyncTask('C',4000)) 
})

  queueExec$.concatMap(value=>value)
    .subscribe(function(data) {
      console.log('onNext', data);
    }, 
    function(error) {
      console.log('onError', error);
    },function(){
 console.log('completed') 
});

回答2:

What you describe sounds like "backpressure". You can read about it in RxJS 4 documentation https://github.com/Reactive-Extensions/RxJS/blob/master/doc/gettingstarted/backpressure.md. However this is mentioning operators that don't exist in RxJS 5. For example have a look at "Controlled Observables" that should refer to what you need.

I think you could achieve the same with concatMap and an instance of Subject:

const asyncOperationEnd = new Subject();

source.concatMap(val => asyncOperationEnd
    .mapTo(void 0)
    .startWith(val)
    .take(2) // that's `val` and the `void 0` that ends this inner Observable
  )
  .filter(Boolean) // Always ignore `void 0`
  .subscribe(val => {
    // do some async operation...
    // call `asyncOperationEnd.next()` and let `concatMap` process another value
  });

Fro your description it actually seems like the "observer" you're mentioning works like Subject so it would make maybe more sense to make a custom Subject class that you could use in any Observable chain.

回答3:

Isn't this just concatMap?

// Requests are coming in a stream, with small intervals or without any.
const requests=Rx.Observable.of(2,1,16,8,16)
    .concatMap(v=>Rx.Observable.timer(1000).mapTo(v));

// Fetch, it takes some time.
function fetch(query){
  return Rx.Observable.timer(100*query)
      .mapTo('!'+query).startWith('?'+query);
}

requests.concatMap(q=>fetch(q));

https://rxviz.com/v/Mog1rmGJ

If you want to allow multiple fetches simultaneously, use mergeMap with concurrency parameter.