78         if (contentLength == -2) {

    79             // HTTP/1.0 content

    80             chunkedContentInitialized = true;

    81             chunkedContent = false;

    82             return chunkedContent;

    83         }

   403     class UnknownLengthBodyParser implements BodyParser {

   404         final Consumer<Throwable> onComplete;

   405         final Logger debug = Utils.getDebugLogger(this::dbgString, Utils.DEBUG);

   406         final String dbgTag = ResponseContent.this.dbgTag + "/UnknownLengthBodyParser";

   407         volatile Throwable closedExceptionally;

   408         volatile AbstractSubscription sub;

   409         volatile int breceived = 0;

   410 

   411         UnknownLengthBodyParser(Consumer<Throwable> onComplete) {

   412             this.onComplete = onComplete;

   413         }

   414 

   415         String dbgString() {

   416             return dbgTag;

   417         }

   418 

   419         @Override

   420         public void onSubscribe(AbstractSubscription sub) {

   421             if (debug.on())

   422                 debug.log("onSubscribe: " + pusher.getClass().getName());

   423             pusher.onSubscribe(this.sub = sub);

   424         }

   425 

   426         @Override

   427         public String currentStateMessage() {

   428             return format("http1_0 content, bytes received: %d", breceived);

   429         }

   430 

   431         @Override

   432         public void accept(ByteBuffer b) {

   433             if (closedExceptionally != null) {

   434                 if (debug.on())

   435                     debug.log("already closed: " + closedExceptionally);

   436                 return;

   437             }

   438             boolean completed = false;

   439             try {

   440                 if (debug.on())

   441                     debug.log("Parser got %d bytes ", b.remaining());

   442 

   443                 if (b.hasRemaining()) {

   444                     // only reduce demand if we actually push something.

   445                     // we would not have come here if there was no

   446                     // demand.

   447                     boolean hasDemand = sub.demand().tryDecrement();

   448                     assert hasDemand;

   449                     breceived += b.remaining();

   450                     pusher.onNext(List.of(b.asReadOnlyBuffer()));

   451                 }

   452             } catch (Throwable t) {

   453                 if (debug.on()) debug.log("Unexpected exception", t);

   454                 closedExceptionally = t;

   455                 if (!completed) {

   456                     onComplete.accept(t);

   457                 }

   458             }

   459         }

   460 

   461         /**

   462          * Must be called externally when connection has closed

   463          * and therefore no more bytes can be read

   464          */

   465         public void complete() {

   466             // We're done! All data has been received.

   467             if (debug.on())

   468                 debug.log("Parser got all expected bytes: completing");

   469             assert closedExceptionally == null;

   470             onFinished.run();

   471             pusher.onComplete();

   472             onComplete.accept(closedExceptionally); // should be null

   473         }

   474     }

   475