RxJS, how to poll an API to continuously check for updated records using a dynamic timestamp

Question

I am new to RxJS and I am trying to write an app that will accomplish the following things:

1. On load, make an AJAX request (faked as fetchItems() for simplicity) to fetch a list of items.
2. Every second after that, make an AJAX request to get the items.
3. When checking for new items, ONLY items changed after the most recent timestamp should be returned.
4. There shouldn't be any state external to the observables.

My first attempt was very straight forward and met goals 1, 2 and 4.

var data$ = Rx.Observable.interval(1000)
  .startWith('run right away')
  .map(function() { 
    // `fetchItems(modifiedSince)` returns an array of items modified after `modifiedSince`, but
    // I am not yet tracking the `modifiedSince` timestamp yet so all items will always be returned
    return fetchItems(); 
  });

Now I'm excited, that was easy, it can't be that much harder to meet goal 3...several hours later this is where I am at:

var modifiedSince = null;
var data$ = Rx.Observable.interval(1000)
  .startWith('run right away')
  .flatMap(function() { 
    // `fetchItems(modifiedSince)` returns an array of items modified after `modifiedSince`
    return fetchItems(modifiedSince);
  })
  .do(function(item) {
    if(item.updatedAt > modifiedSince) {
      modifiedSince = item.updatedAt;
    }
  })
  .scan(function(previous, current) {
    previous.push(current);
    return previous;
  }, []);

This solves goal 3, but regresses on goal 4. I am now storing state outside of the observable.

I'm assuming that global modifiedSince and the .do() block aren't the best way of accomplishing this. Any guidance would be greatly appreciated.

EDIT: hopefully clarified what I am looking for with this question.

Answer 1

You seem to mean that modifiedSince is part of the state you carry, so it should appear in the scan. Why don-t you move the action in do into the scan too?. Your seed would then be {modifiedSince: null, itemArray: []}.

Errr, I just thought that this might not work, as you need to feed modifiedSince back to the fetchItem function which is upstream. Don't you have a cycle here? That means you would have to use a subject to break that cycle. Alternatively you can try to keep modifiedSince encapsulated in a closure. Something like

function pollItems (fetchItems, polling_frequency) {
var modifiedSince = null;
var data$ = Rx.Observable.interval(polling_frequency)
  .startWith('run right away')
  .flatMap(function() { 
    // `fetchItems(modifiedSince)` returns an array of items modified after `modifiedSince`
    return fetchItems(modifiedSince);
  })
  .do(function(item) {
    if(item.updatedAt > modifiedSince) {
      modifiedSince = item.updatedAt;
    }
  })
  .scan(function(previous, current) {
    previous.push(current);
    return previous;
  }, []);

return data$;
}

I have to run out to celebrate the new year, if that does not work, I can give another try later (maybe using the expand operator, the other version of scan).

Answer 2

How about this:

var interval = 1000;
function fetchItems() {
    return items;
}

var data$ = Rx.Observable.interval(interval)
  .map(function() { return fetchItems(); })
  .filter(function(x) {return x.lastModified > Date.now() - interval}
  .skip(1)
  .startWith(fetchItems());

That should filter the source only for new items, plus start you off with the full collection. Just write the filter function to be appropriate for your data source.

Or by passing an argument to fetchItems:

var interval = 1000;
function fetchItems(modifiedSince) {
    var retVal = modifiedSince ? items.filter( function(x) {return x.lastModified > modifiedSince}) : items
    return retVal;
}

var data$ = Rx.Observable.interval(interval)
  .map(function() { return fetchItems(Date.now() - interval); })
  .skip(1)
  .startWith(fetchItems());