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 2025/10/26 14:52:44 UTC
(pekko) branch main updated: chore: Remove the deprecated overloaded aggregateWithBoundary method (#2146)
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/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 8a619dc722 chore: Remove the deprecated overloaded aggregateWithBoundary method (#2146)
8a619dc722 is described below
commit 8a619dc7224ad15705680017f086b67d5f870c92
Author: He-Pin(kerr) <he...@gmail.com>
AuthorDate: Sun Oct 26 22:52:38 2025 +0800
chore: Remove the deprecated overloaded aggregateWithBoundary method (#2146)
---
.../org/apache/pekko/stream/javadsl/Flow.scala | 48 +++-------------------
.../org/apache/pekko/stream/javadsl/Source.scala | 47 +++------------------
.../org/apache/pekko/stream/javadsl/SubFlow.scala | 46 +++------------------
.../apache/pekko/stream/javadsl/SubSource.scala | 46 +++------------------
.../org/apache/pekko/stream/scaladsl/Flow.scala | 1 -
5 files changed, 24 insertions(+), 164 deletions(-)
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
index fb80771b06..35619f491f 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala
@@ -4473,40 +4473,6 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
extractContext: function.Function[Out, CtxOut]): FlowWithContext[U, CtxU, Out, CtxOut, Mat] =
this.asScala.asFlowWithContext((x: U, c: CtxU) => collapseContext.apply(x, c))(x => extractContext.apply(x)).asJava
- /**
- * Aggregate input elements into an arbitrary data structure that can be completed and emitted downstream
- * when custom condition is met which can be triggered by aggregate or timer.
- * It can be thought of a more general [[groupedWeightedWithin]].
- *
- * '''Emits when''' the aggregation function decides the aggregate is complete or the timer function returns true
- *
- * '''Backpressures when''' downstream backpressures and the aggregate is complete
- *
- * '''Completes when''' upstream completes and the last aggregate has been emitted downstream
- *
- * '''Cancels when''' downstream cancels
- *
- * @param allocate allocate the initial data structure for aggregated elements
- * @param aggregate update the aggregated elements, return true if ready to emit after update.
- * @param harvest this is invoked before emit within the current stage/operator
- * @param emitOnTimer decide whether the current aggregated elements can be emitted, the custom function is invoked on every interval
- */
- @deprecated("Use the overloaded one which accepts an Optional instead.", since = "1.2.0")
- def aggregateWithBoundary[Agg, Emit](allocate: function.Creator[Agg],
- aggregate: function.Function2[Agg, Out, Pair[Agg, Boolean]],
- harvest: function.Function[Agg, Emit],
- emitOnTimer: Pair[function.Predicate[Agg], java.time.Duration])
- : javadsl.Flow[In, Emit, Mat] = {
- asScala
- .aggregateWithBoundary(() => allocate.create())(
- aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
- harvest = agg => harvest.apply(agg),
- emitOnTimer = Option(emitOnTimer).map {
- case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
- })
- .asJava
- }
-
/**
* Aggregate input elements into an arbitrary data structure that can be completed and emitted downstream
* when custom condition is met which can be triggered by aggregate or timer.
@@ -4531,14 +4497,12 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr
emitOnTimer: Optional[Pair[function.Predicate[Agg], java.time.Duration]])
: javadsl.Flow[In, Emit, Mat] = {
import scala.jdk.OptionConverters._
- asScala
- .aggregateWithBoundary(() => allocate.create())(
- aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
- harvest = agg => harvest.apply(agg),
- emitOnTimer = emitOnTimer.toScala.map {
- case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
- })
- .asJava
+ new Flow(delegate.aggregateWithBoundary(() => allocate.create())(
+ aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
+ harvest = agg => harvest.apply(agg),
+ emitOnTimer = emitOnTimer.toScala.map {
+ case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
+ }))
}
override def getAttributes: Attributes = delegate.getAttributes
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
index 1ac01920e7..407314b3e0 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala
@@ -4914,39 +4914,6 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
def asSourceWithContext[Ctx](extractContext: function.Function[Out, Ctx]): SourceWithContext[Out, Ctx, Mat] =
new scaladsl.SourceWithContext(this.asScala.map(x => (x, extractContext.apply(x)))).asJava
- /**
- * Aggregate input elements into an arbitrary data structure that can be completed and emitted downstream
- * when custom condition is met which can be triggered by aggregate or timer.
- * It can be thought of a more general [[groupedWeightedWithin]].
- *
- * '''Emits when''' the aggregation function decides the aggregate is complete or the timer function returns true
- *
- * '''Backpressures when''' downstream backpressures and the aggregate is complete
- *
- * '''Completes when''' upstream completes and the last aggregate has been emitted downstream
- *
- * '''Cancels when''' downstream cancels
- *
- * @param allocate allocate the initial data structure for aggregated elements
- * @param aggregate update the aggregated elements, return true if ready to emit after update.
- * @param harvest this is invoked before emit within the current stage/operator
- * @param emitOnTimer decide whether the current aggregated elements can be emitted, the custom function is invoked on every interval
- */
- @deprecated("Use the overloaded one which accepts an Optional instead.", since = "1.2.0")
- def aggregateWithBoundary[Agg, Emit](allocate: function.Creator[Agg],
- aggregate: function.Function2[Agg, Out, Pair[Agg, Boolean]],
- harvest: function.Function[Agg, Emit],
- emitOnTimer: Pair[function.Predicate[Agg], java.time.Duration]): javadsl.Source[Emit, Mat] = {
- asScala
- .aggregateWithBoundary(() => allocate.create())(
- aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
- harvest = agg => harvest.apply(agg),
- emitOnTimer = Option(emitOnTimer).map {
- case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
- })
- .asJava
- }
-
/**
* Aggregate input elements into an arbitrary data structure that can be completed and emitted downstream
* when custom condition is met which can be triggered by aggregate or timer.
@@ -4970,14 +4937,12 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[
harvest: function.Function[Agg, Emit],
emitOnTimer: Optional[Pair[function.Predicate[Agg], java.time.Duration]]): javadsl.Source[Emit, Mat] = {
import scala.jdk.OptionConverters._
- asScala
- .aggregateWithBoundary(() => allocate.create())(
- aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
- harvest = agg => harvest.apply(agg),
- emitOnTimer = emitOnTimer.toScala.map {
- case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
- })
- .asJava
+ new Source(delegate.aggregateWithBoundary(() => allocate.create())(
+ aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
+ harvest = agg => harvest.apply(agg),
+ emitOnTimer = emitOnTimer.toScala.map {
+ case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
+ }))
}
override def getAttributes: Attributes = delegate.getAttributes
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
index cb0738b3c9..f5660192a7 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubFlow.scala
@@ -3049,39 +3049,6 @@ final class SubFlow[In, Out, Mat](
def logWithMarker(name: String, marker: function.Function[Out, LogMarker]): SubFlow[In, Out, Mat] =
this.logWithMarker(name, marker, ConstantFun.javaIdentityFunction[Out], null)
- /**
- * Aggregate input elements into an arbitrary data structure that can be completed and emitted downstream
- * when custom condition is met which can be triggered by aggregate or timer.
- * It can be thought of a more general [[groupedWeightedWithin]].
- *
- * '''Emits when''' the aggregation function decides the aggregate is complete or the timer function returns true
- *
- * '''Backpressures when''' downstream backpressures and the aggregate is complete
- *
- * '''Completes when''' upstream completes and the last aggregate has been emitted downstream
- *
- * '''Cancels when''' downstream cancels
- *
- * @param allocate allocate the initial data structure for aggregated elements
- * @param aggregate update the aggregated elements, return true if ready to emit after update.
- * @param harvest this is invoked before emit within the current stage/operator
- * @param emitOnTimer decide whether the current aggregated elements can be emitted, the custom function is invoked on every interval
- */
- @deprecated("Use the overloaded one which accepts an Optional instead.", since = "1.2.0")
- def aggregateWithBoundary[Agg, Emit](allocate: function.Creator[Agg],
- aggregate: function.Function2[Agg, Out, Pair[Agg, Boolean]],
- harvest: function.Function[Agg, Emit],
- emitOnTimer: Pair[function.Predicate[Agg], java.time.Duration])
- : javadsl.SubFlow[In, Emit, Mat] = {
- new SubFlow(
- asScala.aggregateWithBoundary(() => allocate.create())(
- aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
- harvest = agg => harvest.apply(agg),
- emitOnTimer = Option(emitOnTimer).map {
- case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
- }))
- }
-
/**
* Aggregate input elements into an arbitrary data structure that can be completed and emitted downstream
* when custom condition is met which can be triggered by aggregate or timer.
@@ -3106,13 +3073,12 @@ final class SubFlow[In, Out, Mat](
emitOnTimer: Optional[Pair[function.Predicate[Agg], java.time.Duration]])
: javadsl.SubFlow[In, Emit, Mat] = {
import scala.jdk.OptionConverters._
- new SubFlow(
- asScala.aggregateWithBoundary(() => allocate.create())(
- aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
- harvest = agg => harvest.apply(agg),
- emitOnTimer = emitOnTimer.toScala.map {
- case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
- }))
+ new SubFlow(delegate.aggregateWithBoundary(() => allocate.create())(
+ aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
+ harvest = agg => harvest.apply(agg),
+ emitOnTimer = emitOnTimer.toScala.map {
+ case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
+ }))
}
}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
index 6fead57639..e7806c1f23 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/SubSource.scala
@@ -3016,39 +3016,6 @@ final class SubSource[Out, Mat](
def logWithMarker(name: String, marker: function.Function[Out, LogMarker]): SubSource[Out, Mat] =
this.logWithMarker(name, marker, ConstantFun.javaIdentityFunction[Out], null)
- /**
- * Aggregate input elements into an arbitrary data structure that can be completed and emitted downstream
- * when custom condition is met which can be triggered by aggregate or timer.
- * It can be thought of a more general [[groupedWeightedWithin]].
- *
- * '''Emits when''' the aggregation function decides the aggregate is complete or the timer function returns true
- *
- * '''Backpressures when''' downstream backpressures and the aggregate is complete
- *
- * '''Completes when''' upstream completes and the last aggregate has been emitted downstream
- *
- * '''Cancels when''' downstream cancels
- *
- * @param allocate allocate the initial data structure for aggregated elements
- * @param aggregate update the aggregated elements, return true if ready to emit after update.
- * @param harvest this is invoked before emit within the current stage/operator
- * @param emitOnTimer decide whether the current aggregated elements can be emitted, the custom function is invoked on every interval
- */
- @deprecated("Use the overloaded one which accepts an Optional instead.", since = "1.2.0")
- def aggregateWithBoundary[Agg, Emit](allocate: function.Creator[Agg],
- aggregate: function.Function2[Agg, Out, Pair[Agg, Boolean]],
- harvest: function.Function[Agg, Emit],
- emitOnTimer: Pair[function.Predicate[Agg], java.time.Duration])
- : javadsl.SubSource[Emit, Mat] = {
- new SubSource(
- asScala.aggregateWithBoundary(() => allocate.create())(
- aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
- harvest = agg => harvest.apply(agg),
- emitOnTimer = Option(emitOnTimer).map {
- case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
- }))
- }
-
/**
* Aggregate input elements into an arbitrary data structure that can be completed and emitted downstream
* when custom condition is met which can be triggered by aggregate or timer.
@@ -3073,12 +3040,11 @@ final class SubSource[Out, Mat](
emitOnTimer: Optional[Pair[function.Predicate[Agg], java.time.Duration]])
: javadsl.SubSource[Emit, Mat] = {
import scala.jdk.OptionConverters._
- new SubSource(
- asScala.aggregateWithBoundary(() => allocate.create())(
- aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
- harvest = agg => harvest.apply(agg),
- emitOnTimer = emitOnTimer.toScala.map {
- case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
- }))
+ new SubSource(delegate.aggregateWithBoundary(() => allocate.create())(
+ aggregate = (agg, out) => aggregate.apply(agg, out).toScala,
+ harvest = agg => harvest.apply(agg),
+ emitOnTimer = emitOnTimer.toScala.map {
+ case Pair(predicate, duration) => (agg => predicate.test(agg), duration.toScala)
+ }))
}
}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
index 12997819f4..e67cc1fd67 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala
@@ -4069,7 +4069,6 @@ trait FlowOps[+Out, +Mat] {
* @param harvest this is invoked before emit within the current stage/operator
* @param emitOnTimer decide whether the current aggregated elements can be emitted, the custom function is invoked on every interval
*/
- @ApiMayChange
def aggregateWithBoundary[Agg, Emit](allocate: () => Agg)(
aggregate: (Agg, Out) => (Agg, Boolean),
harvest: Agg => Emit,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org