stompMiddleware.js

import Stomp from 'stompjs';
import {
  APP_STOMP_URL,
  APP_STOMP_LOGIN,
  APP_STOMP_PASSCODE,
} from '../../constants';
import {
  connect,
  connecting,
  connected,
  messageRecevied,
  disconnect,
  subscribe,
  unsubscribe,
  unsubscribeAll,
} from './stompActions';

const subscriptions = {};
let client = null;

const middleware = store => next => (action) => {
  if (process.env.REACT_APP_STOMP_DISABLED) {
    return next(action);
  }

  const ensureConnection = (callback) => {
    setTimeout(() => {
      if (client.connected) {
        if (callback !== null) {
          callback();
        }
      } else {
        ensureConnection(callback);
      }
    }, 5); // wait 5 miliseconds for the connection...
  };

  const connectFailure = (error) => {
    console.log(`STOMP: ${error}`);
    setTimeout(connectToStomp, 5000);
    console.log('STOMP: Reconecting in 5 seconds');
  };

  const connectSuccess = () => {
    const existingSubscriptions = Object.keys(subscriptions);
    store.dispatch(connected());
    // If we are reconnecting we need to resubscribe
    if (existingSubscriptions.length) {
      ensureConnection(() => {
        existingSubscriptions.forEach((subscription) => {
          subscriptions[subscription] = {
            destination: subscriptions[subscription].destination,
            onmessage: subscriptions[subscription].onmessage,
            connection: client.subscribe(subscriptions[subscription].destination, (message) => {
              store.dispatch(messageRecevied(message));
              store.dispatch(subscriptions[subscription].onmessage(message));
            }),
          };
        });
      });
    }
  };

  const connectToStomp = () => {
    console.log('STOMP: Attempting connection');
    client = Stomp.client(APP_STOMP_URL);
    store.dispatch(connecting());
    client.connect(
      APP_STOMP_LOGIN,
      APP_STOMP_PASSCODE,
      connectSuccess,
      connectFailure,
    );
  };

  switch (action.type) {
    case connect.toString(): {
      connectToStomp();

      break;
    }

    case disconnect.toString(): {
      ensureConnection(() => {
        client.disconnect();
      });

      break;
    }

    case subscribe.toString(): {
      const { subscriptionName, destination, onmessage } = action.payload;

      ensureConnection(() => {
        subscriptions[subscriptionName] = {
          destination,
          onmessage,
          connection: client.subscribe(destination, (message) => {
            store.dispatch(messageRecevied(message));
            store.dispatch(onmessage(message));
          }),
        };
      });

      break;
    }

    case unsubscribe.toString(): {
      const { subscriptionName } = action.payload;
      subscriptions[subscriptionName].connection.unsubscribe();
      delete subscriptions[subscriptionName];

      break;
    }

    case unsubscribeAll.toString(): {
      Object.keys(subscriptions).forEach((subscription) => {
        subscriptions[subscription].connection.unsubscribe();
        delete subscriptions[subscription];
      });

      break;
    }

    default: {
      break;
    }
  }

  return next(action);
};

export default middleware;