private Mono<List<TurmsClient>> prepareClients(long firstUserId, int userCount) {
        LOGGER.info("Preparing clients");
        LoopResources loopResources = LoopResources.create(ThreadNameConst.FAKE_CLIENT);
        List<Mono<TurmsNotification>> results = new ArrayList<>(userCount);
        ConcurrentLinkedQueue<TurmsClient> clients = new ConcurrentLinkedQueue<>();
        for (int i = 0; i < userCount; i++) {
            TurmsClient client = TurmsClient.tcp();
            long userId = firstUserId + i;
            Mono<TurmsNotification> loginResult = client
                    .connect(tcpUserSessionDispatcher.getHost(),
                            tcpUserSessionDispatcher.getPort(),
                            loopResources)
                    .retryWhen(Retry.backoff(5, Duration.ofSeconds(15))
                            // e.g. "Connection reset"
                            .filter(ThrowableUtil::isDisconnectedClientError))
                    .then(Mono.defer(() -> client.login(userId, DeviceType.DESKTOP, "123")))
                    .doOnSuccess(notification -> {
                        if (ResponseStatusCode.isSuccessCode(notification.getCode())) {
                            clients.add(client);
                        } else {
                            LOGGER.error(
                                    "Failed to establish the session: {}. The login response is: {}",
                                    client.getSessionIdDesc(),
                                    ProtoFormatter.toJSON5(notification, 128));
                        }
                    })
                    .onErrorResume(t -> {
                        LOGGER.error("Failed to establish the session: "
                                + client.getSessionIdDesc(), t);
                        return Mono.empty();
                    });
            results.add(loginResult);
        }
        return Mono.when(results)
                .then(Mono.fromCallable(() -> {
                    ArrayList<TurmsClient> list = new ArrayList<>(clients);
                    LOGGER.info("Prepared clients. Expected: {}. Prepared: {}",
                            userCount,
                            list.size());
                    return list;
                }));
    }
The main reason to prepare clients in an async way is to recover from "Connection reset" to
retry to connect without blocking the server
turms-gateway/src/main/java/im/turms/gateway/infra/fake/ClientFakingManager.java