private void onMemberConnectionAdded(@Nullable Member member, TurmsConnection connection) {
        String endpointType = connection.isLocalNodeClient()
                ? "Client"
                : "Server";
        String memberIdAndAddress = getMemberIdAndAddress(connection.getNodeId(), member);
        LOGGER.info("[{}] Connected to the member{}",
                endpointType,
                memberIdAndAddress.isEmpty()
                        ? ""
                        : ": "
                                + memberIdAndAddress);
        for (MemberConnectionListener listener : connection.getListeners()) {
            try {
                listener.onConnectionOpened(connection);
            } catch (Exception e) {
                LOGGER.error("Caught an error while notifying the onConnectionOpened listener: "
                        + listener.getClass()
                                .getName(),
                        e);
            }
        }
        ChannelOperations<?, ?> conn = connection.getConnection();
        conn.receiveObject()
                .doOnNext(value -> {
                    for (MemberConnectionListener listener : connection.getListeners()) {
                        try {
                            listener.onDataReceived(value);
                        } catch (Exception e) {
                            LOGGER.error(
                                    "Caught an error while notifying the onDataReceived listener: "
                                            + listener.getClass()
                                                    .getName(),
                                    e);
                        }
                    }
                })
                .onErrorResume(t -> {
                    if (ThrowableUtil.isDisconnectedClientError(t) && connection.isClosing()) {
                        return Mono.empty();
                    }
                    LOGGER.error("[{}] Failed to listen to the connection to the member{}",
                            endpointType,
                            memberIdAndAddress.isEmpty()
                                    ? ""
                                    : ": "
                                            + memberIdAndAddress,
                            t);
                    return Mono.empty();
                })
                .subscribe();
        conn.onDispose()
                .subscribe(null,
                        t -> onConnectionClosed(connection, t),
                        () -> onConnectionClosed(connection, null));
    }
@param member is null when accepting a connection on the server side
turms-server-common/src/main/java/im/turms/server/common/infra/cluster/service/connection/ConnectionService.java