This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch fix/source-with-context-pair-shape in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 0e96e19ca9ca95b657d3023d17caea7e13f56401 Author: 虎鸣 <[email protected]> AuthorDate: Sun Jul 26 18:14:37 2026 +0800 fix: Java DSL SourceWithContext exposes Pair-typed shape in GraphDSL Motivation: javadsl.SourceWithContext delegated to scaladsl.SourceWithContext whose shape is SourceShape[(Out, Ctx)] (Scala Tuple2). Java developers expect SourceShape[Pair[Out, Ctx]] when using it in GraphDSL (#2510). Modification: Change the internal delegate from scaladsl.SourceWithContext to javadsl.Source[Pair[Out, Ctx], Mat], matching the pattern already used by javadsl.FlowWithContext (which does not have this bug). Simplify via, asSource, to, and toMat methods accordingly. Result: SourceWithContext now exposes a Pair-typed SourceShape in GraphDSL, consistent with FlowWithContext and Java developer expectations. Tests: - sbt "stream-tests / Test / testOnly org.apache.pekko.stream.javadsl.SourceWithContextTest" - sbt "stream-tests / Test / testOnly org.apache.pekko.stream.javadsl.FlowWithContextTest" - sbt "stream-tests / Test / testOnly org.apache.pekko.stream.scaladsl.SourceWithContextSpec" - sbt "stream/mimaReportBinaryIssues" (with new filter) - All passed References: Fixes #2510 --- .../stream/javadsl/SourceWithContextTest.java | 22 ++++++++++++++++++++++ .../source-with-context-pair-shape.excludes | 19 +++++++++++++++++++ .../pekko/stream/javadsl/SourceWithContext.scala | 22 ++++++++++++---------- .../pekko/stream/scaladsl/SourceWithContext.scala | 3 ++- 4 files changed, 55 insertions(+), 11 deletions(-) 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..f15c223b95 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 @@ -23,8 +23,10 @@ import java.time.Duration; import java.util.List; import java.util.Optional; import java.util.concurrent.TimeUnit; +import org.apache.pekko.NotUsed; import org.apache.pekko.japi.Pair; import org.apache.pekko.japi.pf.PFBuilder; +import org.apache.pekko.stream.SourceShape; import org.apache.pekko.stream.StreamTestJupiter; import org.apache.pekko.testkit.PekkoJUnitJupiterActorSystemResource; import org.apache.pekko.testkit.PekkoSpec; @@ -155,4 +157,24 @@ public class SourceWithContextTest extends StreamTestJupiter { assertEquals(List.of(new Pair<>(1, "one")), timedResult); } + + @Test + public void mustExposePairTypedShapeInGraphDsl() throws Exception { + SourceWithContext<Integer, String, NotUsed> sourceWithContext = + Source.single(1).asSourceWithContext(elem -> "ctx-" + elem); + + List<Pair<Integer, String>> result = + Source.fromGraph( + GraphDSL.create( + sourceWithContext, + (builder, in) -> { + SourceShape<Pair<Integer, String>> shape = in; + return SourceShape.of(shape.out()); + })) + .runWith(Sink.seq(), system) + .toCompletableFuture() + .get(3, TimeUnit.SECONDS); + + assertEquals(List.of(new Pair<>(1, "ctx-1")), result); + } } diff --git a/stream/src/main/mima-filters/2.0.x.backwards.excludes/source-with-context-pair-shape.excludes b/stream/src/main/mima-filters/2.0.x.backwards.excludes/source-with-context-pair-shape.excludes new file mode 100644 index 0000000000..ed5b70e707 --- /dev/null +++ b/stream/src/main/mima-filters/2.0.x.backwards.excludes/source-with-context-pair-shape.excludes @@ -0,0 +1,19 @@ +# 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. + +# Fix Java DSL SourceWithContext shape to use Pair instead of Tuple2 (#2510) +ProblemFilters.exclude[IncompatibleMethTypeProblem]("org.apache.pekko.stream.javadsl.SourceWithContext.this") 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..ed6ffe4ce7 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 @@ -40,7 +40,7 @@ object SourceWithContext { * Creates a SourceWithContext from a regular flow that operates on `Pair<data, context>` elements. */ def fromPairs[Out, CtxOut, Mat](under: Source[Pair[Out, CtxOut], Mat]): SourceWithContext[Out, CtxOut, Mat] = { - new SourceWithContext(scaladsl.SourceWithContext.fromTuples(under.asScala.map(_.toScala))) + new SourceWithContext(under) } /** @@ -84,8 +84,10 @@ object SourceWithContext { * * Can be created by calling [[Source.asSourceWithContext]] */ -final class SourceWithContext[+Out, +Ctx, +Mat](delegate: scaladsl.SourceWithContext[Out, Ctx, Mat]) - extends GraphDelegate(delegate) { +final class SourceWithContext[+Out, +Ctx, +Mat]( + delegate: Source[Pair[Out @uncheckedVariance, Ctx @uncheckedVariance], Mat @uncheckedVariance]) + extends GraphDelegate[SourceShape[Pair[Out @uncheckedVariance, Ctx @uncheckedVariance]], Mat @uncheckedVariance]( + delegate) { /** * Transform this flow by the regular flow. The given flow must support manual context propagation by @@ -103,8 +105,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 +153,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 + delegate // remaining operations in alphabetic order @@ -565,7 +566,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 +575,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. @@ -596,10 +597,11 @@ final class SourceWithContext[+Out, +Ctx, +Mat](delegate: scaladsl.SourceWithCon materializer: Materializer): M = toMat(sink, Keep.right[Mat, M]).run(materializer) - def asScala: scaladsl.SourceWithContext[Out, Ctx, Mat] = delegate + def asScala: scaladsl.SourceWithContext[Out, Ctx, Mat] = + scaladsl.SourceWithContext.fromTuples(delegate.asScala.map(_.toScala)) private[this] def viaScala[Out2, Ctx2, Mat2]( f: scaladsl.SourceWithContext[Out, Ctx, Mat] => scaladsl.SourceWithContext[Out2, Ctx2, Mat2]) : SourceWithContext[Out2, Ctx2, Mat2] = - new SourceWithContext(f(delegate)) + f(asScala).asJava } diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/SourceWithContext.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/SourceWithContext.scala index e97b206696..7786cfbe12 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/SourceWithContext.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/SourceWithContext.scala @@ -17,6 +17,7 @@ import scala.annotation.unchecked.uncheckedVariance import org.apache.pekko import pekko.annotation.ApiMayChange +import pekko.japi.Pair import pekko.stream._ object SourceWithContext { @@ -196,5 +197,5 @@ final class SourceWithContext[+Out, +Ctx, +Mat] private[stream] (delegate: Sourc def asSource: Source[(Out, Ctx), Mat] = delegate def asJava[JOut >: Out, JCtx >: Ctx, JMat >: Mat]: javadsl.SourceWithContext[JOut, JCtx, JMat] = - new javadsl.SourceWithContext(this) + new javadsl.SourceWithContext(delegate.map { case (o, c) => Pair[JOut, JCtx](o, c) }.asJava) } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
