mikl February 2016

How to tie emitted events events into redux-saga?

I'm trying to use redux-saga to connect events from PouchDB to my React.js application, but I'm struggling to figure out how to connect events emitted from PouchDB to my Saga. Since the event uses a callback function (and I can't pass it a generator), I can't use yield put() inside the callback, it gives weird errors after ES2015 compilation (using Webpack).

So here's what I'm trying to accomplish, the part that doesn't work is inside replication.on('change' (info) => {}).

function * startReplication (wrapper) {
  while (yield take(DATABASE_SET_CONFIGURATION)) {
    yield call(wrapper.connect.bind(wrapper))

    // Returns a promise, or false.
    let replication = wrapper.replicate()

    if (replication) {
      replication.on('change', (info) => {
        yield put(replicationChange(info))
      })
    }
  }
}

export default [ startReplication ]

Answers


Nirrek February 2016

The fundamental problem we have to solve is that event emitters are 'push-based', whereas sagas are 'pull-based'.

If you subscribe to an event like so: replication.on('change', (info) => {}) ,then the callback is executed whenever the replication event emitter decides to push a new value.

With sagas, we need to flip the control around. It is the saga that must be in control of when it decides to respond to new change info being available. Put another way, a saga needs to pull the new info.

Below is an example of one way to achieve this:

function* startReplication(wrapper) {
  while (yield take(DATABASE_SET_CONFIGURATION)) {
    yield apply(wrapper, wrapper.connect);
    let replication = wrapper.replicate()
    if (replication)
      yield call(monitorChangeEvents, replication);
  }
}

function* monitorChangeEvents(replication) {
  const stream = createReadableStreamOfChanges(replication);

  while (true) {
    const info = yield stream.read(); // Blocks until the promise resolves
    yield put(replicationChange(info));
  }
}

// Returns a stream object that has read() method we can use to read new info.
// The read() method returns a Promise that will be resolved when info from a
// change event becomes available. This is what allows us to shift from working
// with a 'push-based' model to a 'pull-based' model.
function createReadableStreamOfChanges(replication) {
  let deferred;

  replication.on('change', info => {
    if (!deferred) return;
    deferred.resolve(info);
    deferred = null;
  });

  return {
    read() {
      if (deferred)
        return deferred.promise;

      deferred = {};
      deferred.promise = new Promise(resolve => deferred.resolve = resolve);
      return deferred.promise;
    }
  };
}

There is a JSbin of the above example here:

Yassine Elouafi February 2016

As Nirrek explained it, when you need to connect to push data sources, you'll have to build an event iterator for that source.

I'd like to add that the above mechanism could be made reusable. So we don't have to recreate an event iterator for each different source.

The solution is to create a generic channel with put and take methods. You can call the take method from inside the Generator and connect the put method to the listener interface of your data source.

Here is a possible implementation. Note that the channel buffers messages if no one is waiting for them (e.g. the Generator is busy doing some remote call)

function createChannel () {
  const messageQueue = []
  const resolveQueue = []

  function put (msg) {
    // anyone waiting for a message ?
    if (resolveQueue.length) {
      // deliver the message to the oldest one waiting (First In First Out)
      const nextResolve = resolveQueue.shift()
      nextResolve(msg)
    } else {
      // no one is waiting ? queue the event
      messageQueue.push(msg)
    }
  }

  // returns a Promise resolved with the next message
  function take () {
    // do we have queued messages ?
    if (messageQueue.length) {
      // deliver the oldest queued message
      return Promise.resolve(messageQueue.shift())
    } else {
      // no queued messages ? queue the taker until a message arrives
      return new Promise((resolve) => resolveQueue.push(resolve))
    }
  }

  return {
    take,
    put
  }
}

Then the above channel can be used anytime you want to listen to an external push data source. For your example

function createChangeChannel (replication) {
  const channel = createChannel()

  // every change event will call put on the channel
  replication.on('change', channel.put)
  return channel
}

f 

Post Status

Asked in February 2016
Viewed 1,526 times
Voted 4
Answered 2 times

Search




Leave an answer