WIP

core/pom.xml

@@ -22,7 +22,9 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>3.11.0</version>
                 <configuration>
-                    <enablePreview>true</enablePreview>
+                    <compilerArgs>
+                        <arg>--enable-preview</arg>
+                    </compilerArgs>
                 </configuration>
             </plugin>
             <plugin>
@@ -42,5 +44,11 @@
             <version>5.10.1</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>4.2.0</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

core/src/test/java/jox/ChannelInterruptionTest.java

@@ -2,7 +2,15 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static jox.TestUtil.*;
+import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
 public class ChannelInterruptionTest {
@@ -65,4 +73,44 @@ void testRaceInterruptAndSend() throws Exception {
             }
         });
     }
+
+    @Test
+    void testReceiveManyInterruptsReceive() throws ExecutionException, InterruptedException {
+        scoped(scope -> {
+            Channel<String> channel = new Channel<>();
+            Set<String> received = ConcurrentHashMap.newKeySet();
+
+            // starting with a single receive
+            forkVoid(scope, () -> {
+                received.add(channel.receive());
+            });
+            // wait for the `receive` to suspend
+            Thread.sleep(100);
+
+            // then, starting subsequent receives, and interrupting them
+            for (int i = 0; i < 1000; i++) {
+                var s = new Semaphore(0);
+                var f = forkCancelable(scope, () -> {
+                    s.release(); // letting the main thread know that the receive has started
+                    channel.receive();
+                });
+                s.acquire();
+                f.cancel();
+            }
+
+            // then, starting one more receive
+            forkVoid(scope, () -> {
+                received.add(channel.receive());
+            });
+            // wait for the `receive` to suspend
+            Thread.sleep(100);
+
+            // send two elements, wait for completion
+            channel.send("a");
+            channel.send("b");
+
+            // check the results
+            await().atMost(1, SECONDS).until(() -> received.equals(Set.of("a", "b")));
+        });
+    }
 }