Observable为何在回调完成前返回？MQTT连接场景排查

现有代码 #

组件A #

testMqtt(){
    console.log('testMqtt is called');
    this.serviceB.connectToBroker().subscribe({
      next: (resp) => console.log('response from connection to mqtt: ', resp.connected),
      error: (err) => console.log('error from connection to mqtt: ', err)
    });
  }

服务B #

connectToBroker(){
    let client = mqtt.connect(this.url, {clientId: 'abc123'});
    return of(client.on('connect', (packet) => {
      console.log('testing value of connected: ', JSON.stringify(packet));
      console.log('value of client connected: ', client.connected);
    }))
  };

问题 #

需要把服务B中client.connected的真实值同步到组件A的resp.connected，但目前组件A日志始终显示resp.connected: false，服务B内却能打印出client.connected: true，请问：

1. 为什么会出现这种状态不一致？
2. 服务B该用哪个RxJS操作符，确保等client.on('connect')回调完成后再向组件A返回响应？

原因分析 #

你当前的代码逻辑有个核心问题：服务B用of()创建Observable，这是同步创建操作——它会立刻把传入的内容发射给订阅者，但此时MQTT连接还没建立完成，client.connected自然是false。

client.on('connect')是异步触发的回调，要等真正连接成功才会执行，但of()根本不会等待这个回调，直接就把初始状态的client相关内容抛给了组件A，所以组件A拿到的永远是连接完成前的false状态。

解决方案 #

要把异步的MQTT连接事件转换成能正确等待的Observable，推荐用**Observable.create()或者更简洁的fromEvent**，两者都能确保在连接成功的回调触发后，再把真实的client.connected值发射给组件A。

修改后的服务B代码 #

方案1：使用Observable.create()（灵活可控）

import { Observable } from 'rxjs';

connectToBroker(){
  return Observable.create(observer => {
    const client = mqtt.connect(this.url, {clientId: 'abc123'});
    
    // 监听连接成功事件
    client.on('connect', () => {
      console.log('value of client connected: ', client.connected);
      // 发射包含连接状态的对象
      observer.next({ connected: client.connected });
      observer.complete(); // 单次事件，完成Observable
    });

    // 监听连接错误
    client.on('error', (err) => {
      observer.error(err); // 向订阅者传递错误信息
    });

    // 清理函数：组件取消订阅时关闭MQTT连接
    return () => {
      client.end();
    };
  });
}

方案2：使用fromEvent（简洁高效）

import { fromEvent, map, first } from 'rxjs';

connectToBroker(){
  const client = mqtt.connect(this.url, {clientId: 'abc123'});
  
  // 将connect事件转成Observable，只取第一次触发结果
  return fromEvent(client, 'connect').pipe(
    first(), // 确保只发射一次连接成功事件
    map(() => ({ connected: client.connected })) // 转换成组件需要的格式
  );
}

关键说明 #

• Observable.create()：手动创建Observable，完全控制事件发射时机，同时能处理错误和订阅清理逻辑，适合复杂异步场景。
• fromEvent：RxJS专门用来将Node.js/DOM事件转换成Observable的操作符，配合first()避免重复发射，map()调整输出格式，代码更简洁。

修改后，组件A的订阅会在MQTT连接成功后才收到resp.connected: true的日志，和服务B内的状态完全同步。

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