This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 3d9e723e22 =str Fix maybe throw for `MinimalStage`. (#822)
3d9e723e22 is described below
commit 3d9e723e228af18c900570f827ecf17628557343
Author: kerr <[email protected]>
AuthorDate: Sat Dec 2 17:42:25 2023 +0800
=str Fix maybe throw for `MinimalStage`. (#822)
---
.../pekko/stream/javadsl/FlowUnfoldAsyncTest.java | 57 ++++++++++++++++++++++
.../org/apache/pekko/stream/impl/Unfold.scala | 26 +++++-----
2 files changed, 69 insertions(+), 14 deletions(-)
diff --git
a/stream-tests/src/test/java-jdk9-only/org/apache/pekko/stream/javadsl/FlowUnfoldAsyncTest.java
b/stream-tests/src/test/java-jdk9-only/org/apache/pekko/stream/javadsl/FlowUnfoldAsyncTest.java
new file mode 100644
index 0000000000..88219f372b
--- /dev/null
+++
b/stream-tests/src/test/java-jdk9-only/org/apache/pekko/stream/javadsl/FlowUnfoldAsyncTest.java
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.stream.javadsl;
+
+import org.apache.pekko.japi.Pair;
+import org.apache.pekko.stream.StreamTest;
+import org.apache.pekko.testkit.PekkoJUnitActorSystemResource;
+import org.apache.pekko.testkit.PekkoSpec;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+public class FlowUnfoldAsyncTest extends StreamTest {
+ @ClassRule
+ public static PekkoJUnitActorSystemResource actorSystemResource =
+ new PekkoJUnitActorSystemResource("SourceTest", PekkoSpec.testConf());
+
+ public FlowUnfoldAsyncTest() {
+ super(actorSystemResource);
+ }
+
+ @Test
+ public void testFoldAsync() throws Exception {
+ final Integer result = Source.unfoldAsync(
+ 0,
+ idx -> {
+ if (idx >= 10) {
+ return
CompletableFuture.completedStage(Optional.empty());
+ } else {
+ return
CompletableFuture.completedStage(Optional.of(Pair.create(idx + 1, idx)));
+ }
+ })
+ .runFold(0, Integer::sum, system)
+ .toCompletableFuture()
+ .get(3, TimeUnit.SECONDS);
+ Assert.assertEquals(45, result.intValue());
+ }
+}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala
index bd12267f8e..ea6a525ec4 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/impl/Unfold.scala
@@ -21,7 +21,6 @@ import pekko.stream.impl.Stages.DefaultAttributes
import pekko.stream.stage.{ GraphStage, GraphStageLogic, OutHandler }
import java.util.Optional
-import java.util.concurrent.CompletableFuture
import java.util.concurrent.CompletionStage
import scala.concurrent.Future
import scala.util.{ Failure, Success, Try }
@@ -124,21 +123,20 @@ import scala.util.{ Failure, Success, Try }
}
def onPull(): Unit = {
- f.apply(state) match {
- case cf: CompletableFuture[Optional[Pair[S, E]] @unchecked] if
cf.isDone && !cf.isCompletedExceptionally =>
- handle(cf.join())
- case future =>
- future.handle((r, ex) => {
- if (ex != null) {
- asyncHandler(Failure(ex))
- } else {
- asyncHandler(Success(r))
- }
- null
- })
+ val future = f.apply(state).toCompletableFuture
+ if (future.isDone && !future.isCompletedExceptionally) {
+ handle(future.getNow(null))
+ } else {
+ future.handle((r, ex) => {
+ if (ex != null) {
+ asyncHandler(Failure(ex))
+ } else {
+ asyncHandler(Success(r))
+ }
+ null
+ })
}
}
-
setHandler(out, this)
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
