subscribing-to-stream-processing-results.js

const pendingItems = new Rx.Subject();
const failedItems = new Rx.Subject();
const syncedItems = new Rx.Subject();
const maxRetries = 3;

const queue = (item) => {
  let workingItem = item;
  
  if (!workingItem.sync) {
    workingItem = {
      sync: { counter: 0 },
      item,
    };
  }
  
  if (item.sync.counter > 0 && item.sync.counter >= maxRetries) {
    failedItems.next(workingItem);
    return;
  }  
 
  const workingItem = Object.assign({}, item, {
    sync: {
      counter: item.sync.counter + 1,
      lastTry: new Date(),
    },
  });

  pendingItems.next(workingItem);
}

pendingItems.subscribe((itemWithMetaData) => {
  api.syncEvent(itemWithMetaData.item)
  .then(result => syncedItems.next(itemWithMetaData))
  .catch((reason) => {
    queue(itemWithMetaData);
  });
});

syncedItems.subscribe(x => console.log('successfully synced a mouse event ', x));
failedItems.subscribe(x => console.error('failed to sync mouse event ', x));