Can I use redux-saga's es6 generators as onmessage listener for websockets or eventsource?

Question

I'm trying to get redux-saga working with the onmessage listener. I don't know why what I have isn't working.

I have the following set-up.

// sagas.js
import { take, put } from 'redux-saga';
import {transactions} from "./actions";

function* foo (txs) {
    console.log("yielding");  // appears in console
    yield put(transactions(txs));  // action *is not* dispatched
    console.log("yielded"); //appears in console
}

const onMessage = (event) => {
  const txs = JSON.parse(event.data);
  const iter = foo(txs);
  iter.next(); // do I really need to do this? 
};

function* getTransactions() {
  while(yield take('APP_LOADED')) {
    const stream = new EventSource(eventSourceUrl);

    stream.onopen = onOpen;
    stream.onmessage = onMessage;
    stream.onerror = onError;

    // this is just testing that `yield put` works 
    yield put(transactions([{baz : 42}])); //this action *is* dispatched
  }
};

When the APP_LOADED action is dispatched getTransactions is called, the stream is opened and the onMessage listener is called as data is received from the server, but I'm not having any luck in dispatching the action when calling yield put(transactions(txs)) in the generator foo.

Can anyone tell me what I'm doing wrong?

Answer 1

A Saga can be invoked only from inside another Saga (using yield foo() or yield call(foo)) .

In your example, the foo Saga is called from inside a normal function (onMessage callback) so it'll just return the iterator object. By yielding an iterator (or a call to a generator) from a Saga, we allow the redux-saga middleware to intercept that call and run the iterator in order to resolve all yielded Effects. But in your code, stream.onmessage = onMessage just do a simple assignement so the middleware won't notice anything.

As for the main question. Sagas typically takes events from the Redux store. You can use runSaga to connect a saga to a custom input/output source but it wont be trivial to apply that to the above use case. So I'll propose another alternative using simply the call effect. However, in order to introduce it, we'll have to shift from the push perspective of Events, to a pull perspective.

The traditional way to handle events is to register some event listener on some event source. Like assigning the onMessage callback to stream.onmessage in the example above. Each Event occurrence is pushed to the listener callback. The event source is in total control.

redux-saga adopts a different model: Sagas pull the desired Event. As callbacks, they typically do some processing. But they have total control on what to do next: they may chose to pull the same event again -which mimics the callback model- but they are not forced to. They may chose to pull another Event, start another Saga to take the relay or even terminate their execution. i.e. they are in control of their own logic of progression. All the event source can do is to resolve the queries for future events.

To integrate external push sources, we'll need to transpose the Event Source from the push model into the pull model; i.e. we'll have to build an event iterator from which we can pull the future events from the event source

Here is an example of deriving an onmessage iterator from an EventSource

function createSource(url) {

  const source = new EventSource(url)
  let deferred

  source.onmessage = event => {
    if(deferred) {
      deferred.resolve(JSON.parse(event.data))
      deferred = null 
    }
  }

  return {
    nextMessage() {
      if(!deferred) {
        deferred = {}
        deferred.promise = 
          new Promise(resolve => deferred.resolve = resolve)
      }
      return deferred.promise
    }
  }
}

The above function returns an object with a nextMessage method that we can use to pull the future messages. Calling it will return a Promise that will resolve with the next incoming message.

Having the createSource API function. We can now use it by a simple call effect

function* watchMessages(msgSource) {
  let txs = yield call(msgSource.nextMessage)
  while(txs) {
    yield put(transactions(txs))
    txs = yield call(msgSource.nextMessage)
  } 
}


function* getTransactionsOnLoad() {
  yield take('APP_LOADED')
  const msgSource = yield call(createSource, '/myurl')
  yield fork(watchMessages, msgSource)
}

You can find a live running demo of the above code.

An advantage of the above approach is that it keeps the code inside Sagas fully declarative (using only declarative forms fork and call)