S2Client.java 文件源码

private S2Client(Builder builder) {

        connectToIp = builder.connectToIp;
        connectToPort = builder.connectToPort;
        traced = builder.traced;
        tracer = builder.tracer;
        game = builder.game;

        log.info("Starting: {}", this);

        Channel channel = channelProvider.getChannel();
        responseStream = channel.outputStream().mergeWith(channel.errorStream())
                .map(this::prepareResponse)
                .toFlowable(BackpressureStrategy.ERROR)
                .onBackpressureBuffer(cfg().getInt(OcraftConfig.CLIENT_BUFFER_SIZE_RESPONSE_BACKPRESSURE))
                .observeOn(Schedulers.computation(), false, cfg().getInt(CLIENT_BUFFER_SIZE_RESPONSE_STREAM))
                .publish()
                .autoConnect()
                .doOnSubscribe(s -> await.register())
                .doOnCancel(await::arriveAndDeregister);

        responseStream().subscribe(this);
        await.arriveAndDeregister();

        Optional.ofNullable(game).ifPresent(s2Controller -> {
            responseStream().subscribe(s2Controller);
            await.arriveAndDeregister();
        });

        channelProvider.start(connectToIp, connectToPort);
    }