Angular中Socket.io在.then()回调内无法正常工作的问题求助

问题根源 #

你的dmService.getCallJoin()方法存在两个核心问题：

1. 每次调用都会创建新的BehaviorSubject：这意味着每次订阅都会得到独立的数据流，之前触发的事件不会被新订阅接收。
2. 重复绑定Socket事件：每次调用getCallJoin()都会执行this.socket.on('userConnected', ...)，导致同一个Socket事件被多次监听，触发时多个Subject同时发值，造成逻辑混乱。

当你在navigator.mediaDevices.getUserMedia().then()回调内调用getCallJoin()时，每次回调执行都会生成新的Subject和重复绑定事件，最终导致订阅只能接收一次事件或无法正常响应。

修复方案 #

重构dmService，将BehaviorSubject改为服务类的成员变量，且仅在服务初始化时绑定一次Socket事件：

修改后的Angular服务代码 #

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

@Injectable({ providedIn: 'root' })
export class DmService {
  private data$ = new BehaviorSubject<any>(null); // 初始化值可根据实际类型调整

  constructor(private socket: Socket) {
    // 仅在服务初始化时绑定一次Socket事件
    this.socket.on('userConnected', (data) => {
      console.log("ZSALMASZOSZ IN");
      this.data$.next(data);
    });
  }

  public joinCall(data: { callId: any, userId: any }) {
    this.socket.emit('joinRoom', data);
  }

  public getCallJoin(): Observable<any> {
    console.log("ZSALMASZOSZ");
    return this.data$.asObservable();
  }
}

组件代码优化（可选但建议） #

为避免内存泄漏，建议在组件销毁时取消订阅：

import { Component, OnDestroy, OnInit } from '@angular/core';
import { Subscription } from 'rxjs';

@Component({
  // 组件元数据
})
export class YourComponent implements OnInit, OnDestroy {
  private callJoinSub!: Subscription;

  ngOnInit(): void {
    const videoholder = <HTMLDivElement>(document.querySelector('#videoholder'));
    const myPeer = new Peer(this.userid, { host: '/', port: 3001 });
    const myVideo = document.createElement('video');
    myVideo.muted = true;

    navigator.mediaDevices
      .getUserMedia({ video: true, audio: true })
      .then((stream) => {
        // 订阅同一个共享数据流
        this.callJoinSub = this.dmService.getCallJoin().subscribe((message: any) => {
          console.log(message);
          console.log(message.userId + ' ' + message.callId);
          console.log(stream);
          if (this.callInfo.callid == message.callId) {
            // 你的业务逻辑
          }
        });
        addVideoStream(myVideo, stream);
      });
  }

  ngOnDestroy(): void {
    // 组件销毁时取消订阅
    this.callJoinSub?.unsubscribe();
  }
}

额外说明 #

• 修改后，无论在getUserMedia回调内还是外部调用getCallJoin()，都会订阅同一个数据流，所有userConnected事件都会被正常接收。
• 如果不需要新订阅者立即获取最后一次事件值的特性，可以将BehaviorSubject替换为普通Subject。

内容的提问来源于stack exchange，提问作者Alms