Repository: softwaremill/jox
Pull request #95 (2025-02-04 09:54:33)

Make channel constructor private, add explicit factory methods
The motivation is to make it crystal clear what kind of buffer the channel is using. The previous `new Channel()` which created a rendezvous channel might be misleading, as it in most cases rendezvous channels are *not* what you want, and are slower than buffered channels. The new design encourages usage of buffered-default channels.


Patch:
diff --git a/README.md b/README.md
index 6191efc..d91829c 100644
--- a/README.md
+++ b/README.md
@@ -70,7 +70,7 @@ class Demo1 {
     public static void main(String[] args) throws InterruptedException {
         // creates a rendezvous channel
         // (a sender & receiver must meet to pass a value: as if the buffer had size 0)
-        var ch = new Channel<Integer>();
+        var ch = Channel.<Integer>newRendezvousChannel();
 
         Thread.ofVirtual().start(() -> {
             try {
@@ -100,8 +100,7 @@ import com.softwaremill.jox.Channel;
 
 class Demo2 {
     public static void main(String[] args) throws InterruptedException {
-        // creates a buffered channel (buffer of size 3)
-        var ch = new Channel<Integer>(3);
+        var ch = Channel.<Integer>newBufferedChannel(3);
 
         // send()-s won't block
         ch.send(1);
@@ -158,8 +157,7 @@ import com.softwaremill.jox.Channel;
 
 class Demo3 {
     public static void main(String[] args) throws InterruptedException {
-        // creates a buffered channel (buffer of size 3)
-        var ch = new Channel<Integer>(3);
+        var ch = Channel.<Integer>newBufferedChannel(3);
 
         // send()-s won't block
         ch.send(1);
@@ -186,9 +184,9 @@ import static com.softwaremill.jox.Select.select;
 class Demo4 {
     public static void main(String[] args) throws InterruptedException {
         // creates a buffered channel (buffer of size 3)
-        var ch1 = new Channel<Integer>(3);
-        var ch2 = new Channel<Integer>(3);
-        var ch3 = new Channel<Integer>(3);
+        var ch1 = Channel.<Integer>newBufferedChannel(3);
+        var ch2 = Channel.<Integer>newBufferedChannel(3);
+        var ch3 = Channel.<Integer>newBufferedChannel(3);
 
         // send a value to two channels
         ch2.send(29);
@@ -218,8 +216,8 @@ import static com.softwaremill.jox.Select.select;
 
 class Demo5 {
     public static void main(String[] args) throws InterruptedException {
-        var ch1 = new Channel<Integer>(1);
-        var ch2 = new Channel<Integer>(1);
+        var ch1 = Channel.<Integer>newBufferedChannel(1);
+        var ch2 = Channel.<Integer>newBufferedChannel(1);
 
         ch1.send(12); // buffer is now full
 
@@ -242,8 +240,8 @@ import static com.softwaremill.jox.Select.select;
 
 class Demo6 {
     public static void main(String[] args) throws InterruptedException {
-        var ch1 = new Channel<Integer>(3);
-        var ch2 = new Channel<Integer>(3);
+        var ch1 = Channel.<Integer>newBufferedChannel(3);
+        var ch2 = Channel.<Integer>newBufferedChannel(3);
 
         var received = select(ch1.receiveClause(), ch2.receiveClause(), defaultClause(52));
 
@@ -613,7 +611,7 @@ import com.softwaremill.jox.structured.Scopes;
 public class Demo {
 
     public static void main(String[] args) throws ExecutionException, InterruptedException {
-        Channel<Integer> ch = new Channel<>(16);
+        Channel<Integer> ch = Channel.<Integer>newBufferedDefaultChannel();
         Scopes.supervised(scope -> {
             scope.fork(() -> {
                 ch.send(1);
diff --git a/bench/bench-java/src/main/java/com/softwaremill/jox/BufferedBenchmark.java b/bench/bench-java/src/main/java/com/softwaremill/jox/BufferedBenchmark.java
index dfde643..3c39a35 100644
--- a/bench/bench-java/src/main/java/com/softwaremill/jox/BufferedBenchmark.java
+++ b/bench/bench-java/src/main/java/com/softwaremill/jox/BufferedBenchmark.java
@@ -55,7 +55,7 @@ public class BufferedBenchmark {
     @Benchmark
     @OperationsPerInvocation(OPERATIONS_PER_INVOCATION)
     public void channel() throws InterruptedException {
-        var ch = new Channel<>(capacity);
+        var ch = Channel.newBufferedChannel(capacity);
         var t1 = Thread.startVirtualThread(() -> {
             for (int i = 0; i < OPERATIONS_PER_INVOCATION; i++) {
                 try {
diff --git a/bench/bench-java/src/main/java/com/softwaremill/jox/ChainedBenchmark.java b/bench/bench-java/src/main/java/com/softwaremill/jox/ChainedBenchmark.java
index 32c3b96..7ff65c8 100644
--- a/bench/bench-java/src/main/java/com/softwaremill/jox/ChainedBenchmark.java
+++ b/bench/bench-java/src/main/java/com/softwaremill/jox/ChainedBenchmark.java
@@ -32,7 +32,7 @@ public class ChainedBenchmark {
         int elements = OPERATIONS_PER_INVOCATION / chainLength;
         Channel<Integer>[] channels = new Channel[chainLength];
         for (int i = 0; i < chainLength; i++) {
-            channels[i] = new Channel<>(capacity);
+            channels[i] = (capacity == 0) ? Channel.newRendezvousChannel() : Channel.newBufferedChannel(capacity);
         }
 
         Thread[] threads = new Thread[chainLength + 1];
diff --git a/bench/bench-java/src/main/java/com/softwaremill/jox/ParallelBenchmark.java b/bench/bench-java/src/main/java/com/softwaremill/jox/ParallelBenchmark.java
index 21ea030..734e289 100644
--- a/bench/bench-java/src/main/java/com/softwaremill/jox/ParallelBenchmark.java
+++ b/bench/bench-java/src/main/java/com/softwaremill/jox/ParallelBenchmark.java
@@ -32,7 +32,7 @@ public class ParallelBenchmark {
         var latch = new CountDownLatch(parallelism);
 
         for (int t = 0; t < parallelism; t++) {
-            var ch = new Channel<Integer>(capacity);
+            var ch = (capacity == 0) ? Channel.<Integer>newRendezvousChannel() : Channel.<Integer>newBufferedChannel(capacity);
             // sender
             Thread.startVirtualThread(() -> {
                 for (int i = 0; i < elementsPerChannel; i++) {
diff --git a/bench/bench-java/src/main/java/com/softwaremill/jox/RendezvousBenchmark.java b/bench/bench-java/src/main/java/com/softwaremill/jox/RendezvousBenchmark.java
index 60998d6..3650760 100644
--- a/bench/bench-java/src/main/java/com/softwaremill/jox/RendezvousBenchmark.java
+++ b/bench/bench-java/src/main/java/com/softwaremill/jox/RendezvousBenchmark.java
@@ -76,7 +76,7 @@ public class RendezvousBenchmark {
     @Benchmark
     @OperationsPerInvocation(OPERATIONS_PER_INVOCATION)
     public void channel() throws InterruptedException {
-        var ch = new Channel<>();
+        var ch = Channel.newRendezvousChannel();
         var t1 = Thread.startVirtualThread(() -> {
             for (int i = 0; i < OPERATIONS_PER_INVOCATION; i++) {
                 try {
diff --git a/bench/bench-java/src/main/java/com/softwaremill/jox/SelectBenchmark.java b/bench/bench-java/src/main/java/com/softwaremill/jox/SelectBenchmark.java
index 5baebc4..72dd564 100644
--- a/bench/bench-java/src/main/java/com/softwaremill/jox/SelectBenchmark.java
+++ b/bench/bench-java/src/main/java/com/softwaremill/jox/SelectBenchmark.java
@@ -20,7 +20,7 @@ public class SelectBenchmark {
     @Benchmark
     @OperationsPerInvocation(OPERATIONS_PER_INVOCATION)
     public void selectWithSingleClause() throws InterruptedException {
-        var ch = new Channel<Integer>();
+        var ch = Channel.newRendezvousChannel();
         var t1 = Thread.startVirtualThread(() -> {
             for (int i = 0; i < OPERATIONS_PER_INVOCATION; i++) {
                 try {
@@ -48,8 +48,8 @@ public class SelectBenchmark {
     @Benchmark
     @OperationsPerInvocation(OPERATIONS_PER_INVOCATION)
     public void selectWithTwoClauses() throws InterruptedException {
-        var ch1 = new Channel<Integer>();
-        var ch2 = new Channel<Integer>();
+        var ch1 = Channel.newRendezvousChannel();
+        var ch2 = Channel.newRendezvousChannel();
         var t1 = Thread.startVirtualThread(() -> {
             for (int i = 0; i < OPERATIONS_PER_INVOCATION / 2; i++) {
                 try {
diff --git a/channels/src/main/java/com/softwaremill/jox/Channel.java b/channels/src/main/java/com/softwaremill/jox/Channel.java
index fe04c96..0cbdf8f 100644
--- a/channels/src/main/java/com/softwaremill/jox/Channel.java
+++ b/channels/src/main/java/com/softwaremill/jox/Channel.java
@@ -24,9 +24,9 @@ import static com.softwaremill.jox.Segment.findAndMoveForward;
  * - buffered channels, where a given number of sent values might be buffered, before subsequent `send`s block
  * - unlimited channels, where an unlimited number of values might be buffered, hence `send` never blocks
  * <p>
- * The no-argument {@link Channel} constructor creates a rendezvous channel, while a buffered channel can be created
- * by providing a positive integer to the constructor. A rendezvous channel behaves like a buffered channel with
- * buffer size 0. An unlimited channel can be created using {@link Channel#newUnlimitedChannel()}.
+ * To create a channel, use {@link Channel#newRendezvousChannel()}, {@link Channel#newBufferedChannel(int)} and
+ * {@link Channel#newUnlimitedChannel()} methods. Additionally, {@link Channel#newBufferedDefaultChannel()} creates
+ * a buffered channel with a "default" capacity of 16, which should be a good starting point for most use-cases.
  * <p>
  * In a rendezvous channel, senders and receivers block, until a matching party arrives (unless one is already waiting).
  * Similarly, buffered channels block if the buffer is full (in case of senders), or in case of receivers, if the
@@ -152,17 +152,10 @@ public final class Channel<T> implements Source<T>, Sink<T> {
         }
     }
 
-    /**
-     * Creates a rendezvous channel.
-     */
-    public Channel() {
-        this(0);
-    }
-
     /**
      * Creates a buffered channel (when capacity is positive), or a rendezvous channel if the capacity is 0.
      */
-    public Channel(int capacity) {
+    private Channel(int capacity) {
         if (capacity < UNLIMITED_CAPACITY) {
             throw new IllegalArgumentException("Capacity must be 0 (rendezvous), positive (buffered) or -1 (unlimited channels).");
         }
@@ -205,6 +198,21 @@ public final class Channel<T> implements Source<T>, Sink<T> {
         }
     }
 
+    public static <T> Channel<T> newRendezvousChannel() {
+        return new Channel<>(0);
+    }
+
+    public static <T> Channel<T> newBufferedChannel(int capacity) {
+        return new Channel<>(capacity);
+    }
+
+    /**
+     * Creates a new buffered channel, with the default buffer size (16).
+     */
+    public static <T> Channel<T> newBufferedDefaultChannel() {
+        return new Channel<>(DEFAULT_BUFFER_SIZE);
+    }
+
     public static <T> Channel<T> newUnlimitedChannel() {
         return new Channel<>(UNLIMITED_CAPACITY);
     }
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 56f9689..5ca1c59 100644
--- a/flows/src/main/java/com/softwaremill/jox/flows/Flow.java
+++ b/flows/src/main/java/com/softwaremill/jox/flows/Flow.java
@@ -108,7 +108,7 @@ public class Flow<T> {
      * @param bufferCapacity Specifies buffer capacity of created channel
      */
     public Source<T> runToChannel(UnsupervisedScope scope, int bufferCapacity) {
-        return runToChannelInternal(scope, () -> new Channel<>(bufferCapacity));
+        return runToChannelInternal(scope, () -> Channel.newBufferedChannel(bufferCapacity));
     }
 
     private Source<T> runToChannelInternal(UnsupervisedScope scope, Supplier<Channel<T>> channelProvider) {
@@ -251,7 +251,7 @@ public class Flow<T> {
      */
     public Flow<T> buffer(int bufferCapacity) {
         return usingEmit(emit -> {
-            Channel<T> ch = new Channel<>(bufferCapacity);
+            Channel<T> ch = Channel.newBufferedChannel(bufferCapacity);
             unsupervised(scope -> {
                 runLastToChannelAsync(scope, ch);
                 FlowEmit.channelToEmit(ch, emit);
@@ -1257,7 +1257,7 @@ public class Flow<T> {
     public <U> Flow<U> mapPar(int parallelism, ThrowingFunction<T, U> f) {
         return usingEmit(emit -> {
             Semaphore semaphore = new Semaphore(parallelism);
-            Channel<Fork<Optional<U>>> inProgress = new Channel<>(parallelism);
+            Channel<Fork<Optional<U>>> inProgress = Channel.newBufferedChannel(parallelism);
             Channel<U> results = Channel.withScopedBufferSize();
 
             // creating a nested scope, so that in case of errors, we can clean up any mapping forks in a "local" fashion,
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 5c6f94d..a4ea83c 100644
--- a/flows/src/main/java/com/softwaremill/jox/flows/Flows.java
+++ b/flows/src/main/java/com/softwaremill/jox/flows/Flows.java
@@ -431,7 +431,7 @@ public final class Flows {
             return flows.getFirst();
         } else {
             return usingEmit(emit -> {
-                Channel<T> results = new Channel<>(bufferCapacity);
+                Channel<T> results = Channel.newBufferedChannel(bufferCapacity);
                 unsupervised(scope -> {
                     scope.forkUnsupervised(() -> {
                         List<Source<T>> availableSources = new ArrayList<>(flows.stream()


Test patch:
diff --git a/channels/src/test/java/com/softwaremill/jox/ChannelBufferedTest.java b/channels/src/test/java/com/softwaremill/jox/ChannelBufferedTest.java
index b723063..b2c21d9 100644
--- a/channels/src/test/java/com/softwaremill/jox/ChannelBufferedTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/ChannelBufferedTest.java
@@ -17,7 +17,7 @@ public class ChannelBufferedTest {
     @Timeout(1)
     void testSimpleSendReceiveBuffer1() throws InterruptedException {
         // given
-        Channel<String> channel = new Channel<>(1);
+        Channel<String> channel = Channel.newBufferedChannel(1);
 
         // when
         channel.send("x"); // should not block
@@ -31,7 +31,7 @@ public class ChannelBufferedTest {
     @Timeout(1)
     void testSimpleSendReceiveBuffer2() throws InterruptedException {
         // given
-        Channel<String> channel = new Channel<>(2);
+        Channel<String> channel = Channel.newBufferedChannel(2);
 
         // when
         channel.send("x"); // should not block
@@ -48,7 +48,7 @@ public class ChannelBufferedTest {
     @Timeout(2)
     void testBufferCapacityStaysTheSameAfterSendsReceives() throws ExecutionException, InterruptedException {
         // given
-        Channel<Integer> channel = new Channel<>(2);
+        Channel<Integer> channel = Channel.newBufferedChannel(2);
 
         // when
         scoped(scope -> {
@@ -78,7 +78,7 @@ public class ChannelBufferedTest {
     @Timeout(1)
     void shouldReceiveFromAChannelUntilDone() throws InterruptedException {
         // given
-        Channel<Integer> c = new Channel<>(3);
+        Channel<Integer> c = Channel.newBufferedChannel(3);
         c.send(1);
         c.send(2);
         c.done();
@@ -98,7 +98,7 @@ public class ChannelBufferedTest {
     @Timeout(1)
     void shouldNotReceiveFromAChannelInCaseOfAnError() throws InterruptedException {
         // given
-        Channel<Integer> c = new Channel<>(3);
+        Channel<Integer> c = Channel.newBufferedChannel(3);
         c.send(1);
         c.send(2);
         c.error(new RuntimeException());
@@ -114,8 +114,8 @@ public class ChannelBufferedTest {
 
     @Test
     void shouldProcessCellsInitially() {
-        assertTrue(new Channel<String>(1).toString().contains("notProcessed=31"));
-        assertTrue(new Channel<String>(31).toString().contains("notProcessed=1"));
-        assertTrue(new Channel<String>(32).toString().contains("notProcessed=0"));
+        assertTrue(Channel.<String>newBufferedChannel(1).toString().contains("notProcessed=31"));
+        assertTrue(Channel.<String>newBufferedChannel(31).toString().contains("notProcessed=1"));
+        assertTrue(Channel.<String>newBufferedChannel(32).toString().contains("notProcessed=0"));
     }
 }
diff --git a/channels/src/test/java/com/softwaremill/jox/ChannelClosedTest.java b/channels/src/test/java/com/softwaremill/jox/ChannelClosedTest.java
index 1d7b03e..8a8878d 100644
--- a/channels/src/test/java/com/softwaremill/jox/ChannelClosedTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/ChannelClosedTest.java
@@ -12,7 +12,7 @@ public class ChannelClosedTest {
     @Test
     void testClosed_noValues_whenError() throws InterruptedException {
         // given
-        Channel<Integer> c = new Channel<>();
+        Channel<Integer> c = Channel.newRendezvousChannel();
         RuntimeException reason = new RuntimeException();
 
         // when
@@ -27,7 +27,7 @@ public class ChannelClosedTest {
     @Test
     void testClosed_noValues_whenDone() throws InterruptedException {
         // given
-        Channel<Integer> c = new Channel<>();
+        Channel<Integer> c = Channel.newRendezvousChannel();
 
         // when
         c.done();
@@ -41,7 +41,7 @@ public class ChannelClosedTest {
     @Test
     void testClosed_hasSuspendedValues_whenDone() throws InterruptedException, ExecutionException {
         // given
-        Channel<Integer> c = new Channel<>();
+        Channel<Integer> c = Channel.newRendezvousChannel();
 
         // when
         scoped(scope -> {
@@ -65,7 +65,7 @@ public class ChannelClosedTest {
     @Test
     void testClosed_hasBufferedValues_whenDone() throws InterruptedException {
         // given
-        Channel<Integer> c = new Channel<>(5);
+        Channel<Integer> c = Channel.newBufferedChannel(5);
 
         // when
         c.send(1);
@@ -80,7 +80,7 @@ public class ChannelClosedTest {
     @Test
     void testClosed_hasValues_whenError() throws InterruptedException {
         // given
-        Channel<Integer> c = new Channel<>(5);
+        Channel<Integer> c = Channel.newBufferedChannel(5);
 
         // when
         c.send(1);
diff --git a/channels/src/test/java/com/softwaremill/jox/ChannelInterruptionTest.java b/channels/src/test/java/com/softwaremill/jox/ChannelInterruptionTest.java
index 3e1976e..259a582 100644
--- a/channels/src/test/java/com/softwaremill/jox/ChannelInterruptionTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/ChannelInterruptionTest.java
@@ -16,7 +16,7 @@ public class ChannelInterruptionTest {
     @Test
     void testSendReceiveAfterSendInterrupt() throws Exception {
         // given
-        Channel<String> channel = new Channel<>();
+        Channel<String> channel = Channel.newRendezvousChannel();
 
         // when
         scoped(scope -> {
@@ -34,7 +34,7 @@ public class ChannelInterruptionTest {
     @Test
     void testSendReceiveAfterReceiveInterrupt() throws Exception {
         // given
-        Channel<String> channel = new Channel<>();
+        Channel<String> channel = Channel.newRendezvousChannel();
 
         // when
         scoped(scope -> {
@@ -55,7 +55,7 @@ public class ChannelInterruptionTest {
         scoped(scope -> {
             for (int i = 0; i < 100; i++) {
                 // given
-                Channel<String> channel = new Channel<>();
+                Channel<String> channel = Channel.newRendezvousChannel();
 
                 var t1 = forkCancelable(scope, () -> channel.send("x"));
                 var t2 = fork(scope, channel::receive);
@@ -76,7 +76,7 @@ public class ChannelInterruptionTest {
     @Test
     void testReceiveManyInterruptsReceive() throws ExecutionException, InterruptedException {
         scoped(scope -> {
-            Channel<String> channel = new Channel<>();
+            Channel<String> channel = Channel.newRendezvousChannel();
             Set<String> received = ConcurrentHashMap.newKeySet();
 
             // starting with a single receive
@@ -115,7 +115,7 @@ public class ChannelInterruptionTest {
 
     @Test
     void testManyInterruptedReceivesShouldNotLeakMemory() throws InterruptedException, ExecutionException {
-        var ch = new Channel<String>();
+        var ch = Channel.newRendezvousChannel();
 
         scoped(scope -> {
             var forks = new Fork[300];
@@ -138,7 +138,7 @@ public class ChannelInterruptionTest {
 
     @Test
     void testManyInterruptedSendsShouldNotLeakMemory() throws InterruptedException, ExecutionException {
-        var ch = new Channel<String>(1);
+        var ch = Channel.<String>newBufferedChannel(1);
         ch.send("x");
 
         scoped(scope -> {
diff --git a/channels/src/test/java/com/softwaremill/jox/ChannelRendezvousTest.java b/channels/src/test/java/com/softwaremill/jox/ChannelRendezvousTest.java
index 06bb24c..924593e 100644
--- a/channels/src/test/java/com/softwaremill/jox/ChannelRendezvousTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/ChannelRendezvousTest.java
@@ -19,7 +19,7 @@ public class ChannelRendezvousTest {
     @Test
     void testSimpleSendReceive() throws InterruptedException, ExecutionException {
         // given
-        Channel<String> channel = new Channel<>();
+        Channel<String> channel = Channel.newRendezvousChannel();
 
         // when
         scoped(scope -> {
@@ -34,7 +34,7 @@ public class ChannelRendezvousTest {
     @Test
     void testSendReceiveInManyForks() throws ExecutionException, InterruptedException {
         // given
-        Channel<Integer> channel = new Channel<>();
+        Channel<Integer> channel = Channel.newRendezvousChannel();
         var fs = new HashSet<Future<Void>>();
         var s = new ConcurrentSkipListSet<Integer>();
 
@@ -59,7 +59,7 @@ public class ChannelRendezvousTest {
     @Test
     void testSendReceiveManyElementsInTwoForks() throws ExecutionException, InterruptedException {
         // given
-        Channel<Integer> channel = new Channel<>();
+        Channel<Integer> channel = Channel.newRendezvousChannel();
         var s = new ConcurrentSkipListSet<Integer>();
 
         // when
@@ -83,7 +83,7 @@ public class ChannelRendezvousTest {
     @Test
     void testSendWaitsForRendezvous() throws ExecutionException, InterruptedException {
         // given
-        Channel<Integer> channel = new Channel<>();
+        Channel<Integer> channel = Channel.newRendezvousChannel();
         var trail = new ConcurrentLinkedQueue<String>();
 
         // when
@@ -116,7 +116,7 @@ public class ChannelRendezvousTest {
     @Test
     void pendingReceivesShouldGetNotifiedThatChannelIsDone() throws InterruptedException, ExecutionException {
         // given
-        Channel<Integer> c = new Channel<>();
+        Channel<Integer> c = Channel.newRendezvousChannel();
         scoped(scope -> {
             var f = fork(scope, c::receiveOrClosed);
 
@@ -135,7 +135,7 @@ public class ChannelRendezvousTest {
     @Test
     void pendingSendsShouldGetNotifiedThatChannelIsErrored() throws InterruptedException, ExecutionException {
         // given
-        Channel<Integer> c = new Channel<>();
+        Channel<Integer> c = Channel.newRendezvousChannel();
         scoped(scope -> {
             var f = fork(scope, () -> c.sendOrClosed(1));
 
@@ -156,7 +156,7 @@ public class ChannelRendezvousTest {
     void performanceTest() throws Exception {
         for (int j = 1; j <= 10; j++) {
             var max = 10_000_000L;
-            var c = new Channel<Integer>();
+            var c = Channel.<Integer>newRendezvousChannel();
             timed("rendezvous", () -> {
                 scoped(scope -> {
                     forkVoid(scope, () -> {
diff --git a/channels/src/test/java/com/softwaremill/jox/ChannelTest.java b/channels/src/test/java/com/softwaremill/jox/ChannelTest.java
index e406af3..da30c59 100644
--- a/channels/src/test/java/com/softwaremill/jox/ChannelTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/ChannelTest.java
@@ -16,7 +16,7 @@ public class ChannelTest {
     @TestWithCapacities
     void testSendReceiveInManyForks(int capacity) throws ExecutionException, InterruptedException {
         // given
-        Channel<Integer> channel = new Channel<>(capacity);
+        Channel<Integer> channel = capacity == 0 ? Channel.newRendezvousChannel() : Channel.newBufferedChannel(capacity);
         var fs = new HashSet<Future<Void>>();
         var s = new ConcurrentSkipListSet<Integer>();
 
@@ -41,7 +41,7 @@ public class ChannelTest {
     @TestWithCapacities
     void testSendReceiveManyElementsInTwoForks(int capacity) throws ExecutionException, InterruptedException {
         // given
-        Channel<Integer> channel = new Channel<>(capacity);
+        Channel<Integer> channel = capacity == 0 ? Channel.newRendezvousChannel() : Channel.newBufferedChannel(capacity);
         var s = new ConcurrentSkipListSet<Integer>();
 
         // when
diff --git a/channels/src/test/java/com/softwaremill/jox/SelectReceiveTest.java b/channels/src/test/java/com/softwaremill/jox/SelectReceiveTest.java
index 1f69028..cabb34c 100644
--- a/channels/src/test/java/com/softwaremill/jox/SelectReceiveTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/SelectReceiveTest.java
@@ -16,8 +16,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectFromFirst_buffered_immediate() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch1.send("v");
 
         // when
@@ -30,8 +30,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectFromSecond_buffered_immediate() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch2.send("v");
 
         // when
@@ -44,8 +44,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectFromFirst_buffered_suspend() throws InterruptedException, ExecutionException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
 
         scoped(scope -> {
             // when
@@ -63,8 +63,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectBiasedTowardsFirst_buffered() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch1.send("v1");
         ch2.send("v2");
 
@@ -78,8 +78,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectBiasedTowardsFirst_whenDone_buffered() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch1.send("x");
         ch2.done();
 
@@ -93,8 +93,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectFromReady_rendezvous_suspend() throws InterruptedException, ExecutionException {
         // given
-        Channel<String> ch1 = new Channel<>();
-        Channel<String> ch2 = new Channel<>();
+        Channel<String> ch1 = Channel.newRendezvousChannel();
+        Channel<String> ch2 = Channel.newRendezvousChannel();
 
         scoped(scope -> {
             forkVoid(scope, () -> {
@@ -113,8 +113,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectFromReady_rendezvous_immediate() throws InterruptedException, ExecutionException {
         // given
-        Channel<String> ch1 = new Channel<>();
-        Channel<String> ch2 = new Channel<>();
+        Channel<String> ch1 = Channel.newRendezvousChannel();
+        Channel<String> ch2 = Channel.newRendezvousChannel();
 
         scoped(scope -> {
             forkVoid(scope, () -> ch2.send("v"));
@@ -131,8 +131,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectWhenDone() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch1.done();
         ch2.send("x");
 
@@ -151,7 +151,7 @@ public class SelectReceiveTest {
 
         var channels = new ArrayList<Channel<String>>();
         for (int i = 0; i < channelsCount; i++) {
-            channels.add(new Channel<>(capacity));
+            channels.add(capacity == 0 ? Channel.newRendezvousChannel() : Channel.newBufferedChannel(capacity));
         }
 
         scoped(scope -> {
@@ -185,8 +185,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectWithTransformation() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch1.send("v");
 
         // when
@@ -199,7 +199,7 @@ public class SelectReceiveTest {
     @Test
     void testBufferExpandedWhenSelecting() throws InterruptedException {
         // given
-        Channel<String> ch = new Channel<>(2);
+        Channel<String> ch = Channel.newBufferedChannel(2);
 
         // when
         ch.send("v1");
@@ -227,8 +227,8 @@ public class SelectReceiveTest {
     @Test
     void testSelectFromNullableList() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch1.send("v");
 
         // when
@@ -239,11 +239,11 @@ public class SelectReceiveTest {
     @Test
     void testSelect_immediate_withError() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(2);
+        Channel<String> ch1 = Channel.newBufferedChannel(2);
         ch1.send("x");
 
         var e = new RuntimeException("boom!");
-        Channel<String> ch2 = new Channel<>(2);
+        Channel<String> ch2 = Channel.newBufferedChannel(2);
         ch2.error(e);
 
         // when
diff --git a/channels/src/test/java/com/softwaremill/jox/SelectSendTest.java b/channels/src/test/java/com/softwaremill/jox/SelectSendTest.java
index 91a2246..df10da7 100644
--- a/channels/src/test/java/com/softwaremill/jox/SelectSendTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/SelectSendTest.java
@@ -17,8 +17,8 @@ public class SelectSendTest {
     @Test
     public void testSelectToFirst_buffered_immediate() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
 
         // when
         select(ch1.sendClause("v1"), ch2.sendClause("v2"));
@@ -30,8 +30,8 @@ public class SelectSendTest {
     @Test
     public void testSelectToSecond_buffered_immediate() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch1.send("v0"); // filling in the buffer
 
         // when
@@ -45,8 +45,8 @@ public class SelectSendTest {
     @Test
     public void testSelectBiasedTowardsToFirst_rendezvous_immediate() throws InterruptedException, ExecutionException {
         // given
-        Channel<String> ch1 = new Channel<>();
-        Channel<String> ch2 = new Channel<>();
+        Channel<String> ch1 = Channel.newRendezvousChannel();
+        Channel<String> ch2 = Channel.newRendezvousChannel();
 
         scoped(scope -> {
             // when
@@ -64,8 +64,8 @@ public class SelectSendTest {
     @Test
     public void testSelect_rendezvous_suspend() throws InterruptedException, ExecutionException {
         // given
-        Channel<String> ch1 = new Channel<>();
-        Channel<String> ch2 = new Channel<>();
+        Channel<String> ch1 = Channel.newRendezvousChannel();
+        Channel<String> ch2 = Channel.newRendezvousChannel();
 
         scoped(scope -> {
             // when
@@ -85,8 +85,8 @@ public class SelectSendTest {
     @Test
     public void testSelectBiasedTowardsFirst_buffered() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
 
         // when
         select(ch1.sendClause("v1"), ch2.sendClause("v2"));
@@ -99,8 +99,8 @@ public class SelectSendTest {
     @Test
     public void testSelectWhenDone() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>();
-        Channel<String> ch2 = new Channel<>();
+        Channel<String> ch1 = Channel.newRendezvousChannel();
+        Channel<String> ch2 = Channel.newRendezvousChannel();
         ch2.done();
 
         // when
@@ -119,7 +119,7 @@ public class SelectSendTest {
         for (int k = 0; k < 100; k++) {
             var channels = new ArrayList<Channel<String>>();
             for (int i = 0; i < channelsCount; i++) {
-                channels.add(new Channel<>(capacity));
+                channels.add(Channel.newBufferedChannel(capacity));
             }
 
             try {
@@ -169,8 +169,8 @@ public class SelectSendTest {
     @Test
     public void testSelectWithTransformation() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
 
         // when
         String sent = select(ch1.sendClause("v1", () -> "1"), ch2.sendClause("v2", () -> "2"));
@@ -182,7 +182,7 @@ public class SelectSendTest {
     @Test
     public void testBufferExpandedWhenSelecting() throws InterruptedException {
         // given
-        Channel<String> ch = new Channel<>(2);
+        Channel<String> ch = Channel.newBufferedChannel(2);
 
         // when
         select(ch.sendClause("v1"));
diff --git a/channels/src/test/java/com/softwaremill/jox/SelectTest.java b/channels/src/test/java/com/softwaremill/jox/SelectTest.java
index b310d92..fc10eea 100644
--- a/channels/src/test/java/com/softwaremill/jox/SelectTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/SelectTest.java
@@ -11,8 +11,8 @@ public class SelectTest {
     @Test
     public void testSelectDefault() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
 
         // when
         String received = select(ch1.receiveClause(), ch2.receiveClause(), defaultClause("x"));
@@ -24,8 +24,8 @@ public class SelectTest {
     @Test
     public void testDoNotSelectDefault() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
         ch2.send("a");
 
         // when
@@ -38,8 +38,8 @@ public class SelectTest {
     @Test
     public void testDefaultCanOnlyBeLast() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<String> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<String> ch2 = Channel.newBufferedChannel(1);
 
         // when
         try {
@@ -53,8 +53,8 @@ public class SelectTest {
     @Test
     public void testSelectObject() throws InterruptedException {
         // given
-        Channel<String> ch1 = new Channel<>(1);
-        Channel<Integer> ch2 = new Channel<>(1);
+        Channel<String> ch1 = Channel.newBufferedChannel(1);
+        Channel<Integer> ch2 = Channel.newBufferedChannel(1);
         ch1.send("x");
 
         // when
diff --git a/channels/src/test/java/com/softwaremill/jox/SourceOpsForEachTest.java b/channels/src/test/java/com/softwaremill/jox/SourceOpsForEachTest.java
index 8a0da87..e262f98 100644
--- a/channels/src/test/java/com/softwaremill/jox/SourceOpsForEachTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/SourceOpsForEachTest.java
@@ -11,7 +11,7 @@ import static org.junit.jupiter.api.Assertions.assertIterableEquals;
 public class SourceOpsForEachTest {
     @Test
     void testIterateOverSource() throws Exception {
-        var c = new Channel<Integer>(10);
+        var c = Channel.<Integer>newBufferedChannel(10);
         c.sendOrClosed(1);
         c.sendOrClosed(2);
         c.sendOrClosed(3);
@@ -25,7 +25,7 @@ public class SourceOpsForEachTest {
 
     @Test
     void testConvertSourceToList() throws Exception {
-        var c = new Channel<Integer>(10);
+        var c = Channel.<Integer>newBufferedChannel(10);
         c.sendOrClosed(1);
         c.sendOrClosed(2);
         c.sendOrClosed(3);
diff --git a/channels/src/test/java/com/softwaremill/jox/StressTest.java b/channels/src/test/java/com/softwaremill/jox/StressTest.java
index bfcbbba..cd39007 100644
--- a/channels/src/test/java/com/softwaremill/jox/StressTest.java
+++ b/channels/src/test/java/com/softwaremill/jox/StressTest.java
@@ -41,7 +41,7 @@ public class StressTest {
         for (int r = 0; r < numberOfRepetitions; r++) {
             var chs = new ArrayList<Channel<String>>();
             for (int i = 0; i < numberOfChannels; i++) {
-                chs.add(new Channel<>(capacity));
+                chs.add(Channel.newBufferedChannel(capacity));
             }
             try {
                 scoped(scope -> {
diff --git a/flows/src/test/java/com/softwaremill/jox/flows/FlowAlsoToTest.java b/flows/src/test/java/com/softwaremill/jox/flows/FlowAlsoToTest.java
index 7c7ab4c..8ccbdf6 100644
--- a/flows/src/test/java/com/softwaremill/jox/flows/FlowAlsoToTest.java
+++ b/flows/src/test/java/com/softwaremill/jox/flows/FlowAlsoToTest.java
@@ -23,7 +23,7 @@ public class FlowAlsoToTest {
     @Test
     void alsoTo_shouldSendToBothSinks() throws Exception {
         // given
-        var c = new Channel<Integer>(10);
+        var c = Channel.<Integer>newBufferedChannel(10);
 
         // when
         List<Integer> result = Flows.fromValues(1, 2, 3)
@@ -39,7 +39,7 @@ public class FlowAlsoToTest {
     void alsoTo_shouldSendToBothSinksAndNotHangWhenOtherSinkIsRendezvousChannel() throws Exception {
         Scopes.supervised(scope -> {
             // given
-            var c = new Channel<Integer>();
+            var c = Channel.<Integer>newRendezvousChannel();
             var f = scope.fork(c::toList);
 
             // when
@@ -58,7 +58,7 @@ public class FlowAlsoToTest {
     void alsoTo_shouldCloseMainFlowWhenOtherCloses() throws Exception {
         Scopes.supervised(scope -> {
             // given
-            var c = new Channel<Integer>();
+            var c = Channel.<Integer>newRendezvousChannel();
             scope.fork(() -> {
                 var list = List.of(c.receiveOrClosed(), c.receiveOrClosed(), c.receiveOrClosed());
                 c.doneOrClosed();
@@ -81,7 +81,7 @@ public class FlowAlsoToTest {
     void alsoTo_shouldCloseMainFlowWithErrorWhenOtherErrors() throws Exception {
         Scopes.supervised(scope -> {
             // given
-            var c = new Channel<Integer>(1);
+            var c = Channel.<Integer>newBufferedChannel(1);
             var f = scope.fork(() -> {
                 c.receiveOrClosed();
                 c.receiveOrClosed();
@@ -105,7 +105,7 @@ public class FlowAlsoToTest {
     void alsoTo_shouldCloseOtherChannelWithErrorWhenMainErrors() throws Exception {
         Scopes.supervised(scope -> {
             // given
-            var other = new Channel<Integer>(0);
+            var other = Channel.<Integer>newRendezvousChannel();
             var forkOther = scope.forkUnsupervised(other::toList);
 
             Flow<Integer> flow = Flows.iterate(1, i -> i + 1)
@@ -123,7 +123,7 @@ public class FlowAlsoToTest {
     @Test
     void alsoToTap_shouldSendToBothSinksWhenOtherIsFaster() throws Exception {
         // given
-        var other = new Channel<Integer>(10);
+        var other = Channel.<Integer>newBufferedChannel(10);
         Flow<Integer> flow = Flows
                 .fromValues(1, 2, 3)
                 .alsoToTap(other)
@@ -138,7 +138,7 @@ public class FlowAlsoToTest {
     void alsoTapTo_shouldSendToBothSinksWhenOtherIsSlower() throws Exception {
         Scopes.supervised(scope -> {
             // given
-            var other = new Channel<Integer>();
+            var other = Channel.<Integer>newRendezvousChannel();
             var slowConsumerFork = scope.fork(() -> {
                 var consumed = new LinkedList<>();
                 while (true) {
@@ -152,7 +152,7 @@ public class FlowAlsoToTest {
                 }
                 return consumed;
             });
-            var main = new Channel<Integer>();
+            var main = Channel.<Integer>newRendezvousChannel();
             scope.fork(() -> {
                 for (int i = 1; i <= 20; i++) {
                     main.send(i);
@@ -176,7 +176,7 @@ public class FlowAlsoToTest {
     void alsoTapTo_shouldNotFailTheFlowWhenTheOtherSinkFails() throws Exception {
         Scopes.supervised(scope -> {
             // given
-            var other = new Channel<Integer>();
+            var other = Channel.<Integer>newRendezvousChannel();
             var f = scope.fork(() -> {
                 var v = other.receiveOrClosed();
                 other.error(new RuntimeException("boom!"));
@@ -202,7 +202,7 @@ public class FlowAlsoToTest {
     void alsoTapTo_shouldNotCloseTheFlowWhenTheOtherSinkCloses() throws Exception {
         Scopes.supervised(scope -> {
             // given
-            var other = new Channel<Integer>();
+            var other = Channel.<Integer>newRendezvousChannel();
             var f = scope.fork(() -> {
                 var v = other.receiveOrClosed();
                 other.done();
diff --git a/flows/src/test/java/com/softwaremill/jox/flows/FlowRunOperationsTest.java b/flows/src/test/java/com/softwaremill/jox/flows/FlowRunOperationsTest.java
index e023024..3c3fa9c 100644
--- a/flows/src/test/java/com/softwaremill/jox/flows/FlowRunOperationsTest.java
+++ b/flows/src/test/java/com/softwaremill/jox/flows/FlowRunOperationsTest.java
@@ -326,7 +326,7 @@ public class FlowRunOperationsTest {
     void runPipeToSink_pipeOneSourceToAnother() throws InterruptedException {
         Scopes.supervised(scope -> {
             Flow<Integer> c1 = Flows.fromValues(1, 2, 3);
-            Channel<Integer> c2 = new Channel<>();
+            Channel<Integer> c2 = Channel.newRendezvousChannel();
 
             scope.fork(() -> {
                 c1.runPipeToSink(c2, false);
@@ -343,7 +343,7 @@ public class FlowRunOperationsTest {
     void runPipeToSink_pipeOneSourceToAnotherWithDonePropagation() throws InterruptedException {
         Scopes.supervised(scope -> {
             Flow<Integer> c1 = Flows.fromValues(1, 2, 3);
-            Channel<Integer> c2 = new Channel<>();
+            Channel<Integer> c2 = Channel.newRendezvousChannel();
 
             scope.fork(() -> {
                 c1.runPipeToSink(c2, true);