From d313eef0a99b7f88d348d7df679f824313666f6c Mon Sep 17 00:00:00 2001
From: "He-Pin(kerr)" <hepin1989@gmail.com>
Date: Tue, 16 Jan 2024 10:13:29 +0800
Subject: [PATCH] chore: Add collect operator test for stream javadsl. (#963)

---
 .../org/apache/pekko/stream/javadsl/FlowTest.java  | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

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.<Integer, Integer>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);