Gateway (Socket.io)

NestJS 中的 Gateway 是一個帶有 @WebSocketGateway() 裝飾器的 class, 並將 Socket.io 封裝在其中, 我們可以透過 @WebSocketGateway() 提供的 API 來操作 Socket.io 與 Client 端做互動

而在 Guard 這幾篇有實作使用 Guard + Passport 來完成登入以及持續驗證, 並且在這幾篇也實作了 Client 的部分來顯示登入/登出的狀態, 接下來實作用 Socket.io 來做同步通知的功能

Server 端建立 socket.io

安裝所需套件

pnpm add @nestjs/platform-socket.io @nestjs/websockets "@socket.io/redis-adapter" redis socket.io --filter PACKAGE_NAME

Socket.io Adapter

socket-io.adapter.ts
import { INestApplicationContext } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { IoAdapter } from '@nestjs/platform-socket.io';
import { ServerOptions } from 'socket.io';

export class SocketIoAdapter extends IoAdapter {
  constructor(
    private app: INestApplicationContext,
    private configService: ConfigService
  ) {
    super(app);
  }

  createIOServer(port: number, options?: ServerOptions) {
    options.transports = ['websocket'];
    options.cors = {
      credentials: true,
      methods: ['GET', 'POST', 'PUT'],
      origin: this.configService.get<string[]>('general.allowedCredentialDomains')
    };

    const server = super.createIOServer(port, options);
    return server;
  }
}

為了在對 socket.io server 連線時進行額外的設定, 這邊自定義了一個 SocketIoAdapter, 這個 adapter 繼承自 IoAdapter 並且實作了 createIOServer 方法, 這個方法會在 server 啟動時被呼叫, 並且在啟動時, 透過 configService 取得需要的參數設定來進行連線, 或者之後如果需要使用到 redis 等也都可以在這邊做額外設定, 詳細可以參考官方文件的 Adapters

然後在 main.ts 中設定

main.ts
// ...
async function bootstrap() {
  const app = await NestFactory.create<NestExpressApplication>(AppModule);
  // ...

  app.useWebSocketAdapter(new SocketIoAdapter(app, configService));
  await app.listen(port);
}
bootstrap();

Socket.io Gateway

socket-io.gateway.ts
import { UseInterceptors } from '@nestjs/common';
import { OnGatewayConnection, OnGatewayDisconnect, OnGatewayInit, SubscribeMessage, WebSocketGateway, WebSocketServer } from '@nestjs/websockets';
import { Server, Socket } from 'socket.io';
import { RequestLogger } from '@/common/request-logger/request-logger.service';
import { SocketIoLoggerInterceptor } from '@/core/interceptors/socket-logger/socket-logger.interceptor';

@UseInterceptors(SocketIoLoggerInterceptor)
@WebSocketGateway()
export class SocketIoGateway implements OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect {
  constructor(private logger: RequestLogger) {}

  @WebSocketServer() server: Server;

  afterInit() {
    // server: Server
    this.logger.log(`[socket.io] - server initialized`);
  }

  handleConnection(client: Socket) {
    this.logger.log(`[socket.io] - Client connected: ${client.id}`);
  }

  handleDisconnect(client: Socket) {
    this.logger.log(`[socket.io] - Client disconnected: ${client.id}`);
  }

  @SubscribeMessage('joinRoom')
  handleJoinRoom(client: Socket, payload: { roomId: string }) {
    client.join(payload.roomId);
  }

  @SubscribeMessage('leaveRoom')
  handleLeaveRoom(client: Socket, payload: { roomId: string }) {
    const room = payload.roomId;
    this.server.to(room).emit('userSignOut', { userId: payload.roomId });
    client.leave(room);
  }
}

這邊建立了一個 SocketIoGateway, 並且去實作 @nestjs/websockets 提供的 OnGatewayInit, OnGatewayConnection, OnGatewayDisconnect 這三個 interface, 然後實作 @nestjs/websockets 所提供的三個 Lifecycle hooks 方法, 分別是 afterInit, handleConnection, handleDisconnect, 這三個方法會在 server 初始化, client 連線, 以及 client 離線時被呼叫

然後這邊也實作了兩個 @SubscribeMessage 方法, 分別是 handleJoinRoom 與 handleLeaveRoom, 這兩個方法會在 client 端呼叫 socket.emit('joinRoom', { roomId: 'ROOM_ID' }) 與 socket.emit('leaveRoom', { roomId: 'ROOM_ID' }) 時被呼叫, 這邊的 client.join 與 client.leave 是 socket.io 提供的方法, 用來讓 client 加入或離開指定的 room, 而 this.server.to(room).emit 則是用來發送訊息給指定的 room 中的所有 client

note

有關 Socket.io 使用的部分, 可以參考官方文件

在使用 SocketIoGateway 的時候, 我們也使用了一個 @UseInterceptors(SocketIoLoggerInterceptor) 的 Interceptor, 這個 Interceptor 會去攔截我們定義的每個 WebSocket 事件 (ex: @SubscribeMessage('joinRoom') )並輸出統一格式的相關 log, Interceptor 的部分可以參考這篇

而原本所提供的 afterInit, handleConnection, handleDisconnect 三個方法我們可以直接透過注入的 RequestLogger 來輸出相關的 log

在 app.module.ts 中的 providers 注入 SocketIoGateway

app.module.ts
// ...
@Module({
  imports: [
    // ...
  ],
  controllers: [AppController],
  providers: [
    // ...
    RequestLogger,
    SocketIoGateway
  ]
})
export class AppModule {}

這樣我們就完成了 server 端的 socket.io 的建立, 接下來實作在 client 端的使用

Client 端建立 socket.io

安裝所需套件

pnpm add socket.io-client --filter PACKAGE_NAME

info

這邊最終會有兩個 web 端來 Demo, 雖然狀態的部分使用不同的方式(參考這裡), 但 socket 的實作上大同小異, 所以只針對其中一個 web 來描述設定

Socket.io connection

vite.config.ts
import path from 'node:path';
import react from '@vitejs/plugin-react';
import { defineConfig, loadEnv } from 'vite';
import svgr from 'vite-plugin-svgr';

export default ({ mode }: { mode: string }) => {
  const env = loadEnv(mode, process.cwd(), '');

  return defineConfig({
    resolve: {
      alias: {
        '@': path.resolve(__dirname, './src')
      }
    },
    plugins: [react(), svgr()],
    server: {
      port: parseInt(env.VITE_PORT),
      host: true,
      proxy: {
        '/socket.io': {
          target: env.VITE_CAS_URL,
          changeOrigin: true,
          ws: true
        }
      }
    }
  });
};

socket.ts
import { io } from 'socket.io-client';

import { casUrl } from '@/config.ts';

// "undefined" means the URL will be computed from the `window.location` object
const URL = casUrl || undefined;

export const socket = io(URL, {
  transports: ['websocket'],
  withCredentials: true
});

因為 demo 的 web 有透過 nginx 去做 reverse proxy 並在 hosts 設定 domain 的攔截, 所以在開發的時候我們可以透過 vite.config.ts 來去設定 proxy, 這樣在開發時就可以直接透過 https://web.YOUR_DOMAIN/socket.io 來連接到 server 端的 socket.io (server 端也需要設定對應的 cors)

useSocket Hook

useSocket.ts
import { useEffect } from 'react';

import { socket } from '@/socket.ts';
import { UserType } from '@/store/slice/user.slice.ts';
import { removeAccessTokenCookie } from '@/utils/token.utility.ts';

const useSocket = (currentUser: UserType | undefined) => {
  useEffect(() => {
    if (!socket.connected) {
      socket.connect();
    }

    return () => {
      if (socket.connected) {
        socket.disconnect();
      }
    };
  }, []);

  useEffect(() => {
    function onUserSignOut({ userId }: { userId: string | undefined }) {
      if (currentUser && userId && currentUser._id === userId) {
        removeAccessTokenCookie();
        window.location.reload();
      }
    }

    socket.on('userSignOut', onUserSignOut);

    return () => {
      socket.off('userSignOut', onUserSignOut);
    };
  }, [currentUser]);

  return {
    emitSignOut: (roomId: string) => {
      socket.emit('leaveRoom', { roomId });
    }
  };
};

export default useSocket;

這邊我們使用 hook 的方式來設定我們的 socket 連線以及要監聽的事件, 透過在 component mount 時去連接 socket, 並且在 component unmount 時去斷開連線, 確保 socket connect 僅在需要時建立, 避免重複建立, 並確保資源釋放正確避免資源浪費

這邊另外寫了一個 effect 來監聽 userSignOut 事件, 當 server 端發送 userSignOut 事件時, 會觸發 onUserSignOut 這個 function, 這個 function 會去判斷是否是自己被通知登出, 如果是的話就會移除 token 並且重新整理頁面來達到登出的效果

最後透過 emitSignOut 這個 function 來發送 leaveRoom 事件, 這個事件會發送通知給 server 端, 讓 server 端去處理 socket client 離開 room 的動作

透過 Socket 加入 Room

user.slice.ts
import { StateCreator } from 'zustand';
import { devtools } from 'zustand/middleware';
import { immer } from 'zustand/middleware/immer';
import { socket } from '@/socket.ts';
import { useMainStore } from '@/store';
import { instance } from '@/utils/request.utility.ts';

// ...

export const currentUserSlice: StateCreator<UserSlice, [], [['zustand/devtools', never], ['zustand/immer', never]]> = devtools(
  immer(set => ({
    user: {
      currentUser: undefined,
      fetchProfile: async () => {
        try {
          const { data } = await instance({
            url: `/auth/profile`,
            method: 'GET'
          });
          set(state => {
            state.user.currentUser = data;
            if (data._id) {
              socket.emit('joinRoom', { roomId: data._id });
            }
          });
        } catch (error) {
          console.error(error);
        }
        useMainStore.getState().fetchStatus.onFetchStatusChange('isProfileFetching', false);
      },
      onCurrentUserChange: value => {
        set(state => {
          state.user.currentUser = value;
        });
      }
    }
  }))
);

在 web1 的 user.slcie.ts 中, 當我們登入成功並取得 user 資料後, 我們可以透過 socket 來發送 joinRoom 事件給 server, server 端接收到後就知道這個 socket 的 client 已經登入成功並取得 user 資料, 因此將這個 socket client 加入到這個 userId 的 room 之中

同樣的, 當進入到 web2 並成功取得 user 資料後, 也會透過 socket 來發送 joinRoom 事件給 server, server 端同樣也會將這個 socket client 加入到這個 userId 的 room 之中 (不同的頁面所建立的 socket client 都是獨立的), 這樣子兩個 web 就都在同一個 room 了

使用 useSocket 來登出

global.provider.tsx
// ...
import useSocket from '@/hooks/use-socket.tsx';

export const GlobalContext = createContext<{}>();
// ...

export const GlobalProvider: FC<{
  children: React.ReactNode;
}> = ({ children }) => {
  const { fetchStatus, currentUser } = useAuth();
  const { emitSignOut } = useSocket(currentUser);

  return (
    <GlobalContext.Provider
      value={{
        currentUser,
        emitSignOut,
        fetchStatus
      }}
    >
      {children}
    </GlobalContext.Provider>
  );
};

在 GlobalProvider 中使用 useSocket 來取得 emitSignOut 這個 function 並傳遞下去, 這樣我們就可以在任何 component 中使用 Context 來取得 emitSignOut 這個 function 來發送 leaveRoom 事件

const { currentUser, emitSignOut } = useContext(GlobalContext);
return (
  <div>
    <Button className="text-error" text={'Sign Out'} onClick={() => emitSignOut(currentUser._id)} />
  </div>
);

當任一個 web 按下登出按鈕發送 leaveRoom 事件, server 端接收到事件後, 就會發送 userSignOut 事件給在這個 room 中的所有 client, 最後 client 透過上面剛剛所說的 useEffect 中所監聽的 userSignOut 事件, 來判斷是否是自己需要被登出, 而因為這些 client 都是屬於這個 userId (同一個使用者登入中), 這樣就可以達到同步登出的效果

Server 端建立 socket.io​

Socket.io Adapter​

Socket.io Gateway​

在 app.module.ts 中的 providers 注入 SocketIoGateway​

Client 端建立 socket.io​

Socket.io connection​

useSocket Hook​

透過 Socket 加入 Room​

使用 useSocket 來登出​