321

public static void main(String[] args) throws Exception {
    CompressorRegistry compressorRegistry = CompressorRegistry.newEmptyInstance();
    compressorRegistry.register(Codec.Identity.NONE);
    DecompressorRegistry decompressorRegistry =
        DecompressorRegistry.emptyInstance().with(Codec.Identity.NONE, true);

    ManagedChannel channel =
        NettyChannelBuilder.forAddress("localhost", 9099)
            .keepAliveTime(5, TimeUnit.SECONDS)
            .keepAliveTimeout(30, TimeUnit.SECONDS)
            .compressorRegistry(compressorRegistry)
            .decompressorRegistry(decompressorRegistry)
            .usePlaintext()
            //.flowControlWindow(1_000_000)
            .build();

    TestServiceGrpc.TestServiceStub client = TestServiceGrpc.newStub(channel);

    client.bidiStream(
        new ClientResponseObserver<Request, Response>() {
          ClientCallStreamObserver<Request> requestStream;
          int windowSize = WINDOW_SIZE;

          @Override
          public void beforeStart(ClientCallStreamObserver<Request> requestStream) {
            this.requestStream = requestStream;
            Request request = Request.newBuilder().setTimestamp(System.nanoTime()).build();
            requestStream.disableAutoRequestWithInitial(WINDOW_SIZE);
            requestStream.setOnReadyHandler(
                () -> {
                  while (requestStream.isReady()) {
                    requestStream.onNext(request);
                  }
                });
          }

          @Override
          public void onNext(Response value) {
            if (--windowSize == WINDOW_SIZE / 2) {
              windowSize += WINDOW_SIZE;
              requestStream.request(WINDOW_SIZE);
            }
          }

          @Override
          public void onError(Throwable t) {
            logger.error("Bidi stream terminated with error", t);
            channel.shutdownNow();
          }

          @Override
          public void onCompleted() {
            logger.info("Bidi stream completed");
            channel.shutdownNow();
          }
        });

    channel.awaitTermination(365, TimeUnit.DAYS);
  }