8367067: Improve exception handling in HttpRequest.BodyPublishers

src/java.net.http/share/classes/jdk/internal/net/http/CheckedIterable.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) 2025, 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
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package jdk.internal.net.http;
+
+import java.util.Iterator;
+
+/**
+ * An {@link Iterable} clone for {@link CheckedIterator}.
+ *
+ * @param <E> the type of elements returned by the produced iterators
+ */
+@FunctionalInterface
+interface CheckedIterable<E> {
+
+    /**
+     * {@return an {@linkplain CheckedIterator iterator} over elements of type {@code E}}
+     */
+    CheckedIterator<E> iterator();
+
+    static <E> CheckedIterable<E> fromIterable(Iterable<E> iterable) {
+        return () -> {
+            Iterator<E> iterator = iterable.iterator();
+            return CheckedIterator.fromIterator(iterator);
+        };
+    }
+
+}

src/java.net.http/share/classes/jdk/internal/net/http/CheckedIterator.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright (c) 2025, 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
+ * particular file as subject to the "Classpath" exception as provided
+ * by Oracle in the LICENSE file that accompanied this code.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ */
+
+package jdk.internal.net.http;
+
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * An {@link java.util.Iterator} clone supporting checked exceptions.
+ *
+ * @param <E> the type of elements returned by this iterator
+ */
+interface CheckedIterator<E> {
+
+    /**
+     * {@return {@code true} if the iteration has more elements}
+     * @throws Exception if operation fails
+     */
+    boolean hasNext() throws Exception;
+
+    /**
+     * {@return the next element in the iteration}
+     *
+     * @throws NoSuchElementException if the iteration has no more elements
+     * @throws Exception if operation fails
+     */
+    E next() throws Exception;
+
+    static <E> CheckedIterator<E> fromIterator(Iterator<E> iterator) {
+        return new CheckedIterator<>() {
+
+            @Override
+            public boolean hasNext() {
+                return iterator.hasNext();
+            }
+
+            @Override
+            public E next() {
+                return iterator.next();
+            }
+
+        };
+    }
+
+}

src/java.net.http/share/classes/jdk/internal/net/http/PullPublisher.java

@@ -25,41 +25,45 @@
 
 package jdk.internal.net.http;
 
-import java.util.Iterator;
 import java.util.concurrent.Flow;
+
 import jdk.internal.net.http.common.Demand;
 import jdk.internal.net.http.common.SequentialScheduler;
 
 /**
- * A Publisher that publishes items obtained from the given Iterable. Each new
- * subscription gets a new Iterator.
+ * A {@linkplain Flow.Publisher publisher} that publishes items obtained from the given {@link CheckedIterator} supplier.
+ * Each new subscription gets a new {@code CheckedIterator}.
  */
 class PullPublisher<T> implements Flow.Publisher<T> {
 
     // Only one of `iterable` and `throwable` can be non-null. throwable is
     // non-null when an error has been encountered, by the creator of
     // PullPublisher, while subscribing the subscriber, but before subscribe has
     // completed.
-    private final Iterable<T> iterable;
+    private final CheckedIterable<T> iterable;
     private final Throwable throwable;
 
-    PullPublisher(Iterable<T> iterable, Throwable throwable) {
+    PullPublisher(CheckedIterable<T> iterable, Throwable throwable) {
+        if ((iterable == null && throwable == null) || (iterable != null && throwable != null)) {
+            String message = String.format(
+                    "only one of `iterable` or `throwable` can be null, but %s are",
+                    throwable == null ? "both" : "none");
+            throw new IllegalArgumentException(message);
+        }
         this.iterable = iterable;
         this.throwable = throwable;
     }
 
-    PullPublisher(Iterable<T> iterable) {
+    PullPublisher(CheckedIterable<T> iterable) {
         this(iterable, null);
     }
 
     @Override
     public void subscribe(Flow.Subscriber<? super T> subscriber) {
         Subscription sub;
         if (throwable != null) {
-            assert iterable == null : "non-null iterable: " + iterable;
             sub = new Subscription(subscriber, null, throwable);
         } else {
-            assert throwable == null : "non-null exception: " + throwable;
             sub = new Subscription(subscriber, iterable.iterator(), null);
         }
         subscriber.onSubscribe(sub);
@@ -72,15 +76,15 @@ public void subscribe(Flow.Subscriber<? super T> subscriber) {
     private class Subscription implements Flow.Subscription {
 
         private final Flow.Subscriber<? super T> subscriber;
-        private final Iterator<T> iter;
+        private final CheckedIterator<T> iter;
         private volatile boolean completed;
         private volatile boolean cancelled;
         private volatile Throwable error;
         final SequentialScheduler pullScheduler = new SequentialScheduler(new PullTask());
         private final Demand demand = new Demand();
 
         Subscription(Flow.Subscriber<? super T> subscriber,
-                     Iterator<T> iter,
+                     CheckedIterator<T> iter,
                      Throwable throwable) {
             this.subscriber = subscriber;
             this.iter = iter;
@@ -117,7 +121,18 @@ protected void run() {
                     }
                     subscriber.onNext(next);
                 }
-                if (!iter.hasNext() && !cancelled) {
+
+                boolean hasNext;
+                try {
+                    hasNext = iter.hasNext();
+                } catch (Exception e) {
+                    completed = true;
+                    pullScheduler.stop();
+                    subscriber.onError(e);
+                    return;
+                }
+
+                if (!hasNext && !cancelled) {
                     completed = true;
                     pullScheduler.stop();
                     subscriber.onComplete();

src/java.net.http/share/classes/jdk/internal/net/http/RequestPublishers.java

@@ -38,7 +38,6 @@
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Iterator;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.Objects;
@@ -98,7 +97,7 @@ List<ByteBuffer> copy(byte[] content, int offset, int length) {
         @Override
         public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
             List<ByteBuffer> copy = copy(content, offset, length);
-            var delegate = new PullPublisher<>(copy);
+            var delegate = new PullPublisher<>(CheckedIterable.fromIterable(copy));
             delegate.subscribe(subscriber);
         }
 
@@ -110,26 +109,30 @@ public long contentLength() {
 
     // This implementation has lots of room for improvement.
     public static class IterablePublisher implements BodyPublisher {
-        private final Iterable<byte[]> content;
+
+        private final CheckedIterable<byte[]> content;
         private volatile long contentLength;
 
         public IterablePublisher(Iterable<byte[]> content) {
-            this.content = Objects.requireNonNull(content);
+            Objects.requireNonNull(content, "content");
+            this.content = CheckedIterable.fromIterable(content);
         }
 
         // The ByteBufferIterator will iterate over the byte[] arrays in
         // the content one at the time.
         //
-        class ByteBufferIterator implements Iterator<ByteBuffer> {
+        class ByteBufferIterator implements CheckedIterator<ByteBuffer> {
+
             final ConcurrentLinkedQueue<ByteBuffer> buffers = new ConcurrentLinkedQueue<>();
-            final Iterator<byte[]> iterator = content.iterator();
+            final CheckedIterator<byte[]> iterator = content.iterator();
+
             @Override
-            public boolean hasNext() {
+            public boolean hasNext() throws Exception {
                 return !buffers.isEmpty() || iterator.hasNext();
             }
 
             @Override
-            public ByteBuffer next() {
+            public ByteBuffer next() throws Exception {
                 ByteBuffer buffer = buffers.poll();
                 while (buffer == null) {
                     copy();
@@ -142,7 +145,7 @@ ByteBuffer getBuffer() {
                 return Utils.getBuffer();
             }
 
-            void copy() {
+            void copy() throws Exception {
                 byte[] bytes = iterator.next();
                 int length = bytes.length;
                 if (length == 0 && iterator.hasNext()) {
@@ -165,18 +168,18 @@ void copy() {
             }
         }
 
-        public Iterator<ByteBuffer> iterator() {
+        CheckedIterator<ByteBuffer> iterator() {
             return new ByteBufferIterator();
         }
 
         @Override
         public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
-            Iterable<ByteBuffer> iterable = this::iterator;
+            CheckedIterable<ByteBuffer> iterable = this::iterator;
             var delegate = new PullPublisher<>(iterable);
             delegate.subscribe(subscriber);
         }
 
-        static long computeLength(Iterable<byte[]> bytes) {
+        static long computeLength(CheckedIterable<byte[]> bytes) {
             // Avoid iterating just for the purpose of computing
             // a length, in case iterating is a costly operation
             // For HTTP/1.1 it means we will be using chunk encoding
@@ -207,7 +210,7 @@ public StringPublisher(String content, Charset charset) {
 
     public static class EmptyPublisher implements BodyPublisher {
         private final Flow.Publisher<ByteBuffer> delegate =
-                new PullPublisher<ByteBuffer>(Collections.emptyList(), null);
+                new PullPublisher<>(CheckedIterable.fromIterable(Collections.emptyList()), null);
 
         @Override
         public long contentLength() {
@@ -289,7 +292,7 @@ public long contentLength() {
     /**
      * Reads one buffer ahead all the time, blocking in hasNext()
      */
-    public static class StreamIterator implements Iterator<ByteBuffer> {
+    public static class StreamIterator implements CheckedIterator<ByteBuffer> {
         final InputStream is;
         final Supplier<? extends ByteBuffer> bufSupplier;
         private volatile boolean eof;
@@ -330,20 +333,8 @@ private int read() throws IOException {
             return n;
         }
 
-        /**
-         * Close stream in this instance.
-         * UncheckedIOException may be thrown if IOE happens at InputStream::close.
-         */
-        private void closeStream() {
-            try {
-                is.close();
-            } catch (IOException e) {
-                throw new UncheckedIOException(e);
-            }
-        }
-
         @Override
-        public boolean hasNext() {
+        public boolean hasNext() throws IOException {
             stateLock.lock();
             try {
                 return hasNext0();
@@ -352,7 +343,7 @@ public boolean hasNext() {
             }
         }
 
-        private boolean hasNext0() {
+        private boolean hasNext0() throws IOException {
             if (need2Read) {
                 try {
                     haveNext = read() != -1;
@@ -362,18 +353,18 @@ private boolean hasNext0() {
                 } catch (IOException e) {
                     haveNext = false;
                     need2Read = false;
-                    throw new UncheckedIOException(e);
+                    throw new IOException(e);
                 } finally {
                     if (!haveNext) {
-                        closeStream();
+                        is.close();
                     }
                 }
             }
             return haveNext;
         }
 
         @Override
-        public ByteBuffer next() {
+        public ByteBuffer next() throws IOException {
             stateLock.lock();
             try {
                 if (!hasNext()) {
@@ -408,7 +399,7 @@ public void subscribe(Flow.Subscriber<? super ByteBuffer> subscriber) {
             publisher.subscribe(subscriber);
         }
 
-        protected Iterable<ByteBuffer> iterableOf(InputStream is) {
+        CheckedIterable<ByteBuffer> iterableOf(InputStream is) {
             return () -> new StreamIterator(is);
         }