private void connectMemberUntilSucceedOrRemoved0(Member member) {
        String nodeId = member.getNodeId();
        LOGGER.info("[Client] Connecting to the member: {id={}, host={}, port={}}. Retry times: {}",
                nodeId,
                member.getMemberHost(),
                member.getMemberPort(),
                nodeIdToConnectionRetries.getOrDefault(nodeId, 0));
        initTcpConnection(member.getMemberHost(), member.getMemberPort()).doOnSuccess(conn -> {
            TurmsConnection connection = new TurmsConnection(
                    nodeId,
                    (ChannelOperations<?, ?>) conn,
                    true,
                    newMemberConnectionListeners());
            onMemberConnectionAdded(member, connection);
            String localNodeId = discoveryService.getLocalMember()
                    .getNodeId();
            LOGGER.info(
                    "[Client] Sending an opening handshake request to the member: {id={}, host={}, port={}}",
                    nodeId,
                    member.getMemberHost(),
                    member.getMemberPort());
            rpcService
                    .requestResponse(nodeId,
                            new OpeningHandshakeRequest(localNodeId),
                            null,
                            connection)
                    .subscribe(code -> {
                        if (code == OpeningHandshakeRequest.RESPONSE_CODE_SUCCESS) {
                            onMemberConnectionHandshakeCompleted(member, connection, true);
                        } else {
                            throw new RuntimeException(
                                    "Failure code: "
                                            + code);
                        }
                    }, t -> {
                        LOGGER.error(
                                "[Client] Failed to complete the opening handshake with the member: {id={}, host={}, port={}}. Closing connection to reconnect",
                                nodeId,
                                member.getMemberHost(),
                                member.getMemberPort(),
                                t);
                        // To keep logic simple, just disconnect to
                        // connect and start a handshake again.
                        // After disposed, the listener to onDispose will reconnect
                        disconnectConnection(connection);
                    });
        })
                .onErrorResume(throwable -> {
                    if (!discoveryService.isKnownMember(nodeId)) {
                        return Mono.empty();
                    }
                    int retryTimes = nodeIdToConnectionRetries.getOrDefault(nodeId, 0);
                    LOGGER.error(
                            "[Client] Failed to connect to member: {id={}, host={}, port={}}. Retry times: {}",
                            nodeId,
                            member.getMemberHost(),
                            member.getMemberPort(),
                            retryTimes,
                            throwable);
                    retryTimes++;
                    nodeIdToConnectionRetries.put(nodeId, retryTimes);
                    connectionRetryScheduler.schedule(() -> {
                        if (!isMemberConnected(nodeId) && discoveryService.isKnownMember(nodeId)) {
                            connectMemberUntilSucceedOrRemoved0(member);
                        } else {
                            nodeIdToConnectionRetries.remove(nodeId);
                        }
                    }, Math.min(retryTimes * 10, 60), TimeUnit.SECONDS);
                    return Mono.empty();
                })
                .subscribe();
    }
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