private void sendKeepalive(Iterator<Map.Entry<String, TurmsConnection>> iterator) {
        Map.Entry<String, TurmsConnection> entry = iterator.next();
        String nodeId = entry.getKey();
        TurmsConnection connection = entry.getValue();
        Connection conn = connection.getConnection();
        if (conn.isDisposed()) {
            iterator.remove();
            return;
        }
        if (!connection.isLocalNodeClient()) {
            return;
        }
        long elapsedTime = System.nanoTime() - connection.getLastKeepaliveTimestampNanos();
        if (elapsedTime > keepaliveTimeoutNanos) {
            LOGGER.warn("Reconnecting to the member ({}) due to keepalive timeout", nodeId);
            // onConnectionClosed() will reconnect the member
            disconnectConnection(connection);
            iterator.remove();
            return;
        }
        if (elapsedTime < keepaliveIntervalNanos) {
            return;
        }
        rpcService.requestResponse(nodeId, new KeepaliveRequest())
                .subscribe(null,
                        t -> LOGGER.warn("Failed to send a keepalive request to the member: "
                                + nodeId, t),
                        () -> connection.setLastKeepaliveTimestampNanos(System.nanoTime()));
    }
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