Call redux action within redux-saga inside websocket callback (stomp + sockjs)

Question

I am using redux and redux-saga in my project. Right now using WebSocket I have a problem calling a FETCH_SUCCESS redux action inside a callback of socket response. I tried making the callback a generator as well but didn't work as well.

function* websocketSaga() {
  const socket = new SockJS(`${CONFIG.API_URL}/ws`);
  const stomp = Stomp.over(socket);
  const token = yield select(selectToken);
  stomp.connect(
    {
      Authorization: `Bearer ${token}`,
    },
    frame => {
      stomp.subscribe('/queue/data', message => {
        const response = JSON.parse(message.body);
        console.log(response); // here is the proper response, it works
        put({
          type: FETCH_SUCCESS, // here the FETCH_SUCCESS action is not called
          payload: response.dataResponse,
        });
      });
      ...
      ....
    }
  );
}

Or maybe this WebSocket should be implemented in a completely different way in redux-saga?

Answer 1

I have the same stack over the years and only recently I faced websockets over Stomp client. None of the above solutions doesn't work for me both technically and mentally

Reasons:

• I don't like channels with Stomp because the only way to manipulate connections in more surgical way you have to use global state object (for me - it's redux). It doesn't seems right even if you storing only random generated IDS (with unsubscribe function it will be... read more here about store serialization
• the way with container another pain in the ... (you know where). Again redux and a lot of under-the-hood functionality used without any reason
• another way with promises: again without storing helpful connection info and some DI by using promises inside generators. This narrows down the implementation choice

So:

• I need to have connection info (I decided to use state but not in: redux, component state. Singleton state). Stomp doesn't force you to place ID but I do because I want to manage connections by myself
• I need one entry point without: promises, iterators and a lot of things that will be pain for future-me. One place to "rule them all" (as I want) - activate: login - deactivate: logout - subscribe: componentDidMount - unsubscribe: componentWillUnmount
• DI by request in one place (passing store.dispatch to constructor only if need it) // main topic of the question

And I wrote this implementation that perfectly works for me:

import SockJS from 'sockjs-client';
import {
    Client,
    IMessage,
    messageCallbackType,
    StompHeaders,
} from '@stomp/stompjs';

import { Action, Dispatch } from 'redux';

type ConnectionId = string;

interface IServiceConfig {
    url: string;
    dispatch?: Dispatch;
}

export default class Stomp {
    serviceConfig: IServiceConfig = {
        dispatch: null,
        url: null,
    };

    ids: ConnectionId[] = [];

    stomp: Client;

    constructor(config: IServiceConfig) {
        this.serviceConfig = { ...config };

        this.stomp = new Client();

        this.stomp.webSocketFactory = () => {
            return (new SockJS(config.url));
        };
    }

    alreadyInQueue = (id: ConnectionId): boolean => {
        return Boolean(this.ids.find(_id => id === _id));
    };

    subscribeByDispatchAction = (
        destination: string,
        callback: (message: IMessage) => Action,
        headers: StompHeaders & {
            id: ConnectionId;
        },
    ): void => {
        const alreadyInQueue = this.alreadyInQueue(headers.id);

        if (!alreadyInQueue) {
            this.stomp.subscribe(
                destination,
                (message) => {
                    this.serviceConfig.dispatch(callback(message));
                },
                headers,
            );

            this.ids.push(headers.id);

            return;
        }

        console.warn(`Already in queue #${headers.id}`);
    };

    subscribe = (
        destination: string,
        callback: messageCallbackType,
        headers: StompHeaders & {
            id: ConnectionId;
        },
    ): void => {
        const alreadyInQueue = this.alreadyInQueue(headers.id);

        if (!alreadyInQueue) {
            this.stomp.subscribe(
                destination,
                (message) => callback(message),
                headers,
            );

            this.ids.push(headers.id);

            this.logState('subscribe');

            return;
        }

        console.warn(`Failed to subscribe over Socks by #${headers.id}`);
    };

    unsubscribe = (id: ConnectionId, headers?: StompHeaders): void => {
        this.stomp.unsubscribe(id, headers);

        this.ids.splice(this.ids.indexOf(id), 1);
    };

    activate = (): void => {
        this.stomp.activate();
    };

    deactivate = (): void => {
        if (this.ids.length === 0) {
            this.stomp.deactivate();

            return;
        }

        for (let i = 0; i < this.ids.length; i++) {
            this.unsubscribe(this.ids[i]);
        }

        /**
         * it seems like it's overkil but
         * for me it works only if i do all
         * the things as you see below
         * - stomp deactivation
         * - closing webSockets manually by using native constant // sockjs-client
         * - closing webSockets instance by using returned value fron factory
         */
        this.stomp.deactivate();

        this.stomp.webSocket.close(
            this.stomp.webSocket.CLOSED,
        );

        this.stomp.webSocketFactory().close();
    };

    getAllIds = (): readonly ConnectionId[] => {
        return this.ids;
    };

    // debug method
    logState = (method: string): void => {
        /* eslint-disable */
        console.group(`Stomp.${method}`);
        console.log('this', this);
        console.log('this.ids', this.getAllIds());
        console.log('this.stomp', this.stomp);
        console.groupEnd();
        /* eslint-enable */
    };
}

My configuration file

import { store } from '~/index';

import Stomp from '~/modules/_Core/services/Stomp';
import appConfig from '~/modules/Common/services/appConfig';

export const StompService = new Stomp({
    dispatch: store?.dispatch,
    url: `${appConfig.apiV1}/websocket`,
});

I hope that it will help someone

Answer 2

I will give you another way of managing this: create a component connected to redux where you will handle the WS subscription. This component will not render anything to the UI but will be useful for handling redux store interactions.

The main idea is, don't put everything into redux-saga, try and split it into multiple parts to make it easier to maintain.

const socket = new SockJS(`${CONFIG.API_URL}/ws`);

function WSConnection(props) {
  const {token, fetchDone} = props;
  const [stomp, setStomp] = React.useState();

  const onMessage = React.useCallback(message => {
    const response = JSON.parse(message.body);
    fetchDone(response.dataResponse);
  }, [fetchDone]);

  const onConnect = React.useCallback(frame => {
    const subscription = stomp.subscribe('/queue/data', onMessage);
    // cleanup subscription
    return () => subscription.unsubscribe();
  }, [stomp, onMessage]);

  const onError = React.useCallback(error => {
    // some error happened, handle it here
  }, []);

  React.useEffect(() => {
    const header = {Authorization: `Bearer ${token}`};
    stomp.connect(header, onConnect, onError);
    // cleanup function
    return () => stomp.disconnect();
  }, [stomp])

  React.useEffect(() => {
    setStomp(Stomp.over(socket));
  }, []);

  return null;
}

const mapStateToProps = state => ({
... // whatever you need from redux store
});

const mapDispatchToProps = dispatch => ({
... // whatever actions you need to dispatch
});

export default connect(mapStateToProps, mapDispatchToProps)(WSConnection);

You can also take it a step further and extract the stomp logic into another file and reuse it wherever you will need it.

It's not wrong to put everything into redux-saga but it's a nice alternative to handle WS connections inside components connected to redux (and easier to understand to people who are not completely familiar with redux-saga and channels etc).

Answer 3

Promise should be the perfect fit. Just wrap the callback related code in a promise and resolve it in the callback function. After that use the yield to get the data from the promise. I have modified your code with the Promise below.

function* websocketSaga() {
  const socket = new SockJS(`${CONFIG.API_URL}/ws`);
  const stomp = Stomp.over(socket);
  const token = yield select(selectToken);

  const p = new Promise((resolve, reject) => {
    stomp.connect(
      {
        Authorization: `Bearer ${token}`,
      },
      frame => {
        stomp.subscribe('/queue/data', message => {
          const response = JSON.parse(message.body);
          console.log(response); // here is the proper response, it works
          resolve(response); // here resolve the promise, or reject if any error
        });
        ...
        ....
      }
    );
  });

  try {
    const response = yield p;  // here you will get the resolved data
    yield put({
      type: FETCH_SUCCESS, // here the FETCH_SUCCESS action is not called
      payload: response.dataResponse,
    });
  } catch (ex) {
    // handle error here, with rejected value
  }

}

Answer 4

You won't be able to use yield put inside a callback function. Stompjs knows nothing about sagas, so it doesn't know what it's supposed to do when given a generator function.

The simplest approach, though not necessarily the best, is to go directly to the redux store in the callback, and dispatch the action without involving redux-saga. For example:

import store from 'wherever you setup your store'

// ...
stomp.subscribe('/queue/data', message => {
  const response = JSON.parse(message.body);
  store.dispatch({
    type: FETCH_SUCCESS,
    payload: response.dataResponse,
  });
});

If you'd like to use a more redux-saga-y approach, I would recommend wrapping the subscription in an event channel. Event channels take a callback-based API and turn it into something that you can interact with using redux-saga's effects such as take

Here's how you might create the event channel:

import { eventChannel } from 'redux-saga';

function createChannel(token) {
  return eventChannel(emitter => {
    const socket = new SockJS(`${CONFIG.API_URL}/ws`);
    const stomp = Stomp.over(socket);
    stomp.connect(
      {
        Authorization: `Bearer ${token}`,
      },
      frame => {
        stomp.subscribe('/queue/data', message => {
          const response = JSON.parse(message.body);
          emitter(response); // This is the value which will be made available to your saga
        });
      }
    );

    // Returning a cleanup function, to be called if the saga completes or is cancelled
    return () => stomp.disconnect();
  });
}

And then you'd use it like this:

function* websocketSaga() {
  const token = yield select(selectToken);
  const channel = createChannel(token);
  while (true) {
    const response = yield take(channel);
    yield put({
      type: FETCH_SUCCESS,
      payload: response.dataResponse,
    });
  }
}