From cb3e6803dc46a6ac062c2ea66f34db09ee8ca05e Mon Sep 17 00:00:00 2001 From: Julien Ponge Date: Thu, 28 May 2026 14:31:26 +0200 Subject: [PATCH 1/2] Spread (aka flatMap) operator --- .../java/mutiny/zero/operators/Spread.java | 304 +++++++++++ .../mutiny/zero/operators/SpreadTest.java | 489 ++++++++++++++++++ .../zero/operators/tck/SpreadTckTest.java | 40 ++ 3 files changed, 833 insertions(+) create mode 100644 mutiny-zero/src/main/java/mutiny/zero/operators/Spread.java create mode 100644 mutiny-zero/src/test/java/mutiny/zero/operators/SpreadTest.java create mode 100644 mutiny-zero/src/test/java/mutiny/zero/operators/tck/SpreadTckTest.java diff --git a/mutiny-zero/src/main/java/mutiny/zero/operators/Spread.java b/mutiny-zero/src/main/java/mutiny/zero/operators/Spread.java new file mode 100644 index 0000000..81471e0 --- /dev/null +++ b/mutiny-zero/src/main/java/mutiny/zero/operators/Spread.java @@ -0,0 +1,304 @@ +package mutiny.zero.operators; + +import static java.util.Objects.requireNonNull; + +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.Flow; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; + +import org.jspecify.annotations.Nullable; + +import mutiny.zero.internal.Helper; + +/** + * A {@link Flow.Publisher} that maps each upstream item to a {@link Flow.Publisher} and flattens the results + * with bounded concurrency. + *

+ * With {@code concurrency = 1} this behaves as a concatMap (strict ordering). + * With {@code concurrency > 1} items from different inner publishers may interleave. + * + * @param the upstream item type + * @param the downstream (flattened) item type + */ +public class Spread implements Flow.Publisher { + + private final Flow.Publisher upstream; + private final Function> mapper; + private final int concurrency; + private final int prefetch; + + /** + * Create a new spread (flatMap) publisher. + * + * @param upstream the upstream publisher, must not be {@code null} + * @param mapper the mapping function, must not be {@code null} + * @param concurrency the maximum number of concurrent inner subscriptions, must be positive + * @param prefetch the per-inner request batch size, must be positive + */ + public Spread(Flow.Publisher upstream, Function> mapper, int concurrency, int prefetch) { + this.upstream = requireNonNull(upstream, "The upstream publisher cannot be null"); + this.mapper = requireNonNull(mapper, "The mapper function cannot be null"); + if (concurrency <= 0) { + throw new IllegalArgumentException("Concurrency must be positive, was " + concurrency); + } + if (prefetch <= 0) { + throw new IllegalArgumentException("Prefetch must be positive, was " + prefetch); + } + this.concurrency = concurrency; + this.prefetch = prefetch; + } + + @Override + public void subscribe(Flow.Subscriber subscriber) { + requireNonNull(subscriber, "The subscriber cannot be null"); + SpreadProcessor processor = new SpreadProcessor(subscriber); + subscriber.onSubscribe(processor); + upstream.subscribe(processor); + } + + private class SpreadProcessor implements Flow.Subscriber, Flow.Subscription { + + private final Flow.Subscriber downstream; + private volatile Flow.@Nullable Subscription upstreamSubscription; + + private final AtomicBoolean cancelled = new AtomicBoolean(); + private final AtomicLong demand = new AtomicLong(); + private final AtomicInteger wip = new AtomicInteger(); + private final AtomicInteger activeCount = new AtomicInteger(); + + private final ConcurrentLinkedQueue queue = new ConcurrentLinkedQueue<>(); + private final ConcurrentLinkedQueue innerSubscribers = new ConcurrentLinkedQueue<>(); + + private final AtomicReference<@Nullable Throwable> error = new AtomicReference<>(); + private volatile boolean upstreamDone; + + SpreadProcessor(Flow.Subscriber downstream) { + this.downstream = downstream; + } + + // ---- Flow.Subscriber (upstream) ---- // + + @Override + public void onSubscribe(Flow.Subscription subscription) { + this.upstreamSubscription = subscription; + if (cancelled.get()) { + subscription.cancel(); + return; + } + subscription.request(concurrency); + } + + @Override + public void onNext(T item) { + if (cancelled.get() || error.get() != null) { + return; + } + Flow.Publisher inner; + try { + inner = mapper.apply(item); + } catch (Throwable t) { + onError(t); + return; + } + if (inner == null) { + onError(new NullPointerException("The mapper returned a null publisher")); + return; + } + activeCount.incrementAndGet(); + InnerSubscriber innerSubscriber = new InnerSubscriber(); + innerSubscribers.offer(innerSubscriber); + inner.subscribe(innerSubscriber); + } + + @Override + public void onError(Throwable throwable) { + if (error.compareAndSet(null, throwable)) { + upstreamDone = true; + Flow.Subscription us = upstreamSubscription; + if (us != null) { + us.cancel(); + } + cancelAllInners(); + drain(); + } + } + + @Override + public void onComplete() { + upstreamDone = true; + drain(); + } + + // ---- Flow.Subscription (downstream) ---- // + + @Override + public void request(long n) { + if (cancelled.get()) { + return; + } + if (n <= 0L) { + onError(Helper.negativeRequest(n)); + } else { + Helper.add(demand, n); + drain(); + } + } + + @Override + public void cancel() { + if (cancelled.compareAndSet(false, true)) { + Flow.Subscription us = upstreamSubscription; + if (us != null) { + us.cancel(); + } + cancelAllInners(); + queue.clear(); + } + } + + // ---- Drain loop ---- // + + private void drain() { + if (wip.getAndIncrement() != 0) { + return; + } + int missed = 1; + while (true) { + if (cancelled.get()) { + queue.clear(); + return; + } + + Throwable err = error.get(); + if (err != null) { + queue.clear(); + cancelled.set(true); + downstream.onError(err); + return; + } + + long emitted = 0L; + long currentDemand = demand.get(); + + while (emitted < currentDemand) { + if (cancelled.get()) { + queue.clear(); + return; + } + err = error.get(); + if (err != null) { + queue.clear(); + cancelled.set(true); + downstream.onError(err); + return; + } + R item = queue.poll(); + if (item == null) { + break; + } + downstream.onNext(item); + emitted++; + } + + if (emitted > 0L) { + if (currentDemand != Long.MAX_VALUE) { + demand.addAndGet(-emitted); + } + } + + if (cancelled.get()) { + queue.clear(); + return; + } + + if (upstreamDone && activeCount.get() == 0 && queue.isEmpty()) { + err = error.get(); + cancelled.set(true); + if (err != null) { + downstream.onError(err); + } else { + downstream.onComplete(); + } + return; + } + + missed = wip.addAndGet(-missed); + if (missed == 0) { + return; + } + } + } + + private void cancelAllInners() { + InnerSubscriber inner; + while ((inner = innerSubscribers.poll()) != null) { + inner.cancel(); + } + } + + // ---- Inner subscriber ---- // + + private class InnerSubscriber implements Flow.Subscriber { + + private volatile Flow.@Nullable Subscription subscription; + private int produced; + private final int limit = prefetch - (prefetch >> 2); + + @Override + public void onSubscribe(Flow.Subscription s) { + this.subscription = s; + if (cancelled.get() || error.get() != null) { + s.cancel(); + return; + } + s.request(prefetch); + } + + @Override + public void onNext(R item) { + if (cancelled.get() || error.get() != null) { + return; + } + queue.offer(item); + produced++; + if (produced >= limit) { + Flow.Subscription s = subscription; + if (s != null) { + s.request(produced); + } + produced = 0; + } + drain(); + } + + @Override + public void onError(Throwable throwable) { + SpreadProcessor.this.onError(throwable); + } + + @Override + public void onComplete() { + activeCount.decrementAndGet(); + innerSubscribers.remove(this); + if (!upstreamDone && !cancelled.get()) { + Flow.Subscription us = upstreamSubscription; + if (us != null) { + us.request(1); + } + } + drain(); + } + + void cancel() { + Flow.Subscription s = subscription; + if (s != null) { + s.cancel(); + } + } + } + } +} diff --git a/mutiny-zero/src/test/java/mutiny/zero/operators/SpreadTest.java b/mutiny-zero/src/test/java/mutiny/zero/operators/SpreadTest.java new file mode 100644 index 0000000..8978b76 --- /dev/null +++ b/mutiny-zero/src/test/java/mutiny/zero/operators/SpreadTest.java @@ -0,0 +1,489 @@ +package mutiny.zero.operators; + +import static org.junit.jupiter.api.Assertions.*; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Flow; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.RepeatedTest; +import org.junit.jupiter.api.Test; + +import io.smallrye.mutiny.helpers.test.AssertSubscriber; +import mutiny.zero.ZeroPublisher; + +@DisplayName("Tests for the Spread operator") +class SpreadTest { + + // ---- Basic functionality ---- // + + @Test + @DisplayName("Spread with concurrency=1 preserves order (concatMap)") + void concatMapOrdering() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2, 3); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems(i + "a", i + "b"), 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted().assertItems("1a", "1b", "2a", "2b", "3a", "3b"); + } + + @Test + @DisplayName("Spread with concurrency > 1 delivers all items") + void concurrentSpreadDeliversAll() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2, 3); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems(i + "a", i + "b"), 4, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted(); + List items = sub.getItems(); + assertEquals(6, items.size()); + assertTrue(items.containsAll(List.of("1a", "1b", "2a", "2b", "3a", "3b"))); + } + + @Test + @DisplayName("Empty upstream completes immediately") + void emptyUpstream() { + Flow.Publisher upstream = ZeroPublisher.empty(); + Spread spread = new Spread<>(upstream, ZeroPublisher::fromItems, 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted().assertHasNotReceivedAnyItem(); + } + + @Test + @DisplayName("Empty inner publishers produce no items") + void emptyInnerPublishers() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2, 3); + Spread spread = new Spread<>(upstream, i -> ZeroPublisher.empty(), 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted().assertHasNotReceivedAnyItem(); + } + + @Test + @DisplayName("Mix of empty and non-empty inner publishers") + void mixedEmptyAndNonEmpty() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2, 3); + Spread spread = new Spread<>(upstream, + i -> (i % 2 == 0) ? ZeroPublisher.empty() : ZeroPublisher.fromItems(i + "x"), 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted().assertItems("1x", "3x"); + } + + // ---- Backpressure ---- // + + @Test + @DisplayName("Paced demand (request one at a time)") + void pacedDemand() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems(i + "a", i + "b"), 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(); + spread.subscribe(sub); + + sub.request(1); + sub.assertItems("1a"); + + sub.request(1); + sub.assertItems("1a", "1b"); + + sub.request(1); + sub.assertItems("1a", "1b", "2a"); + + sub.request(1); + sub.assertItems("1a", "1b", "2a", "2b"); + + sub.request(1); + sub.assertCompleted(); + } + + @Test + @DisplayName("Unbounded demand (Long.MAX_VALUE)") + void unboundedDemand() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2, 3); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems(i * 10, i * 100), 2, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted(); + assertEquals(6, sub.getItems().size()); + } + + // ---- Error handling ---- // + + @Test + @DisplayName("Mapper function throws") + void mapperThrows() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2, 3); + Spread spread = new Spread<>(upstream, + i -> { + if (i == 2) { + throw new RuntimeException("mapper boom"); + } + return ZeroPublisher.fromItems(i); + }, 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertFailedWith(RuntimeException.class, "mapper boom"); + } + + @SuppressWarnings("DataFlowIssue") + @Test + @DisplayName("Mapper returns null") + void mapperReturnsNull() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1); + Spread spread = new Spread<>(upstream, i -> null, 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertFailedWith(NullPointerException.class, null); + } + + @Test + @DisplayName("Inner publisher error propagates downstream") + void innerPublisherError() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2); + Spread spread = new Spread<>(upstream, + i -> { + if (i == 2) { + return ZeroPublisher.fromFailure(new IOException("inner boom")); + } + return ZeroPublisher.fromItems(i); + }, 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertFailedWith(IOException.class, "inner boom"); + } + + @Test + @DisplayName("Upstream error cancels inners and propagates") + void upstreamError() { + Flow.Publisher upstream = ZeroPublisher.fromFailure(new IOException("upstream boom")); + Spread spread = new Spread<>(upstream, ZeroPublisher::fromItems, 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertFailedWith(IOException.class, "upstream boom"); + } + + // ---- Cancellation ---- // + + @Test + @DisplayName("Cancel propagates to upstream") + void cancelPropagates() { + AtomicBoolean upstreamCancelled = new AtomicBoolean(); + Flow.Publisher upstream = subscriber -> subscriber.onSubscribe(new Flow.Subscription() { + @Override + public void request(long n) { + // no-op + } + + @Override + public void cancel() { + upstreamCancelled.set(true); + } + }); + + Spread spread = new Spread<>(upstream, + ZeroPublisher::fromItems, 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + sub.cancel(); + + assertTrue(upstreamCancelled.get()); + } + + @RepeatedTest(1000) + @DisplayName("Concurrent cancel vs request does not throw") + void concurrentCancelVsRequest() throws InterruptedException { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2, 3, 4, 5); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems(i, i * 10), 2, 8); + + AssertSubscriber sub = AssertSubscriber.create(); + spread.subscribe(sub); + + CountDownLatch latch = new CountDownLatch(2); + Thread t1 = new Thread(() -> { + sub.request(10); + latch.countDown(); + }); + Thread t2 = new Thread(() -> { + sub.cancel(); + latch.countDown(); + }); + t1.start(); + t2.start(); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + t1.join(5000); + t2.join(5000); + } + + // ---- Concurrency limits ---- // + + @Test + @DisplayName("Concurrency=1 is sequential") + void concurrencyOneIsSequential() { + AtomicInteger maxConcurrent = new AtomicInteger(); + AtomicInteger currentConcurrent = new AtomicInteger(); + + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2, 3); + Spread spread = new Spread<>(upstream, + i -> subscriber -> { + int c = currentConcurrent.incrementAndGet(); + maxConcurrent.updateAndGet(prev -> Math.max(prev, c)); + subscriber.onSubscribe(new Flow.Subscription() { + @Override + public void request(long n) { + subscriber.onNext(i); + currentConcurrent.decrementAndGet(); + subscriber.onComplete(); + } + + @Override + public void cancel() { + } + }); + }, 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted(); + assertEquals(1, maxConcurrent.get(), "Max concurrent inner subscriptions should be 1"); + } + + // ---- Validation ---- // + + @SuppressWarnings("DataFlowIssue") + @Test + @DisplayName("Reject null upstream") + void rejectNullUpstream() { + assertThrows(NullPointerException.class, () -> new Spread<>(null, i -> ZeroPublisher.empty(), 1, 8)); + } + + @SuppressWarnings("DataFlowIssue") + @Test + @DisplayName("Reject null mapper") + void rejectNullMapper() { + assertThrows(NullPointerException.class, + () -> new Spread<>(ZeroPublisher.empty(), null, 1, 8)); + } + + @Test + @DisplayName("Reject concurrency <= 0") + void rejectZeroConcurrency() { + assertThrows(IllegalArgumentException.class, + () -> new Spread<>(ZeroPublisher.empty(), i -> ZeroPublisher.empty(), 0, 8)); + } + + @Test + @DisplayName("Reject prefetch <= 0") + void rejectZeroPrefetch() { + assertThrows(IllegalArgumentException.class, + () -> new Spread<>(ZeroPublisher.empty(), i -> ZeroPublisher.empty(), 1, 0)); + } + + // ---- Additional tests from audit ---- // + + @Test + @DisplayName("Negative request signals error") + void negativeRequest() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1); + Spread spread = new Spread<>(upstream, + ZeroPublisher::fromItems, 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(); + spread.subscribe(sub); + + sub.request(-1); + sub.assertFailedWith(IllegalArgumentException.class, "non-positive"); + } + + @Test + @DisplayName("Prefetch=1 works correctly") + void prefetchOne() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems(i + "a", i + "b", i + "c"), 1, 1); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted().assertItems("1a", "1b", "1c", "2a", "2b", "2c"); + } + + @Test + @DisplayName("Cancel propagates to inner subscribers") + void cancelPropagatestoInners() { + AtomicBoolean innerCancelled = new AtomicBoolean(); + Flow.Publisher upstream = ZeroPublisher.fromItems(1); + Spread spread = new Spread<>(upstream, + i -> subscriber -> subscriber.onSubscribe(new Flow.Subscription() { + @Override + public void request(long n) { + subscriber.onNext(i); + } + + @Override + public void cancel() { + innerCancelled.set(true); + } + }), 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(1); + spread.subscribe(sub); + sub.cancel(); + + assertTrue(innerCancelled.get()); + } + + @Test + @DisplayName("Mapper error with concurrency > 1 cancels active inners") + void mapperErrorWithConcurrency() { + AtomicBoolean innerCancelled = new AtomicBoolean(); + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2); + Spread spread = new Spread<>(upstream, + i -> { + if (i == 2) { + throw new RuntimeException("boom"); + } + return subscriber -> subscriber.onSubscribe(new Flow.Subscription() { + @Override + public void request(long n) { + subscriber.onNext(i); + } + + @Override + public void cancel() { + innerCancelled.set(true); + } + }); + }, 4, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertFailedWith(RuntimeException.class, "boom"); + assertTrue(innerCancelled.get()); + } + + @Test + @DisplayName("Resubscription creates independent streams") + void resubscription() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1, 2); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems(i * 10), 1, 8); + + AssertSubscriber sub1 = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub1); + sub1.assertCompleted().assertItems(10, 20); + + AssertSubscriber sub2 = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub2); + sub2.assertCompleted().assertItems(10, 20); + } + + @Test + @DisplayName("Cancel inside onNext does not deliver onComplete") + void cancelInsideOnNextNoComplete() { + Flow.Publisher upstream = ZeroPublisher.fromItems(1); + Spread spread = new Spread<>(upstream, ZeroPublisher::fromItems, 1, 8); + + AtomicBoolean completeCalled = new AtomicBoolean(); + AtomicBoolean errorCalled = new AtomicBoolean(); + Flow.Subscriber sub = new Flow.Subscriber<>() { + Flow.Subscription subscription; + + @Override + public void onSubscribe(Flow.Subscription s) { + this.subscription = s; + s.request(1); + } + + @Override + public void onNext(Integer item) { + subscription.cancel(); + } + + @Override + public void onError(Throwable throwable) { + errorCalled.set(true); + } + + @Override + public void onComplete() { + completeCalled.set(true); + } + }; + spread.subscribe(sub); + + assertFalse(completeCalled.get(), "onComplete must not be called after cancel"); + assertFalse(errorCalled.get(), "onError must not be called after cancel"); + } + + @Test + @DisplayName("Single item upstream with single item inner") + void singleItemSingleInner() { + Flow.Publisher upstream = ZeroPublisher.fromItems(42); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems("val-" + i), 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted().assertItems("val-42"); + } + + @Test + @DisplayName("Large number of items with concurrency=1") + void manyItemsConcatMap() { + int count = 100; + Integer[] items = new Integer[count]; + for (int i = 0; i < count; i++) { + items[i] = i; + } + Flow.Publisher upstream = ZeroPublisher.fromItems(items); + Spread spread = new Spread<>(upstream, + i -> ZeroPublisher.fromItems(i, i + 1000), 1, 8); + + AssertSubscriber sub = AssertSubscriber.create(Long.MAX_VALUE); + spread.subscribe(sub); + + sub.assertCompleted(); + assertEquals(200, sub.getItems().size()); + assertEquals(0, sub.getItems().get(0)); + assertEquals(1000, sub.getItems().get(1)); + assertEquals(1, sub.getItems().get(2)); + assertEquals(1001, sub.getItems().get(3)); + } +} diff --git a/mutiny-zero/src/test/java/mutiny/zero/operators/tck/SpreadTckTest.java b/mutiny-zero/src/test/java/mutiny/zero/operators/tck/SpreadTckTest.java new file mode 100644 index 0000000..8b90a51 --- /dev/null +++ b/mutiny-zero/src/test/java/mutiny/zero/operators/tck/SpreadTckTest.java @@ -0,0 +1,40 @@ +package mutiny.zero.operators.tck; + +import java.util.Random; +import java.util.concurrent.Flow; + +import org.reactivestreams.tck.TestEnvironment; +import org.reactivestreams.tck.flow.FlowPublisherVerification; + +import io.smallrye.mutiny.Multi; +import mutiny.zero.ZeroPublisher; +import mutiny.zero.operators.Spread; + +public class SpreadTckTest extends FlowPublisherVerification { + + public SpreadTckTest() { + super(new TestEnvironment()); + } + + @Override + public Flow.Publisher createFlowPublisher(long count) { + Flow.Publisher source; + if (count > 0) { + Random random = new Random(); + source = Multi.createBy().repeating().supplier(random::nextLong).atMost(count); + } else { + source = Multi.createFrom().empty(); + } + return new Spread<>(source, ZeroPublisher::fromItems, 2, 8); + } + + @Override + public Flow.Publisher createFailedFlowPublisher() { + return null; + } + + @Override + public long maxElementsFromPublisher() { + return 1024; + } +} From 435b76a1d21ff73a81d7cf66308eadf5e2a8a1b2 Mon Sep 17 00:00:00 2001 From: Julien Ponge Date: Thu, 28 May 2026 14:33:17 +0200 Subject: [PATCH 2/2] Build PRs on Java 17 and 25 --- .github/workflows/build-pull-request.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/build-pull-request.yml b/.github/workflows/build-pull-request.yml index 3fc437a..041df45 100644 --- a/.github/workflows/build-pull-request.yml +++ b/.github/workflows/build-pull-request.yml @@ -9,7 +9,7 @@ jobs: matrix: java: [ { 'version': '17', 'opts': '' }, - { 'version': '23', 'opts': '' } + { 'version': '25', 'opts': '' } ] name: Build with Java ${{ matrix.java.version }} (OpenJDK) steps: