Redux_Saga_Event_Channel_for_Biderctional_Websockets.js

// Gist for https://medium.com/@ebakhtarov/bidirectional-websockets-with-redux-saga

import { eventChannel } from "redux-saga";
import { all, call, put, take, select, race } from "redux-saga/effects";

import { actionTypes } from "./constants";

function watchMessages(socket) {
  return eventChannel(emit => {
    /* eslint-disable no-param-reassign */
    socket.onmessage = event => {
      const msg = JSON.parse(event.data);
      if (msg.type === "online"){
        emit(msg);
      }
      // emit(END) // this causes the channel to close
    };
    /* eslint-enable no-param-reassign */

    return () => {
      socket.close();
      // .then(() => console.log('closing socket... logout'));
    };
  });
}

function* backgroundTask(socketChannel) {
  while (true) {
    const payload = yield take(socketChannel);
    if (payload.type === "online") {
      yield put({
        type: "UPDATE_ONLINE",
        on: payload.on
      });
    }
  }
}

/*  User Created Message (e.g. dispatch({ type: 'EXE_TASK', taskid: 5 })) sent to ws server  */
function* executeTask(socketChannel) {
  while (true) {
    const data = yield take("EXE_TASK");
    socketChannel.send(JSON.stringify({ type: "setTask", task: data.data.taskid }));
  }
}

export default function* establishWSConSaga() {
  while (true) {
    yield take(actionTypes.START_WEBSOCKET);
    const socket  = new WebSocket('wss://ws.xyz.com/items/abc') // your websocket server

    socket.onopen = () => {
      // console.log('ws connected');
      socket.send(JSON.stringify({ firstMessageToServer: 1 }));
    };

    const socketChannel = yield call(watchMessages, socket);
    const { cancel } = yield race({
      task: all([
        call(backgroundTask, socketChannel),
        call(executeTask, socket)
      ]),
      cancel: take("STOP_WEBSOCKET")
    });
    if (cancel) {
      // console.log('channel cancelled');
      socketChannel.close();
    }
  }
}