code: protected CountDownLatch startStreamingCalls(int callsPerChannel, final AtomicLong counter,
        final AtomicBoolean record, final AtomicBoolean done, final long counterDelta) {
      final CountDownLatch latch = new CountDownLatch(callsPerChannel * channels.length);
      for (final ManagedChannel channel : channels) {
        for (int i = 0; i < callsPerChannel; i++) {
          final ClientCall<ByteBuf, ByteBuf> streamingCall =
              channel.newCall(pingPongMethod, CALL_OPTIONS);
          final AtomicReference<StreamObserver<ByteBuf>> requestObserverRef =
              new AtomicReference<>();
          final AtomicBoolean ignoreMessages = new AtomicBoolean();
          StreamObserver<ByteBuf> requestObserver = ClientCalls.asyncBidiStreamingCall(
              streamingCall,
              new StreamObserver<ByteBuf>() {
                @Override
                public void onNext(ByteBuf value) {
                  if (done.get()) {
                    if (!ignoreMessages.getAndSet(true)) {
                      requestObserverRef.get().onCompleted();
                    }
                    return;
                  }
                  requestObserverRef.get().onNext(request.slice());
                  if (record.get()) {
                    counter.addAndGet(counterDelta);
                  }
                  // request is called automatically because the observer implicitly has auto
                  // inbound flow control
                }
  
                @Override
                public void onError(Throwable t) {
                  logger.log(Level.WARNING, "call error", t);
                  latch.countDown();
                }
  
                @Override
                public void onCompleted() {
                  latch.countDown();
                }
              });
          requestObserverRef.set(requestObserver);
          requestObserver.onNext(request.slice());
          requestObserver.onNext(request.slice());
        }
      }
      return latch;
    }
docs: Start a continuously executing set of duplex streaming ping-pong calls that will terminate when
  {@code done.get()} is true. Each completed call will increment the counter by the specified
  delta which benchmarks can use to measure messages per second or bandwidth.