src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/ResponseProcessors.java

*** 1,7 ****
  /*
!  * Copyright (c) 2016, Oracle and/or its affiliates. All rights reserved.
   * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   *
   * This code is free software; you can redistribute it and/or modify it
   * under the terms of the GNU General Public License version 2 only, as
   * published by the Free Software Foundation.  Oracle designates this
--- 1,7 ----
  /*
!  * Copyright (c) 2016, 2017, Oracle and/or its affiliates. All rights reserved.
   * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   *
   * This code is free software; you can redistribute it and/or modify it
   * under the terms of the GNU General Public License version 2 only, as
   * published by the Free Software Foundation.  Oracle designates this
*** 70,83 ****
              this.subscription = subscription;
              subscription.request(1);
          }
  
          @Override
!         public void onNext(ByteBuffer item) {
              byte[] buf = new byte[item.remaining()];
              item.get(buf);
              consumer.accept(Optional.of(buf));
              subscription.request(1);
          }
  
          @Override
          public void onError(Throwable throwable) {
--- 70,85 ----
              this.subscription = subscription;
              subscription.request(1);
          }
  
          @Override
!         public void onNext(List<ByteBuffer> items) {
!             for (ByteBuffer item : items) {
                  byte[] buf = new byte[item.remaining()];
                  item.get(buf);
                  consumer.accept(Optional.of(buf));
+             }
              subscription.request(1);
          }
  
          @Override
          public void onError(Throwable throwable) {
*** 118,130 ****
              }
              subscription.request(1);
          }
  
          @Override
!         public void onNext(ByteBuffer item) {
              try {
!                 out.write(item);
              } catch (IOException ex) {
                  Utils.close(out);
                  subscription.cancel();
                  result.completeExceptionally(ex);
              }
--- 120,132 ----
              }
              subscription.request(1);
          }
  
          @Override
!         public void onNext(List<ByteBuffer> items) {
              try {
!                 out.write(items.toArray(new ByteBuffer[0]));
              } catch (IOException ex) {
                  Utils.close(out);
                  subscription.cancel();
                  result.completeExceptionally(ex);
              }
*** 170,186 ****
              // We can handle whatever you've got
              subscription.request(Long.MAX_VALUE);
          }
  
          @Override
!         public void onNext(ByteBuffer item) {
              // incoming buffers are allocated by http client internally,
              // and won't be used anywhere except this place.
              // So it's free simply to store them for further processing.
!             if(item.hasRemaining()) {
!                 received.add(item);
!             }
          }
  
          @Override
          public void onError(Throwable throwable) {
              received.clear();
--- 172,187 ----
              // We can handle whatever you've got
              subscription.request(Long.MAX_VALUE);
          }
  
          @Override
!         public void onNext(List<ByteBuffer> items) {
              // incoming buffers are allocated by http client internally,
              // and won't be used anywhere except this place.
              // So it's free simply to store them for further processing.
!             assert Utils.remaining(items) > 0;   // TODO: is this really necessary?
!             received.addAll(items);
          }
  
          @Override
          public void onError(Throwable throwable) {
              received.clear();
*** 302,314 ****
              this.subscription = subscription;
              subscription.request(Long.MAX_VALUE);
          }
  
          @Override
!         public void onNext(ByteBuffer item) {
!             // TODO: check whether this should consume the buffer, as in:
!             item.position(item.limit());
          }
  
          @Override
          public void onError(Throwable throwable) {
              cf.completeExceptionally(throwable);
--- 303,314 ----
              this.subscription = subscription;
              subscription.request(Long.MAX_VALUE);
          }
  
          @Override
!         public void onNext(List<ByteBuffer> items) {
!             // NO-OP
          }
  
          @Override
          public void onError(Throwable throwable) {
              cf.completeExceptionally(throwable);