How to tie emitted events events into redux-saga?

Question

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 ]

Answer 1

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
}

function * startReplication (getState) {
  // Wait for the configuration to be set. This can happen multiple
  // times during the life cycle, for example when the user wants to
  // switch database/workspace.
  while (yield take(DATABASE_SET_CONFIGURATION)) {
    let state = getState()
    let wrapper = state.database.wrapper

    // Wait for a connection to work.
    yield apply(wrapper, wrapper.connect)

    // Trigger replication, and keep the promise.
    let replication = wrapper.replicate()

    if (replication) {
      yield call(monitorChangeEvents, createChangeChannel(replication))
    }
  }
}

function * monitorChangeEvents (channel) {
  while (true) {
    const info = yield call(channel.take) // Blocks until the promise resolves
    yield put(databaseActions.replicationChange(info))
  }
}

Answer 2

We can use eventChannel of redux-saga

Here is my example

// fetch history messages
function* watchMessageEventChannel(client) {
  const chan = eventChannel(emitter => {
    client.on('message', (message) => emitter(message));
    return () => {
      client.close().then(() => console.log('logout'));
    };
  });
  while (true) {
    const message = yield take(chan);
    yield put(receiveMessage(message));
  }
}

function* fetchMessageHistory(action) {
  const client = yield realtime.createIMClient('demo_uuid');
  // listen message event
  yield fork(watchMessageEventChannel, client);
}

Please Note:

messages on an eventChannel are not buffered by default. If you want to process message event only one by one, you cannot use blocking call after const message = yield take(chan);

Or You have to provide a buffer to the eventChannel factory in order to specify buffering strategy for the channel (e.g. eventChannel(subscriber, buffer)). See redux-saga API docs for more info