Realtime communication with Socket.IO

Author: Łukasz Malinowski

Problem context

We got to implement some sort of realtime communication in our app, it could be for some kind of multiplayer game or chat or anything that relies on achieving bi-directional connection. With such a problem we've got couple of solutions, but in this article we will focus on trusted and stablished library called Socket.IO

What is Websocket and SocketIO

Websocket is a web communication protocol that allows for the transfer of data from the client side to the server side and vice versa. Socket.IO primarily uses the WebSocket protocol with polling as a fallback option, while providing the same interface. Although it can be used simply as a wrapper for WebSockets, it provides many more features, including broadcasting to multiple sockets, storing data associated with each client, and asynchronous I/O. It consists of two parts: a client-side library that runs in the browser, and a server-side library for Node.js. Both components have a nearly identical API.

Pros and cons of using SocketIO

pros:

multiplexing support - Socket.IO allows to create several Namespaces
room support - within each Namespace, we can define arbitrary channels, called Rooms
handles various support level and inconsistencies from the browser
deals with firewalls and proxies
mature and stable

cons:

initial connection is longer compared to WebSockets, this is due to its first establishing a connection using long polling and xhr-polling, and then upgrading to WebSockets if available.
pure Websocket is faster than SocketIO

Basic implementation of socketIO

Backend:

import http from "http";
import { Server as SocketServer } from "socket.io";
import express from "express";

const app = express();
const httpServer = http.createServer(app);

// for example cors settings, pingInterval ...
const socketIoOptions = {
  cors: { origin: "http://localhost:3000", credentials: true },
};
const socketIo: ServerSocket = new SocketServer(httpServer, socketIoOptions);

socketIo.on("connection", (socket) => {
  socket.emit("messageFromServer", "Just a text message");

  socket.on("messageFromClient", (data) => {
    // handle message with data from client
  });
});

// start server
httpServer.listen(8000);

Frontend:

import socketIOClient from "socket.io-client";

const socket = socketIOClient("http://localhost:8000");

socket.on("connect", () => {
  // socket is connected
});

socket.on("disconnect", () => {
  // socket is disconnected
});

socket.on("messageFromServer", () => {
  // received message from server
});

socket.emit("messageFromClient", () => {
  // sending message to server
});

Example implementation with tests

AddUserActivityAction class is handling POST request on path /api/user/activity

export class AddUserActivityAction implements HttpAction {
  constructor(
    private readonly dependencies: AddUserActivityActionDependencies
  ) {}

  async invoke({ body, params }: Request, res: Response) {
    const { message } = body;
    const { userId } = params;

    const { result: usersId } = await this.dependencies.queryBus.execute(
      new GetUserByIdQuery({ userId })
    );

    // we would rather delegate this logic to command bus, but let's keep it simple
    const userRoomName = getSocketUserRoomName(userId);
    this.dependencies.socketIo
      .to(userRoomName)
      .emit(UserActivityEvents.ACTIVITY_CREATED, { message });

    res.json(Result.ok().toJSON());
  }
}

Unit tests

import { mockRequest, mockResponse } from "../../../../../tests/mocks";
import { GET_USER_BY_ID_QUERY_TYPE } from "../../queries";
import { AddUserActivityAction } from "../add-user-activity.action";

describe("USER/ACTION Add user activity to specific user", () => {
  const queryBusMock = jest.fn();
  const socketIoMock = {
    to: jest.fn().mockReturnThis(),
    emit: jest.fn(),
  };

  let action: AddUserActivityAction;

  beforeEach(() => {
    action = new AddUserActivityAction({
      socketIo: socketIoMock as any,
      queryBus: {
        execute: queryBusMock,
      } as any,
    });
  });

  it("should add user activity", async () => {
    const res = mockResponse();
    const req = mockRequest({
      params: {
        userId: "userId",
      },
      body: {
        message: "Text message",
      },
    });

    const user = {
      _id: "userId",
    };

    queryBusMock.mockResolvedValueOnce({
      result: user,
    });

    await action.invoke(req, res);

    expect(queryBusMock).toBeCalledWith(
      expect.objectContaining({
        payload: { id: req.params.userId },
        type: GET_USER_BY_ID_QUERY_TYPE,
      })
    );

    expect(socketIoMock.to).toBeCalledWith(`user/${user._id}`);
    expect(socketIoMock.emit).toBeCalledWith({ message: req.body.message });

    expect(res.getContext().body).toEqual({
      status: "success",
    });
  });
});

Integration tests

First of all let's create helper function, it's job is to return stablished connection with socket client using access_token as authentication.

import { io, type Socket as SocketIOClient } from "socket.io-client";

export const getSocketClient = async (
  accessToken: string
): Promise<SocketIOClient> => {
  return await new Promise<SocketIOClient>((resolve) => {
    const socket = io(`ws://localhost:8000`, {
      auth: {
        access_token: accessToken,
      },
    });

    socket.on("connect", () => {
      resolve(socket);
    });
  });
};

Another helper for listening on socket event

import type { Socket } from "socket.io-client";

export const listenOnSocket = (eventName: string, socket: Socket) => {
  const listener = jest.fn();
  socket.on(eventName, listener);

  return listener;
};

Actual test

// great module for awaiting instead of using somekind of setTimeout
import waitForExpect from "wait-for-expect";
...

describe("[Integration] Add user activity", () => {
  it("should add user activity", async () => {
    // creates users and log them in
    const { 0: user1, 1: user2 } = await createUsers(2);

    const res = mockResponse();
    const req = mockRequest({
      params: {
        userId: user2.userId,
      },
      body: {
        message: "Text message",
      },
    });

    // listen on socket event
    const userSocket = await getSocketClient(user2.accessToken);
    const activityCreatedListener = listenOnSocket(
      "activityCreated",
      userSocket
    );

    action = new AddUserActivityAction({
      socketIo: this.container.cradle.socketIo,
      queryBus: this.container.cradle.queryBus,
    });

    await action.invoke(req, res);

    await waitForExpect(() => {
      expect(activityCreatedListener).toBeCalledWith({message: req.body.message});
    });

    expect(result).toBeDefined();
  });
});

E2E tests

it("should successfully get activity created event", async () => {
  const { 0: user1, 1: user2 } = await createUsers(2);

  const userSocket = await getSocketClient(user2.accessToken);

  // listen on socket event
  const activityCreatedListener = listenOnSocket("activityCreated", userSocket);

  await user1.agent.post(`/api/users/activity`).expect(200);

  await waitForExpect(() => {
    expect(activityCreatedListener).toBeCalledWith({
      message: req.body.message,
    });
  });

  // disconnect socket after it's job is done
  userSocket.disconnect();
});