Implement non-blocking trySend and tryReceive methods (#262)

## Summary

- Implement non-blocking `trySend`/`tryReceive` (and `OrClosed`
variants) on `Channel`, `Sink`, and `Source` for integration with
non-blocking frameworks (Netty, Vert.x, etc.)
- Add optimized implementations in `Channel` using CAS-based cell
reservation with pre-check optimization to avoid unnecessary CAS
attempts
- Update downstream usages in `FromFlowPublisher`, `KafkaDrain`, and
`KafkaStage` to use `trySend` instead of blocking `send`
- Add `Channel.estimateSize()` for monitoring and observability
- Add Fray deterministic concurrency tests and comprehensive unit tests
covering all channel types
- Add contention caveat to Javadoc and `channels.md` documentation

Resolves #187

---------

Co-authored-by: Andrzej Kobyliński <andrzej.kobylinski@extern.corify.de>
Co-authored-by: adamw <adam@warski.org>

Author: 3 people

.gitignore

@@ -4,6 +4,7 @@ target
 .cursor
 tmp
 .vscode
+.metals
 
 # virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
 hs_err_pid*

channels-fray-tests/src/test/java/com/softwaremill/jox/fray/FrayTrySendReceiveTest.java

@@ -0,0 +1,150 @@
+package com.softwaremill.jox.fray;
+
+import static com.softwaremill.jox.fray.Config.CHANNEL_SIZE;
+
+import java.util.ArrayList;
+
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.pastalab.fray.junit.junit5.FrayTestExtension;
+import org.pastalab.fray.junit.junit5.annotations.ConcurrencyTest;
+
+import com.softwaremill.jox.Channel;
+import com.softwaremill.jox.ChannelDone;
+
+@ExtendWith(FrayTestExtension.class)
+public class FrayTrySendReceiveTest {
+
+    // trySend | receive
+
+    @ConcurrencyTest
+    public void trySendReceiveTest() throws InterruptedException {
+        Channel<Integer> ch = Channel.newBufferedChannel(CHANNEL_SIZE);
+
+        Fork<Void> f1 =
+                Fork.newNoResult(
+                        () -> {
+                            if (!ch.trySend(10)) {
+                                ch.send(10);
+                            }
+                        });
+        Fork<Integer> f2 = Fork.newWithResult(ch::receive);
+
+        Fork.startAll(f1, f2);
+        f1.join();
+
+        assert (f2.join() == 10);
+    }
+
+    // send | tryReceive
+
+    @ConcurrencyTest
+    public void sendTryReceiveTest() throws InterruptedException {
+        Channel<Integer> ch = Channel.newBufferedChannel(CHANNEL_SIZE);
+
+        Fork<Void> f1 = Fork.newNoResult(() -> ch.send(10));
+        Fork<Integer> f2 =
+                Fork.newWithResult(
+                        () -> {
+                            Integer result = ch.tryReceive();
+                            if (result != null) return result;
+                            return ch.receive();
+                        });
+
+        Fork.startAll(f1, f2);
+        f1.join();
+
+        assert (f2.join() == 10);
+    }
+
+    // trySend | tryReceive
+
+    @ConcurrencyTest
+    public void trySendTryReceiveTest() throws InterruptedException {
+        Channel<Integer> ch = Channel.newBufferedChannel(CHANNEL_SIZE);
+
+        Fork<Boolean> f1 = Fork.newWithResult(() -> ch.trySend(10));
+        Fork<Integer> f2 = Fork.newWithResult(ch::tryReceive);
+
+        Fork.startAll(f1, f2);
+        boolean sent = f1.join();
+        Integer received = f2.join();
+
+        if (received != null) {
+            assert sent;
+            assert (received == 10);
+        }
+    }
+
+    // multiple trySend | multiple tryReceive
+
+    @ConcurrencyTest
+    public void multiTrySendMultiTryReceiveTest() throws InterruptedException {
+        Channel<Integer> ch = Channel.newBufferedChannel(CHANNEL_SIZE);
+
+        int concurrency = 4;
+
+        var sendForks = new ArrayList<Fork<Void>>();
+        var receiveForks = new ArrayList<Fork<Integer>>();
+
+        for (int i = 0; i < concurrency; i++) {
+            final var finalI = i;
+            sendForks.add(
+                    Fork.newNoResult(
+                            () -> {
+                                if (!ch.trySend(finalI)) {
+                                    ch.send(finalI);
+                                }
+                            }));
+            receiveForks.add(
+                    Fork.newWithResult(
+                            () -> {
+                                Integer result = ch.tryReceive();
+                                if (result != null) return result;
+                                return ch.receive();
+                            }));
+        }
+
+        Fork.startAll(sendForks.toArray(new Fork<?>[0]));
+        Fork.startAll(receiveForks.toArray(new Fork<?>[0]));
+
+        for (Fork<Void> sendFork : sendForks) {
+            sendFork.join();
+        }
+
+        var result = 0;
+        for (Fork<Integer> receiveFork : receiveForks) {
+            result += receiveFork.join();
+        }
+
+        assert (result == concurrency * (concurrency - 1) / 2);
+    }
+
+    // trySend | close | tryReceive
+
+    @ConcurrencyTest
+    public void trySendCloseTryReceiveTest() throws InterruptedException {
+        Channel<Integer> ch = Channel.newBufferedChannel(CHANNEL_SIZE);
+
+        Fork<Boolean> f1 =
+                Fork.newWithResult(
+                        () -> {
+                            Object r = ch.trySendOrClosed(10);
+                            return r == null;
+                        });
+
+        Fork<Void> f2 = Fork.newNoResult(ch::done);
+
+        Fork<Object> f3 = Fork.newWithResult(ch::tryReceiveOrClosed);
+
+        Fork.startAll(f1, f2, f3);
+
+        boolean sent = f1.join();
+        f2.join();
+        Object received = f3.join();
+
+        if (received != null && !(received instanceof ChannelDone)) {
+            assert (received.equals(10));
+            assert sent;
+        }
+    }
+}