Observable本身反复出现,我不知道为什么

我是一个新手 rxjs 库,仍在学习,我正试图使用在node.js中实现TCP服务器 Observable ,并使用我在公司看到的模式,该模式包括具有 可观察 作为输入并返回另一个输出 可观察 将被发送回客户端,所以我想出了这个代码

import net from "net";
import * as rx from "rxjs";

type MaybePromise<T> = T | PromiseLike<T>;
type ConnectionHandler = (address: net.AddressInfo, input$: rx.Observable<Buffer>) => MaybePromise<rx.Observable<Buffer>>;

const createServer = (port: number, handler: ConnectionHandler, opts?: net.ServerOpts) => {
  const server = net.createServer(opts);

  new rx.Observable<net.Socket>(subscriber => {
    server
      .on("connection", socket => subscriber.next(socket))
      .on("error", err => subscriber.error(err));
  }).subscribe(async socket =>
    (await handler(
      socket.address() as net.SocketAddress,
      new rx.Observable<Buffer>(subscriber => {
        socket
          .on("end", () => socket.destroySoon())
          .on("error", err => subscriber.error(err))
          .on("data", request => subscriber.next(request))
          .on("close", () => subscriber.complete());
      }),
    )).subscribe({
      next: response => {
        console.log(response.toString());
        socket.emit("data", response);
      },
      error: () => socket.destroySoon(),
    })
  );

  server.listen(port);
};

createServer(3000, async (address, input$) => {
  console.log(`New client: ${address.address}`);

  const login = await rx.firstValueFrom(input$);

  if (login.toString() != "login\n")
    return rx.throwError(() => new Error("login failure"));

  return input$.pipe(
    rx.map(data => {
      return Buffer.from(`ACK: ${data.toString()}`, "utf-8");
    })
  );
});

所以基本上在这个例子中,我想等待 login\n 请求,然后对于每个请求,我想将其回显到客户端,添加 ACK: 在前面,但当创建可观察的输出时,它添加了 ACK 在同一次按摩中进行了数千次,所以每次输出似乎都会作为输入返回到函数中,但我不知道为什么,我没有更改输入 可观察 ,我只是退回一个新的,将被处理后发送给客户,你能帮我指出错误的来源吗?