Repository: softwaremill/jox
Pull request #88 (2025-01-22 10:42:52)

Channel reference in the ClosedChannel
Closes #73

adamw: -> merge conflicts, sorry :)
plus, a failed build, but seems unrelated; I'll add a timeout so that we can investigate

Patch:
diff --git a/channels/src/main/java/com/softwaremill/jox/Channel.java b/channels/src/main/java/com/softwaremill/jox/Channel.java
index 7e3ab4b..68cf987 100644
--- a/channels/src/main/java/com/softwaremill/jox/Channel.java
+++ b/channels/src/main/java/com/softwaremill/jox/Channel.java
@@ -706,7 +706,7 @@ public final class Channel<T> implements Source<T>, Sink<T> {
 
     @Override
     public Object doneOrClosed() {
-        return closeOrClosed(new ChannelDone());
+        return closeOrClosed(new ChannelDone(this));
     }
 
     @Override
@@ -722,7 +722,7 @@ public final class Channel<T> implements Source<T>, Sink<T> {
 
     @Override
     public Object errorOrClosed(Throwable reason) {
-        return closeOrClosed(new ChannelError(reason));
+        return closeOrClosed(new ChannelError(reason, this));
     }
 
     private Object closeOrClosed(ChannelClosed channelClosed) {
diff --git a/channels/src/main/java/com/softwaremill/jox/ChannelClosed.java b/channels/src/main/java/com/softwaremill/jox/ChannelClosed.java
index 46d7738..ef4a0d0 100644
--- a/channels/src/main/java/com/softwaremill/jox/ChannelClosed.java
+++ b/channels/src/main/java/com/softwaremill/jox/ChannelClosed.java
@@ -5,4 +5,5 @@ package com.softwaremill.jox;
  */
 public sealed interface ChannelClosed permits ChannelDone, ChannelError {
     ChannelClosedException toException();
+    Channel<?> channel();
 }
diff --git a/channels/src/main/java/com/softwaremill/jox/ChannelDone.java b/channels/src/main/java/com/softwaremill/jox/ChannelDone.java
index 2608b48..9bd529d 100644
--- a/channels/src/main/java/com/softwaremill/jox/ChannelDone.java
+++ b/channels/src/main/java/com/softwaremill/jox/ChannelDone.java
@@ -1,6 +1,6 @@
 package com.softwaremill.jox;
 
-public record ChannelDone() implements ChannelClosed {
+public record ChannelDone(Channel<?> channel) implements ChannelClosed {
     @Override
     public ChannelClosedException toException() {
         return new ChannelDoneException();
diff --git a/channels/src/main/java/com/softwaremill/jox/ChannelError.java b/channels/src/main/java/com/softwaremill/jox/ChannelError.java
index 36a9451..3bc5d91 100644
--- a/channels/src/main/java/com/softwaremill/jox/ChannelError.java
+++ b/channels/src/main/java/com/softwaremill/jox/ChannelError.java
@@ -1,6 +1,6 @@
 package com.softwaremill.jox;
 
-public record ChannelError(Throwable cause) implements ChannelClosed {
+public record ChannelError(Throwable cause, Channel<?> channel) implements ChannelClosed {
     @Override
     public ChannelClosedException toException() {
         return new ChannelErrorException(cause);
diff --git a/channels/src/main/java/com/softwaremill/jox/Select.java b/channels/src/main/java/com/softwaremill/jox/Select.java
index 34587bf..e677929 100644
--- a/channels/src/main/java/com/softwaremill/jox/Select.java
+++ b/channels/src/main/java/com/softwaremill/jox/Select.java
@@ -48,7 +48,8 @@ public class Select {
      * <p>
      * If no clauses are given, throws {@link ChannelDoneException}.
      *
-     * @param clauses The clauses, from which one will be selected. Not {@code null}.
+     * @param clauses The clauses, from which one will be selected. Array must not be empty or {@code null} and
+     *                can't contain {@code null} values.
      * @return The value returned by the selected clause.
      * @throws ChannelClosedException When any of the channels is closed (done or in error).
      */
@@ -72,16 +73,21 @@ public class Select {
      * <p>
      * If no clauses are given, returns {@link ChannelDone}.
      *
-     * @param clauses The clauses, from which one will be selected. Not {@code null}.
+     * @param clauses The clauses, from which one will be selected. Array must not be empty or {@code null} and
+     *                can't contain {@code null} values.
      * @return Either the value returned by the selected clause, or {@link ChannelClosed}, when any of the channels
      * is closed (done or in error).
      */
     @SafeVarargs
     public static <U> Object selectOrClosed(SelectClause<? extends U>... clauses) throws InterruptedException {
         while (true) {
-            if (clauses.length == 0) {
+            if (clauses == null || clauses.length == 0) {
                 // no clauses given
-                return new ChannelDone();
+                throw new IllegalArgumentException("No clauses given");
+            }
+            if (Arrays.stream(clauses).anyMatch(Objects::isNull)){
+                // null clauses given
+                throw new IllegalArgumentException("Null clauses are not supported");
             }
 
             var r = doSelectOrClosed(clauses);
diff --git a/flows/src/main/java/com/softwaremill/jox/flows/Flow.java b/flows/src/main/java/com/softwaremill/jox/flows/Flow.java
index 420ee21..5dfd2e0 100644
--- a/flows/src/main/java/com/softwaremill/jox/flows/Flow.java
+++ b/flows/src/main/java/com/softwaremill/jox/flows/Flow.java
@@ -879,7 +879,7 @@ public class Flow<T> {
                                 if (!buffer.isEmpty()) outputChannel.send(buffer);
                                 outputChannel.done();
                                 yield false;
-                            case ChannelError(Throwable cause):
+                            case ChannelError(Throwable cause, Channel<?> _):
                                 // source returned error, propagate it and finish
                                 if (timeoutFork != null) timeoutFork.cancelNow();
                                 outputChannel.error(cause);
@@ -1319,7 +1319,7 @@ public class Flow<T> {
                                 results.done();
                                 return null;
                             }
-                            case ChannelError(Throwable e) -> throw new IllegalStateException("inProgress should never be closed with an error", e);
+                            case ChannelError(Throwable e, Channel<?> _) -> throw new IllegalStateException("inProgress should never be closed with an error", e);
                             case Object fork -> {
                                 //noinspection unchecked
                                 Optional<U> result = ((Fork<Optional<U>>) fork).join();
@@ -1676,10 +1676,6 @@ public class Flow<T> {
                     }
                 });
                 close(channel, null);
-            } catch (JoxScopeExecutionException e) {
-                Exception cause = (Exception) e.getCause();
-                close(channel, cause);
-                throw cause;
             } catch (Exception t) {
                 close(channel, t);
                 throw t;
diff --git a/flows/src/main/java/com/softwaremill/jox/flows/Flows.java b/flows/src/main/java/com/softwaremill/jox/flows/Flows.java
index 3c299b8..3752f5f 100644
--- a/flows/src/main/java/com/softwaremill/jox/flows/Flows.java
+++ b/flows/src/main/java/com/softwaremill/jox/flows/Flows.java
@@ -463,7 +463,7 @@ public final class Flows {
                                     currentSourceIndex = (currentSourceIndex + 1) % availableSources.size();
                                     elementsRead = 0;
                                 }
-                            } else if (received instanceof ChannelError(Throwable cause)) {
+                            } else if (received instanceof ChannelError(Throwable cause, Channel<?> _)) {
                                 // if any source fails, propagate the error
                                 results.errorOrClosed(cause);
                                 break;


Test patch:
diff --git a/channel-ops/src/test/java/com/softwaremill/jox/ops/SourceOpsCollectTest.java b/channel-ops/src/test/java/com/softwaremill/jox/ops/SourceOpsCollectTest.java
index 8e09e12..5ec7be2 100644
--- a/channel-ops/src/test/java/com/softwaremill/jox/ops/SourceOpsCollectTest.java
+++ b/channel-ops/src/test/java/com/softwaremill/jox/ops/SourceOpsCollectTest.java
@@ -7,6 +7,7 @@ import com.softwaremill.jox.structured.Scopes;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 
 public class SourceOpsCollectTest {
     @Test
@@ -26,7 +27,7 @@ public class SourceOpsCollectTest {
             assertEquals(10, s.receive());
             assertEquals(20, s.receive());
             assertEquals(30, s.receive());
-            assertEquals(new ChannelDone(), s.receiveOrClosed());
+            assertInstanceOf(ChannelDone.class, s.receiveOrClosed());
             return null;
         });
     }
@@ -52,7 +53,7 @@ public class SourceOpsCollectTest {
 
             assertEquals(20, s.receive());
             assertEquals(40, s.receive());
-            assertEquals(new ChannelDone(), s.receiveOrClosed());
+            assertInstanceOf(ChannelDone.class, s.receiveOrClosed());
             return null;
         });
     }
@@ -71,7 +72,7 @@ public class SourceOpsCollectTest {
                 Source<Integer> s = SourceOps.forSource(scope, c).collect(x -> x * 10).toSource();
 
                 assertEquals(10, s.receive());
-                assertEquals(new ChannelDone(), s.receiveOrClosed());
+                assertInstanceOf(ChannelDone.class, s.receiveOrClosed());
                 return null;
             });
         }
@@ -94,7 +95,7 @@ public class SourceOpsCollectTest {
             assertEquals(2, s.receive());
             assertEquals(4, s.receive());
             assertEquals(6, s.receive());
-            assertEquals(new ChannelDone(), s.receiveOrClosed());
+            assertInstanceOf(ChannelDone.class, s.receiveOrClosed());
             return null;
         });
     }
diff --git a/channels/src/test/java/com/softwaremill/jox/ChannelClosedTest.java b/channels/src/test/java/com/softwaremill/jox/ChannelClosedTest.java
index dee1676..1d7b03e 100644
--- a/channels/src/test/java/com/softwaremill/jox/ChannelClosedTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/ChannelClosedTest.java
@@ -3,28 +3,29 @@ package com.softwaremill.jox;
 import org.junit.jupiter.api.Test;
 
 import java.util.concurrent.ExecutionException;
-import java.util.concurrent.Future;
-import java.util.concurrent.Semaphore;
 
-import static com.softwaremill.jox.TestUtil.*;
+import static com.softwaremill.jox.TestUtil.forkCancelable;
+import static com.softwaremill.jox.TestUtil.scoped;
 import static org.junit.jupiter.api.Assertions.*;
 
 public class ChannelClosedTest {
     @Test
-    void testClosed_noValues_whenError() {
+    void testClosed_noValues_whenError() throws InterruptedException {
         // given
         Channel<Integer> c = new Channel<>();
+        RuntimeException reason = new RuntimeException();
 
         // when
-        c.error(new RuntimeException());
+        c.error(reason);
 
         // then
         assertTrue(c.isClosedForReceive());
         assertTrue(c.isClosedForSend());
+        assertEquals(new ChannelError(reason, c), c.receiveOrClosed());
     }
 
     @Test
-    void testClosed_noValues_whenDone() {
+    void testClosed_noValues_whenDone() throws InterruptedException {
         // given
         Channel<Integer> c = new Channel<>();
 
@@ -34,6 +35,7 @@ public class ChannelClosedTest {
         // then
         assertTrue(c.isClosedForReceive());
         assertTrue(c.isClosedForSend());
+        assertEquals(new ChannelDone(c), c.receiveOrClosed());
     }
 
     @Test
diff --git a/channels/src/test/java/com/softwaremill/jox/ChannelRendezvousTest.java b/channels/src/test/java/com/softwaremill/jox/ChannelRendezvousTest.java
index c33bcb4..253216f 100644
--- a/channels/src/test/java/com/softwaremill/jox/ChannelRendezvousTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/ChannelRendezvousTest.java
@@ -121,10 +121,10 @@ public class ChannelRendezvousTest {
             c.done();
 
             // then
-            assertEquals(new ChannelDone(), f.get());
+            assertEquals(new ChannelDone(c), f.get());
 
             // should be rejected immediately
-            assertEquals(new ChannelDone(), c.receiveOrClosed());
+            assertEquals(new ChannelDone(c), c.receiveOrClosed());
         });
     }
 
diff --git a/channels/src/test/java/com/softwaremill/jox/SelectReceiveTest.java b/channels/src/test/java/com/softwaremill/jox/SelectReceiveTest.java
index d41ec5b..1f69028 100644
--- a/channels/src/test/java/com/softwaremill/jox/SelectReceiveTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/SelectReceiveTest.java
@@ -10,6 +10,7 @@ import static com.softwaremill.jox.Select.select;
 import static com.softwaremill.jox.Select.selectOrClosed;
 import static com.softwaremill.jox.TestUtil.*;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class SelectReceiveTest {
     @Test
@@ -139,7 +140,7 @@ public class SelectReceiveTest {
         Object received = selectOrClosed(ch1.receiveClause(), ch2.receiveClause());
 
         // then
-        assertEquals(new ChannelDone(), received);
+        assertEquals(new ChannelDone(ch1), received);
     }
 
     @TestWithCapacities
@@ -219,12 +220,24 @@ public class SelectReceiveTest {
     }
 
     @Test
-    void testSelectFromNone() throws InterruptedException {
-        assertEquals(new ChannelDone(), selectOrClosed());
+    void testSelectFromNone() {
+        assertThrows(IllegalArgumentException.class, Select::selectOrClosed);
     }
 
     @Test
-    public void testSelect_immediate_withError() throws InterruptedException {
+    void testSelectFromNullableList() throws InterruptedException {
+        // given
+        Channel<String> ch1 = new Channel<>(1);
+        Channel<String> ch2 = new Channel<>(1);
+        ch1.send("v");
+
+        // when
+        assertThrows(IllegalArgumentException.class,
+                () -> select(ch1.receiveClause(), null, ch2.receiveClause()));
+    }
+
+    @Test
+    void testSelect_immediate_withError() throws InterruptedException {
         // given
         Channel<String> ch1 = new Channel<>(2);
         ch1.send("x");
@@ -237,6 +250,6 @@ public class SelectReceiveTest {
         var result = selectOrClosed(ch1.receiveClause(), ch2.receiveClause());
 
         // then
-        assertEquals(new ChannelError(e), result);
+        assertEquals(new ChannelError(e, ch2), result);
     }
 }
diff --git a/channels/src/test/java/com/softwaremill/jox/SelectSendTest.java b/channels/src/test/java/com/softwaremill/jox/SelectSendTest.java
index 3d139a8..91a2246 100644
--- a/channels/src/test/java/com/softwaremill/jox/SelectSendTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/SelectSendTest.java
@@ -107,7 +107,7 @@ public class SelectSendTest {
         Object received = selectOrClosed(ch1.sendClause("v1"), ch2.sendClause("v2"));
 
         // then
-        assertEquals(new ChannelDone(), received);
+        assertEquals(new ChannelDone(ch2), received);
     }
 
     @TestWithCapacities
diff --git a/flows/src/test/java/com/softwaremill/jox/flows/FlowMapTest.java b/flows/src/test/java/com/softwaremill/jox/flows/FlowMapTest.java
index 5d278da..0d14a8f 100644
--- a/flows/src/test/java/com/softwaremill/jox/flows/FlowMapTest.java
+++ b/flows/src/test/java/com/softwaremill/jox/flows/FlowMapTest.java
@@ -1,5 +1,6 @@
 package com.softwaremill.jox.flows;
 
+import com.softwaremill.jox.Channel;
 import com.softwaremill.jox.ChannelError;
 import com.softwaremill.jox.Source;
 import com.softwaremill.jox.structured.JoxScopeExecutionException;
@@ -144,7 +145,7 @@ public class FlowMapTest {
             assertEquals("b", s.receive());
             assertEquals("c", s.receive());
             var result = s.receiveOrClosed();
-            if (result instanceof ChannelError(Throwable error)) {
+            if (result instanceof ChannelError(Throwable error, Channel<?> _)) {
                 assertEquals(boom, error);
             }
             return null;