Как связать события, связанные с событиями, в саунд-редукцию?

Ответ 1

Как объяснил Ниррк, когда вам нужно подключиться к источникам данных push, вам нужно будет построить итератор событий для этого источника.

Я хотел бы добавить, что вышеупомянутый механизм можно сделать многоразовым. Поэтому нам не нужно воссоздавать итератор событий для каждого другого источника.

Решение состоит в создании общего канала с методами put и take. Вы можете вызвать метод take изнутри генератора и подключить метод put к интерфейсу прослушивателя вашего источника данных.

Вот возможная реализация. Обратите внимание, что канал буферизует сообщения, если их никто не ждет (например, генератор занят удаленным вызовом)

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

Затем вышеупомянутый канал можно использовать в любое время, когда вы хотите прослушивать внешний источник данных push. Для вашего примера

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

Ответ 2

Основная проблема, которую мы должны решить, состоит в том, что излучатели событий являются "push-based", тогда как саги являются "основанными на тяге".

Если вы подписаны на такое событие: replication.on('change', (info) => {}), тогда обратный вызов выполняется всякий раз, когда эмиттер события replication решает нажимать новое значение.

С сагами нам нужно перевернуть контроль. Сага должна контролироваться, когда она решает ответить на доступную новую информацию об изменении. Иными словами, сага должна тянуть новую информацию.

Ниже приведен пример одного из способов достижения этого:

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

Здесь приведен пример JSbin: http://jsbin.com/cujudes/edit?js,console

Вы также должны взглянуть на ответ Яссина Элуафи на аналогичный вопрос: Могу ли я использовать генераторы es6 из редукционной саги в качестве приемника onmessage для websockets или eventource?

Ответ 3

Мы можем использовать eventChannel сокращения-саги

Вот мой пример

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

Обратите внимание:

сообщения в eventChannel по умолчанию не буферизуются. Если вы хотите обрабатывать message event только один за другим, вы не можете использовать блокирующий вызов после const message = yield take(chan);

Или вы должны предоставить буфер eventChannel factory, чтобы указать стратегию буферизации для канала (например, eventChannel (подписчик, буфер)). Подробнее см. документы API-интерфейсов redux-saga для получения дополнительной информации

Ответ 4

Благодаря @Yassine Elouafi

Я создал короткую лицензию на использование общих каналов MIT в качестве расширения redux-saga для языка TypeScript на основе решения @Yassine Elouafi.

// redux-saga/channels.ts
import { Saga } from 'redux-saga';
import { call, fork } from 'redux-saga/effects';

export interface IChannel<TMessage> {
    take(): Promise<TMessage>;
    put(message: TMessage): void;
}

export function* takeEvery<TMessage>(channel: IChannel<TMessage>, saga: Saga) {
    while (true) {
        const message: TMessage = yield call(channel.take);
        yield fork(saga, message);
    }
}

export function createChannel<TMessage>(): IChannel<TMessage> {
    const messageQueue: TMessage[] = [];
    const resolveQueue: ((message: TMessage) => void)[] = [];

    function put(message: TMessage): void {
        if (resolveQueue.length) {
            const nextResolve = resolveQueue.shift();
            nextResolve(message);
        } else {
            messageQueue.push(message);
        }
    }

    function take(): Promise<TMessage> {
        if (messageQueue.length) {
            return Promise.resolve(messageQueue.shift());
        } else {
            return new Promise((resolve: (message: TMessage) => void) => resolveQueue.push(resolve));
        }
    }

    return {
        take,
        put
    };
}

И пример использования, аналогичный сокращению-сага * takeEvery construction

// example-socket-action-binding.ts
import { put } from 'redux-saga/effects';
import {
    createChannel,
    takeEvery as takeEveryChannelMessage
} from './redux-saga/channels';

export function* socketBindActions(
    socket: SocketIOClient.Socket
) {
    const socketChannel = createSocketChannel(socket);
    yield* takeEveryChannelMessage(socketChannel, function* (action: IAction) {
        yield put(action);
    });
}

function createSocketChannel(socket: SocketIOClient.Socket) {
    const socketChannel = createChannel<IAction>();
    socket.on('action', (action: IAction) => socketChannel.put(action));
    return socketChannel;
}

Ответ 5

У меня была такая же проблема и с использованием PouchDB, и я нашел ответы чрезвычайно полезными и интересными. Однако есть много способов сделать то же самое в PouchDB, и я немного подкопался и нашел другой подход, который, возможно, проще рассуждать.

Если вы не подключаете слушателей к запросу db.change, тогда он возвращает данные об изменении непосредственно вызывающему, а добавление continuous: true к опции приведет к выпуску longpoll и не будет возвращаться до тех пор, пока не произойдет какое-либо изменение. Таким образом, тот же результат может быть достигнут с помощью следующих

export function * monitorDbChanges() {
  var info = yield call([db, db.info]); // get reference to last change 
  let lastSeq = info.update_seq;

  while(true){
    try{
      var changes = yield call([db, db.changes], { since: lastSeq, continuous: true, include_docs: true, heartbeat: 20000 });
      if (changes){
        for(let i = 0; i < changes.results.length; i++){
          yield put({type: 'CHANGED_DOC', doc: changes.results[i].doc});
        }
        lastSeq = changes.last_seq;
      }
    }catch (error){
      yield put({type: 'monitor-changes-error', err: error})
    }
  }
}

Есть одна вещь, которую я не добрался до сути. Если я заменил цикл for на change.results.forEach((change)=>{...}), тогда я получу недопустимую синтаксическую ошибку на yield. Я предполагаю, что это связано с некоторым столкновением в использовании итераторов.