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/09/05 02:54:14 UTC
(pekko) 01/01: chore: Remove the deprecated overloaded aggregateWithBoundary method
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch agg
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit a9375525c7129f9586a33a7e96687cf1bb2d685b
Author: He-Pin <he...@gmail.com>
AuthorDate: Fri Sep 5 10:53:53 2025 +0800
chore: Remove the deprecated overloaded aggregateWithBoundary method
---
.../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 6e76d89579..be4fbd1554 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
@@ -4287,40 +4287,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.asScala)
- })
- .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.
@@ -4345,14 +4311,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 org.apache.pekko.util.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.asScala)
- })
- .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.asScala)
+ }))
}
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 96e5f852c7..8f5e158672 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
@@ -4729,39 +4729,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.asScala)
- })
- .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.
@@ -4785,14 +4752,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 org.apache.pekko.util.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.asScala)
- })
- .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.asScala)
+ }))
}
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 947432a6f1..c422b299b9 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
@@ -2770,39 +2770,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.asScala)
- }))
- }
-
/**
* 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.
@@ -2827,13 +2794,12 @@ final class SubFlow[In, Out, Mat](
emitOnTimer: Optional[Pair[function.Predicate[Agg], java.time.Duration]])
: javadsl.SubFlow[In, Emit, Mat] = {
import org.apache.pekko.util.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.asScala)
- }))
+ 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.asScala)
+ }))
}
}
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 387c4f1b19..4cc5a35508 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
@@ -2744,39 +2744,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.asScala)
- }))
- }
-
/**
* 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.
@@ -2801,12 +2768,11 @@ final class SubSource[Out, Mat](
emitOnTimer: Optional[Pair[function.Predicate[Agg], java.time.Duration]])
: javadsl.SubSource[Emit, Mat] = {
import org.apache.pekko.util.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.asScala)
- }))
+ 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.asScala)
+ }))
}
}
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 b77be6cc03..a8cf3d2c3f 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
@@ -3972,7 +3972,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