Node.js Stream模拟双工Socket对接Aedes MQTT首次交互断连问题

在Node.js环境中使用 mqtt-connection 与 aedes mqtt 库开发，基于aedes搭建MQTT服务器，期望不创建TCP Socket，而是在Node.js进程内部通过Stream连接到MQTT服务器，已知两个库均支持接收Duplex双工流作为入参。

• 基于net.createConnection创建双工流的实现可正常运行，代码如下：

const mqttCon = require('mqtt-connection');
const duplex = require('net').createConnection(1883);
const client = mqttCon(duplex, {
  protocolVersion: 3
});

• 自定义Transform流的实现会在首次数据交互后关闭流，运行失败，代码如下：

const mqttCon = require('mqtt-connection');
const duplex = new stream.Transform();
duplex._transform = (chunk, encoding, callback) => {
  duplex.push(chunk);
  console.log(chunk);
  callback();
};

const client = mqttCon(duplex, {
  protocolVersion: 3
});
aedes.handle(duplex);

• 核心诉求：创建行为与TCP Socket一致的通信通道，实现两个逻辑模块在Node.js进程内部的通信。
• aedes常规TCP服务用法参考：

const aedes = require('aedes')(
  {
    concurrency: 500,
    maxClientsIdLength: 100
  }
);
const server = require('net').createServer(aedes.handle);

• 故障细节：客户端向流写入数据后，aedes完成响应的瞬间就会触发duplex.on('close')事件，全程无错误抛出，通信两端均无错误提示，流直接关闭后两端优雅退出，初步推测是某一端检测到流结束事件从而触发了关闭逻辑。

核心错误是对Node.js双工流和TCP Socket的行为差异存在认知偏差：

1. 单个Duplex/Transform流不是全双工通信管道。TCP Socket的全双工本质是两条互相独立的单向传输通道：A端写入的数据仅会被B端读取，B端写入的数据仅会被A端读取，两个通道的状态（包括半关闭、缓冲区）完全独立。
2. 问题实现中，同一个Transform流被同时传给MQTT客户端和aedes服务端，相当于两端共享同一个读写通道：客户端发送的MQTT握手包会被客户端自己先读取，服务端返回的响应包也会被服务端自己读取，数据路由完全错误。
3. 自定义Transform流没有做通道隔离，任意一端触发流结束动作时，会直接关闭整个共享流实例，导致两端同时收到close事件，这就是首次交互后连接直接优雅断开的原因。

用两个独立的PassThrough流交叉连接，构造一对符合TCP Socket行为的进程内全双工流，两个流分别作为客户端侧和服务端的通信载体，不需要引入第三方依赖：

const { PassThrough, Duplex } = require('stream');

/**
 * 创建一对进程内全双工通信流，行为与TCP Socket一致
 * @returns {[Duplex, Duplex]} 依次为客户端侧流、服务端侧流
 */
function createInProcessStreamPair() {
  const clientToServer = new PassThrough();
  const serverToClient = new PassThrough();

  // 构造客户端侧双工流：读服务端写入的数据，写给服务端
  const clientStream = new Duplex({
    read: () => {},
    write: (chunk, enc, cb) => clientToServer.write(chunk, enc, cb),
    final: (cb) => {
      clientToServer.end();
      cb();
    },
    destroy: (err, cb) => {
      clientToServer.destroy(err);
      serverToClient.destroy(err);
      cb(err);
    }
  });
  serverToClient.on('data', chunk => clientStream.push(chunk));
  serverToClient.on('end', () => clientStream.push(null));
  serverToClient.on('error', err => clientStream.destroy(err));

  // 构造服务端侧双工流：读客户端写入的数据，写给客户端
  const serverStream = new Duplex({
    read: () => {},
    write: (chunk, enc, cb) => serverToClient.write(chunk, enc, cb),
    final: (cb) => {
      serverToClient.end();
      cb();
    },
    destroy: (err, cb) => {
      clientToServer.destroy(err);
      serverToClient.destroy(err);
      cb(err);
    }
  });
  clientToServer.on('data', chunk => serverStream.push(chunk));
  clientToServer.on('end', () => serverStream.push(null));
  clientToServer.on('error', err => serverStream.destroy(err));

  return [clientStream, serverStream];
}

使用方式和TCP连接完全一致：

const mqttCon = require('mqtt-connection');
const aedes = require('aedes')({ concurrency: 500, maxClientsIdLength: 100 });

const [clientStream, serverStream] = createInProcessStreamPair();
aedes.handle(serverStream);
const client = mqttCon(clientStream, { protocolVersion: 3 });

// 后续正常使用client发起订阅、发布即可，行为和TCP连接无差异

注意：不要尝试复用同一个流实例给通信两端，这种写法只适合单生产者多消费者的场景，完全不符合双向通信的要求。

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