You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2026/07/26 15:52:48 UTC
(pekko) branch main updated: fix: expose Pair shape for Java SourceWithContext (#3388)
This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 663b9a9a2f fix: expose Pair shape for Java SourceWithContext (#3388)
663b9a9a2f is described below
commit 663b9a9a2fbbdfb590d9cfff374a823ca91c01dd
Author: He-Pin(kerr) <he...@gmail.com>
AuthorDate: Sun Jul 26 23:52:42 2026 +0800
fix: expose Pair shape for Java SourceWithContext (#3388)
Motivation:
Java GraphDSL sees a tuple-shaped outlet from SourceWithContext, so it cannot connect the source to Pair-typed Java stages.
Modification:
Keep the existing Scala delegate and public constructor, and expose a Pair-typed graph view only at the Java graph boundary. Reuse and document that view for Java graph operations, add migration guidance, and add directional shape, materialized-value, and exact graph-size tests.
Result:
Java GraphDSL can consume SourceWithContext directly without binary compatibility filters or repeated conversion stages in operator chains.
Tests:
- sbt focused SourceWithContext and adjacent context-stream tests - passed, 45 tests
- sbt stream MiMa on Scala 2.13 and Scala 3 - passed
- sbt +headerCheckAll and checkCodeStyle - passed
- sbt docs / paradox - passed
- scalafmt, javafmtCheckAll, and git diff --check - passed
- sbt +mimaReportBinaryIssues - Scala 2.13 passed; the full task then hit pre-existing remote Scala 3 cyclic compilation errors
- sbt validatePullRequest - stopped after an x86-only leveldbjni binary failed to load on arm64
References:
Fixes #2510
---
.../paradox/migration/migration-guide-1.x-2.x.md | 5 ++++
.../stream/javadsl/SourceWithContextTest.java | 30 ++++++++++++++++++++++
.../stream/scaladsl/SourceWithContextSpec.scala | 24 +++++++++++++++++
.../pekko/stream/javadsl/SourceWithContext.scala | 14 +++++-----
4 files changed, 67 insertions(+), 6 deletions(-)
diff --git a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
index f19565324c..4b7ebd32a1 100644
--- a/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
+++ b/docs/src/main/paradox/migration/migration-guide-1.x-2.x.md
@@ -37,3 +37,8 @@ list before a second param list. Instead of `watchTermination(){ ... }`, you now
In the Java API, `FSMTransitionHandlerBuilder.build` and `FSMTransitionHandlerBuilder.state` now use
`pekko.japi.Pair` instead of `scala.Tuple2`. Java users need to update their code to use `Pair` instead of `Tuple2`.
([PR3378](https://github.com/apache/pekko/pull/3378))
+
+The Java DSL `SourceWithContext` graph shape now uses `pekko.japi.Pair` instead of `scala.Tuple2`.
+Java code that passes a `SourceWithContext` directly to `GraphDSL` must use `Pair`-typed stages.
+Code that converts it with `asSource()` already uses `Pair` and requires no changes.
+([PR3388](https://github.com/apache/pekko/pull/3388))
diff --git a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceWithContextTest.java b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceWithContextTest.java
index ea47272e6f..82bbb21708 100644
--- a/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceWithContextTest.java
+++ b/stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SourceWithContextTest.java
@@ -22,9 +22,11 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import java.time.Duration;
import java.util.List;
import java.util.Optional;
+import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import org.apache.pekko.japi.Pair;
import org.apache.pekko.japi.pf.PFBuilder;
+import org.apache.pekko.stream.ClosedShape;
import org.apache.pekko.stream.StreamTestJupiter;
import org.apache.pekko.testkit.PekkoJUnitJupiterActorSystemResource;
import org.apache.pekko.testkit.PekkoSpec;
@@ -155,4 +157,32 @@ public class SourceWithContextTest extends StreamTestJupiter {
assertEquals(List.of(new Pair<>(1, "one")), timedResult);
}
+
+ @Test
+ public void mustConnectPairTypedOutletToJavaSinkInGraphDsl() throws Exception {
+ SourceWithContext<Integer, String, String> contextSource =
+ Source.from(List.of(3, 5))
+ .mapMaterializedValue(ignored -> "upstream-control")
+ .asSourceWithContext(value -> "request-" + value)
+ .map(value -> value * 10);
+ Sink<Pair<Integer, String>, CompletionStage<List<Pair<Integer, String>>>> pairSink = Sink.seq();
+
+ RunnableGraph<Pair<String, CompletionStage<List<Pair<Integer, String>>>>> graph =
+ RunnableGraph.fromGraph(
+ GraphDSL.create(
+ contextSource,
+ pairSink,
+ Keep.both(),
+ (builder, sourceShape, sinkShape) -> {
+ builder.from(sourceShape).to(sinkShape);
+ return ClosedShape.getInstance();
+ }));
+
+ Pair<String, CompletionStage<List<Pair<Integer, String>>>> materialized = graph.run(system);
+
+ assertEquals("upstream-control", materialized.first());
+ assertEquals(
+ List.of(new Pair<>(30, "request-3"), new Pair<>(50, "request-5")),
+ materialized.second().toCompletableFuture().get(3, TimeUnit.SECONDS));
+ }
}
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceWithContextSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceWithContextSpec.scala
index 363a0133cc..a649144fb2 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceWithContextSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SourceWithContextSpec.scala
@@ -20,6 +20,8 @@ import scala.jdk.CollectionConverters._
import scala.util.control.NoStackTrace
import org.apache.pekko
+import pekko.NotUsed
+import pekko.stream.impl.TraversalTestUtils
import pekko.stream.testkit.StreamSpec
import pekko.stream.testkit.scaladsl.TestSink
@@ -64,6 +66,28 @@ class SourceWithContextSpec extends StreamSpec {
.expectComplete()
}
+ "add only the requested operators to the Java DSL graph" in {
+ def moduleCount[T](source: Source[T, NotUsed]): Int =
+ TraversalTestUtils
+ .testMaterialize(source.to(Sink.ignore).traversalBuilder)
+ .attributesAssignments
+ .size
+
+ val source = Source.single(1).asSourceWithContext(_.toString).asJava
+ val transformed = source
+ .map((value: Int) => value + 1)
+ .map((value: Int) => value + 1)
+ .map((value: Int) => value + 1)
+
+ val scalaModules = moduleCount(source.asScala.asSource)
+ val sourceModules = moduleCount(source.asSource().asScala)
+ val transformedModules = moduleCount(transformed.asSource().asScala)
+ withClue(s"scala=$scalaModules, java=$sourceModules, transformed=$transformedModules: ") {
+ sourceModules - scalaModules shouldBe 1
+ transformedModules - sourceModules shouldBe 3
+ }
+ }
+
"pass through contexts using map and filter" in {
Source(Vector(Message("A", 1L), Message("B", 2L), Message("D", 3L), Message("C", 4L)))
.asSourceWithContext(_.offset)
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SourceWithContext.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SourceWithContext.scala
index e46bff7ae0..ece8cdeb21 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SourceWithContext.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SourceWithContext.scala
@@ -82,10 +82,13 @@ object SourceWithContext {
* use [[SourceWithContext#via]] to manually provide the context propagation for otherwise unsupported
* operations.
*
+ * When used as a [[Graph]], its outlet emits [[pekko.japi.Pair]] elements, matching [[SourceWithContext#asSource]].
+ *
* Can be created by calling [[Source.asSourceWithContext]]
*/
final class SourceWithContext[+Out, +Ctx, +Mat](delegate: scaladsl.SourceWithContext[Out, Ctx, Mat])
- extends GraphDelegate(delegate) {
+ extends GraphDelegate[SourceShape[Pair[Out @uncheckedVariance, Ctx @uncheckedVariance]], Mat @uncheckedVariance](
+ delegate.asSource.map { case (out, ctx) => Pair(out, ctx) }.asJava) {
/**
* Transform this flow by the regular flow. The given flow must support manual context propagation by
@@ -103,8 +106,7 @@ final class SourceWithContext[+Out, +Ctx, +Mat](delegate: scaladsl.SourceWithCon
def via[Out2, Ctx2, Mat2](
viaFlow: Graph[FlowShape[Pair[Out @uncheckedVariance, Ctx @uncheckedVariance], Pair[Out2, Ctx2]], Mat2])
: SourceWithContext[Out2, Ctx2, Mat] =
- viaScala(_.via(pekko.stream.scaladsl.Flow[(Out, Ctx)].map { case (o, c) => Pair(o, c) }.via(viaFlow).map(
- _.toScala)))
+ SourceWithContext.fromPairs(asSource().via(viaFlow))
/**
* Transform this flow by the regular flow. The given flow works on the data portion of the stream and
@@ -152,7 +154,7 @@ final class SourceWithContext[+Out, +Ctx, +Mat](delegate: scaladsl.SourceWithCon
* stream of a pair of (data, context).
*/
def asSource(): Source[Pair[Out @uncheckedVariance, Ctx @uncheckedVariance], Mat @uncheckedVariance] =
- delegate.asSource.map { case (o, c) => Pair(o, c) }.asJava
+ Source.fromGraph(this)
// remaining operations in alphabetic order
@@ -565,7 +567,7 @@ final class SourceWithContext[+Out, +Ctx, +Mat](delegate: scaladsl.SourceWithCon
*/
def to[Mat2](
sink: Graph[SinkShape[Pair[Out @uncheckedVariance, Ctx @uncheckedVariance]], Mat2]): javadsl.RunnableGraph[Mat] =
- RunnableGraph.fromGraph(asScala.asSource.map { case (o, e) => Pair(o, e) }.to(sink))
+ asSource().to(sink)
/**
* Connect this [[pekko.stream.javadsl.SourceWithContext]] to a [[pekko.stream.javadsl.Sink]],
@@ -574,7 +576,7 @@ final class SourceWithContext[+Out, +Ctx, +Mat](delegate: scaladsl.SourceWithCon
def toMat[Mat2, Mat3](
sink: Graph[SinkShape[Pair[Out @uncheckedVariance, Ctx @uncheckedVariance]], Mat2],
combine: function.Function2[Mat, Mat2, Mat3]): javadsl.RunnableGraph[Mat3] =
- RunnableGraph.fromGraph(asScala.asSource.map { case (o, e) => Pair(o, e) }.toMat(sink)(combinerToScala(combine)))
+ asSource().toMat(sink, combine)
/**
* Connect this [[pekko.stream.javadsl.SourceWithContext]] to a [[pekko.stream.javadsl.Sink]] and run it.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org