一个 Saga 只能从另一个 Saga 内部调用（使用yield foo()或yield call(foo)）。

在您的示例中，fooSaga 是从普通函数（onMessage回调）内部调用的，因此它只会返回迭代器对象。通过从 Saga 产生迭代器（或对生成器的调用），我们允许 redux-saga 中间件拦截该调用并运行迭代器以解决所有产生的效果。但是在您的代码中，stream.onmessage = onMessage只需进行简单的赋值，中间件就不会注意到任何事情。

至于主要问题。Sagas 通常从 Redux 商店获取事件。您可以使用runSaga将 saga 连接到自定义输入/输出源，但将其应用于上述用例并非易事。所以我将提出另一种仅使用call效果的替代方案。但是，为了介绍它，我们必须从Events的push视角转换为pull视角。

处理事件的传统方法是在某些事件源上注册某些事件侦听器。就像在上面的例子中分配onMessage回调一样stream.onmessage。每个事件发生都被推送到侦听器回调。事件源处于完全控制之中。

redux-saga 采用了不同的模型：Sagas拉取所需的 Event。作为回调，它们通常会做一些处理。但是他们可以完全控制下一步做什么：他们可能会选择再次拉动同一个事件——这模仿了回调模型——但他们不是被迫的。他们可能选择拉另一个事件，启动另一个 Saga 来接力，甚至终止他们的执行。即他们可以控制自己的进展逻辑。事件源所能做的就是解决对未来事件的查询。

要集成外部推送源，我们需要将事件源从推送模型转换为拉取模型；即我们必须构建一个事件迭代器，从中我们可以从事件源中提取未来的事件

这是onmessage从一个迭代器派生的例子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
    }
  }
}

上面的函数返回一个带有nextMessage方法的对象，我们可以用它来拉取未来的消息。调用它会返回一个 Promise，它将用下一个传入的消息解析。

具有createSourceAPI函数。我们现在可以通过一个简单的call效果来使用它

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)
}

您可以找到上述代码的实时运行演示。

上述方法的一个优点是它使 Sagas 中的代码保持完全声明性（仅使用声明性形式fork和call）