diff --git a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java index fce0a461e2..aa5f08a65d 100644 --- a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java +++ b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/FlowTest.java @@ -948,6 +948,20 @@ public class FlowTest extends StreamTest { probe.expectMsgEquals("2C"); } + @Test + public void mustBeAbleToUseCollect() { + Source.from(Arrays.asList(1, 2, 3, 4, 5)) + .collect( + PFBuilder.create() + .match(Integer.class, elem -> elem % 2 != 0, elem -> elem) + .build()) + .runWith(TestSink.create(system), system) + .ensureSubscription() + .request(5) + .expectNext(1, 3, 5) + .expectComplete(); + } + @Test public void mustBeAbleToUseCollectType() throws Exception { final TestKit probe = new TestKit(system);