code: @Override
      public void operationComplete(ChannelFuture future) throws Exception {
          // If a message has been sent out, cancel the timeout for starting a request.
          cancelTimeout();
  
          if (future.isSuccess()) {
              // The first write is always the first headers, so log that we finished our first transfer over the
              // wire.
              if (!loggedRequestFirstBytesTransferred) {
                  logBuilder.requestFirstBytesTransferred();
                  loggedRequestFirstBytesTransferred = true;
              }
  
              if (state == State.DONE) {
                  // Successfully sent the request; schedule the response timeout.
                  response.scheduleTimeout(ch.eventLoop());
              }
  
              // Request more messages regardless whether the state is DONE. It makes the producer have
              // a chance to produce the last call such as 'onComplete' and 'onError' when there are
              // no more messages it can produce.
              if (!isSubscriptionCompleted) {
                  assert subscription != null;
                  subscription.request(1);
              }
              return;
          }
  
          fail(future.cause());
  
          final Throwable cause = future.cause();
          if (!(cause instanceof ClosedPublisherException)) {
              final Channel ch = future.channel();
              Exceptions.logIfUnexpected(logger, ch, HttpSession.get(ch).protocol(), cause);
              ch.close();
          }
      }
docs: Invoked on each write of an {@link HttpObject}.