关于这个 Angular + ngRx（效果）的建议 - websocket 事件呢？答案

【问题标题】：Advice on this Angular + ngRx (effects) - what about websocket events?关于这个 Angular + ngRx（效果）的建议 - websocket 事件呢？
【发布时间】：2018-08-11 02:28:37
【问题描述】：

所以我正在通过构建以下沙箱来试验 ngrx 和 ngrx/效果：

https://stackblitz.com/edit/ngrx-vanilla

快速介绍：

它在 app/store 中有一个根存储区
它在 app/features 中有两个模块的延迟加载
它在 app/commons 中有单例服务

三页：

行动项目：路由到此页面会触发随机生成三个愚蠢的公司行动项目
用户：一个基本的 master > 详细的 redux 实现与路由器支持
会议：提出我的问题的地方，点击“开始会议”，见证相关的思想交流。

问题和背景：

我了解 redux 中的所有数据更新都是通过操作发生的
“效果”库用于处理异步事件，以便根据第 3 方事件和异步调用调度新操作。
app/common/meeting/service 模仿例如 websocket 或 firebase 实时数据库推送更新的行为。

收到更新后（在 app/store/effects/meeting.effects.ts 中说明），将调度一个新操作。

最后，问题是：让一个共同的服务了解商店是一种干净的做法吗？在哪里注册监听器到 websocket/firebase 实时数据库以便在数据被推送时调度操作的最佳位置？

在这里，我这样做是为了让效果（meeting.effects）对 meetingActions.START_MEETING 动作类型做出反应，并且每当推送数据时，向商店发送更新订单，但这感觉不对，因为我来的一系列原因最多：

很难单独进行单元测试（需要比自身更多的上下文）
在“停止会议”操作的情况下，此方法需要存储订阅（或？）以停止订阅。在我的方法中，无法控制在荒野中创建的 observable。

此类案件通常如何处理？

【问题讨论】：

标签： angular redux ngrx ngrx-store ngrx-effects

【解决方案1】：

假设websocket 发出不同类型的事件，将每个事件映射到 websocket 服务中的不同操作，例如

@Injectable()
class WebsocketService{
    private socket$:Observable<Action>
    getActions():Observable<Action>{
        if(!this.socket$) {
           this.socket$ = Observable.webSocket(url).pipe(
               map(functionToMapActions)
            ,shareReplay(1)
           );
        }
        return this.socket$;
    }
}

functionToMapActions 将 webSocket 事件映射到操作，我建议在末尾添加 shareReplay 运算符，以便我们只读取一次 webSocket。

Observable.webSocket 连接到 webSocket，并在事件到达时发出事件

订阅webService.getActions()会建立webSocket连接

您可以在@Effects初始化see here订阅websocket操作

@Effect()
init$ = this.websocketService.getActions();

这将在您的应用启动（如果在根模块中生效）或您的模块在延迟加载模块中加载时立即发出所有操作；

或者，如果您对有限的一组操作感兴趣，您可以这样做

@Effect()
init$ = this.websocketService.getActions().pipe(filter(filterNeededActionsHere));

你也可以在这样的特定事件之后才开始收听动作

@Effect()
init$ = this.actions$.pipe(
          ofType('Event which marks start')
          ,swichMapTo(this.websocketService.getActions())
     );

像前面的例子一样，你也可以像以前一样过滤掉这里的动作

希望这能回答你的问题

【讨论】：

【解决方案2】：

NgRx v9 稍微改变了语法。对 root-level 效果类使用以下代码：

init$ = createEffect(() => 
  this.actions$.pipe(
    ofType(ROOT_EFFECTS_INIT),
    // websocket events handling logic
    switchMap(() => webSocketEvents$)
  )
);

这只是来自Docs 的稍微改编的示例。

如果你使用特征级效果类，ROOT_EFFECTS_INIT 将不起作用，你需要使用OnRunEffects生命周期钩子：

class UserEffects implements OnRunEffects  {
  ngrxOnRunEffects(resolvedEffects$: Observable<EffectNotification>) {
    // websocket events handling logic
    return webSocketEvents$.pipe(
      exhaustMap(() => resolvedEffects$)
    );
  }
}

更详细的例子在Docs。

两个示例中的webSocketEvents$ 都是Observable，可以使用其中一个 rxjs 函数来构造：

webSocket(socketUrl) // if you want to manage websocket connection via rxjs
fromEvent(socketIoClient) // useful if websocket connection handled by socket.io-client

【讨论】：

【解决方案3】：

使用适配器为实体 Vehicle 发出不同类型事件的 WebSocket 自定义示例

1.创建 Vehicle-Socket-Adapter.ts

import { fromEvent, merge, Observable } from 'rxjs';
import { map } from 'rxjs/operators';
import { SocketIoService } from 'src/app/shared/services/socket-io.service';

export class VehicleSocketAdapter {
  public onVehicleEngineStateChange: Observable<boolean>;
  constructor(vehicleId: string, private socketIoService: SocketIoService) {

   //using merge from rxjs/operators to concat multiple ws events to one Observable 

    this.onVehicleEngineStateChange = merge(
      fromEvent(this.socketIoService.socket, `event.vehicle.${vehicleId}.start`).pipe(
        map(vehicleStatus => true)
      ),
      fromEvent(this.socketIoService.socket,`event.vehicle.${vehicleId}.stop`).pipe(
        map(vehicleStatus => false)
      ),
    )
  }
}

2.稍后将适配器导入您想使用的任何地方，例如 app.component.ts

private subscriptions = new Subscription();
private listenForVehiclesState(vehicles) {
    vehicles.forEach((vehicle) => {
      const vehicleAdapter = new VehicleSocketAdapter(vehicle.id, this.webSocket);
      this.subscriptions.add( 
        vehicleAdapter.onVehicleEngineStateChange.subscribe(vehicleStatus => {
        
       // dispatch action when adapter commands you

        this.store.dispatch(
          VehiclesActions.BeginUpdateVehiclesStatusAction({
            payload: {
              vehicleId: vehicle.id,
              status: vehicleStatus
            }
          })
        );

      }));
    });
  }

当您的视图或组件被销毁时，不要忘记取消订阅所有订阅 this.subscriptions.unsubscribe();

奖励：套接字即服务 sn-p :)

import { Injectable } from '@angular/core';
import * as io from 'socket.io-client';

@Injectable({
  providedIn: 'root'
})
export class SocketIoService {

  public socket: SocketIOClient.Socket = io('/', { path: '/api/livedata', transportOptions: { polling: extraHeaders: {'AuthToken': 'ifAny'} }});

constructor() { }
}

快乐编码:)

【讨论】：