在 Angular 应用程序中将 Socket.IO 与 NgRx 集成

本文分享如何在 Angular 中集成 Socket.IO 和 NgRx 实现实时通信。

1. 项目设置：

• 确保有一个现有的 Angular 项目，或使用 Angular CLI 创建一个新项目：

ng new socket-io-angular-ngrx

• 安装依赖项：

cd socket-io-angular-ngrx npm install socket.io-client @ngrx/store @ngrx/effects

2. 服务器端设置（假设使用 Node.js）：

• 创建服务器文件（如 server.js）并设置 Socket.IO：

const express = require('express');
const http = require('http');
const { Server } = require('socket.io');

const app = express();
const server = http.createServer(app);
const io = new Server(server);

// Routes and logic for your application

io.on('connection', (socket) => {
    console.log('A user connected');

    // Handle events from the client
    socket.on('clientEvent', (data) => {
        console.log('Received data from client:', data);

        // Emit events back to the client(s) or other sockets as needed
        io.emit('serverEvent', { message: 'Hello from the server!' });
    });

    // Handle disconnections
    socket.on('disconnect', () => {
        console.log('A user disconnected');
    });
});

server.listen(3000, () => {
    console.log('Server listening on port 3000');
});

• 用自定义事件名称替换 clientEvent 和 serverEvent。

• 执行您的应用程序逻辑，以处理接收到的事件并将相关数据发送回客户端。

3. Socket.IO 服务（Angular 客户端）：

创建一个服务（如 socket.service.ts）来处理 Socket.IO 交互：

import { Injectable } from '@angular/core';
import { io, Socket } from 'socket.io-client';
import { Observable, from } from 'rxjs';

@Injectable({
    providedIn: 'root'
})
export class SocketService {
    private socket: Socket;

    constructor() {
        this.socket = io('http://localhost:3000'); // Replace with your server URL
    }

    onEvent(eventName: string): Observable<any> {
        return from(new Observable(observer => {
            this.socket.on(eventName, (data) => observer.next(data));
        }));
    }

    emitEvent(eventName: string, data: any): void {
        this.socket.emit(eventName, data);
    }
}

该服务提供了连接、监听事件和向服务器发送事件的方法。

4. NgRx 动作和减速器：

• 创建表示事件的动作（如 socket-io.actions.ts）：

import { createAction, props } from '@ngrx/store';

export const connect = createAction('[Socket.IO] Connect');
export const disconnect = createAction('[Socket.IO] Disconnect');
export const receiveData = createAction('[Socket.IO] Receive Data', props<{ data: any }>());

• 创建一个减速器（例如socket-io.reducer.ts）来管理状态：

该减速器跟踪连接状态和接收到的数据。

import { createReducer, on } from '@ngrx/store';

export interface SocketIoState {
    isConnected: boolean;
    data: any;
}

const initialState: SocketIoState = {
    isConnected: false,
    data: null,
};

const socketIoReducer = createReducer(
    initialState,
    on(connect, (state) => ({ ...state, isConnected: true })),
    on(disconnect, (state) => ({ ...state, isConnected: false })),
    on(receiveData, (state, { data }) => ({ ...state, data }))
);

export default socketIoReducer;

5. NgRx Effects(NgRx 效果)

import { Injectable } from '@angular/core';
import { Actions, createEffect, ofType } from '@ngrx/effects';
import { catchError, map, mergeMap, tap } from 'rxjs/operators';
import { of } from 'rxjs';
import { SocketService } from './socket.service';
import * as SocketIoActions from './socket-io.actions';

@Injectable()
export class SocketIoEffects {
  connect$ = createEffect(
    () => this.actions$.pipe(
      ofType(SocketIoActions.connect),
      tap(() => console.log('Connecting to Socket.IO server...')),
      mergeMap(() => this.socketService.onEvent('connect').pipe(
        map(() => SocketIoActions.connect()),
        catchError((error) => of(SocketIoActions.disconnect({ error })))
      ))
    )
  );

  disconnect$ = createEffect(
    () => this.actions$.pipe(
      ofType(SocketIoActions.disconnect),
      tap(() => console.log('Disconnecting from Socket.IO server...')),
      mergeMap(() => of(SocketIoActions.disconnect()))
    )
  );

  receiveData$ = createEffect(
    () => this.actions$.pipe(
      ofType(SocketIoActions.connect),
      mergeMap(() =>
        this.socketService.onEvent('serverEvent').pipe( // Replace with your event name
          map((data) => SocketIoActions.receiveData({ data }))
        )
      )
    )
  );

  constructor(private actions$: Actions, private socketService: SocketService) {}
}

• 这些效果处理与 Socket.IO 交互相关的副作用。
• connect$：在派发连接操作时启动连接。
• 将消息记录到控制台。
• 使用 mergeMap 订阅来自 Socket.IO 服务的连接事件。
• 将接收到的数据映射到connect操作。
• 捕捉错误并发送带错误的disconnect操作。
• disconnect$：在发出断开连接操作时断开连接。
• 将消息记录到控制台。
• 简单地发送一个disconnect动作，没有进一步的逻辑。
• receiveData$：连接时监听服务器数据。
• 因 mergeMap 而触发连接操作。
• 订阅来自 Socket.IO 服务的 serverEvent（用实际事件名称代替）。
• 将接收到的数据映射到receiveData操作。

6. 与组件集成：

在组件中注入 SocketService 和 Store：

import { Component, OnInit } from '@angular/core';
import { SocketService } from './socket.service';
import { Store } from '@ngrx/store';
import { selectIsConnected, selectData } from './socket-io.selectors';

@Component({
    selector: 'app-my-component',
    templateUrl: './my-component.component.html',
    styleUrls: ['./my-component.component.css']
})
export class MyComponent implements OnInit {
    isConnected$ = this.store.select(selectIsConnected);
    data$ = this.store.select(selectData);

    constructor(
        private socketService: SocketService,
        private store: Store<any>
    ) {}

    ngOnInit() {
        // Dispatch the connect action to initiate the connection
        this.store.dispatch(SocketIoActions.connect());
    }

    // ... component logic
}

• 使用异步管道从商店订阅状态和数据，并将其显示在模板中。
• 根据需要使用 SocketService 向服务器发送事件。

7. 在应用程序模块中注册效果：

从 @ngrx/effects 导入 EffectsModule，并在 AppModule 中注册 SocketIoEffects：

import { BrowserModule } from '@angular/platform-browser';
import { NgModule } from '@angular/core';

import { AppComponent } from './app.component';
import { SocketService } from './socket.service';
import { EffectsModule } from '@ngrx/effects';
import { SocketIoEffects } from './socket-io.effects';

@NgModule({
      declarations: [AppComponent],
      imports: [
          BrowserModule,
          // ... other imports
          EffectsModule.forRoot([SocketIoEffects])
      ],
      providers: [SocketService],
      bootstrap: [AppComponent]
  })
  export class AppModule { }

8. 在App模块中注册Reducers：

  // ... other imports
  import { StoreModule } from '@ngrx/store';
  import { socketIoReducer } from './socket-io.reducer';

  @NgModule({
      declarations: [AppComponent],
      imports: [
          BrowserModule,
          // ... other imports
          StoreModule.forRoot({ socketIo: socketIoReducer }),
          EffectsModule.forRoot([SocketIoEffects])
      ],
      providers: [SocketService],
      bootstrap: [AppComponent]
  })
  export class AppModule { }

• 从 @ngrx/store 导入 StoreModule。

• 使用 StoreModule.forRoot 为状态的 socketIo 片注册 socketIoReducer。

• 导入 EffectsModule.forRoot 以在应用程序的根模块中注册 SocketIoEffects。

9. 创建选择器（可选）：

• 如果希望在组件中访问状态的特定部分，可创建选择器（如 socket-io.selectors.ts）：

这些选择器会从 socketIo 状态片中提取 isConnected 和数据。

import { createSelector } from '@ngrx/store';
import { SocketIoState } from './socket-io.reducer';

export const selectIsConnected = createSelector(
    (state: any) => state.socketIo,
    (state: SocketIoState) => state.isConnected
);

export const selectData = createSelector(
    (state: any) => state.socketIo,
    (state: SocketIoState) => state.data
);

10. 组件使用示例：

<p *ngIf="isConnected$ | async">Connected to Socket.IO server!</p>
<p>Received data: {{ data$ | async }}</p>

• 使用 async 管道订阅模板中的 isConnected$ 和 data$ 观察项。

• 根据状态显示消息或数据。

11. 其他注意事项

错误处理：

• 在整个代码中实施适当的错误处理，以便从容应对连接、断开连接或数据接收过程中可能出现的错误。

自定义事件：

• 在客户端和服务器端定义自定义事件以传达特定数据或操作。
• 使用SocketService发出和侦听这些自定义事件。

可扩展性：

• 考虑使用房间或命名空间来有效管理多个连接和事件，特别是对于大型应用程序。

测试：

• 实施单元和集成测试，以确保 Socket.IO 集成按预期工作并有效处理各种场景。

部署：

• 配置服务器以在适当的部署环境中运行，考虑生产与开发以及负载平衡等因素。

文档：

• 彻底记录 Socket.IO 集成，为未来从事该项目的开发人员提供清晰的说明和示例。