Throw checked IOException in input/output stream methods

Author: manuelbl

java-does-usb/src/main/java/net/codecrete/usb/common/EndpointInputStream.java

@@ -15,9 +15,12 @@
 import java.io.InputStream;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
+import java.util.Objects;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import static net.codecrete.usb.common.EndpointStreams.toIOException;
+
 import static java.lang.System.Logger.Level.WARNING;
 import static java.lang.foreign.ValueLayout.JAVA_BYTE;
 
@@ -134,44 +137,66 @@ public void close() throws IOException {
 
     @Override
     public int read() throws IOException {
-        if (isClosed())
-            return -1;
+        ensureOpen();
+
+        try {
+            if (bufferedBytes() == 0)
+                receiveMoreData();
 
-        if (available() == 0)
-            receiveMoreData();
+            var b = currentTransfer.data().get(JAVA_BYTE, readOffset) & 0xff;
+            readOffset += 1;
+            return b;
 
-        var b = currentTransfer.data().get(JAVA_BYTE, readOffset) & 0xff;
-        readOffset += 1;
-        return b;
+        } catch (UsbException e) {
+            throw toIOException(e);
+        }
     }
 
     @Override
     public int read(byte @NotNull [] b, int off, int len) throws IOException {
-        if (isClosed())
-            return -1;
+        Objects.checkFromIndexSize(off, len, b.length);
+        ensureOpen();
+        if (len == 0)
+            return 0;
 
-        var numRead = 0;
-        do {
-            if (available() == 0)
-                receiveMoreData();
+        try {
+            var numRead = 0;
+            do {
+                if (bufferedBytes() == 0)
+                    receiveMoreData();
+
+                // copy data to receiving buffer
+                var n = Math.min(len - numRead, currentTransfer.resultSize() - readOffset);
+                MemorySegment.copy(currentTransfer.data(), readOffset, MemorySegment.ofArray(b), (long) off + numRead, n);
+                readOffset += n;
+                numRead += n;
 
-            // copy data to receiving buffer
-            var n = Math.min(len - numRead, currentTransfer.resultSize() - readOffset);
-            MemorySegment.copy(currentTransfer.data(), readOffset, MemorySegment.ofArray(b), (long) off + numRead, n);
-            readOffset += n;
-            numRead += n;
+            } while (numRead < len && hasMoreTransfers());
 
-        } while (numRead < len && hasMoreTransfers());
+            return numRead;
 
-        return numRead;
+        } catch (UsbException e) {
+            throw toIOException(e);
+        }
     }
 
-    @SuppressWarnings("RedundantThrows")
     @Override
     public int available() throws IOException {
+        ensureOpen();
+        return bufferedBytes();
+    }
+
+    // Bytes buffered in the current transfer, without a closed-stream check.
+    // Callers on the read path guard with ensureOpen() first.
+    private int bufferedBytes() {
         return currentTransfer.resultSize() - readOffset;
     }
 
+    private void ensureOpen() throws IOException {
+        if (isClosed())
+            throw new IOException("input stream has been closed");
+    }
+
     private boolean hasMoreTransfers() {
         return !completedTransferQueue.isEmpty();
     }

java-does-usb/src/main/java/net/codecrete/usb/common/EndpointOutputStream.java

@@ -8,16 +8,20 @@
 package net.codecrete.usb.common;
 
 import net.codecrete.usb.UsbDirection;
+import net.codecrete.usb.UsbException;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.IOException;
 import java.io.OutputStream;
 import java.lang.foreign.Arena;
 import java.lang.foreign.MemorySegment;
 import java.util.Arrays;
+import java.util.Objects;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import static net.codecrete.usb.common.EndpointStreams.toIOException;
+
 import static java.lang.System.Logger.Level.WARNING;
 import static java.lang.foreign.ValueLayout.JAVA_BYTE;
 
@@ -239,41 +243,57 @@ private Transfer pollAvailableTransfer(long deadline) {
 
     @Override
     public void write(int b) throws IOException {
-        checkIsOpen();
+        ensureOpen();
 
-        currentTransfer.data().set(JAVA_BYTE, writeOffset, (byte) b);
-        writeOffset += 1;
-        if (writeOffset == transferSize)
-            submitTransfer(writeOffset);
+        try {
+            currentTransfer.data().set(JAVA_BYTE, writeOffset, (byte) b);
+            writeOffset += 1;
+            if (writeOffset == transferSize)
+                submitTransfer(writeOffset);
+
+        } catch (UsbException e) {
+            throw toIOException(e);
+        }
     }
 
     @Override
     public void write(byte @NotNull [] b, int off, int len) throws IOException {
-        checkIsOpen();
+        Objects.checkFromIndexSize(off, len, b.length);
+        ensureOpen();
 
-        while (len > 0) {
-            var chunkSize = Math.min(len, transferSize - writeOffset);
-            MemorySegment.copy(b, off, currentTransfer.data(), JAVA_BYTE, writeOffset, chunkSize);
-            writeOffset += chunkSize;
-            off += chunkSize;
-            len -= chunkSize;
+        try {
+            while (len > 0) {
+                var chunkSize = Math.min(len, transferSize - writeOffset);
+                MemorySegment.copy(b, off, currentTransfer.data(), JAVA_BYTE, writeOffset, chunkSize);
+                writeOffset += chunkSize;
+                off += chunkSize;
+                len -= chunkSize;
+
+                if (writeOffset == transferSize)
+                    submitTransfer(writeOffset);
+            }
 
-            if (writeOffset == transferSize)
-                submitTransfer(writeOffset);
+        } catch (UsbException e) {
+            throw toIOException(e);
         }
     }
 
     @Override
     public void flush() throws IOException {
-        checkIsOpen();
+        ensureOpen();
 
-        if (writeOffset > 0)
-            submitTransfer(writeOffset);
+        try {
+            if (writeOffset > 0)
+                submitTransfer(writeOffset);
+
+            if (needsZlp)
+                submitTransfer(0);
 
-        if (needsZlp)
-            submitTransfer(0);
+            waitForOutstandingTransfers();
 
-        waitForOutstandingTransfers();
+        } catch (UsbException e) {
+            throw toIOException(e);
+        }
     }
 
     /**
@@ -387,8 +407,8 @@ private synchronized void onCompletion(Transfer transfer) {
     protected void configureEndpoint() {
     }
 
-    private void checkIsOpen() throws IOException {
+    private void ensureOpen() throws IOException {
         if (isClosed())
-            throw new IOException("endpoint output stream has been closed");
+            throw new IOException("output stream has been closed");
     }
 }

java-does-usb/src/main/java/net/codecrete/usb/common/EndpointStreams.java

@@ -0,0 +1,44 @@
+//
+// Java Does USB
+// Copyright (c) 2023 Manuel Bleichenbacher
+// Licensed under MIT License
+// https://opensource.org/licenses/MIT
+//
+
+package net.codecrete.usb.common;
+
+import net.codecrete.usb.UsbException;
+import net.codecrete.usb.UsbTimeoutException;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+
+/**
+ * Helpers shared by {@link EndpointInputStream} and {@link EndpointOutputStream}.
+ */
+final class EndpointStreams {
+
+    private EndpointStreams() {
+    }
+
+    /**
+     * Wraps a USB error in an {@link IOException} so it is surfaced through the
+     * {@link java.io.InputStream}/{@link java.io.OutputStream} contract.
+     * <p>
+     * A transfer timeout is mapped to {@link InterruptedIOException} (java.io's "I/O timed out").
+     * All other USB errors become a plain {@link IOException} with the {@link UsbException} as cause,
+     * which preserves the USB error code and any stall/timeout subtype for callers that inspect it.
+     * </p>
+     *
+     * @param e the USB error
+     * @return the corresponding I/O exception
+     */
+    static IOException toIOException(UsbException e) {
+        if (e instanceof UsbTimeoutException) {
+            var ioException = new InterruptedIOException(e.getMessage());
+            ioException.initCause(e);
+            return ioException;
+        }
+        return new IOException(e.getMessage(), e);
+    }
+}

java-does-usb/src/test/java/net/codecrete/usb/StreamTest.java

@@ -86,7 +86,7 @@ void largeTransferBigChunks_succeeds() {
     void blockedWriter_canBeAborted() throws InterruptedException {
         // A writer that fills the pipe faster than it is drained eventually blocks in write().
         // Aborting the outstanding transfers from another thread must terminate it promptly and
-        // safely (with a USB exception) rather than leave it wedged forever.
+        // safely (with an IOException wrapping the USB error) rather than leave it wedged forever.
 
         final var data = generateRandomBytes(1_000_000, 0x5c7f10ebL);
 
@@ -152,7 +152,10 @@ void blockedWriter_canBeAborted() throws InterruptedException {
             assertFalse(writer.isAlive(), "writer thread did not terminate within 1.5 s after abort");
 
             // It must have unwound because of the abort, not by finishing or dying some other way.
-            assertInstanceOf(UsbException.class, writerError.get());
+            // The stream surfaces the USB error as an IOException (with the UsbException as cause).
+            var err = writerError.get();
+            assertInstanceOf(IOException.class, err);
+            assertInstanceOf(UsbException.class, err.getCause());
 
         } finally {
             // Restore a clean device state so subsequent tests are not affected.