Channels and websockets

create-reconnecting-websocket.js

/**
 * Import dependencies
 */
const { EventEmitter } = require('events');
const ReconnectingWebSocket = require('reconnecting-websocket');

/**
 * Export a function that creates a websocket connection interface
 */
export default function createReconnectingWebsocket(uri) {
  /**
   * Flag to determine if socket has been opened and is ready for sending messages
   */
  let isOpen = false;

  /**
   * Create a message queue for all messages that have been requested sent before the socket opened
   */
  let messageQueue = [];

  /**
   * Create an event emitter
   */
  const emitter = new EventEmitter();

  /**
   * Create error handler
   */
  emitter.on('error', console.log);

  /**
   * Create a new websocket connection instance
   */
  const websocket = new ReconnectingWebSocket(uri);

  /**
   * Handle opening of the websocket
   */
  websocket.onopen = (evt) => {
    isOpen = true;
    messageQueue.forEach((msg) => websocket.send(JSON.stringify(msg)));
  };

  /**
   * Handle closing of the websocket
   */
  websocket.onclose = (evt) => {
    isOpen = false;
  };

  /**
   * Handle websocket messages
   */
  websocket.onmessage = (evt) => {
    /**
     * Try to parse the message if it is JSON
     */
    let parsedMessage;

    try {
      parsedMessage = JSON.parse(evt.data);
    } catch (e) {}

    emitter.emit('message', parsedMessage || evt.data);
  };

  /**
   * Handle errors on the websocket
   */
  websocket.onerror = (evt) => {
    isOpen = false;
  };

  /**
   * Return the interface
   */
  return {
    send(msg) {
      if (isOpen) {
        websocket.send(JSON.stringify(msg))
      } else {
        messageQueue.push(msg);
      }
    },
    close() {
      websocket.close();
      return emitter.removeAllListeners();
    },
    on(chan, listener) {
      return emitter.on(chan, listener);
    },
    removeListener(chan, listener) {
      return emitter.removeListener(chan, listener);
    }
  }
}

create-socket-channel.js

/**
 * Dependencies
 */
import { eventChannel } from 'redux-saga';

/**
 * Abstraction for creating socket channels
 */
function createSocketChannel(socket) {
  return eventChannel(emit => {
    const messageHandler = (data) => emit(data);
    socket.on('message', messageHandler);
    return () => socket.removeListener('message', messageHandler);
  });
}

websocket-saga.js

/**
 * Import dependencies
 */
import { take, call, apply, put, fork } from 'redux-saga/effects';

/**
 * Import helpers
 */
import createReconnectingWebsocket from '../utils/create-reconnecting-websocket';
import createSocketChannel from '../utils/create-socket-channel';

/**
 * Import action types
 */
import {
  SEND_SOCKET_MESSAGE,
} from '../ducks/whatever';

/**
 * Import action creators
 */
import {
  someAction,
  someOtherAction,
} from '../ducks/whatever';

/**
 * Initialization saga
 */
export default function* initializeSocket() {
  const socket = yield call(createWebsocket, 'ws://socket.url');
  const socketChannel = yield call(createSocketChannel, socket);

  yield [
    fork(sendSocketMessages, socket),
    fork(receiveSocketMessages, socketChannel),
  ];
}

/**
 * Receive socket messages
 */
function* receiveSocketMessages(socketChannel) {
  while (true) {
    const message = yield take(socketChannel);

    /**
     * Map websocket messages to actions
     */
    switch (message.type) {
      case 'some-message':
      yield put(someAction(message.payload));
      break;

      case 'some-other-message':
      yield put(someOtherAction(message.payload));
      break;

      default:
      break;
    }
  }
}

/**
 * Send socket messages
 */
function* sendSocketMessages(socket) {
  while (true) {
    const message = yield take(SEND_SOCKET_MESSAGE);
    yield apply(socket, socket.send, [message])
  }
}