@Override
    public Mono<Void> stop(long timeoutMillis) {
        List<Mono<Void>> monos = new LinkedList<>();
        monos.add(keepaliveThread.terminate());
        if (server != null) {
            monos.add(server.dispose()
                    .onErrorMap(
                            t -> new RuntimeException("Failed to stop the local node server", t)));
        }
        ClosingHandshakeRequest request = new ClosingHandshakeRequest(
                ClosingHandshakeRequest.CLOSE_STATUS_CODE_SERVER_SHUTTING_DOWN);
        for (TurmsConnection connection : nodeIdToConnection.values()) {
            connection.setClosing(true);
            Connection conn = connection.getConnection();
            if (conn.isDisposed()) {
                continue;
            }
            String nodeId = connection.getNodeId();
            if (nodeId == null) {
                conn.dispose();
                monos.add(conn.onDispose());
            } else {
                monos.add(rpcService.requestResponse(nodeId, request)
                        .onErrorMap(t -> new RuntimeException(
                                "Failed to send the closing handshake to the member: "
                                        + nodeId,
                                t))
                        .flatMap(unused -> {
                            conn.dispose();
                            return conn.onDispose();
                        }));
            }
        }
        nodeIdToConnection.clear();
        return Mono.whenDelayError(monos);
    }
Use supplier rather than just listener so that we can bind each generated listener to a
specific TcpConnection to make logic simple
turms-server-common/src/main/java/im/turms/server/common/infra/cluster/service/connection/ConnectionService.java