Class Source<Out,Mat>
- java.lang.Object
-
- org.apache.pekko.stream.javadsl.Source<Out,Mat>
-
- All Implemented Interfaces:
Graph<SourceShape<Out>,Mat>
public final class Source<Out,Mat> extends java.lang.Object implements Graph<SourceShape<Out>,Mat>
Java APIA
Source
is a set of stream processing steps that has one open output and an attached input. Can be used as aPublisher
-
-
Nested Class Summary
-
Nested classes/interfaces inherited from interface org.apache.pekko.stream.Graph
Graph.GraphMapMatVal<S extends Shape,M>
-
-
Method Summary
All Methods Static Methods Instance Methods Concrete Methods Deprecated Methods Modifier and Type Method Description static <T> Source<T,ActorRef>
actorRef(int bufferSize, OverflowStrategy overflowStrategy)
Deprecated.Use variant accepting completion and failure matchers.static <T> Source<T,ActorRef>
actorRef(Function<java.lang.Object,java.util.Optional<CompletionStrategy>> completionMatcher, Function<java.lang.Object,java.util.Optional<java.lang.Throwable>> failureMatcher, int bufferSize, OverflowStrategy overflowStrategy)
Creates aSource
that is materialized as anpekko.actor.ActorRef
.static <T> Source<T,ActorRef>
actorRefWithAck(java.lang.Object ackMessage)
Deprecated.Use actorRefWithBackpressure accepting completion and failure matchers.static <T> Source<T,ActorRef>
actorRefWithAck(java.lang.Object ackMessage, Function<java.lang.Object,java.util.Optional<CompletionStrategy>> completionMatcher, Function<java.lang.Object,java.util.Optional<java.lang.Throwable>> failureMatcher)
Deprecated.Use actorRefWithBackpressure insteadstatic <T> Source<T,ActorRef>
actorRefWithBackpressure(java.lang.Object ackMessage, Function<java.lang.Object,java.util.Optional<CompletionStrategy>> completionMatcher, Function<java.lang.Object,java.util.Optional<java.lang.Throwable>> failureMatcher)
Creates aSource
that is materialized as anpekko.actor.ActorRef
.Source<Out,Mat>
addAttributes(Attributes attr)
Add the given attributes to thisSource
.<Agg,Emit>
Source<Emit,Mat>aggregateWithBoundary(java.util.function.Supplier<Agg> allocate, Function2<Agg,Out,Pair<Agg,java.lang.Object>> aggregate, Function<Agg,Emit> harvest, Pair<java.util.function.Predicate<Agg>,java.time.Duration> emitOnTimer)
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.Source<Out,Mat>
alsoTo(Graph<SinkShape<Out>,?> that)
Source<Out,Mat>
alsoToAll(Graph<SinkShape<Out>,?>... those)
Source<Out,Mat>
alsoToAll(scala.collection.immutable.Seq<Graph<SinkShape<Out>,?>> those)
<M2,M3>
Source<Out,M3>alsoToMat(Graph<SinkShape<Out>,M2> that, Function2<Mat,M2,M3> matF)
<S> Source<S,Mat>
ask(int parallelism, ActorRef ref, java.lang.Class<S> mapTo, Timeout timeout)
Use theask
pattern to send a request-reply message to the targetref
actor.<S> Source<S,Mat>
ask(ActorRef ref, java.lang.Class<S> mapTo, Timeout timeout)
Use theask
pattern to send a request-reply message to the targetref
actor.Source<Out,Mat>
asScala()
Converts this Java DSL element to its Scala DSL counterpart.<Ctx> SourceWithContext<Out,Ctx,Mat>
asSourceWithContext(Function<Out,Ctx> extractContext)
Transform this source whose element ise
into a source producing tuple
(e, f(e))
static <T> Source<T,org.reactivestreams.Subscriber<T>>
asSubscriber()
Creates aSource
that is materialized as aSubscriber
Source<Out,Mat>
async()
Put an asynchronous boundary around thisSource
Source<Out,Mat>
async(java.lang.String dispatcher)
Put an asynchronous boundary around thisSource
Source<Out,Mat>
async(java.lang.String dispatcher, int inputBufferSize)
Put an asynchronous boundary around thisSource
Source<Out,Mat>
backpressureTimeout(java.time.Duration timeout)
If the time between the emission of an element and the following downstream demand exceeds the provided timeout, the stream is failed with aBackpressureTimeoutException
.Source<Out,Mat>
backpressureTimeout(scala.concurrent.duration.FiniteDuration timeout)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.<S> Source<S,Mat>
batch(long max, Function<Out,S> seed, Function2<S,Out,S> aggregate)
Allows a faster upstream to progress independently of a slower subscriber by aggregating elements into batches until the subscriber is ready to accept them.<S> Source<S,Mat>
batchWeighted(long max, Function<Out,java.lang.Long> costFn, Function<Out,S> seed, Function2<S,Out,S> aggregate)
Allows a faster upstream to progress independently of a slower subscriber by aggregating elements into batches until the subscriber is ready to accept them.Source<Out,Mat>
buffer(int size, OverflowStrategy overflowStrategy)
Adds a fixed size buffer in the flow that allows to store elements from a faster upstream until it becomes full.<T> Source<T,Mat>
collect(scala.PartialFunction<Out,T> pf)
Transform this stream by applying the given partial function to each of the elements on which the function is defined as they pass through this processing step.<T> Source<T,Mat>
collectFirst(scala.PartialFunction<Out,T> pf)
Transform this stream by applying the given partial function to the first element on which the function is defined as it pass through this processing step, and cancel the upstream publisher after the first element is emitted.<T> Source<T,Mat>
collectType(java.lang.Class<T> clazz)
Transform this stream by testing the type of each of the elements on which the element is an instance of the provided type as they pass through this processing step.<T> Source<T,Mat>
collectWhile(scala.PartialFunction<Out,T> pf)
Transform this stream by applying the given partial function to each of the elements on which the function is defined as they pass through this processing step, and cancel the upstream publisher after the partial function is not applied.static <T,U,M>
Source<U,java.util.List<M>>combine(java.util.List<? extends Graph<SourceShape<T>,M>> sources, Function<java.lang.Integer,Graph<UniformFanInShape<T,U>,NotUsed>> fanInStrategy)
static <T,U>
Source<U,NotUsed>combine(Source<T,?> first, Source<T,?> second, java.util.List<Source<T,?>> rest, Function<java.lang.Integer,? extends Graph<UniformFanInShape<T,U>,NotUsed>> fanInStrategy)
static <T,U,M1,M2,M>
Source<U,M>combineMat(Source<T,M1> first, Source<T,M2> second, Function<java.lang.Integer,? extends Graph<UniformFanInShape<T,U>,NotUsed>> fanInStrategy, Function2<M1,M2,M> combine)
Combines two sources with fan-in strategy likeMerge
orConcat
and returnsSource
with a materialized value.static <T> Source<T,NotUsed>
completionStage(java.util.concurrent.CompletionStage<T> completionStage)
Emits a single value when the givenCompletionStage
is successfully completed and then completes the stream.static <T,M>
Source<T,java.util.concurrent.CompletionStage<M>>completionStageSource(java.util.concurrent.CompletionStage<Source<T,M>> completionStageSource)
Turn aCompletionStage[Source]
into a source that will emit the values of the source when the future completes successfully.Source<Out,Mat>
completionTimeout(java.time.Duration timeout)
If the completion of the stream does not happen until the provided timeout, the stream is failed with aCompletionTimeoutException
.Source<Out,Mat>
completionTimeout(scala.concurrent.duration.FiniteDuration timeout)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.<M> Source<Out,Mat>
concat(Graph<SourceShape<Out>,M> that)
Concatenate thisSource
with the given one, meaning that once current is exhausted and all result elements have been generated, the given source elements will be produced.Source<Out,Mat>
concatAllLazy(Graph<SourceShape<Out>,?>... those)
Concatenate the givenSource
s to this one, meaning that once this Flow’s input is exhausted and all result elements have been generated, the Source’s elements will be produced.Source<Out,Mat>
concatAllLazy(scala.collection.immutable.Seq<Graph<SourceShape<Out>,?>> those)
Concatenate the givenSource
s to this one, meaning that once this Flow’s input is exhausted and all result elements have been generated, the Source’s elements will be produced.<M> Source<Out,Mat>
concatLazy(Graph<SourceShape<Out>,M> that)
<M,M2>
Source<Out,M2>concatLazyMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
<M,M2>
Source<Out,M2>concatMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
Concatenate thisSource
with the given one, meaning that once current is exhausted and all result elements have been generated, the given source elements will be produced.Source<Out,Mat>
conflate(Function2<Out,Out,Out> aggregate)
Allows a faster upstream to progress independently of a slower subscriber by conflating elements into a summary until the subscriber is ready to accept them.<S> Source<S,Mat>
conflateWithSeed(Function<Out,S> seed, Function2<S,Out,S> aggregate)
Allows a faster upstream to progress independently of a slower subscriber by conflating elements into a summary until the subscriber is ready to accept them.static <O> Source<O,NotUsed>
cycle(Creator<java.util.Iterator<O>> f)
Helper to create 'cycled'Source
from iterator provider.Source<Out,Mat>
delay(java.time.Duration of, DelayOverflowStrategy strategy)
Shifts elements emission in time by a specified amount.Source<Out,Mat>
delay(scala.concurrent.duration.FiniteDuration of, DelayOverflowStrategy strategy)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<Out,Mat>
delayWith(java.util.function.Supplier<DelayStrategy<Out>> delayStrategySupplier, DelayOverflowStrategy overFlowStrategy)
Shifts elements emission in time by an amount individually determined through delay strategy a specified amount.Source<Out,Mat>
detach()
Detaches upstream demand from downstream demand without detaching the stream rates; in other words acts like a buffer of size 1.Source<Out,Mat>
divertTo(Graph<SinkShape<Out>,?> that, Predicate<Out> when)
<M2,M3>
Source<Out,M3>divertToMat(Graph<SinkShape<Out>,M2> that, Predicate<Out> when, Function2<Mat,M2,M3> matF)
Source<Out,Mat>
drop(long n)
Discard the given number of elements at the beginning of the stream.Source<Out,Mat>
dropWhile(Predicate<Out> p)
Discard elements at the beginning of the stream while predicate is true.Source<Out,Mat>
dropWithin(java.time.Duration duration)
Discard the elements received within the given duration at beginning of the stream.Source<Out,Mat>
dropWithin(scala.concurrent.duration.FiniteDuration duration)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.static <O> Source<O,NotUsed>
empty()
Create aSource
with no elements, i.e.static <T> Source<T,NotUsed>
empty(java.lang.Class<T> clazz)
Create aSource
with no elements.<U> Source<U,Mat>
expand(Function<Out,java.util.Iterator<U>> expander)
Allows a faster downstream to progress independently of a slower publisher by extrapolating elements from an older element until new element comes from the upstream.Source<Out,Mat>
extrapolate(Function<Out,java.util.Iterator<Out>> extrapolator)
Allows a faster downstream to progress independent of a slower upstream.Source<Out,Mat>
extrapolate(Function<Out,java.util.Iterator<Out>> extrapolator, Out initial)
Allows a faster downstream to progress independent of a slower upstream.static <T> Source<T,NotUsed>
failed(java.lang.Throwable cause)
Create aSource
that immediately ends the stream with thecause
failure to every connectedSink
.Source<Out,Mat>
filter(Predicate<Out> p)
Only pass on those elements that satisfy the given predicate.Source<Out,Mat>
filterNot(Predicate<Out> p)
Only pass on those elements that NOT satisfy the given predicate.<T,M>
Source<T,Mat>flatMapConcat(Function<Out,? extends Graph<SourceShape<T>,M>> f)
Transform each input element into aSource
of output elements that is then flattened into the output stream by concatenation, fully consuming one Source after the other.<T,M>
Source<T,Mat>flatMapMerge(int breadth, Function<Out,? extends Graph<SourceShape<T>,M>> f)
Transform each input element into aSource
of output elements that is then flattened into the output stream by merging, where at mostbreadth
substreams are being consumed at any given time.<Out2,Mat2>
Source<Out2,Mat>flatMapPrefix(int n, Function<java.lang.Iterable<Out>,Flow<Out,Out2,Mat2>> f)
Takes up ton
elements from the stream (less thann
only if the upstream completes before emittingn
elements), then applyf
on these elements in order to obtain a flow, this flow is then materialized and the rest of the input is processed by this flow (similar to via).<Out2,Mat2,Mat3>
Source<Out2,Mat3>flatMapPrefixMat(int n, Function<java.lang.Iterable<Out>,Flow<Out,Out2,Mat2>> f, Function2<Mat,java.util.concurrent.CompletionStage<Mat2>,Mat3> matF)
mat version offlatMapPrefix(int, org.apache.pekko.japi.function.Function<java.lang.Iterable<Out>, org.apache.pekko.stream.javadsl.Flow<Out, Out2, Mat2>>)
, this method gives access to a future materialized value of the downstream flow (as a completion stage).<T> Source<T,Mat>
fold(T zero, Function2<T,Out,T> f)
Similar toscan
but only emits its result when the upstream completes, after which it also completes.<T> Source<T,Mat>
foldAsync(T zero, Function2<T,Out,java.util.concurrent.CompletionStage<T>> f)
Similar tofold
but with an asynchronous function.<T> Source<T,Mat>
foldWhile(T zero, Predicate<T> p, Function2<T,Out,T> f)
Similar toscan
but only emits its result when the upstream completes or the predicatep
returnsfalse
.static <O> Source<O,NotUsed>
from(java.lang.Iterable<O> iterable)
Helper to createSource
fromIterable
.static <T> Source<T,NotUsed>
fromArray(java.lang.Object array)
Creates aSource
from an array, if the array is empty, the stream is completed immediately, otherwise, every element of the array will be emitted sequentially.static <O> Source<O,NotUsed>
fromCompletionStage(java.util.concurrent.CompletionStage<O> future)
Deprecated.Use 'Source.completionStage' instead.static <O> Source<O,NotUsed>
fromFuture(scala.concurrent.Future<O> future)
Deprecated.Use 'Source.future' instead.static <T,M>
Source<T,scala.concurrent.Future<M>>fromFutureSource(scala.concurrent.Future<? extends Graph<SourceShape<T>,M>> future)
Deprecated.Use 'Source.futureSource' (potentially together with `Source.fromGraph`) instead.static <T,M>
Source<T,M>fromGraph(Graph<SourceShape<T>,M> g)
A graph with the shape of a source logically is a source, this method makes it so also in type.static <O> Source<O,NotUsed>
fromIterator(Creator<java.util.Iterator<O>> f)
Helper to createSource
fromIterator
.static <O,S extends java.util.stream.BaseStream<O,S>>
Source<O,NotUsed>fromJavaStream(Creator<java.util.stream.BaseStream<O,S>> stream)
Creates a source that wraps a Java 8Stream
.
static <T,M>
Source<T,java.util.concurrent.CompletionStage<M>>fromMaterializer(java.util.function.BiFunction<Materializer,Attributes,Source<T,M>> factory)
Defers the creation of aSource
until materialization.static <O> Source<O,NotUsed>
fromPublisher(org.reactivestreams.Publisher<O> publisher)
Helper to createSource
fromPublisher
.static <T,M>
Source<T,java.util.concurrent.CompletionStage<M>>fromSourceCompletionStage(java.util.concurrent.CompletionStage<? extends Graph<SourceShape<T>,M>> completion)
Deprecated.Use 'Source.completionStageSource' (potentially together with `Source.fromGraph`) instead.static <T> Source<T,NotUsed>
future(scala.concurrent.Future<T> futureElement)
Emits a single value when the given ScalaFuture
is successfully completed and then completes the stream.Attributes
getAttributes()
<K> SubSource<Out,Mat>
groupBy(int maxSubstreams, Function<Out,K> f)
This operation demultiplexes the incoming stream into separate output streams, one for each element key.<K> SubSource<Out,Mat>
groupBy(int maxSubstreams, Function<Out,K> f, boolean allowClosedSubstreamRecreation)
This operation demultiplexes the incoming stream into separate output streams, one for each element key.Source<java.util.List<Out>,Mat>
grouped(int n)
Chunk up this stream into groups of the given size, with the last group possibly smaller than requested due to end-of-stream.Source<java.util.List<Out>,Mat>
groupedWeighted(long minWeight, java.util.function.Function<Out,java.lang.Long> costFn)
Chunk up this stream into groups of elements that have a cumulative weight greater than or equal to theminWeight
, with the last group possibly smaller than requestedminWeight
due to end-of-stream.Source<java.util.List<Out>,Mat>
groupedWeightedWithin(long maxWeight, int maxNumber, Function<Out,java.lang.Long> costFn, java.time.Duration duration)
Chunk up this stream into groups of elements received within a time window, or limited by the weight and number of the elements, whatever happens first.Source<java.util.List<Out>,Mat>
groupedWeightedWithin(long maxWeight, Function<Out,java.lang.Long> costFn, java.time.Duration duration)
Chunk up this stream into groups of elements received within a time window, or limited by the weight of the elements, whatever happens first.Source<java.util.List<Out>,Mat>
groupedWeightedWithin(long maxWeight, Function<Out,java.lang.Long> costFn, scala.concurrent.duration.FiniteDuration duration)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<java.util.List<Out>,Mat>
groupedWithin(int maxNumber, java.time.Duration duration)
Chunk up this stream into groups of elements received within a time window, or limited by the given number of elements, whatever happens first.Source<java.util.List<Out>,Mat>
groupedWithin(int maxNumber, scala.concurrent.duration.FiniteDuration duration)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<Out,Mat>
idleTimeout(java.time.Duration timeout)
If the time between two processed elements exceeds the provided timeout, the stream is failed with aStreamIdleTimeoutException
.Source<Out,Mat>
idleTimeout(scala.concurrent.duration.FiniteDuration timeout)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<Out,Mat>
initialDelay(java.time.Duration delay)
Delays the initial element by the specified duration.Source<Out,Mat>
initialDelay(scala.concurrent.duration.FiniteDuration delay)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<Out,Mat>
initialTimeout(java.time.Duration timeout)
If the first element has not passed through this operator before the provided timeout, the stream is failed with aInitialTimeoutException
.Source<Out,Mat>
initialTimeout(scala.concurrent.duration.FiniteDuration timeout)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<Out,Mat>
interleave(Graph<SourceShape<Out>,?> that, int segmentSize)
Source<Out,Mat>
interleave(Graph<SourceShape<Out>,?> that, int segmentSize, boolean eagerClose)
Source<Out,Mat>
interleaveAll(java.util.List<? extends Graph<SourceShape<Out>,?>> those, int segmentSize, boolean eagerClose)
<M,M2>
Source<Out,M2>interleaveMat(Graph<SourceShape<Out>,M> that, int segmentSize, boolean eagerClose, Function2<Mat,M,M2> matF)
<M,M2>
Source<Out,M2>interleaveMat(Graph<SourceShape<Out>,M> that, int segmentSize, Function2<Mat,M,M2> matF)
Source<Out,Mat>
intersperse(Out inject)
Intersperses stream with provided element, similar to howscala.collection.immutable.List.mkString
injects a separator between a List's elements.Source<Out,Mat>
intersperse(Out start, Out inject, Out end)
Intersperses stream with provided element, similar to howscala.collection.immutable.List.mkString
injects a separator between a List's elements.static <T> Source<T,NotUsed>
iterate(T seed, Function<T,T> f)
Creates an infinite sequentialSource
by iterating with the given function, starting with the givenseed
value.static <T> Source<T,NotUsed>
iterate(T seed, Predicate<T> p, Function<T,T> f)
Creates a sequentialSource
by iterating with the given predicate and function, starting with the givenseed
value.Source<Out,Mat>
keepAlive(java.time.Duration maxIdle, Creator<Out> injectedElem)
Injects additional elements if upstream does not emit for a configured amount of time.Source<Out,Mat>
keepAlive(scala.concurrent.duration.FiniteDuration maxIdle, Creator<Out> injectedElem)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.static <T,M>
Source<T,java.util.concurrent.CompletionStage<M>>lazily(Creator<Source<T,M>> create)
Deprecated.Use 'Source.lazySource' instead.static <T> Source<T,scala.concurrent.Future<NotUsed>>
lazilyAsync(Creator<java.util.concurrent.CompletionStage<T>> create)
Deprecated.Use 'Source.lazyCompletionStage' instead.static <T> Source<T,NotUsed>
lazyCompletionStage(Creator<java.util.concurrent.CompletionStage<T>> create)
Defers invoking thecreate
function to create a future element until there is downstream demand.static <T,M>
Source<T,java.util.concurrent.CompletionStage<M>>lazyCompletionStageSource(Creator<java.util.concurrent.CompletionStage<Source<T,M>>> create)
Defers invoking thecreate
function to create a future source until there is downstream demand.static <T> Source<T,NotUsed>
lazySingle(Creator<T> create)
Defers invoking thecreate
function to create a single element until there is downstream demand.static <T,M>
Source<T,java.util.concurrent.CompletionStage<M>>lazySource(Creator<Source<T,M>> create)
Defers invoking thecreate
function to create a future source until there is downstream demand.Source<Out,Mat>
limit(int n)
Ensure stream boundedness by limiting the number of elements from upstream.Source<Out,Mat>
limitWeighted(long n, Function<Out,java.lang.Long> costFn)
Ensure stream boundedness by evaluating the cost of incoming elements using a cost function.Source<Out,Mat>
log(java.lang.String name)
Logs elements flowing through the stream as well as completion and erroring.Source<Out,Mat>
log(java.lang.String name, LoggingAdapter log)
Logs elements flowing through the stream as well as completion and erroring.Source<Out,Mat>
log(java.lang.String name, Function<Out,java.lang.Object> extract)
Logs elements flowing through the stream as well as completion and erroring.Source<Out,Mat>
log(java.lang.String name, Function<Out,java.lang.Object> extract, LoggingAdapter log)
Logs elements flowing through the stream as well as completion and erroring.Source<Out,Mat>
logWithMarker(java.lang.String name, Function<Out,LogMarker> marker)
Logs elements flowing through the stream as well as completion and erroring.Source<Out,Mat>
logWithMarker(java.lang.String name, Function<Out,LogMarker> marker, MarkerLoggingAdapter log)
Logs elements flowing through the stream as well as completion and erroring.Source<Out,Mat>
logWithMarker(java.lang.String name, Function<Out,LogMarker> marker, Function<Out,java.lang.Object> extract)
Logs elements flowing through the stream as well as completion and erroring.Source<Out,Mat>
logWithMarker(java.lang.String name, Function<Out,LogMarker> marker, Function<Out,java.lang.Object> extract, MarkerLoggingAdapter log)
Logs elements flowing through the stream as well as completion and erroring.<T> Source<T,Mat>
map(Function<Out,T> f)
Transform this stream by applying the given function to each of the elements as they pass through this processing step.<T> Source<T,Mat>
mapAsync(int parallelism, Function<Out,java.util.concurrent.CompletionStage<T>> f)
Transform this stream by applying the given function to each of the elements as they pass through this processing step.<T,P>
Source<T,Mat>mapAsyncPartitioned(int parallelism, Function<Out,P> partitioner, Function2<Out,P,java.util.concurrent.CompletionStage<T>> f)
Transforms this stream.<T,P>
Source<T,Mat>mapAsyncPartitionedUnordered(int parallelism, Function<Out,P> partitioner, Function2<Out,P,java.util.concurrent.CompletionStage<T>> f)
Transforms this stream.<T> Source<T,Mat>
mapAsyncUnordered(int parallelism, Function<Out,java.util.concurrent.CompletionStage<T>> f)
Transform this stream by applying the given function to each of the elements as they pass through this processing step.<T> Source<T,Mat>
mapConcat(Function<Out,? extends java.lang.Iterable<T>> f)
Transform each input element into anIterable
of output elements that is then flattened into the output stream.<E extends java.lang.Throwable>
Source<Out,Mat>mapError(java.lang.Class<E> clazz, Function<E,java.lang.Throwable> f)
While similar torecover(scala.PartialFunction<java.lang.Throwable,Out>)
this operator can be used to transform an error signal to a different one *without* logging it as an error in the process.Source<Out,Mat>
mapError(scala.PartialFunction<java.lang.Throwable,java.lang.Throwable> pf)
While similar torecover(scala.PartialFunction<java.lang.Throwable,Out>)
this operator can be used to transform an error signal to a different one *without* logging it as an error in the process.<Mat2> Source<Out,Mat2>
mapMaterializedValue(Function<Mat,Mat2> f)
Transform only the materialized value of this Source, leaving all other properties as they were.<R extends java.lang.AutoCloseable,T>
Source<T,Mat>mapWithResource(Creator<R> create, Function2<R,Out,T> f)
Transform each stream element with the help of anAutoCloseable
resource and close it when the stream finishes or fails.<R,T>
Source<T,Mat>mapWithResource(Creator<R> create, Function2<R,Out,T> f, Function<R,java.util.Optional<T>> close)
Transform each stream element with the help of a resource.static <T> Source<T,java.util.concurrent.CompletableFuture<java.util.Optional<T>>>
maybe()
Create aSource
which materializes aCompletableFuture
which controls what element will be emitted by the Source.Source<Out,Mat>
merge(Graph<SourceShape<Out>,?> that)
Merge the givenSource
to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.Source<Out,Mat>
merge(Graph<SourceShape<Out>,?> that, boolean eagerComplete)
Merge the givenSource
to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.Source<Out,Mat>
mergeAll(java.util.List<? extends Graph<SourceShape<Out>,?>> those, boolean eagerComplete)
Merge the givenSource
s to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.<M> Source<java.util.List<Out>,Mat>
mergeLatest(Graph<SourceShape<Out>,M> that, boolean eagerComplete)
MergeLatest joins elements from N input streams into stream of lists of size N.<Mat2,Mat3>
Source<java.util.List<Out>,Mat3>mergeLatestMat(Graph<SourceShape<Out>,Mat2> that, boolean eagerComplete, Function2<Mat,Mat2,Mat3> matF)
MergeLatest joins elements from N input streams into stream of lists of size N.<M,M2>
Source<Out,M2>mergeMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
Merge the givenSource
to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.<M,M2>
Source<Out,M2>mergeMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF, boolean eagerComplete)
Merge the givenSource
to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.<M> Source<Out,Mat>
mergePreferred(Graph<SourceShape<Out>,M> that, boolean preferred, boolean eagerComplete)
Merge two sources.<Mat2,Mat3>
Source<Out,Mat3>mergePreferredMat(Graph<SourceShape<Out>,Mat2> that, boolean preferred, boolean eagerComplete, Function2<Mat,Mat2,Mat3> matF)
Merge two sources.<M> Source<Out,Mat>
mergePrioritized(Graph<SourceShape<Out>,M> that, int leftPriority, int rightPriority, boolean eagerComplete)
Merge two sources.<Mat2,Mat3>
Source<Out,Mat3>mergePrioritizedMat(Graph<SourceShape<Out>,Mat2> that, int leftPriority, int rightPriority, boolean eagerComplete, Function2<Mat,Mat2,Mat3> matF)
Merge multiple sources.static <T> Source<T,NotUsed>
mergePrioritizedN(java.util.List<Pair<Source<T,?>,java.lang.Integer>> sourcesAndPriorities, boolean eagerComplete)
Merge multipleSource
s.<M> Source<Out,Mat>
mergeSorted(Graph<SourceShape<Out>,M> that, java.util.Comparator<Out> comp)
<Mat2,Mat3>
Source<Out,Mat3>mergeSortedMat(Graph<SourceShape<Out>,Mat2> that, java.util.Comparator<Out> comp, Function2<Mat,Mat2,Mat3> matF)
Source<Out,Pair<Mat,FlowMonitor<Out>>>
monitor()
Materializes toPair
, which is unlike most other operators (!), in which usually the default materialized value keeping semantics is to keep the left value (by passing> Keep.left()
to a*Mat
version of a method).<M> Source<Out,M>
monitor(Function2<Mat,FlowMonitor<Out>,M> combine)
Deprecated.Use monitor() or monitorMat(combine) instead.<M> Source<Out,M>
monitorMat(Function2<Mat,FlowMonitor<Out>,M> combine)
Materializes toFlowMonitor[Out]
that allows monitoring of the current flow.Source<Out,Mat>
named(java.lang.String name)
Add aname
attribute to this Source.
static <T> Source<T,NotUsed>
never()
Never emits any elements, never completes and never fails.Source<Out,Mat>
onErrorComplete()
onErrorComplete allows to complete the stream when an upstream error occurs.Source<Out,Mat>
onErrorComplete(java.lang.Class<? extends java.lang.Throwable> clazz)
onErrorComplete allows to complete the stream when an upstream error occurs.Source<Out,Mat>
onErrorComplete(java.util.function.Predicate<? super java.lang.Throwable> predicate)
onErrorComplete allows to complete the stream when an upstream error occurs.static <SOut,FOut,SMat,FMat,Mat>
Source<java.util.Optional<FOut>,Mat>optionalVia(Source<java.util.Optional<SOut>,SMat> source, Flow<SOut,FOut,FMat> viaFlow, Function2<SMat,FMat,Mat> combine)
Creates a Source from an existing base Source outputting an optional element and applying an additional viaFlow only if the element in the stream is defined.<M> Source<Out,Mat>
orElse(Graph<SourceShape<Out>,M> secondary)
Provides a secondary source that will be consumed if this source completes without any elements passing by.<M,M2>
Source<Out,M2>orElseMat(Graph<SourceShape<Out>,M> secondary, Function2<Mat,M,M2> matF)
Provides a secondary source that will be consumed if this source completes without any elements passing by.Source<Pair<java.util.List<Out>,Source<Out,NotUsed>>,Mat>
prefixAndTail(int n)
Takes up ton
elements from the stream (less thann
if the upstream completes before emittingn
elements) and returns a pair containing a strict sequence of the taken element and a stream representing the remaining elements.Pair<Mat,Source<Out,NotUsed>>
preMaterialize(ClassicActorSystemProvider systemProvider)
Materializes this Source, immediately returning (1) its materialized value, and (2) a new Source that can be used to consume elements from the newly materialized Source.Pair<Mat,Source<Out,NotUsed>>
preMaterialize(Materializer materializer)
Materializes this Source, immediately returning (1) its materialized value, and (2) a new Source that can be used to consume elements from the newly materialized Source.<M> Source<Out,Mat>
prepend(Graph<SourceShape<Out>,M> that)
Prepend the givenSource
to this one, meaning that once the given source is exhausted and all result elements have been generated, the current source's elements will be produced.<M> Source<Out,Mat>
prependLazy(Graph<SourceShape<Out>,M> that)
<M,M2>
Source<Out,M2>prependLazyMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
<M,M2>
Source<Out,M2>prependMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
Prepend the givenSource
to this one, meaning that once the given source is exhausted and all result elements have been generated, the current source's elements will be produced.static <T> Source<T,BoundedSourceQueue<T>>
queue(int bufferSize)
Creates aSource
that is materialized as anpekko.stream.BoundedSourceQueue
.static <T> Source<T,SourceQueueWithComplete<T>>
queue(int bufferSize, OverflowStrategy overflowStrategy)
Creates aSource
that is materialized as anpekko.stream.javadsl.SourceQueueWithComplete
.static <T> Source<T,SourceQueueWithComplete<T>>
queue(int bufferSize, OverflowStrategy overflowStrategy, int maxConcurrentOffers)
Creates aSource
that is materialized as anpekko.stream.javadsl.SourceQueueWithComplete
.static Source<java.lang.Integer,NotUsed>
range(int start, int end)
CreatesSource
that represents integer values in range ''[start;end]'', step equals to 1.static Source<java.lang.Integer,NotUsed>
range(int start, int end, int step)
CreatesSource
that represents integer values in range ''[start;end]'', with the given step.Source<Out,Mat>
recover(java.lang.Class<? extends java.lang.Throwable> clazz, java.util.function.Supplier<Out> supplier)
Recover allows to send last element on failure and gracefully complete the stream Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements.Source<Out,Mat>
recover(scala.PartialFunction<java.lang.Throwable,Out> pf)
Recover allows to send last element on failure and gracefully complete the stream Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements.Source<Out,Mat>
recoverWith(java.lang.Class<? extends java.lang.Throwable> clazz, java.util.function.Supplier<Graph<SourceShape<Out>,NotUsed>> supplier)
Deprecated.userecoverWithRetries
insteadSource<Out,Mat>
recoverWith(scala.PartialFunction<java.lang.Throwable,? extends Graph<SourceShape<Out>,NotUsed>> pf)
Deprecated.userecoverWithRetries
insteadSource<Out,Mat>
recoverWithRetries(int attempts, java.lang.Class<? extends java.lang.Throwable> clazz, java.util.function.Supplier<Graph<SourceShape<Out>,NotUsed>> supplier)
RecoverWithRetries allows to switch to alternative Source on flow failure.Source<Out,Mat>
recoverWithRetries(int attempts, scala.PartialFunction<java.lang.Throwable,? extends Graph<SourceShape<Out>,NotUsed>> pf)
RecoverWithRetries allows to switch to alternative Source on flow failure.Source<Out,Mat>
reduce(Function2<Out,Out,Out> f)
Similar tofold
but uses first element as zero element.static <T> Source<T,NotUsed>
repeat(T element)
Create aSource
that will continually emit the given element.java.util.concurrent.CompletionStage<Done>
run(ClassicActorSystemProvider systemProvider)
Connect thisSource
to theSink.ignore
and run it.java.util.concurrent.CompletionStage<Done>
run(Materializer materializer)
Connect thisSource
to theSink.ignore
and run it.<U> java.util.concurrent.CompletionStage<U>
runFold(U zero, Function2<U,Out,U> f, ClassicActorSystemProvider systemProvider)
Shortcut for running thisSource
with a fold function.<U> java.util.concurrent.CompletionStage<U>
runFold(U zero, Function2<U,Out,U> f, Materializer materializer)
Shortcut for running thisSource
with a fold function.<U> java.util.concurrent.CompletionStage<U>
runFoldAsync(U zero, Function2<U,Out,java.util.concurrent.CompletionStage<U>> f, ClassicActorSystemProvider systemProvider)
Shortcut for running thisSource
with an asynchronous fold function.<U> java.util.concurrent.CompletionStage<U>
runFoldAsync(U zero, Function2<U,Out,java.util.concurrent.CompletionStage<U>> f, Materializer materializer)
Shortcut for running thisSource
with an asynchronous fold function.java.util.concurrent.CompletionStage<Done>
runForeach(Procedure<Out> f, ClassicActorSystemProvider systemProvider)
Shortcut for running thisSource
with a foreach procedure.java.util.concurrent.CompletionStage<Done>
runForeach(Procedure<Out> f, Materializer materializer)
Shortcut for running thisSource
with a foreach procedure.java.util.concurrent.CompletionStage<Out>
runReduce(Function2<Out,Out,Out> f, ClassicActorSystemProvider systemProvider)
Shortcut for running thisSource
with a reduce function.java.util.concurrent.CompletionStage<Out>
runReduce(Function2<Out,Out,Out> f, Materializer materializer)
Shortcut for running thisSource
with a reduce function.<M> M
runWith(Graph<SinkShape<Out>,M> sink, ClassicActorSystemProvider systemProvider)
Connect thisSource
to aSink
and run it.<M> M
runWith(Graph<SinkShape<Out>,M> sink, Materializer materializer)
Connect thisSource
to aSink
and run it.<T> Source<T,Mat>
scan(T zero, Function2<T,Out,T> f)
Similar tofold
but is not a terminal operation, emits its current value which starts atzero
and then applies the current and next value to the given functionf
, emitting the next current value.<T> Source<T,Mat>
scanAsync(T zero, Function2<T,Out,java.util.concurrent.CompletionStage<T>> f)
Similar toscan
but with an asynchronous function, emits its current value which starts atzero
and then applies the current and next value to the given functionf
, emitting aFuture
that resolves to the next current value.static <T,M>
Source<T,java.util.concurrent.CompletionStage<M>>setup(java.util.function.BiFunction<ActorMaterializer,Attributes,Source<T,M>> factory)
Deprecated.Use 'fromMaterializer' instead.SourceShape<Out>
shape()
The shape of a graph is all that is externally visible: its inlets and outlets.static <T> Source<T,NotUsed>
single(T element)
Create aSource
with one element.Source<java.util.List<Out>,Mat>
sliding(int n, int step)
Apply a sliding window over the stream and return the windows as groups of elements, with the last group possibly smaller than requested due to end-of-stream.SubSource<Out,Mat>
splitAfter(Predicate<Out> p)
This operation applies the given predicate to all incoming elements and emits them to a stream of output streams.SubSource<Out,Mat>
splitAfter(SubstreamCancelStrategy substreamCancelStrategy, Predicate<Out> p)
Deprecated.Use .withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather than a SubstreamCancelStrategy.SubSource<Out,Mat>
splitWhen(Predicate<Out> p)
This operation applies the given predicate to all incoming elements and emits them to a stream of output streams, always beginning a new one with the current element if the given predicate returns true for it.SubSource<Out,Mat>
splitWhen(SubstreamCancelStrategy substreamCancelStrategy, Predicate<Out> p)
Deprecated.Use .withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather than a SubstreamCancelStrategy.<S,T>
Source<T,Mat>statefulMap(Creator<S> create, Function2<S,Out,Pair<S,T>> f, Function<S,java.util.Optional<T>> onComplete)
Transform each stream element with the help of a state.<T> Source<T,Mat>
statefulMapConcat(Creator<Function<Out,java.lang.Iterable<T>>> f)
Transform each input element into anIterable
of output elements that is then flattened into the output stream.Source<Out,Mat>
take(long n)
Terminate processing (and cancel the upstream publisher) after the given number of elements.Source<Out,Mat>
takeWhile(Predicate<Out> p)
Terminate processing (and cancel the upstream publisher) after predicate returns false for the first time.Source<Out,Mat>
takeWhile(Predicate<Out> p, boolean inclusive)
Terminate processing (and cancel the upstream publisher) after predicate returns false for the first time, including the first failed element if inclusive is true Due to input buffering some elements may have been requested from upstream publishers that will then not be processed downstream of this step.Source<Out,Mat>
takeWithin(java.time.Duration duration)
Terminate processing (and cancel the upstream publisher) after the given duration.Source<Out,Mat>
takeWithin(scala.concurrent.duration.FiniteDuration duration)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<Out,Mat>
throttle(int elements, java.time.Duration per)
Sends elements downstream with speed limited toelements/per
.Source<Out,Mat>
throttle(int cost, java.time.Duration per, int maximumBurst, Function<Out,java.lang.Integer> costCalculation, ThrottleMode mode)
Sends elements downstream with speed limited tocost/per
.Source<Out,Mat>
throttle(int elements, java.time.Duration per, int maximumBurst, ThrottleMode mode)
Sends elements downstream with speed limited toelements/per
.Source<Out,Mat>
throttle(int cost, java.time.Duration per, Function<Out,java.lang.Integer> costCalculation)
Sends elements downstream with speed limited tocost/per
.Source<Out,Mat>
throttle(int cost, scala.concurrent.duration.FiniteDuration per, int maximumBurst, Function<Out,java.lang.Integer> costCalculation, ThrottleMode mode)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<Out,Mat>
throttle(int elements, scala.concurrent.duration.FiniteDuration per, int maximumBurst, ThrottleMode mode)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.Source<Out,Mat>
throttleEven(int elements, java.time.Duration per, ThrottleMode mode)
Deprecated.Use throttle without `maximumBurst` parameter instead.Source<Out,Mat>
throttleEven(int cost, java.time.Duration per, scala.Function1<Out,java.lang.Object> costCalculation, ThrottleMode mode)
Deprecated.Use throttle without `maximumBurst` parameter instead.Source<Out,Mat>
throttleEven(int elements, scala.concurrent.duration.FiniteDuration per, ThrottleMode mode)
Deprecated.Use throttle without `maximumBurst` parameter instead.Source<Out,Mat>
throttleEven(int cost, scala.concurrent.duration.FiniteDuration per, scala.Function1<Out,java.lang.Object> costCalculation, ThrottleMode mode)
Deprecated.Use throttle without `maximumBurst` parameter instead.static <O> Source<O,Cancellable>
tick(java.time.Duration initialDelay, java.time.Duration interval, O tick)
Elements are emitted periodically with the specified interval.static <O> Source<O,Cancellable>
tick(scala.concurrent.duration.FiniteDuration initialDelay, scala.concurrent.duration.FiniteDuration interval, O tick)
Deprecated.Use the overloaded one which accepts java.time.Duration instead.<M> RunnableGraph<Mat>
to(Graph<SinkShape<Out>,M> sink)
<M,M2>
RunnableGraph<M2>toMat(Graph<SinkShape<Out>,M> sink, Function2<Mat,M,M2> combine)
java.lang.String
toString()
org.apache.pekko.stream.impl.LinearTraversalBuilder
traversalBuilder()
INTERNAL API.static <S,E>
Source<E,NotUsed>unfold(S s, Function<S,java.util.Optional<Pair<S,E>>> f)
Create aSource
that will unfold a value of typeS
into a pair of the next stateS
and output elements of typeE
.static <S,E>
Source<E,NotUsed>unfoldAsync(S s, Function<S,java.util.concurrent.CompletionStage<java.util.Optional<Pair<S,E>>>> f)
Same as<S,E>unfold(S,org.apache.pekko.japi.function.Function<S,java.util.Optional<org.apache.pekko.japi.Pair<S,E>>>)
, but uses an async function to generate the next state-element tuple.static <T,S>
Source<T,NotUsed>unfoldResource(Creator<S> create, Function<S,java.util.Optional<T>> read, Procedure<S> close)
Start a newSource
from some resource which can be opened, read and closed.static <T,S>
Source<T,NotUsed>unfoldResourceAsync(Creator<java.util.concurrent.CompletionStage<S>> create, Function<S,java.util.concurrent.CompletionStage<java.util.Optional<T>>> read, Function<S,java.util.concurrent.CompletionStage<Done>> close)
Start a newSource
from some resource which can be opened, read and closed.static <SuperOut,Out extends SuperOut,Mat>
Source<SuperOut,Mat>upcast(Source<Out,Mat> source)
Upcast a stream of elements to a stream of supertypes of that element.<T,M>
Source<T,Mat>via(Graph<FlowShape<Out,T>,M> flow)
Transform thisSource
by appending the given processing operators.<T,M,M2>
Source<T,M2>viaMat(Graph<FlowShape<Out,T>,M> flow, Function2<Mat,M,M2> combine)
Transform thisSource
by appending the given processing operators.Source<Out,Mat>
watch(ActorRef ref)
The operator fails with anpekko.stream.WatchedActorTerminatedException
if the target actor is terminated.<M> Source<Out,M>
watchTermination(Function2<Mat,java.util.concurrent.CompletionStage<Done>,M> matF)
Materializes toFuture[Done]
that completes on getting termination message.Source<Out,Mat>
wireTap(Procedure<Out> f)
This is a simplified version ofwireTap(Sink)
that takes only a simple procedure.Source<Out,Mat>
wireTap(Graph<SinkShape<Out>,?> that)
<M2,M3>
Source<Out,M3>wireTapMat(Graph<SinkShape<Out>,M2> that, Function2<Mat,M2,M3> matF)
Source<Out,Mat>
withAttributes(Attributes attr)
Replace the attributes of thisSource
with the given ones.<T> Source<Pair<Out,T>,Mat>
zip(Graph<SourceShape<T>,?> that)
Combine the elements of currentSource
and the given one into a stream of tuples.<U,A>
Source<Pair<A,U>,Mat>zipAll(Graph<SourceShape<U>,?> that, A thisElem, U thatElem)
Combine the elements of current flow and the givenSource
into a stream of tuples.<U,Mat2,Mat3,A>
Source<Pair<A,U>,Mat3>zipAllMat(Graph<SourceShape<U>,Mat2> that, A thisElem, U thatElem, scala.Function2<Mat,Mat2,Mat3> matF)
Combine the elements of current flow and the givenSource
into a stream of tuples.<T> Source<Pair<Out,T>,Mat>
zipLatest(Graph<SourceShape<T>,?> that)
Combine the elements of 2 streams into a stream of tuples, picking always the latest element of each.<T,M,M2>
Source<Pair<Out,T>,M2>zipLatestMat(Graph<SourceShape<T>,M> that, Function2<Mat,M,M2> matF)
Combine the elements of currentSource
and the given one into a stream of tuples, picking always the latest element of each.<Out2,Out3>
Source<Out3,Mat>zipLatestWith(Graph<SourceShape<Out2>,?> that, boolean eagerComplete, Function2<Out,Out2,Out3> combine)
Combine the elements of multiple streams into a stream of combined elements using a combiner function, picking always the latest of the elements of each source.<Out2,Out3>
Source<Out3,Mat>zipLatestWith(Graph<SourceShape<Out2>,?> that, Function2<Out,Out2,Out3> combine)
Combine the elements of multiple streams into a stream of combined elements using a combiner function, picking always the latest of the elements of each source.<Out2,Out3,M,M2>
Source<Out3,M2>zipLatestWithMat(Graph<SourceShape<Out2>,M> that, boolean eagerComplete, Function2<Out,Out2,Out3> combine, Function2<Mat,M,M2> matF)
Put together the elements of currentSource
and the given one into a stream of combined elements using a combiner function, picking always the latest of the elements of each source.<Out2,Out3,M,M2>
Source<Out3,M2>zipLatestWithMat(Graph<SourceShape<Out2>,M> that, Function2<Out,Out2,Out3> combine, Function2<Mat,M,M2> matF)
Put together the elements of currentSource
and the given one into a stream of combined elements using a combiner function, picking always the latest of the elements of each source.<T,M,M2>
Source<Pair<Out,T>,M2>zipMat(Graph<SourceShape<T>,M> that, Function2<Mat,M,M2> matF)
Combine the elements of currentSource
and the given one into a stream of tuples.static <T> Source<java.util.List<T>,NotUsed>
zipN(java.util.List<Source<T,?>> sources)
Combine the elements of multiple streams into a stream of lists.<Out2,Out3>
Source<Out3,Mat>zipWith(Graph<SourceShape<Out2>,?> that, Function2<Out,Out2,Out3> combine)
Put together the elements of currentSource
and the given one into a stream of combined elements using a combiner function.Source<Pair<Out,java.lang.Long>,Mat>
zipWithIndex()
Combine the elements of currentSource
into a stream of tuples consisting of all elements paired with their index.<Out2,Out3,M,M2>
Source<Out3,M2>zipWithMat(Graph<SourceShape<Out2>,M> that, Function2<Out,Out2,Out3> combine, Function2<Mat,M,M2> matF)
Put together the elements of currentSource
and the given one into a stream of combined elements using a combiner function.static <T,O>
Source<O,NotUsed>zipWithN(Function<java.util.List<T>,O> zipper, java.util.List<Source<T,?>> sources)
Combine the elements of multiple streams into a stream of lists using a combiner function.
-
-
-
Method Detail
-
empty
public static <O> Source<O,NotUsed> empty()
Create aSource
with no elements, i.e. an empty stream that is completed immediately for every connectedSink
.
-
empty
public static <T> Source<T,NotUsed> empty(java.lang.Class<T> clazz)
Create aSource
with no elements. The result is the same as callingSource.
empty()
-
maybe
public static <T> Source<T,java.util.concurrent.CompletableFuture<java.util.Optional<T>>> maybe()
Create aSource
which materializes aCompletableFuture
which controls what element will be emitted by the Source. If the materialized promise is completed with a filled Optional, that value will be produced downstream, followed by completion. If the materialized promise is completed with an empty Optional, no value will be produced downstream and completion will be signalled immediately. If the materialized promise is completed with a failure, then the source will fail with that error. If the downstream of this source cancels or fails before the promise has been completed, then the promise will be completed with an empty Optional.
-
fromPublisher
public static <O> Source<O,NotUsed> fromPublisher(org.reactivestreams.Publisher<O> publisher)
Helper to createSource
fromPublisher
.Construct a transformation starting with given publisher. The transformation steps are executed by a series of
Processor
instances that mediate the flow of elements downstream and the propagation of back-pressure upstream.
-
fromIterator
public static <O> Source<O,NotUsed> fromIterator(Creator<java.util.Iterator<O>> f)
Helper to createSource
fromIterator
. Example usage:List<Integer> data = new ArrayList<Integer>(); data.add(1); data.add(2); data.add(3); Source.from(() -> data.iterator());
Start a new
Source
from the given function that produces an Iterator. The produced stream of elements will continue until the iterator runs empty or fails during evaluation of thenext()
method. Elements are pulled out of the iterator in accordance with the demand coming from the downstream transformation steps.
-
fromJavaStream
public static <O,S extends java.util.stream.BaseStream<O,S>> Source<O,NotUsed> fromJavaStream(Creator<java.util.stream.BaseStream<O,S>> stream)
Creates a source that wraps a Java 8Stream
.
Source
uses a stream iterator to get all its elements and send them downstream on demand.
You can use
Source.async
to create asynchronous boundaries between synchronous java stream and the rest of flow.
-
cycle
public static <O> Source<O,NotUsed> cycle(Creator<java.util.Iterator<O>> f)
Helper to create 'cycled'Source
from iterator provider. Example usage:Source.cycle(() -> Arrays.asList(1, 2, 3).iterator());
Start a new 'cycled'
Source
from the given elements. The producer stream of elements will continue infinitely by repeating the sequence of elements provided by function parameter.
-
optionalVia
public static <SOut,FOut,SMat,FMat,Mat> Source<java.util.Optional<FOut>,Mat> optionalVia(Source<java.util.Optional<SOut>,SMat> source, Flow<SOut,FOut,FMat> viaFlow, Function2<SMat,FMat,Mat> combine)
Creates a Source from an existing base Source outputting an optional element and applying an additional viaFlow only if the element in the stream is defined.'''Emits when''' the provided viaFlow runs with defined elements
'''Backpressures when''' the viaFlow runs for the defined elements and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
- Parameters:
source
- The base source that outputs an optional elementviaFlow
- The flow that gets used if the optional element in is defined.combine
- How to combine the materialized values of source and viaFlow- Returns:
- a Source with the viaFlow applied onto defined elements of the flow. The output value is contained within an Optional which indicates whether the original source's element had viaFlow applied.
- Since:
- 1.1.0
-
from
public static <O> Source<O,NotUsed> from(java.lang.Iterable<O> iterable)
Helper to createSource
fromIterable
. Example usage:List<Integer> data = new ArrayList<Integer>(); data.add(1); data.add(2); data.add(3); Source.from(data);
Starts a new
Source
from the givenIterable
. This is like starting from an Iterator, but every Subscriber directly attached to the Publisher of this stream will see an individual flow of elements (always starting from the beginning) regardless of when they subscribed.Make sure that the
Iterable
is immutable or at least not modified after being used as aSource
. Otherwise the stream may fail withConcurrentModificationException
or other more subtle errors may occur.
-
fromArray
public static <T> Source<T,NotUsed> fromArray(java.lang.Object array)
Creates aSource
from an array, if the array is empty, the stream is completed immediately, otherwise, every element of the array will be emitted sequentially.- Since:
- 1.1.0
-
range
public static Source<java.lang.Integer,NotUsed> range(int start, int end)
CreatesSource
that represents integer values in range ''[start;end]'', step equals to 1. It allows to createSource
out of range as simply as on ScalaSource(1 to N)
Uses {@link scala.collection.immutable.Range.inclusive(Int, Int)} internally
-
range
public static Source<java.lang.Integer,NotUsed> range(int start, int end, int step)
CreatesSource
that represents integer values in range ''[start;end]'', with the given step. It allows to createSource
out of range as simply as on ScalaSource(1 to N)
Uses {@link scala.collection.immutable.Range.inclusive(Int, Int, Int)} internally
-
fromFuture
public static <O> Source<O,NotUsed> fromFuture(scala.concurrent.Future<O> future)
Deprecated.Use 'Source.future' instead. Since Akka 2.6.0.Start a newSource
from the givenFuture
. The stream will consist of one element when theFuture
is completed with a successful value, which may happen before or after materializing theFlow
. The stream terminates with a failure if theFuture
is completed with a failure.
-
fromCompletionStage
public static <O> Source<O,NotUsed> fromCompletionStage(java.util.concurrent.CompletionStage<O> future)
Deprecated.Use 'Source.completionStage' instead. Since Akka 2.6.0.Starts a newSource
from the givenCompletionStage
. The stream will consist of one element when theCompletionStage
is completed with a successful value, which may happen before or after materializing theFlow
. The stream terminates with a failure if theCompletionStage
is completed with a failure.
-
fromFutureSource
public static <T,M> Source<T,scala.concurrent.Future<M>> fromFutureSource(scala.concurrent.Future<? extends Graph<SourceShape<T>,M>> future)
Deprecated.Use 'Source.futureSource' (potentially together with `Source.fromGraph`) instead. Since Akka 2.6.0.Streams the elements of the given future source once it successfully completes. If theFuture
fails the stream is failed with the exception from the future. If downstream cancels before the stream completes the materializedFuture
will be failed with aStreamDetachedException
.
-
fromSourceCompletionStage
public static <T,M> Source<T,java.util.concurrent.CompletionStage<M>> fromSourceCompletionStage(java.util.concurrent.CompletionStage<? extends Graph<SourceShape<T>,M>> completion)
Deprecated.Use 'Source.completionStageSource' (potentially together with `Source.fromGraph`) instead. Since Akka 2.6.0.Streams the elements of an asynchronous source once its givenCompletionStage
completes. If theCompletionStage
fails the stream is failed with the exception from the future. If downstream cancels before the stream completes the materializedCompletionStage
will be failed with aStreamDetachedException
-
tick
public static <O> Source<O,Cancellable> tick(scala.concurrent.duration.FiniteDuration initialDelay, scala.concurrent.duration.FiniteDuration interval, O tick)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Elements are emitted periodically with the specified interval. The tick element will be delivered to downstream consumers that has requested any elements. If a consumer has not requested any elements at the point in time when the tick element is produced it will not receive that tick element later. It will receive new tick elements as soon as it has requested more elements.
-
tick
public static <O> Source<O,Cancellable> tick(java.time.Duration initialDelay, java.time.Duration interval, O tick)
Elements are emitted periodically with the specified interval. The tick element will be delivered to downstream consumers that has requested any elements. If a consumer has not requested any elements at the point in time when the tick element is produced it will not receive that tick element later. It will receive new tick elements as soon as it has requested more elements.
-
single
public static <T> Source<T,NotUsed> single(T element)
Create aSource
with one element. Every connectedSink
of this stream will see an individual stream consisting of one element.
-
repeat
public static <T> Source<T,NotUsed> repeat(T element)
Create aSource
that will continually emit the given element.
-
unfold
public static <S,E> Source<E,NotUsed> unfold(S s, Function<S,java.util.Optional<Pair<S,E>>> f)
Create aSource
that will unfold a value of typeS
into a pair of the next stateS
and output elements of typeE
.
-
unfoldAsync
public static <S,E> Source<E,NotUsed> unfoldAsync(S s, Function<S,java.util.concurrent.CompletionStage<java.util.Optional<Pair<S,E>>>> f)
Same as<S,E>unfold(S,org.apache.pekko.japi.function.Function<S,java.util.Optional<org.apache.pekko.japi.Pair<S,E>>>)
, but uses an async function to generate the next state-element tuple.
-
iterate
public static <T> Source<T,NotUsed> iterate(T seed, Predicate<T> p, Function<T,T> f)
Creates a sequentialSource
by iterating with the given predicate and function, starting with the givenseed
value. If the predicate returnsfalse
for the seed, theSource
completes with empty.
-
iterate
public static <T> Source<T,NotUsed> iterate(T seed, Function<T,T> f)
Creates an infinite sequentialSource
by iterating with the given function, starting with the givenseed
value.
-
failed
public static <T> Source<T,NotUsed> failed(java.lang.Throwable cause)
Create aSource
that immediately ends the stream with thecause
failure to every connectedSink
.
-
lazily
public static <T,M> Source<T,java.util.concurrent.CompletionStage<M>> lazily(Creator<Source<T,M>> create)
Deprecated.Use 'Source.lazySource' instead. Since Akka 2.6.0.Creates aSource
that is not materialized until there is downstream demand, when the source gets materialized the materialized future is completed with its value, if downstream cancels or fails without any demand thecreate
factory is never called and the materializedCompletionStage
is failed.
-
lazilyAsync
public static <T> Source<T,scala.concurrent.Future<NotUsed>> lazilyAsync(Creator<java.util.concurrent.CompletionStage<T>> create)
Deprecated.Use 'Source.lazyCompletionStage' instead. Since Akka 2.6.0.Creates aSource
from supplied future factory that is not called until downstream demand. When source gets materialized the materialized future is completed with the value from the factory. If downstream cancels or fails without any demand the create factory is never called and the materializedFuture
is failed.- See Also:
Source.lazily
-
future
public static <T> Source<T,NotUsed> future(scala.concurrent.Future<T> futureElement)
Emits a single value when the given ScalaFuture
is successfully completed and then completes the stream. The stream fails if theFuture
is completed with a failure.Here for Java interoperability, the normal use from Java should be
Source.completionStage
-
never
public static <T> Source<T,NotUsed> never()
Never emits any elements, never completes and never fails. This stream could be useful in tests.
-
completionStage
public static <T> Source<T,NotUsed> completionStage(java.util.concurrent.CompletionStage<T> completionStage)
Emits a single value when the givenCompletionStage
is successfully completed and then completes the stream. If theCompletionStage
is completed with a failure the stream is failed.
-
completionStageSource
public static <T,M> Source<T,java.util.concurrent.CompletionStage<M>> completionStageSource(java.util.concurrent.CompletionStage<Source<T,M>> completionStageSource)
Turn aCompletionStage[Source]
into a source that will emit the values of the source when the future completes successfully. If theCompletionStage
is completed with a failure the stream is failed.
-
lazySingle
public static <T> Source<T,NotUsed> lazySingle(Creator<T> create)
Defers invoking thecreate
function to create a single element until there is downstream demand.If the
create
function fails when invoked the stream is failed.Note that asynchronous boundaries (and other operators) in the stream may do pre-fetching which counter acts the laziness and will trigger the factory immediately.
The materialized future
Done
value is completed when thecreate
function has successfully been invoked, if the function throws the future materialized value is failed with that exception. If downstream cancels or fails before the function is invoked the materialized value is failed with apekko.stream.NeverMaterializedException
-
lazyCompletionStage
public static <T> Source<T,NotUsed> lazyCompletionStage(Creator<java.util.concurrent.CompletionStage<T>> create)
Defers invoking thecreate
function to create a future element until there is downstream demand.The returned future element will be emitted downstream when it completes, or fail the stream if the future is failed or the
create
function itself fails.Note that asynchronous boundaries (and other operators) in the stream may do pre-fetching which counter acts the laziness and will trigger the factory immediately.
The materialized future
Done
value is completed when thecreate
function has successfully been invoked and the future completes, if the function throws or the future fails the future materialized value is failed with that exception. If downstream cancels or fails before the function is invoked the materialized value is failed with apekko.stream.NeverMaterializedException
-
lazySource
public static <T,M> Source<T,java.util.concurrent.CompletionStage<M>> lazySource(Creator<Source<T,M>> create)
Defers invoking thecreate
function to create a future source until there is downstream demand.The returned source will emit downstream and behave just like it was the outer source. Downstream completes when the created source completes and fails when the created source fails.
Note that asynchronous boundaries (and other operators) in the stream may do pre-fetching which counter acts the laziness and will trigger the factory immediately.
The materialized future value is completed with the materialized value of the created source when it has been materialized. If the function throws or the source materialization fails the future materialized value is failed with the thrown exception.
If downstream cancels or fails before the function is invoked the materialized value is failed with a
pekko.stream.NeverMaterializedException
-
lazyCompletionStageSource
public static <T,M> Source<T,java.util.concurrent.CompletionStage<M>> lazyCompletionStageSource(Creator<java.util.concurrent.CompletionStage<Source<T,M>>> create)
Defers invoking thecreate
function to create a future source until there is downstream demand.The returned future source will emit downstream and behave just like it was the outer source when the
CompletionStage
completes successfully. Downstream completes when the created source completes and fails when the created source fails. If theCompletionStage
or thecreate
function fails the stream is failed.Note that asynchronous boundaries (and other operators) in the stream may do pre-fetching which counter acts the laziness and triggers the factory immediately.
The materialized
CompletionStage
value is completed with the materialized value of the created source when it has been materialized. If the function throws or the source materialization fails the future materialized value is failed with the thrown exception.If downstream cancels or fails before the function is invoked the materialized value is failed with a
pekko.stream.NeverMaterializedException
-
asSubscriber
public static <T> Source<T,org.reactivestreams.Subscriber<T>> asSubscriber()
Creates aSource
that is materialized as aSubscriber
-
actorRef
public static <T> Source<T,ActorRef> actorRef(Function<java.lang.Object,java.util.Optional<CompletionStrategy>> completionMatcher, Function<java.lang.Object,java.util.Optional<java.lang.Throwable>> failureMatcher, int bufferSize, OverflowStrategy overflowStrategy)
Creates aSource
that is materialized as anpekko.actor.ActorRef
. Messages sent to this actor will be emitted to the stream if there is demand from downstream, otherwise they will be buffered until request for demand is received.Depending on the defined
pekko.stream.OverflowStrategy
it might drop elements if there is no space available in the buffer.The strategy
pekko.stream.OverflowStrategy.backpressure
is not supported, and an IllegalArgument("Backpressure overflowStrategy not supported") will be thrown if it is passed as argument.The buffer can be disabled by using
bufferSize
of 0 and then received messages are dropped if there is no demand from downstream. WhenbufferSize
is 0 theoverflowStrategy
does not matter.The stream can be completed successfully by sending the actor reference a message that is matched by
completionMatcher
in which case already buffered elements will be signaled before signaling completion.The stream can be completed with failure by sending a message that is matched by
failureMatcher
. The extractedThrowable
will be used to fail the stream. In case the Actor is still draining its internal buffer (after having received a message matched bycompletionMatcher
) before signaling completion and it receives a message matched byfailureMatcher
, the failure will be signaled downstream immediately (instead of the completion signal).Note that terminating the actor without first completing it, either with a success or a failure, will prevent the actor triggering downstream completion and the stream will continue to run even though the source actor is dead. Therefore you should **not** attempt to manually terminate the actor such as with a
pekko.actor.PoisonPill
.The actor will be stopped when the stream is completed, failed or canceled from downstream, i.e. you can watch it to get notified when that happens.
See also
pekko.stream.scaladsl.Source.queue
.- Parameters:
completionMatcher
- catches the completion message to end the streamfailureMatcher
- catches the failure message to fail the streambufferSize
- The size of the buffer in element countoverflowStrategy
- Strategy that is used when incoming elements cannot fit inside the buffer
-
actorRef
public static <T> Source<T,ActorRef> actorRef(int bufferSize, OverflowStrategy overflowStrategy)
Deprecated.Use variant accepting completion and failure matchers. Since Akka 2.6.0.Creates aSource
that is materialized as anpekko.actor.ActorRef
. Messages sent to this actor will be emitted to the stream if there is demand from downstream, otherwise they will be buffered until request for demand is received.Depending on the defined
pekko.stream.OverflowStrategy
it might drop elements if there is no space available in the buffer.The strategy
pekko.stream.OverflowStrategy.backpressure
is not supported, and an IllegalArgument("Backpressure overflowStrategy not supported") will be thrown if it is passed as argument.The buffer can be disabled by using
bufferSize
of 0 and then received messages are dropped if there is no demand from downstream. WhenbufferSize
is 0 theoverflowStrategy
does not matter.The stream can be completed successfully by sending the actor reference a
pekko.actor.Status.Success
(whose content will be ignored) in which case already buffered elements will be signaled before signaling completion.The stream can be completed successfully by sending the actor reference a
pekko.actor.Status.Success
. If the content ispekko.stream.CompletionStrategy.immediately
the completion will be signaled immediately, otherwise if the content ispekko.stream.CompletionStrategy.draining
(or anything else) already buffered elements will be signaled before signaling completion. Sendingpekko.actor.PoisonPill
will signal completion immediately but this behavior is deprecated and scheduled to be removed.The stream can be completed with failure by sending a
pekko.actor.Status.Failure
to the actor reference. In case the Actor is still draining its internal buffer (after having received apekko.actor.Status.Success
) before signaling completion and it receives apekko.actor.Status.Failure
, the failure will be signaled downstream immediately (instead of the completion signal).Note that terminating the actor without first completing it, either with a success or a failure, will prevent the actor triggering downstream completion and the stream will continue to run even though the source actor is dead. Therefore you should **not** attempt to manually terminate the actor such as with a
pekko.actor.PoisonPill
.The actor will be stopped when the stream is completed, failed or canceled from downstream, i.e. you can watch it to get notified when that happens.
See also
pekko.stream.javadsl.Source.queue
.- Parameters:
bufferSize
- The size of the buffer in element countoverflowStrategy
- Strategy that is used when incoming elements cannot fit inside the buffer
-
actorRefWithBackpressure
public static <T> Source<T,ActorRef> actorRefWithBackpressure(java.lang.Object ackMessage, Function<java.lang.Object,java.util.Optional<CompletionStrategy>> completionMatcher, Function<java.lang.Object,java.util.Optional<java.lang.Throwable>> failureMatcher)
Creates aSource
that is materialized as anpekko.actor.ActorRef
. Messages sent to this actor will be emitted to the stream if there is demand from downstream, and a new message will only be accepted after the previous messages has been consumed and acknowledged back. The stream will complete with failure if a message is sent before the acknowledgement has been replied back.The stream can be completed with failure by sending a message that is matched by
failureMatcher
. The extractedThrowable
will be used to fail the stream. In case the Actor is still draining its internal buffer (after having received a message matched bycompletionMatcher
) before signaling completion and it receives a message matched byfailureMatcher
, the failure will be signaled downstream immediately (instead of the completion signal).The actor will be stopped when the stream is completed, failed or canceled from downstream, i.e. you can watch it to get notified when that happens.
-
actorRefWithAck
public static <T> Source<T,ActorRef> actorRefWithAck(java.lang.Object ackMessage, Function<java.lang.Object,java.util.Optional<CompletionStrategy>> completionMatcher, Function<java.lang.Object,java.util.Optional<java.lang.Throwable>> failureMatcher)
Deprecated.Use actorRefWithBackpressure insteadCreates aSource
that is materialized as anpekko.actor.ActorRef
. Messages sent to this actor will be emitted to the stream if there is demand from downstream, and a new message will only be accepted after the previous messages has been consumed and acknowledged back. The stream will complete with failure if a message is sent before the acknowledgement has been replied back.The stream can be completed with failure by sending a message that is matched by
failureMatcher
. The extractedThrowable
will be used to fail the stream. In case the Actor is still draining its internal buffer (after having received a message matched bycompletionMatcher
) before signaling completion and it receives a message matched byfailureMatcher
, the failure will be signaled downstream immediately (instead of the completion signal).The actor will be stopped when the stream is completed, failed or canceled from downstream, i.e. you can watch it to get notified when that happens.
-
actorRefWithAck
public static <T> Source<T,ActorRef> actorRefWithAck(java.lang.Object ackMessage)
Deprecated.Use actorRefWithBackpressure accepting completion and failure matchers. Since Akka 2.6.0.Creates aSource
that is materialized as anpekko.actor.ActorRef
. Messages sent to this actor will be emitted to the stream if there is demand from downstream, and a new message will only be accepted after the previous messages has been consumed and acknowledged back. The stream will complete with failure if a message is sent before the acknowledgement has been replied back.The stream can be completed successfully by sending the actor reference a
pekko.actor.Status.Success
. If the content ispekko.stream.CompletionStrategy.immediately
the completion will be signaled immediately, otherwise if the content ispekko.stream.CompletionStrategy.draining
(or anything else) already buffered element will be signaled before signaling completion.The stream can be completed with failure by sending a
pekko.actor.Status.Failure
to the actor reference. In case the Actor is still draining its internal buffer (after having received apekko.actor.Status.Success
) before signaling completion and it receives apekko.actor.Status.Failure
, the failure will be signaled downstream immediately (instead of the completion signal).The actor will be stopped when the stream is completed, failed or canceled from downstream, i.e. you can watch it to get notified when that happens.
-
fromGraph
public static <T,M> Source<T,M> fromGraph(Graph<SourceShape<T>,M> g)
A graph with the shape of a source logically is a source, this method makes it so also in type.
-
fromMaterializer
public static <T,M> Source<T,java.util.concurrent.CompletionStage<M>> fromMaterializer(java.util.function.BiFunction<Materializer,Attributes,Source<T,M>> factory)
-
setup
public static <T,M> Source<T,java.util.concurrent.CompletionStage<M>> setup(java.util.function.BiFunction<ActorMaterializer,Attributes,Source<T,M>> factory)
Deprecated.Use 'fromMaterializer' instead. Since Akka 2.6.0.
-
combine
public static <T,U> Source<U,NotUsed> combine(Source<T,?> first, Source<T,?> second, java.util.List<Source<T,?>> rest, Function<java.lang.Integer,? extends Graph<UniformFanInShape<T,U>,NotUsed>> fanInStrategy)
-
combineMat
public static <T,U,M1,M2,M> Source<U,M> combineMat(Source<T,M1> first, Source<T,M2> second, Function<java.lang.Integer,? extends Graph<UniformFanInShape<T,U>,NotUsed>> fanInStrategy, Function2<M1,M2,M> combine)
Combines two sources with fan-in strategy likeMerge
orConcat
and returnsSource
with a materialized value.
-
combine
public static <T,U,M> Source<U,java.util.List<M>> combine(java.util.List<? extends Graph<SourceShape<T>,M>> sources, Function<java.lang.Integer,Graph<UniformFanInShape<T,U>,NotUsed>> fanInStrategy)
- Since:
- 1.1.0
-
zipN
public static <T> Source<java.util.List<T>,NotUsed> zipN(java.util.List<Source<T,?>> sources)
Combine the elements of multiple streams into a stream of lists.
-
zipWithN
public static <T,O> Source<O,NotUsed> zipWithN(Function<java.util.List<T>,O> zipper, java.util.List<Source<T,?>> sources)
Combine the elements of multiple streams into a stream of lists using a combiner function.
-
queue
public static <T> Source<T,BoundedSourceQueue<T>> queue(int bufferSize)
Creates aSource
that is materialized as anpekko.stream.BoundedSourceQueue
. You can push elements to the queue and they will be emitted to the stream if there is demand from downstream, otherwise they will be buffered until request for demand is received. The buffer size is passed in as a parameter. Elements in the buffer will be discarded if downstream is terminated.Pushed elements may be dropped if there is no space available in the buffer. Elements will also be dropped if the queue is failed through the materialized
BoundedQueueSource
or theSource
is cancelled by the downstream. An element that was reported to beenqueued
is not guaranteed to be processed by the rest of the stream. If the queue is failed by callingBoundedQueueSource.fail
or the downstream cancels the stream, elements in the buffer are discarded.Acknowledgement of pushed elements is immediate.
pekko.stream.BoundedSourceQueue.offer
returnspekko.stream.QueueOfferResult
which is implemented as:QueueOfferResult.enqueued()
element was added to buffer, but may still be discarded later when the queue is failed or cancelledQueueOfferResult.dropped()
element was droppedQueueOfferResult.QueueClosed
the queue was completed withpekko.stream.BoundedSourceQueue.complete
QueueOfferResult.Failure
the queue was failed withpekko.stream.BoundedSourceQueue.fail
or if the stream failed- Parameters:
bufferSize
- size of the buffer in number of elements
-
queue
public static <T> Source<T,SourceQueueWithComplete<T>> queue(int bufferSize, OverflowStrategy overflowStrategy)
Creates aSource
that is materialized as anpekko.stream.javadsl.SourceQueueWithComplete
. You can push elements to the queue and they will be emitted to the stream if there is demand from downstream, otherwise they will be buffered until request for demand is received. Elements in the buffer will be discarded if downstream is terminated.Depending on the defined
pekko.stream.OverflowStrategy
it might drop elements if there is no space available in the buffer.Acknowledgement mechanism is available.
pekko.stream.javadsl.SourceQueueWithComplete.offer
returnsCompletionStage
which completes withQueueOfferResult.enqueued()
if element was added to buffer or sent downstream. It completes withQueueOfferResult.dropped()
if element was dropped. Can also complete withQueueOfferResult.Failure
- when stream failed orQueueOfferResult.QueueClosed
when downstream is completed.The strategy
pekko.stream.OverflowStrategy.backpressure
will not complete lastoffer():CompletionStage
call when buffer is full.Instead of using the strategy
pekko.stream.OverflowStrategy.dropNew
it's recommended to useSource.queue(bufferSize)
instead which returns aQueueOfferResult
synchronously.You can watch accessibility of stream with
pekko.stream.javadsl.SourceQueueWithComplete.watchCompletion
. It returns a future that completes with success when this operator is completed or fails when stream is failed.The buffer can be disabled by using
bufferSize
of 0 and then received message will wait for downstream demand unless there is another message waiting for downstream demand, in that case offer result will be completed according to the overflow strategy.The materialized SourceQueue may only be used from a single producer.
- Parameters:
bufferSize
- size of buffer in element countoverflowStrategy
- Strategy that is used when incoming elements cannot fit inside the buffer
-
queue
public static <T> Source<T,SourceQueueWithComplete<T>> queue(int bufferSize, OverflowStrategy overflowStrategy, int maxConcurrentOffers)
Creates aSource
that is materialized as anpekko.stream.javadsl.SourceQueueWithComplete
. You can push elements to the queue and they will be emitted to the stream if there is demand from downstream, otherwise they will be buffered until request for demand is received. Elements in the buffer will be discarded if downstream is terminated.Depending on the defined
pekko.stream.OverflowStrategy
it might drop elements if there is no space available in the buffer.Acknowledgement mechanism is available.
pekko.stream.javadsl.SourceQueueWithComplete.offer
returnsCompletionStage
which completes withQueueOfferResult.enqueued()
if element was added to buffer or sent downstream. It completes withQueueOfferResult.dropped()
if element was dropped. Can also complete withQueueOfferResult.Failure
- when stream failed orQueueOfferResult.QueueClosed
when downstream is completed.The strategy
pekko.stream.OverflowStrategy.backpressure
will not completemaxConcurrentOffers
number ofoffer():CompletionStage
call when buffer is full.Instead of using the strategy
pekko.stream.OverflowStrategy.dropNew
it's recommended to useSource.queue(bufferSize)
instead which returns aQueueOfferResult
synchronously.You can watch accessibility of stream with
pekko.stream.javadsl.SourceQueueWithComplete.watchCompletion
. It returns a future that completes with success when this operator is completed or fails when stream is failed.The buffer can be disabled by using
bufferSize
of 0 and then received message will wait for downstream demand unless there is another message waiting for downstream demand, in that case offer result will be completed according to the overflow strategy.The materialized SourceQueue may be used by up to maxConcurrentOffers concurrent producers.
- Parameters:
bufferSize
- size of buffer in element countoverflowStrategy
- Strategy that is used when incoming elements cannot fit inside the buffermaxConcurrentOffers
- maximum number of pending offers when buffer is full, should be greater than 0, not applicable whenOverflowStrategy.dropNew
is used
-
unfoldResource
public static <T,S> Source<T,NotUsed> unfoldResource(Creator<S> create, Function<S,java.util.Optional<T>> read, Procedure<S> close)
Start a newSource
from some resource which can be opened, read and closed. Interaction with resource happens in a blocking way.Example:
Source.unfoldResource( () -> new BufferedReader(new FileReader("...")), reader -> reader.readLine(), reader -> reader.close())
You can use the supervision strategy to handle exceptions for
read
function. All exceptions thrown bycreate
orclose
will fail the stream.Restart
supervision strategy will close and create blocking IO again. Default strategy isStop
which means that stream will be terminated on error inread
function by default.You can configure the default dispatcher for this Source by changing the
pekko.stream.materializer.blocking-io-dispatcher
or set it for a given Source by usingActorAttributes
.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.- Parameters:
create
- - function that is called on stream start and creates/opens resource.read
- - function that reads data from opened resource. It is called each time backpressure signal is received. Stream calls close and completes whenread
returns an empty Optional.close
- - function that closes resource
-
unfoldResourceAsync
public static <T,S> Source<T,NotUsed> unfoldResourceAsync(Creator<java.util.concurrent.CompletionStage<S>> create, Function<S,java.util.concurrent.CompletionStage<java.util.Optional<T>>> read, Function<S,java.util.concurrent.CompletionStage<Done>> close)
Start a newSource
from some resource which can be opened, read and closed. It's similar tounfoldResource
but takes functions that returnCompletionStage
instead of plain values.You can use the supervision strategy to handle exceptions for
read
function or failures of producedFutures
. All exceptions thrown bycreate
orclose
as well as fails of returned futures will fail the stream.Restart
supervision strategy will close and create resource. Default strategy isStop
which means that stream will be terminated on error inread
function (or future) by default.You can configure the default dispatcher for this Source by changing the
pekko.stream.materializer.blocking-io-dispatcher
or set it for a given Source by usingActorAttributes
.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.- Parameters:
create
- - function that is called on stream start and creates/opens resource.read
- - function that reads data from opened resource. It is called each time backpressure signal is received. Stream calls close and completes whenCompletionStage
from read function returns an empty Optional.close
- - function that closes resource
-
upcast
public static <SuperOut,Out extends SuperOut,Mat> Source<SuperOut,Mat> upcast(Source<Out,Mat> source)
Upcast a stream of elements to a stream of supertypes of that element. Useful in combination with fan-in operators where you do not want to pay the cost of casting each element in amap
.Example:
Source<Apple, NotUsed> apples = Source.single(new Apple()); Source<Orange, NotUsed> oranges = Source.single(new Orange()); Source<Fruit, NotUsed> appleFruits = Source.upcast(apples); Source<Fruit, NotUsed> orangeFruits = Source.upcast(oranges); Source<Fruit, NotUsed> fruits = appleFruits.merge(orangeFruits);
- Returns:
- A source with the supertype as elements
-
mergePrioritizedN
public static <T> Source<T,NotUsed> mergePrioritizedN(java.util.List<Pair<Source<T,?>,java.lang.Integer>> sourcesAndPriorities, boolean eagerComplete)
Merge multipleSource
s. Prefer the sources depending on the 'priority' parameters. The provided sources and priorities must have the same size and order.'''emits''' when one of the inputs has an element available, preferring inputs based on the 'priority' parameters if both have elements available
'''backpressures''' when downstream backpressures
'''completes''' when both upstreams complete (This behavior is changeable to completing when any upstream completes by setting
eagerComplete=true
.)'''Cancels when''' downstream cancels
-
concatAllLazy
public Source<Out,Mat> concatAllLazy(Graph<SourceShape<Out>,?>... those)
Concatenate the givenSource
s to this one, meaning that once this Flow’s input is exhausted and all result elements have been generated, the Source’s elements will be produced.Note that the
Source
s are materialized together with this Flow. Iflazy
materialization is what is needed the operator can be combined with for exampleSource.lazySource
to defer materialization ofthat
until the time when this source completes.The second source is then kept from producing elements by asserting back-pressure until its time comes.
For a concat operator that is detached, use
concat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>)
If this
Source
gets upstream error - no elements from the givenSource
s will be pulled.'''Emits when''' element is available from current stream or from the given
Source
s when current is completed'''Backpressures when''' downstream backpressures
'''Completes when''' all the given
Source
s completes'''Cancels when''' downstream cancels
-
alsoToAll
public Source<Out,Mat> alsoToAll(Graph<SinkShape<Out>,?>... those)
Attaches the givenSink
s to thisSource
, meaning that elements that passes through will also be sent to all thoseSink
s.It is similar to
wireTap(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SinkShape<Out>, ?>)
but will backpressure instead of dropping elements when the givenSink
s is not ready.'''Emits when''' element is available and demand exists both from the Sinks and the downstream.
'''Backpressures when''' downstream or any of the
Sink
s backpressures'''Completes when''' upstream completes
'''Cancels when''' downstream or any of the
Sink
s cancels
-
shape
public SourceShape<Out> shape()
Description copied from interface:Graph
The shape of a graph is all that is externally visible: its inlets and outlets.
-
traversalBuilder
public org.apache.pekko.stream.impl.LinearTraversalBuilder traversalBuilder()
Description copied from interface:Graph
INTERNAL API.Every materializable element must be backed by a stream layout module
- Specified by:
traversalBuilder
in interfaceGraph<Out,Mat>
-
toString
public java.lang.String toString()
- Overrides:
toString
in classjava.lang.Object
-
asScala
public Source<Out,Mat> asScala()
Converts this Java DSL element to its Scala DSL counterpart.
-
mapMaterializedValue
public <Mat2> Source<Out,Mat2> mapMaterializedValue(Function<Mat,Mat2> f)
Transform only the materialized value of this Source, leaving all other properties as they were.
-
preMaterialize
public Pair<Mat,Source<Out,NotUsed>> preMaterialize(ClassicActorSystemProvider systemProvider)
Materializes this Source, immediately returning (1) its materialized value, and (2) a new Source that can be used to consume elements from the newly materialized Source.Note that the
ActorSystem
can be used as thesystemProvider
parameter.
-
preMaterialize
public Pair<Mat,Source<Out,NotUsed>> preMaterialize(Materializer materializer)
Materializes this Source, immediately returning (1) its materialized value, and (2) a new Source that can be used to consume elements from the newly materialized Source.Prefer the method taking an
ActorSystem
unless you have special requirements.
-
via
public <T,M> Source<T,Mat> via(Graph<FlowShape<Out,T>,M> flow)
Transform thisSource
by appending the given processing operators.
The materialized value of the combined+----------------------------+ | Resulting Source | | | | +------+ +------+ | | | | | | | | | this | ~Out~> | flow | ~~> T | | | | | | | +------+ +------+ | +----------------------------+
Flow
will be the materialized value of the current flow (ignoring the other Flow’s value), useviaMat
if a different strategy is needed.
-
viaMat
public <T,M,M2> Source<T,M2> viaMat(Graph<FlowShape<Out,T>,M> flow, Function2<Mat,M,M2> combine)
Transform thisSource
by appending the given processing operators.
The+----------------------------+ | Resulting Source | | | | +------+ +------+ | | | | | | | | | this | ~Out~> | flow | ~~> T | | | | | | | +------+ +------+ | +----------------------------+
combine
function is used to compose the materialized values of this flow and that flow into the materialized value of the resulting Flow.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
to
public <M> RunnableGraph<Mat> to(Graph<SinkShape<Out>,M> sink)
Connect thisSource
to aSink
, concatenating the processing steps of both.
The materialized value of the combined+----------------------------+ | Resulting RunnableGraph | | | | +------+ +------+ | | | | | | | | | this | ~Out~> | sink | | | | | | | | | +------+ +------+ | +----------------------------+
Sink
will be the materialized value of the current flow (ignoring the given Sink’s value), usetoMat
if a different strategy is needed.
-
toMat
public <M,M2> RunnableGraph<M2> toMat(Graph<SinkShape<Out>,M> sink, Function2<Mat,M,M2> combine)
Connect thisSource
to aSink
, concatenating the processing steps of both.
The+----------------------------+ | Resulting RunnableGraph | | | | +------+ +------+ | | | | | | | | | this | ~Out~> | sink | | | | | | | | | +------+ +------+ | +----------------------------+
combine
function is used to compose the materialized values of this flow and that Sink into the materialized value of the resulting Sink.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
run
public java.util.concurrent.CompletionStage<Done> run(Materializer materializer)
Connect thisSource
to theSink.ignore
and run it. Elements from the stream will be consumed and discarded.Note that the
ActorSystem
can be used as thematerializer
parameter to use thepekko.stream.SystemMaterializer
for running the stream.
-
run
public java.util.concurrent.CompletionStage<Done> run(ClassicActorSystemProvider systemProvider)
Connect thisSource
to theSink.ignore
and run it. Elements from the stream will be consumed and discarded.Note that the
ActorSystem
can be used as thesystemProvider
parameter to use thepekko.stream.SystemMaterializer
for running the stream.
-
runWith
public <M> M runWith(Graph<SinkShape<Out>,M> sink, ClassicActorSystemProvider systemProvider)
Connect thisSource
to aSink
and run it. The returned value is the materialized value of theSink
, e.g. thePublisher
of aSink.asPublisher
.Note that the classic or typed
ActorSystem
can be used as thesystemProvider
parameter.
-
runWith
public <M> M runWith(Graph<SinkShape<Out>,M> sink, Materializer materializer)
Connect thisSource
to aSink
and run it. The returned value is the materialized value of theSink
, e.g. thePublisher
of aSink.asPublisher
.Prefer the method taking an
ActorSystem
unless you have special requirements
-
runFold
public <U> java.util.concurrent.CompletionStage<U> runFold(U zero, Function2<U,Out,U> f, ClassicActorSystemProvider systemProvider)
Shortcut for running thisSource
with a fold function. The given function is invoked for every received element, giving it its previous output (or the givenzero
value) and the element as input. The returnedCompletionStage
will be completed with value of the final function evaluation when the input stream ends, or completed withFailure
if there is a failure is signaled in the stream.Note that the classic or typed
ActorSystem
can be used as thesystemProvider
parameter.
-
runFold
public <U> java.util.concurrent.CompletionStage<U> runFold(U zero, Function2<U,Out,U> f, Materializer materializer)
Shortcut for running thisSource
with a fold function. The given function is invoked for every received element, giving it its previous output (or the givenzero
value) and the element as input. The returnedCompletionStage
will be completed with value of the final function evaluation when the input stream ends, or completed withFailure
if there is a failure is signaled in the stream.Prefer the method taking an ActorSystem unless you have special requirements.
-
runFoldAsync
public <U> java.util.concurrent.CompletionStage<U> runFoldAsync(U zero, Function2<U,Out,java.util.concurrent.CompletionStage<U>> f, ClassicActorSystemProvider systemProvider)
Shortcut for running thisSource
with an asynchronous fold function. The given function is invoked for every received element, giving it its previous output (or the givenzero
value) and the element as input. The returnedCompletionStage
will be completed with value of the final function evaluation when the input stream ends, or completed withFailure
if there is a failure is signaled in the stream.Note that the classic or typed
ActorSystem
can be used as thesystemProvider
parameter.
-
runFoldAsync
public <U> java.util.concurrent.CompletionStage<U> runFoldAsync(U zero, Function2<U,Out,java.util.concurrent.CompletionStage<U>> f, Materializer materializer)
Shortcut for running thisSource
with an asynchronous fold function. The given function is invoked for every received element, giving it its previous output (or the givenzero
value) and the element as input. The returnedCompletionStage
will be completed with value of the final function evaluation when the input stream ends, or completed withFailure
if there is a failure is signaled in the stream.Prefer the method taking an
ActorSystem
unless you have special requirements
-
runReduce
public java.util.concurrent.CompletionStage<Out> runReduce(Function2<Out,Out,Out> f, ClassicActorSystemProvider systemProvider)
Shortcut for running thisSource
with a reduce function. The given function is invoked for every received element, giving it its previous output (from the second ones) an the element as input. The returnedCompletionStage
will be completed with value of the final function evaluation when the input stream ends, or completed withFailure
if there is a failure is signaled in the stream.If the stream is empty (i.e. completes before signalling any elements), the reduce operator will fail its downstream with a
NoSuchElementException
, which is semantically in-line with that Scala's standard library collections do in such situations.Note that the classic or typed
ActorSystem
can be used as thesystemProvider
parameter.
-
runReduce
public java.util.concurrent.CompletionStage<Out> runReduce(Function2<Out,Out,Out> f, Materializer materializer)
Shortcut for running thisSource
with a reduce function. The given function is invoked for every received element, giving it its previous output (from the second ones) an the element as input. The returnedCompletionStage
will be completed with value of the final function evaluation when the input stream ends, or completed withFailure
if there is a failure is signaled in the stream.If the stream is empty (i.e. completes before signalling any elements), the reduce operator will fail its downstream with a
NoSuchElementException
, which is semantically in-line with that Scala's standard library collections do in such situations.Prefer the method taking an
ActorSystem
unless you have special requirements
-
concat
public <M> Source<Out,Mat> concat(Graph<SourceShape<Out>,M> that)
Concatenate thisSource
with the given one, meaning that once current is exhausted and all result elements have been generated, the given source elements will be produced.Note that the
Source
is materialized together with this Flow and is "detached" meaning it will in effect behave as a one element buffer in front of both the sources, that eagerly demands an element on start (so it can not be combined withSource.lazy
to defer materialization ofthat
).The second source is then kept from producing elements by asserting back-pressure until its time comes.
When needing a concat operator that is not detached use
concatLazy(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>)
'''Emits when''' element is available from current source or from the given
Source
when current is completed'''Backpressures when''' downstream backpressures
'''Completes when''' given
Source
completes'''Cancels when''' downstream cancels
-
concatLazy
public <M> Source<Out,Mat> concatLazy(Graph<SourceShape<Out>,M> that)
Concatenate the givenSource
to thisFlow
, meaning that once this Flow’s input is exhausted and all result elements have been generated, the Source’s elements will be produced.Note that the
Source
is materialized together with this Flow. Iflazy
materialization is what is needed the operator can be combined with for exampleSource.lazySource
to defer materialization ofthat
until the time when this source completes.The second source is then kept from producing elements by asserting back-pressure until its time comes.
For a concat operator that is detached, use
concat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>)
If this
Source
gets upstream error - no elements from the givenSource
will be pulled.'''Emits when''' element is available from current stream or from the given
Source
when current is completed'''Backpressures when''' downstream backpressures
'''Completes when''' given
Source
completes'''Cancels when''' downstream cancels
-
concatAllLazy
public Source<Out,Mat> concatAllLazy(scala.collection.immutable.Seq<Graph<SourceShape<Out>,?>> those)
Concatenate the givenSource
s to this one, meaning that once this Flow’s input is exhausted and all result elements have been generated, the Source’s elements will be produced.Note that the
Source
s are materialized together with this Flow. Iflazy
materialization is what is needed the operator can be combined with for exampleSource.lazySource
to defer materialization ofthat
until the time when this source completes.The second source is then kept from producing elements by asserting back-pressure until its time comes.
For a concat operator that is detached, use
concat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>)
If this
Source
gets upstream error - no elements from the givenSource
s will be pulled.'''Emits when''' element is available from current stream or from the given
Source
s when current is completed'''Backpressures when''' downstream backpressures
'''Completes when''' all the given
Source
s completes'''Cancels when''' downstream cancels
-
concatMat
public <M,M2> Source<Out,M2> concatMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
Concatenate thisSource
with the given one, meaning that once current is exhausted and all result elements have been generated, the given source elements will be produced.Note that the
Source
is materialized together with this Flow and is "detached" meaning it will in effect behave as a one element buffer in front of both the sources, that eagerly demands an element on start (so it can not be combined withSource.lazy
to defer materialization ofthat
).The second source is then kept from producing elements by asserting back-pressure until its time comes.
When needing a concat operator that is not detached use
concatLazyMat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>, org.apache.pekko.japi.function.Function2<Mat, M, M2>)
It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
concatLazyMat
public <M,M2> Source<Out,M2> concatLazyMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
Concatenate the givenSource
to thisFlow
, meaning that once this Flow’s input is exhausted and all result elements have been generated, the Source’s elements will be produced.Note that the
Source
is materialized together with this Flow, iflazy
materialization is what is needed the operator can be combined withSource.lazy
to defer materialization ofthat
.The second source is then kept from producing elements by asserting back-pressure until its time comes.
For a concat operator that is detached, use
concatMat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>, org.apache.pekko.japi.function.Function2<Mat, M, M2>)
-
prepend
public <M> Source<Out,Mat> prepend(Graph<SourceShape<Out>,M> that)
Prepend the givenSource
to this one, meaning that once the given source is exhausted and all result elements have been generated, the current source's elements will be produced.Note that the
Source
is materialized together with this Flow and is "detached" meaning in effect behave as a one element buffer in front of both the sources, that eagerly demands an element on start (so it can not be combined withSource.lazy
to defer materialization ofthat
).This flow will then be kept from producing elements by asserting back-pressure until its time comes.
When needing a prepend operator that is not detached use
prependLazy(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>)
'''Emits when''' element is available from current source or from the given
Source
when current is completed'''Backpressures when''' downstream backpressures
'''Completes when''' given
Source
completes'''Cancels when''' downstream cancels
-
prependLazy
public <M> Source<Out,Mat> prependLazy(Graph<SourceShape<Out>,M> that)
Prepend the givenSource
to thisFlow
, meaning that before elements are generated from this Flow, the Source's elements will be produced until it is exhausted, at which point Flow elements will start being produced.Note that the
Source
is materialized together with this Flow and will then be kept from producing elements by asserting back-pressure until its time comes.When needing a prepend operator that is also detached use
prepend(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>)
If the given
Source
gets upstream error - no elements from thisFlow
will be pulled.'''Emits when''' element is available from the given
Source
or from current stream when theSource
is completed'''Backpressures when''' downstream backpressures
'''Completes when''' this
Flow
completes'''Cancels when''' downstream cancels
-
prependMat
public <M,M2> Source<Out,M2> prependMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
Prepend the givenSource
to this one, meaning that once the given source is exhausted and all result elements have been generated, the current source's elements will be produced.Note that this Flow will be materialized together with the
Source
and just kept from producing elements by asserting back-pressure until its time comes.When needing a prepend operator that is not detached use
prependLazyMat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>, org.apache.pekko.japi.function.Function2<Mat, M, M2>)
It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
prependLazyMat
public <M,M2> Source<Out,M2> prependLazyMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
Prepend the givenSource
to thisFlow
, meaning that before elements are generated from this Flow, the Source's elements will be produced until it is exhausted, at which point Flow elements will start being produced.Note that the
Source
is materialized together with this Flow.This flow will then be kept from producing elements by asserting back-pressure until its time comes.
When needing a prepend operator that is detached use
prependMat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SourceShape<Out>, M>, org.apache.pekko.japi.function.Function2<Mat, M, M2>)
-
orElse
public <M> Source<Out,Mat> orElse(Graph<SourceShape<Out>,M> secondary)
Provides a secondary source that will be consumed if this source completes without any elements passing by. As soon as the first element comes through this stream, the alternative will be cancelled.Note that this Flow will be materialized together with the
Source
and just kept from producing elements by asserting back-pressure until its time comes or it gets cancelled.On errors the operator is failed regardless of source of the error.
'''Emits when''' element is available from first stream or first stream closed without emitting any elements and an element is available from the second stream
'''Backpressures when''' downstream backpressures
'''Completes when''' the primary stream completes after emitting at least one element, when the primary stream completes without emitting and the secondary stream already has completed or when the secondary stream completes
'''Cancels when''' downstream cancels and additionally the alternative is cancelled as soon as an element passes by from this stream.
-
orElseMat
public <M,M2> Source<Out,M2> orElseMat(Graph<SourceShape<Out>,M> secondary, Function2<Mat,M,M2> matF)
Provides a secondary source that will be consumed if this source completes without any elements passing by. As soon as the first element comes through this stream, the alternative will be cancelled.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
alsoTo
public Source<Out,Mat> alsoTo(Graph<SinkShape<Out>,?> that)
Attaches the givenSink
to thisSource
, meaning that elements that passes through will also be sent to theSink
.It is similar to
wireTap(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SinkShape<Out>, ?>)
but will backpressure instead of dropping elements when the givenSink
is not ready.'''Emits when''' element is available and demand exists both from the Sink and the downstream.
'''Backpressures when''' downstream or Sink backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream or Sink cancels
-
alsoToAll
public Source<Out,Mat> alsoToAll(scala.collection.immutable.Seq<Graph<SinkShape<Out>,?>> those)
Attaches the givenSink
s to thisSource
, meaning that elements that passes through will also be sent to all thoseSink
s.It is similar to
wireTap(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SinkShape<Out>, ?>)
but will backpressure instead of dropping elements when the givenSink
s is not ready.'''Emits when''' element is available and demand exists both from the Sinks and the downstream.
'''Backpressures when''' downstream or any of the
Sink
s backpressures'''Completes when''' upstream completes
'''Cancels when''' downstream or any of the
Sink
s cancels
-
alsoToMat
public <M2,M3> Source<Out,M3> alsoToMat(Graph<SinkShape<Out>,M2> that, Function2<Mat,M2,M3> matF)
Attaches the givenSink
to thisFlow
, meaning that elements that passes through will also be sent to theSink
.It is similar to
wireTapMat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SinkShape<Out>, M2>, org.apache.pekko.japi.function.Function2<Mat, M2, M3>)
but will backpressure instead of dropping elements when the givenSink
is not ready.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
divertTo
public Source<Out,Mat> divertTo(Graph<SinkShape<Out>,?> that, Predicate<Out> when)
Attaches the givenSink
to thisFlow
, meaning that elements will be sent to theSink
instead of being passed through if the predicatewhen
returnstrue
.'''Emits when''' emits when an element is available from the input and the chosen output has demand
'''Backpressures when''' the currently chosen output back-pressures
'''Completes when''' upstream completes and no output is pending
'''Cancels when''' any of the downstreams cancel
-
divertToMat
public <M2,M3> Source<Out,M3> divertToMat(Graph<SinkShape<Out>,M2> that, Predicate<Out> when, Function2<Mat,M2,M3> matF)
-
wireTap
public Source<Out,Mat> wireTap(Graph<SinkShape<Out>,?> that)
Attaches the givenSink
to thisFlow
as a wire tap, meaning that elements that pass through will also be sent to the wire-tap Sink, without the latter affecting the mainline flow. If the wire-tap Sink backpressures, elements that would've been sent to it will be dropped instead.It is similar to
alsoTo(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SinkShape<Out>, ?>)
which does backpressure instead of dropping elements.'''Emits when''' element is available and demand exists from the downstream; the element will also be sent to the wire-tap Sink if there is demand.
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
wireTapMat
public <M2,M3> Source<Out,M3> wireTapMat(Graph<SinkShape<Out>,M2> that, Function2<Mat,M2,M3> matF)
Attaches the givenSink
to thisFlow
as a wire tap, meaning that elements that pass through will also be sent to the wire-tap Sink, without the latter affecting the mainline flow. If the wire-tap Sink backpressures, elements that would've been sent to it will be dropped instead.It is similar to
alsoToMat(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SinkShape<Out>, M2>, org.apache.pekko.japi.function.Function2<Mat, M2, M3>)
which does backpressure instead of dropping elements.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
interleave
public Source<Out,Mat> interleave(Graph<SourceShape<Out>,?> that, int segmentSize)
Interleave is a deterministic merge of the givenSource
with elements of thisSource
. It first emitssegmentSize
number of elements from this flow to downstream, then - same amount forthat
source, then repeat process.Example:
Source.from(Arrays.asList(1, 2, 3)).interleave(Source.from(Arrays.asList(4, 5, 6, 7), 2) // 1, 2, 4, 5, 3, 6, 7
After one of sources is complete than all the rest elements will be emitted from the second one
If one of sources gets upstream error - stream completes with failure.
'''Emits when''' element is available from the currently consumed upstream
'''Backpressures when''' downstream backpressures. Signal to current upstream, switch to next upstream when received
segmentSize
elements'''Completes when''' this
Source
and given one completes'''Cancels when''' downstream cancels
-
interleave
public Source<Out,Mat> interleave(Graph<SourceShape<Out>,?> that, int segmentSize, boolean eagerClose)
Interleave is a deterministic merge of the givenSource
with elements of thisFlow
. It first emitssegmentSize
number of elements from this flow to downstream, then - same amount forthat
source, then repeat process.If eagerClose is false and one of the upstreams complete the elements from the other upstream will continue passing through the interleave operator. If eagerClose is true and one of the upstream complete interleave will cancel the other upstream and complete itself.
If this
Flow
orSource
gets upstream error - stream completes with failure.'''Emits when''' element is available from the currently consumed upstream
'''Backpressures when''' downstream backpressures. Signal to current upstream, switch to next upstream when received
segmentSize
elements'''Completes when''' the
Flow
and givenSource
completes'''Cancels when''' downstream cancels
-
interleaveMat
public <M,M2> Source<Out,M2> interleaveMat(Graph<SourceShape<Out>,M> that, int segmentSize, Function2<Mat,M,M2> matF)
Interleave is a deterministic merge of the givenSource
with elements of thisSource
. It first emitssegmentSize
number of elements from this flow to downstream, then - same amount forthat
source, then repeat process.After one of sources is complete than all the rest elements will be emitted from the second one
If one of sources gets upstream error - stream completes with failure.
It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
interleaveMat
public <M,M2> Source<Out,M2> interleaveMat(Graph<SourceShape<Out>,M> that, int segmentSize, boolean eagerClose, Function2<Mat,M,M2> matF)
Interleave is a deterministic merge of the givenSource
with elements of thisSource
. It first emitssegmentSize
number of elements from this flow to downstream, then - same amount forthat
source, then repeat process.If eagerClose is false and one of the upstreams complete the elements from the other upstream will continue passing through the interleave operator. If eagerClose is true and one of the upstream complete interleave will cancel the other upstream and complete itself.
If this
Flow
orSource
gets upstream error - stream completes with failure.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
interleaveAll
public Source<Out,Mat> interleaveAll(java.util.List<? extends Graph<SourceShape<Out>,?>> those, int segmentSize, boolean eagerClose)
Interleave is a deterministic merge of the givenSource
with elements of thisFlow
. It first emitssegmentSize
number of elements from this flow to downstream, then - same amount forthat
source, then repeat process.If eagerClose is false and one of the upstreams complete the elements from the other upstream will continue passing through the interleave operator. If eagerClose is true and one of the upstream complete interleave will cancel the other upstream and complete itself.
If this
Flow
orSource
gets upstream error - stream completes with failure.'''Emits when''' element is available from the currently consumed upstream
'''Backpressures when''' downstream backpressures. Signal to current upstream, switch to next upstream when received
segmentSize
elements'''Completes when''' the
Flow
and givenSource
completes'''Cancels when''' downstream cancels
-
merge
public Source<Out,Mat> merge(Graph<SourceShape<Out>,?> that)
Merge the givenSource
to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.'''Emits when''' one of the inputs has an element available
'''Backpressures when''' downstream backpressures
'''Completes when''' all upstreams complete
'''Cancels when''' downstream cancels
-
merge
public Source<Out,Mat> merge(Graph<SourceShape<Out>,?> that, boolean eagerComplete)
Merge the givenSource
to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.'''Emits when''' one of the inputs has an element available
'''Backpressures when''' downstream backpressures
'''Completes when''' all upstreams complete (eagerComplete=false) or one upstream completes (eagerComplete=true), default value is
false
'''Cancels when''' downstream cancels
-
mergeMat
public <M,M2> Source<Out,M2> mergeMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF)
Merge the givenSource
to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
mergeMat
public <M,M2> Source<Out,M2> mergeMat(Graph<SourceShape<Out>,M> that, Function2<Mat,M,M2> matF, boolean eagerComplete)
Merge the givenSource
to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
mergeAll
public Source<Out,Mat> mergeAll(java.util.List<? extends Graph<SourceShape<Out>,?>> those, boolean eagerComplete)
Merge the givenSource
s to the current one, taking elements as they arrive from input streams, picking randomly when several elements ready.'''Emits when''' one of the inputs has an element available
'''Backpressures when''' downstream backpressures
'''Completes when''' all upstreams complete (eagerComplete=false) or one upstream completes (eagerComplete=true), default value is
false
'''Cancels when''' downstream cancels
-
mergeLatest
public <M> Source<java.util.List<Out>,Mat> mergeLatest(Graph<SourceShape<Out>,M> that, boolean eagerComplete)
MergeLatest joins elements from N input streams into stream of lists of size N. i-th element in list is the latest emitted element from i-th input stream. MergeLatest emits list for each element emitted from some input stream, but only after each input stream emitted at least one element.'''Emits when''' an element is available from some input and each input emits at least one element from stream start
'''Completes when''' all upstreams complete (eagerClose=false) or one upstream completes (eagerClose=true)
-
mergeLatestMat
public <Mat2,Mat3> Source<java.util.List<Out>,Mat3> mergeLatestMat(Graph<SourceShape<Out>,Mat2> that, boolean eagerComplete, Function2<Mat,Mat2,Mat3> matF)
MergeLatest joins elements from N input streams into stream of lists of size N. i-th element in list is the latest emitted element from i-th input stream. MergeLatest emits list for each element emitted from some input stream, but only after each input stream emitted at least one element.
-
mergePreferred
public <M> Source<Out,Mat> mergePreferred(Graph<SourceShape<Out>,M> that, boolean preferred, boolean eagerComplete)
Merge two sources. Prefer one source if both sources have elements ready.'''emits''' when one of the inputs has an element available. If multiple have elements available, prefer the 'right' one when 'preferred' is 'true', or the 'left' one when 'preferred' is 'false'.
'''backpressures''' when downstream backpressures
'''completes''' when all upstreams complete (This behavior is changeable to completing when any upstream completes by setting
eagerComplete=true
.)
-
mergePreferredMat
public <Mat2,Mat3> Source<Out,Mat3> mergePreferredMat(Graph<SourceShape<Out>,Mat2> that, boolean preferred, boolean eagerComplete, Function2<Mat,Mat2,Mat3> matF)
Merge two sources. Prefer one source if both sources have elements ready.
-
mergePrioritized
public <M> Source<Out,Mat> mergePrioritized(Graph<SourceShape<Out>,M> that, int leftPriority, int rightPriority, boolean eagerComplete)
Merge two sources. Prefer the sources depending on the 'priority' parameters.'''emits''' when one of the inputs has an element available, preferring inputs based on the 'priority' parameters if both have elements available
'''backpressures''' when downstream backpressures
'''completes''' when both upstreams complete (This behavior is changeable to completing when any upstream completes by setting
eagerComplete=true
.)
-
mergePrioritizedMat
public <Mat2,Mat3> Source<Out,Mat3> mergePrioritizedMat(Graph<SourceShape<Out>,Mat2> that, int leftPriority, int rightPriority, boolean eagerComplete, Function2<Mat,Mat2,Mat3> matF)
Merge multiple sources. Prefer the sources depending on the 'priority' parameters.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
mergeSorted
public <M> Source<Out,Mat> mergeSorted(Graph<SourceShape<Out>,M> that, java.util.Comparator<Out> comp)
Merge the givenSource
to thisSource
, taking elements as they arrive from input streams, picking always the smallest of the available elements (waiting for one element from each side to be available). This means that possible contiguity of the input streams is not exploited to avoid waiting for elements, this merge will block when one of the inputs does not have more elements (and does not complete).'''Emits when''' all of the inputs have an element available
'''Backpressures when''' downstream backpressures
'''Completes when''' all upstreams complete
'''Cancels when''' downstream cancels
-
mergeSortedMat
public <Mat2,Mat3> Source<Out,Mat3> mergeSortedMat(Graph<SourceShape<Out>,Mat2> that, java.util.Comparator<Out> comp, Function2<Mat,Mat2,Mat3> matF)
Merge the givenSource
to thisSource
, taking elements as they arrive from input streams, picking always the smallest of the available elements (waiting for one element from each side to be available). This means that possible contiguity of the input streams is not exploited to avoid waiting for elements, this merge will block when one of the inputs does not have more elements (and does not complete).It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
zip
public <T> Source<Pair<Out,T>,Mat> zip(Graph<SourceShape<T>,?> that)
Combine the elements of currentSource
and the given one into a stream of tuples.'''Emits when''' all of the inputs has an element available
'''Backpressures when''' downstream backpressures
'''Completes when''' any upstream completes
'''Cancels when''' downstream cancels
-
zipMat
public <T,M,M2> Source<Pair<Out,T>,M2> zipMat(Graph<SourceShape<T>,M> that, Function2<Mat,M,M2> matF)
Combine the elements of currentSource
and the given one into a stream of tuples.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
zipAll
public <U,A> Source<Pair<A,U>,Mat> zipAll(Graph<SourceShape<U>,?> that, A thisElem, U thatElem)
Combine the elements of current flow and the givenSource
into a stream of tuples.'''Emits when''' at first emits when both inputs emit, and then as long as any input emits (coupled to the default value of the completed input).
'''Backpressures when''' downstream backpressures
'''Completes when''' all upstream completes
'''Cancels when''' downstream cancels
-
zipAllMat
public <U,Mat2,Mat3,A> Source<Pair<A,U>,Mat3> zipAllMat(Graph<SourceShape<U>,Mat2> that, A thisElem, U thatElem, scala.Function2<Mat,Mat2,Mat3> matF)
Combine the elements of current flow and the givenSource
into a stream of tuples.
-
zipLatest
public <T> Source<Pair<Out,T>,Mat> zipLatest(Graph<SourceShape<T>,?> that)
Combine the elements of 2 streams into a stream of tuples, picking always the latest element of each.A
ZipLatest
has aleft
and aright
input port and oneout
port.No element is emitted until at least one element from each Source becomes available.
'''Emits when''' all of the inputs have at least an element available, and then each time an element becomes * available on either of the inputs
'''Backpressures when''' downstream backpressures
'''Completes when''' any upstream completes
'''Cancels when''' downstream cancels
-
zipLatestMat
public <T,M,M2> Source<Pair<Out,T>,M2> zipLatestMat(Graph<SourceShape<T>,M> that, Function2<Mat,M,M2> matF)
Combine the elements of currentSource
and the given one into a stream of tuples, picking always the latest element of each.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
zipWith
public <Out2,Out3> Source<Out3,Mat> zipWith(Graph<SourceShape<Out2>,?> that, Function2<Out,Out2,Out3> combine)
Put together the elements of currentSource
and the given one into a stream of combined elements using a combiner function.'''Emits when''' all of the inputs has an element available
'''Backpressures when''' downstream backpressures
'''Completes when''' any upstream completes
'''Cancels when''' downstream cancels
-
zipWithMat
public <Out2,Out3,M,M2> Source<Out3,M2> zipWithMat(Graph<SourceShape<Out2>,M> that, Function2<Out,Out2,Out3> combine, Function2<Mat,M,M2> matF)
Put together the elements of currentSource
and the given one into a stream of combined elements using a combiner function.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
zipLatestWith
public <Out2,Out3> Source<Out3,Mat> zipLatestWith(Graph<SourceShape<Out2>,?> that, Function2<Out,Out2,Out3> combine)
Combine the elements of multiple streams into a stream of combined elements using a combiner function, picking always the latest of the elements of each source.No element is emitted until at least one element from each Source becomes available. Whenever a new element appears, the zipping function is invoked with a tuple containing the new element and the other last seen elements.
'''Emits when''' all of the inputs have at least an element available, and then each time an element becomes available on either of the inputs
'''Backpressures when''' downstream backpressures
'''Completes when''' any of the upstreams completes
'''Cancels when''' downstream cancels
-
zipLatestWith
public <Out2,Out3> Source<Out3,Mat> zipLatestWith(Graph<SourceShape<Out2>,?> that, boolean eagerComplete, Function2<Out,Out2,Out3> combine)
Combine the elements of multiple streams into a stream of combined elements using a combiner function, picking always the latest of the elements of each source.No element is emitted until at least one element from each Source becomes available. Whenever a new element appears, the zipping function is invoked with a tuple containing the new element and the other last seen elements.
'''Emits when''' all of the inputs have at least an element available, and then each time an element becomes available on either of the inputs
'''Backpressures when''' downstream backpressures
'''Completes when''' any upstream completes if
eagerComplete
is enabled or wait for all upstreams to complete'''Cancels when''' downstream cancels
-
zipLatestWithMat
public <Out2,Out3,M,M2> Source<Out3,M2> zipLatestWithMat(Graph<SourceShape<Out2>,M> that, Function2<Out,Out2,Out3> combine, Function2<Mat,M,M2> matF)
Put together the elements of currentSource
and the given one into a stream of combined elements using a combiner function, picking always the latest of the elements of each source.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
zipLatestWithMat
public <Out2,Out3,M,M2> Source<Out3,M2> zipLatestWithMat(Graph<SourceShape<Out2>,M> that, boolean eagerComplete, Function2<Out,Out2,Out3> combine, Function2<Mat,M,M2> matF)
Put together the elements of currentSource
and the given one into a stream of combined elements using a combiner function, picking always the latest of the elements of each source.It is recommended to use the internally optimized
Keep.left
andKeep.right
combiners where appropriate instead of manually writing functions that pass through one of the values.
-
zipWithIndex
public Source<Pair<Out,java.lang.Long>,Mat> zipWithIndex()
Combine the elements of currentSource
into a stream of tuples consisting of all elements paired with their index. Indices start at 0.'''Emits when''' upstream emits an element and is paired with their index
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
runForeach
public java.util.concurrent.CompletionStage<Done> runForeach(Procedure<Out> f, ClassicActorSystemProvider systemProvider)
Shortcut for running thisSource
with a foreach procedure. The given procedure is invoked for each received element. The returnedCompletionStage
will be completed normally when reaching the normal end of the stream, or completed exceptionally if there is a failure is signaled in the stream.Note that the classic or typed
ActorSystem
can be used as thesystemProvider
parameter.
-
runForeach
public java.util.concurrent.CompletionStage<Done> runForeach(Procedure<Out> f, Materializer materializer)
Shortcut for running thisSource
with a foreach procedure. The given procedure is invoked for each received element. The returnedCompletionStage
will be completed normally when reaching the normal end of the stream, or completed exceptionally if there is a failure is signaled in the stream.Prefer the method taking an
ActorSystem
unless you have special requirements
-
map
public <T> Source<T,Mat> map(Function<Out,T> f)
Transform this stream by applying the given function to each of the elements as they pass through this processing step.'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
wireTap
public Source<Out,Mat> wireTap(Procedure<Out> f)
This is a simplified version ofwireTap(Sink)
that takes only a simple procedure. Elements will be passed into this "side channel" function, and any of its results will be ignored.If the wire-tap operation is slow (it backpressures), elements that would've been sent to it will be dropped instead.
It is similar to
alsoTo(org.apache.pekko.stream.Graph<org.apache.pekko.stream.SinkShape<Out>, ?>)
which does backpressure instead of dropping elements.This operation is useful for inspecting the passed through element, usually by means of side-effecting operations (such as
println
, or emitting metrics), for each element without having to modify it.For logging signals (elements, completion, error) consider using the
log(java.lang.String,org.apache.pekko.japi.function.Function<Out,java.lang.Object>,org.apache.pekko.event.LoggingAdapter)
operator instead, along with appropriateActorAttributes.createLogLevels
.'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels; Note that failures of the
f
function will not cause cancellation
-
recover
public Source<Out,Mat> recover(scala.PartialFunction<java.lang.Throwable,Out> pf)
Recover allows to send last element on failure and gracefully complete the stream Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.Throwing an exception inside
recover
_will_ be logged on ERROR level automatically.'''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or upstream failed with exception pf can handle
'''Cancels when''' downstream cancels
-
recover
public Source<Out,Mat> recover(java.lang.Class<? extends java.lang.Throwable> clazz, java.util.function.Supplier<Out> supplier)
Recover allows to send last element on failure and gracefully complete the stream Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.Throwing an exception inside
recover
_will_ be logged on ERROR level automatically.'''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or upstream failed with exception pf can handle
'''Cancels when''' downstream cancels
-
mapError
public Source<Out,Mat> mapError(scala.PartialFunction<java.lang.Throwable,java.lang.Throwable> pf)
While similar torecover(scala.PartialFunction<java.lang.Throwable,Out>)
this operator can be used to transform an error signal to a different one *without* logging it as an error in the process. So in that sense it is NOT exactly equivalent torecover(t => throw t2)
since recover would log thet2
error.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
Similarly to
recover(scala.PartialFunction<java.lang.Throwable,Out>)
throwing an exception insidemapError
_will_ be logged.'''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or upstream failed with exception pf can handle
'''Cancels when''' downstream cancels
-
mapError
public <E extends java.lang.Throwable> Source<Out,Mat> mapError(java.lang.Class<E> clazz, Function<E,java.lang.Throwable> f)
While similar torecover(scala.PartialFunction<java.lang.Throwable,Out>)
this operator can be used to transform an error signal to a different one *without* logging it as an error in the process. So in that sense it is NOT exactly equivalent torecover(t => throw t2)
since recover would log thet2
error.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
Similarly to
recover(scala.PartialFunction<java.lang.Throwable,Out>)
throwing an exception insidemapError
_will_ be logged.'''Emits when''' element is available from the upstream or upstream is failed and pf returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or upstream failed with exception pf can handle
'''Cancels when''' downstream cancels
-
recoverWith
public Source<Out,Mat> recoverWith(scala.PartialFunction<java.lang.Throwable,? extends Graph<SourceShape<Out>,NotUsed>> pf)
Deprecated.userecoverWithRetries
insteadRecoverWith allows to switch to alternative Source on flow failure. It will stay in effect after a failure has been recovered so that each time there is a failure it is fed into thepf
and a new Source may be materialized.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
Throwing an exception inside
recoverWith
_will_ be logged on ERROR level automatically.'''Emits when''' element is available from the upstream or upstream is failed and element is available from alternative Source
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or upstream failed with exception pf can handle
'''Cancels when''' downstream cancels
-
recoverWith
public Source<Out,Mat> recoverWith(java.lang.Class<? extends java.lang.Throwable> clazz, java.util.function.Supplier<Graph<SourceShape<Out>,NotUsed>> supplier)
Deprecated.userecoverWithRetries
insteadRecoverWith allows to switch to alternative Source on flow failure. It will stay in effect after a failure has been recovered so that each time there is a failure it is fed into thepf
and a new Source may be materialized.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
Throwing an exception inside
recoverWith
_will_ be logged on ERROR level automatically.'''Emits when''' element is available from the upstream or upstream is failed and element is available from alternative Source
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or upstream failed with exception pf can handle
'''Cancels when''' downstream cancels
-
recoverWithRetries
public Source<Out,Mat> recoverWithRetries(int attempts, scala.PartialFunction<java.lang.Throwable,? extends Graph<SourceShape<Out>,NotUsed>> pf)
RecoverWithRetries allows to switch to alternative Source on flow failure. It will stay in effect after a failure has been recovered up toattempts
number of times so that each time there is a failure it is fed into thepf
and a new Source may be materialized. Note that if you pass in 0, this won't attempt to recover at all.A negative
attempts
number is interpreted as "infinite", which results in the exact same behavior asrecoverWith
.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
Throwing an exception inside
recoverWithRetries
_will_ be logged on ERROR level automatically.'''Emits when''' element is available from the upstream or upstream is failed and element is available from alternative Source
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or upstream failed with exception pf can handle
'''Cancels when''' downstream cancels
-
recoverWithRetries
public Source<Out,Mat> recoverWithRetries(int attempts, java.lang.Class<? extends java.lang.Throwable> clazz, java.util.function.Supplier<Graph<SourceShape<Out>,NotUsed>> supplier)
RecoverWithRetries allows to switch to alternative Source on flow failure. It will stay in effect after a failure has been recovered up toattempts
number of times so that each time there is a failure it is fed into thepf
and a new Source may be materialized. Note that if you pass in 0, this won't attempt to recover at all.A negative
attempts
number is interpreted as "infinite", which results in the exact same behavior asrecoverWith
.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
Throwing an exception inside
recoverWithRetries
_will_ be logged on ERROR level automatically.'''Emits when''' element is available from the upstream or upstream is failed and element is available from alternative Source
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or upstream failed with exception pf can handle
'''Cancels when''' downstream cancels
- Parameters:
attempts
- Maximum number of retries or -1 to retry indefinitelyclazz
- the class object of the failure causesupplier
- supply the new Source to be materialized
-
onErrorComplete
public Source<Out,Mat> onErrorComplete()
onErrorComplete allows to complete the stream when an upstream error occurs.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
'''Emits when''' element is available from the upstream
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or failed with exception is an instance of the provided type
'''Cancels when''' downstream cancels
- Since:
- 1.1.0
-
onErrorComplete
public Source<Out,Mat> onErrorComplete(java.lang.Class<? extends java.lang.Throwable> clazz)
onErrorComplete allows to complete the stream when an upstream error occurs.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
'''Emits when''' element is available from the upstream
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or failed with exception is an instance of the provided type
'''Cancels when''' downstream cancels
- Since:
- 1.1.0
-
onErrorComplete
public Source<Out,Mat> onErrorComplete(java.util.function.Predicate<? super java.lang.Throwable> predicate)
onErrorComplete allows to complete the stream when an upstream error occurs.Since the underlying failure signal onError arrives out-of-band, it might jump over existing elements. This operator can recover the failure signal, but not the skipped elements, which will be dropped.
'''Emits when''' element is available from the upstream
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or failed with predicate return ture
'''Cancels when''' downstream cancels
- Since:
- 1.1.0
-
mapConcat
public <T> Source<T,Mat> mapConcat(Function<Out,? extends java.lang.Iterable<T>> f)
Transform each input element into anIterable
of output elements that is then flattened into the output stream.Make sure that the
Iterable
is immutable or at least not modified after being used as an output sequence. Otherwise the stream may fail withConcurrentModificationException
or other more subtle errors may occur.The returned
Iterable
MUST NOT containnull
values, as they are illegal as stream elements - according to the Reactive Streams specification.'''Emits when''' the mapping function returns an element or there are still remaining elements from the previously calculated collection
'''Backpressures when''' downstream backpressures or there are still remaining elements from the previously calculated collection
'''Completes when''' upstream completes and all remaining elements has been emitted
'''Cancels when''' downstream cancels
-
statefulMap
public <S,T> Source<T,Mat> statefulMap(Creator<S> create, Function2<S,Out,Pair<S,T>> f, Function<S,java.util.Optional<T>> onComplete)
Transform each stream element with the help of a state.The state creation function is invoked once when the stream is materialized and the returned state is passed to the mapping function for mapping the first element. The mapping function returns a mapped element to emit downstream and a state to pass to the next mapping function. The state can be the same for each mapping return, be a new immutable state but it is also safe to use a mutable state. The returned
T
MUST NOT benull
as it is illegal as stream element - according to the Reactive Streams specification.For stateless variant see
<T>map(org.apache.pekko.japi.function.Function<Out,T>)
.The
onComplete
function is called only once when the upstream or downstream finished, You can do some clean-up here, and if the returned value is not empty, it will be emitted to the downstream if available, otherwise the value will be dropped.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the mapping function returns an element and downstream is ready to consume it
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
- Parameters:
create
- a function that creates the initial statef
- a function that transforms the upstream element and the state into a pair of next state and output elementonComplete
- a function that transforms the ongoing state into an optional output element
-
mapWithResource
public <R,T> Source<T,Mat> mapWithResource(Creator<R> create, Function2<R,Out,T> f, Function<R,java.util.Optional<T>> close)
Transform each stream element with the help of a resource.The resource creation function is invoked once when the stream is materialized and the returned resource is passed to the mapping function for mapping the first element. The mapping function returns a mapped element to emit downstream. The returned
T
MUST NOT benull
as it is illegal as stream element - according to the Reactive Streams specification.The
close
function is called only once when the upstream or downstream finishes or fails. You can do some clean-up here, and if the returned value is not empty, it will be emitted to the downstream if available, otherwise the value will be dropped.Early completion can be done with combination of the
takeWhile(org.apache.pekko.japi.function.Predicate<Out>,boolean)
operator.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.You can configure the default dispatcher for this Source by changing the
pekko.stream.materializer.blocking-io-dispatcher
or set it for a given Source by usingActorAttributes
.'''Emits when''' the mapping function returns an element and downstream is ready to consume it
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
- Parameters:
create
- function that creates the resourcef
- function that transforms the upstream element and the resource to output elementclose
- function that closes the resource, optionally outputting a last element- Since:
- 1.1.0
-
mapWithResource
public <R extends java.lang.AutoCloseable,T> Source<T,Mat> mapWithResource(Creator<R> create, Function2<R,Out,T> f)
Transform each stream element with the help of anAutoCloseable
resource and close it when the stream finishes or fails.The resource creation function is invoked once when the stream is materialized and the returned resource is passed to the mapping function for mapping the first element. The mapping function returns a mapped element to emit downstream. The returned
T
MUST NOT benull
as it is illegal as stream element - according to the Reactive Streams specification.The
AutoCloseable
resource is closed only once when the upstream or downstream finishes or fails.Early completion can be done with combination of the
takeWhile(org.apache.pekko.japi.function.Predicate<Out>,boolean)
operator.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.You can configure the default dispatcher for this Source by changing the
pekko.stream.materializer.blocking-io-dispatcher
or set it for a given Source by usingActorAttributes
.'''Emits when''' the mapping function returns an element and downstream is ready to consume it
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
- Parameters:
create
- function that creates the resourcef
- function that transforms the upstream element and the resource to output element- Since:
- 1.1.0
-
statefulMapConcat
public <T> Source<T,Mat> statefulMapConcat(Creator<Function<Out,java.lang.Iterable<T>>> f)
Transform each input element into anIterable
of output elements that is then flattened into the output stream. The transformation is meant to be stateful, which is enabled by creating the transformation function anew for every materialization — the returned function will typically close over mutable objects to store state between invocations. For the stateless variant seemapConcat(org.apache.pekko.japi.function.Function<Out, ? extends java.lang.Iterable<T>>)
.Make sure that the
Iterable
is immutable or at least not modified after being used as an output sequence. Otherwise the stream may fail withConcurrentModificationException
or other more subtle errors may occur.The returned
Iterable
MUST NOT containnull
values, as they are illegal as stream elements - according to the Reactive Streams specification.This operator doesn't handle upstream's completion signal since the state kept in the closure can be lost. Use
FlowOps.statefulMap
instead.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the mapping function returns an element or there are still remaining elements from the previously calculated collection
'''Backpressures when''' downstream backpressures or there are still remaining elements from the previously calculated collection
'''Completes when''' upstream completes and all remaining elements has been emitted
'''Cancels when''' downstream cancels
-
mapAsync
public <T> Source<T,Mat> mapAsync(int parallelism, Function<Out,java.util.concurrent.CompletionStage<T>> f)
Transform this stream by applying the given function to each of the elements as they pass through this processing step. The function returns aCompletionStage
and the value of that future will be emitted downstream. The number of CompletionStages that shall run in parallel is given as the first argument tomapAsync
. These CompletionStages may complete in any order, but the elements that are emitted downstream are in the same order as received from upstream.
If the function
f
throws an exception or if theCompletionStage
is completed with failure and the supervision decision ispekko.stream.Supervision#stop
the stream will be completed with failure.If the function
f
throws an exception or if theCompletionStage
is completed with failure and the supervision decision ispekko.stream.Supervision#resume
orpekko.stream.Supervision#restart
or theCompletionStage
completed withnull
, the element is dropped and the stream continues.The function
f
is always invoked on the elements in the order they arrive.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the CompletionStage returned by the provided function finishes for the next element in sequence
'''Backpressures when''' the number of CompletionStages reaches the configured parallelism and the downstream backpressures or the first CompletionStage is not completed
'''Completes when''' upstream completes and all CompletionStages has been completed and all elements has been emitted
'''Cancels when''' downstream cancels
-
mapAsyncPartitioned
public <T,P> Source<T,Mat> mapAsyncPartitioned(int parallelism, Function<Out,P> partitioner, Function2<Out,P,java.util.concurrent.CompletionStage<T>> f)
Transforms this stream. Works very similarly tomapAsync(int, org.apache.pekko.japi.function.Function<Out, java.util.concurrent.CompletionStage<T>>)
but with an additional partition step before the transform step. The transform function receives the an individual stream entry and the calculated partition value for that entry. The max parallelism of per partition is 1.The function
partitioner
is always invoked on the elements in the order they arrive. The functionf
is always invoked on the elements which in the same partition in the order they arrive.If the function
partitioner
orf
throws an exception or if theCompletionStage
is completed with failure and the supervision decision ispekko.stream.Supervision.Stop
the stream will be completed with failure, otherwise the stream continues and the current element is dropped.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the Future returned by the provided function finishes for the next element in sequence
'''Backpressures when''' the number of futures reaches the configured parallelism and the downstream backpressures
'''Completes when''' upstream completes and all futures have been completed and all elements have been emitted
'''Cancels when''' downstream cancels
-
mapAsyncPartitionedUnordered
public <T,P> Source<T,Mat> mapAsyncPartitionedUnordered(int parallelism, Function<Out,P> partitioner, Function2<Out,P,java.util.concurrent.CompletionStage<T>> f)
Transforms this stream. Works very similarly tomapAsyncUnordered(int, org.apache.pekko.japi.function.Function<Out, java.util.concurrent.CompletionStage<T>>)
but with an additional partition step before the transform step. The transform function receives the an individual stream entry and the calculated partition value for that entry.The max parallelism of per partition is 1.The function
partitioner
is always invoked on the elements in the order they arrive. The functionf
is always invoked on the elements which in the same partition in the order they arrive.If the function
partitioner
orf
throws an exception or if theCompletionStage
is completed with failure and the supervision decision ispekko.stream.Supervision.Stop
the stream will be completed with failure, otherwise the stream continues and the current element is dropped.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the Future returned by the provided function finishes and downstream available.
'''Backpressures when''' the number of futures reaches the configured parallelism and the downstream backpressures
'''Completes when''' upstream completes and all futures have been completed and all elements have been emitted
'''Cancels when''' downstream cancels
-
mapAsyncUnordered
public <T> Source<T,Mat> mapAsyncUnordered(int parallelism, Function<Out,java.util.concurrent.CompletionStage<T>> f)
Transform this stream by applying the given function to each of the elements as they pass through this processing step. The function returns aCompletionStage
and the value of that future will be emitted downstream. The number of CompletionStages that shall run in parallel is given as the first argument tomapAsyncUnordered
. Each processed element will be emitted downstream as soon as it is ready, i.e. it is possible that the elements are not emitted downstream in the same order as received from upstream.
If the function
f
throws an exception or if theCompletionStage
is completed with failure and the supervision decision ispekko.stream.Supervision#stop
the stream will be completed with failure.If the function
f
throws an exception or if theCompletionStage
is completed with failure and the supervision decision ispekko.stream.Supervision#resume
orpekko.stream.Supervision#restart
or theCompletionStage
completed withnull
, the element is dropped and the stream continues.The function
f
is always invoked on the elements in the order they arrive (even though the result of the CompletionStages returned byf
might be emitted in a different order).Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' any of the CompletionStages returned by the provided function complete
'''Backpressures when''' the number of CompletionStages reaches the configured parallelism and the downstream backpressures
'''Completes when''' upstream completes and all CompletionStages has been completed and all elements has been emitted
'''Cancels when''' downstream cancels
-
ask
public <S> Source<S,Mat> ask(ActorRef ref, java.lang.Class<S> mapTo, Timeout timeout)
Use theask
pattern to send a request-reply message to the targetref
actor. If any of the asks times out it will fail the stream with apekko.pattern.AskTimeoutException
.The
mapTo
class parameter is used to cast the incoming responses to the expected response type.Similar to the plain ask pattern, the target actor is allowed to reply with
org.apache.pekko.util.Status
. Anorg.apache.pekko.util.Status#Failure
will cause the operator to fail with the cause carried in theFailure
message.Defaults to parallelism of 2 messages in flight, since while one ask message may be being worked on, the second one still be in the mailbox, so defaulting to sending the second one a bit earlier than when first ask has replied maintains a slightly healthier throughput.
The operator fails with an
pekko.stream.WatchedActorTerminatedException
if the target actor is terminated.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' any of the CompletionStages returned by the provided function complete
'''Backpressures when''' the number of futures reaches the configured parallelism and the downstream backpressures
'''Completes when''' upstream completes and all futures have been completed and all elements have been emitted
'''Fails when''' the passed in actor terminates, or a timeout is exceeded in any of the asks performed
'''Cancels when''' downstream cancels
-
ask
public <S> Source<S,Mat> ask(int parallelism, ActorRef ref, java.lang.Class<S> mapTo, Timeout timeout)
Use theask
pattern to send a request-reply message to the targetref
actor. If any of the asks times out it will fail the stream with apekko.pattern.AskTimeoutException
.The
mapTo
class parameter is used to cast the incoming responses to the expected response type.Similar to the plain ask pattern, the target actor is allowed to reply with
org.apache.pekko.util.Status
. Anorg.apache.pekko.util.Status#Failure
will cause the operator to fail with the cause carried in theFailure
message.Parallelism limits the number of how many asks can be "in flight" at the same time. Please note that the elements emitted by this operator are in-order with regards to the asks being issued (i.e. same behavior as mapAsync).
The operator fails with an
pekko.stream.WatchedActorTerminatedException
if the target actor is terminated.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' any of the CompletionStages returned by the provided function complete
'''Backpressures when''' the number of futures reaches the configured parallelism and the downstream backpressures
'''Completes when''' upstream completes and all futures have been completed and all elements have been emitted
'''Fails when''' the passed in actor terminates, or a timeout is exceeded in any of the asks performed
'''Cancels when''' downstream cancels
-
watch
public Source<Out,Mat> watch(ActorRef ref)
The operator fails with anpekko.stream.WatchedActorTerminatedException
if the target actor is terminated.'''Emits when''' upstream emits
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Fails when''' the watched actor terminates
'''Cancels when''' downstream cancels
-
filter
public Source<Out,Mat> filter(Predicate<Out> p)
Only pass on those elements that satisfy the given predicate.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the given predicate returns true for the element
'''Backpressures when''' the given predicate returns true for the element and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
filterNot
public Source<Out,Mat> filterNot(Predicate<Out> p)
Only pass on those elements that NOT satisfy the given predicate.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the given predicate returns false for the element
'''Backpressures when''' the given predicate returns false for the element and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
collect
public <T> Source<T,Mat> collect(scala.PartialFunction<Out,T> pf)
Transform this stream by applying the given partial function to each of the elements on which the function is defined as they pass through this processing step. Non-matching elements are filtered out.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the provided partial function is defined for the element
'''Backpressures when''' the partial function is defined for the element and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
collectFirst
public <T> Source<T,Mat> collectFirst(scala.PartialFunction<Out,T> pf)
Transform this stream by applying the given partial function to the first element on which the function is defined as it pass through this processing step, and cancel the upstream publisher after the first element is emitted.Non-matching elements are filtered out.
Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the provided partial function is defined for the first element
'''Backpressures when''' the partial function is defined for the element and downstream backpressures
'''Completes when''' upstream completes or the first element is emitted
'''Cancels when''' downstream cancels
-
collectWhile
public <T> Source<T,Mat> collectWhile(scala.PartialFunction<Out,T> pf)
Transform this stream by applying the given partial function to each of the elements on which the function is defined as they pass through this processing step, and cancel the upstream publisher after the partial function is not applied.The stream will be completed without producing any elements if the partial function is not applied for the first stream element, eg: there is a downstream buffer.
Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the provided partial function is defined for the element
'''Backpressures when''' the partial function is defined for the element and downstream backpressures
'''Completes when''' upstream completes or the partial function is not applied.
'''Cancels when''' downstream cancels
- Since:
- 1.1.0
-
collectType
public <T> Source<T,Mat> collectType(java.lang.Class<T> clazz)
Transform this stream by testing the type of each of the elements on which the element is an instance of the provided type as they pass through this processing step. Non-matching elements are filtered out.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the element is an instance of the provided type
'''Backpressures when''' the element is an instance of the provided type and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
grouped
public Source<java.util.List<Out>,Mat> grouped(int n)
Chunk up this stream into groups of the given size, with the last group possibly smaller than requested due to end-of-stream.n
must be positive, otherwise IllegalArgumentException is thrown.'''Emits when''' the specified number of elements has been accumulated or upstream completed
'''Backpressures when''' a group has been assembled and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
groupedWeighted
public Source<java.util.List<Out>,Mat> groupedWeighted(long minWeight, java.util.function.Function<Out,java.lang.Long> costFn)
Chunk up this stream into groups of elements that have a cumulative weight greater than or equal to theminWeight
, with the last group possibly smaller than requestedminWeight
due to end-of-stream.minWeight
must be positive, otherwise IllegalArgumentException is thrown.costFn
must return a non-negative result for all inputs, otherwise the stage will fail with an IllegalArgumentException.'''Emits when''' the cumulative weight of elements is greater than or equal to the
minWeight
or upstream completed'''Backpressures when''' a buffered group weighs more than
minWeight
and downstream backpressures'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
limit
public Source<Out,Mat> limit(int n)
Ensure stream boundedness by limiting the number of elements from upstream. If the number of incoming elements exceeds max, it will signal upstream failureStreamLimitException
downstream.Due to input buffering some elements may have been requested from upstream publishers that will then not be processed downstream of this step.
The stream will be completed without producing any elements if
n
is zero or negative.'''Emits when''' the specified number of elements to take has not yet been reached
'''Backpressures when''' downstream backpressures
'''Completes when''' the defined number of elements has been taken or upstream completes
'''Cancels when''' the defined number of elements has been taken or downstream cancels
See also
Flow.take
,Flow.takeWithin
,Flow.takeWhile
-
limitWeighted
public Source<Out,Mat> limitWeighted(long n, Function<Out,java.lang.Long> costFn)
Ensure stream boundedness by evaluating the cost of incoming elements using a cost function. Exactly how many elements will be allowed to travel downstream depends on the evaluated cost of each element. If the accumulated cost exceeds max, it will signal upstream failureStreamLimitException
downstream.Due to input buffering some elements may have been requested from upstream publishers that will then not be processed downstream of this step.
The stream will be completed without producing any elements if
n
is zero or negative.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the specified number of elements to take has not yet been reached
'''Backpressures when''' downstream backpressures
'''Completes when''' the defined number of elements has been taken or upstream completes
'''Cancels when''' the defined number of elements has been taken or downstream cancels
See also
Flow.take
,Flow.takeWithin
,Flow.takeWhile
-
sliding
public Source<java.util.List<Out>,Mat> sliding(int n, int step)
Apply a sliding window over the stream and return the windows as groups of elements, with the last group possibly smaller than requested due to end-of-stream.n
must be positive, otherwise IllegalArgumentException is thrown.step
must be positive, otherwise IllegalArgumentException is thrown.'''Emits when''' enough elements have been collected within the window or upstream completed
'''Backpressures when''' a window has been assembled and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
scan
public <T> Source<T,Mat> scan(T zero, Function2<T,Out,T> f)
Similar tofold
but is not a terminal operation, emits its current value which starts atzero
and then applies the current and next value to the given functionf
, emitting the next current value.If the function
f
throws an exception and the supervision decision ispekko.stream.Supervision#restart
current value starts atzero
again the stream will continue.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.Note that the
zero
value must be immutable.'''Emits when''' the function scanning the element returns a new element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
scanAsync
public <T> Source<T,Mat> scanAsync(T zero, Function2<T,Out,java.util.concurrent.CompletionStage<T>> f)
Similar toscan
but with an asynchronous function, emits its current value which starts atzero
and then applies the current and next value to the given functionf
, emitting aFuture
that resolves to the next current value.If the function
f
throws an exception and the supervision decision ispekko.stream.Supervision.Restart
current value starts atzero
again the stream will continue.If the function
f
throws an exception and the supervision decision ispekko.stream.Supervision.Resume
current value starts at the previous current value, or zero when it doesn't have one, and the stream will continue.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.Note that the
zero
value must be immutable.'''Emits when''' the future returned by f
completes
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes and the last future returned by
f
completes'''Cancels when''' downstream cancels
See also
FlowOps#scan
-
fold
public <T> Source<T,Mat> fold(T zero, Function2<T,Out,T> f)
Similar toscan
but only emits its result when the upstream completes, after which it also completes. Applies the given functionf
towards its current and next value, yielding the next current value.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.If the function
f
throws an exception and the supervision decision ispekko.stream.Supervision#restart
current value starts atzero
again the stream will continue.Note that the
zero
value must be immutable.'''Emits when''' upstream completes
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
foldWhile
public <T> Source<T,Mat> foldWhile(T zero, Predicate<T> p, Function2<T,Out,T> f)
Similar toscan
but only emits its result when the upstream completes or the predicatep
returnsfalse
. after which it also completes. Applies the given function towards its current and next value, yielding the next current value.If the function
f
throws an exception and the supervision decision ispekko.stream.Supervision.Restart
current value starts atzero
again the stream will continue.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.Note that the
zero
value must be immutable.'''Emits when''' upstream completes or the predicate
p
returnsfalse
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
See also
FlowOps.fold
-
foldAsync
public <T> Source<T,Mat> foldAsync(T zero, Function2<T,Out,java.util.concurrent.CompletionStage<T>> f)
Similar tofold
but with an asynchronous function. Applies the given function towards its current and next value, yielding the next current value.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.If the function
f
returns a failure and the supervision decision ispekko.stream.Supervision.Restart
current value starts atzero
again the stream will continue.Note that the
zero
value must be immutable.'''Emits when''' upstream completes
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
reduce
public Source<Out,Mat> reduce(Function2<Out,Out,Out> f)
Similar tofold
but uses first element as zero element. Applies the given function towards its current and next value, yielding the next current value.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' upstream completes
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
intersperse
public Source<Out,Mat> intersperse(Out start, Out inject, Out end)
Intersperses stream with provided element, similar to howscala.collection.immutable.List.mkString
injects a separator between a List's elements.Additionally can inject start and end marker elements to stream.
Examples:
Source<Integer, ?> nums = Source.from(Arrays.asList(0, 1, 2, 3)); nums.intersperse(","); // 1 , 2 , 3 nums.intersperse("[", ",", "]"); // [ 1 , 2 , 3 ]
In case you want to only prepend or only append an element (yet still use the
intercept
feature to inject a separator between elements, you may want to use the following pattern instead of the 3-argument version of intersperse (SeeSource.concat
for semantics details):
'''Emits when''' upstream emits (or before with theSource.single(">> ").concat(list.intersperse(",")) list.intersperse(",").concat(Source.single("END"))
start
element if provided)'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
intersperse
public Source<Out,Mat> intersperse(Out inject)
Intersperses stream with provided element, similar to howscala.collection.immutable.List.mkString
injects a separator between a List's elements.Additionally can inject start and end marker elements to stream.
Examples:
Source<Integer, ?> nums = Source.from(Arrays.asList(0, 1, 2, 3)); nums.intersperse(","); // 1 , 2 , 3 nums.intersperse("[", ",", "]"); // [ 1 , 2 , 3 ]
'''Emits when''' upstream emits (or before with the
start
element if provided)'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
groupedWithin
public Source<java.util.List<Out>,Mat> groupedWithin(int maxNumber, scala.concurrent.duration.FiniteDuration duration)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Chunk up this stream into groups of elements received within a time window, or limited by the given number of elements, whatever happens first. Empty groups will not be emitted if no elements are received from upstream. The last group before end-of-stream will contain the buffered elements since the previously emitted group.'''Emits when''' the configured time elapses since the last group has been emitted or
n
elements is buffered'''Backpressures when''' downstream backpressures, and there are
n+1
buffered elements'''Completes when''' upstream completes (emits last group)
'''Cancels when''' downstream completes
maxNumber
must be positive, andduration
must be greater than 0 seconds, otherwise IllegalArgumentException is thrown.
-
groupedWithin
public Source<java.util.List<Out>,Mat> groupedWithin(int maxNumber, java.time.Duration duration)
Chunk up this stream into groups of elements received within a time window, or limited by the given number of elements, whatever happens first. Empty groups will not be emitted if no elements are received from upstream. The last group before end-of-stream will contain the buffered elements since the previously emitted group.'''Emits when''' the configured time elapses since the last group has been emitted or
n
elements is buffered'''Backpressures when''' downstream backpressures, and there are
n+1
buffered elements'''Completes when''' upstream completes (emits last group)
'''Cancels when''' downstream completes
maxNumber
must be positive, andduration
must be greater than 0 seconds, otherwise IllegalArgumentException is thrown.
-
groupedWeightedWithin
public Source<java.util.List<Out>,Mat> groupedWeightedWithin(long maxWeight, Function<Out,java.lang.Long> costFn, scala.concurrent.duration.FiniteDuration duration)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Chunk up this stream into groups of elements received within a time window, or limited by the weight of the elements, whatever happens first. Empty groups will not be emitted if no elements are received from upstream. The last group before end-of-stream will contain the buffered elements since the previously emitted group.'''Emits when''' the configured time elapses since the last group has been emitted or weight limit reached
'''Backpressures when''' downstream backpressures, and buffered group (+ pending element) weighs more than
maxWeight
'''Completes when''' upstream completes (emits last group)
'''Cancels when''' downstream completes
maxWeight
must be positive, andduration
must be greater than 0 seconds, otherwise IllegalArgumentException is thrown.
-
groupedWeightedWithin
public Source<java.util.List<Out>,Mat> groupedWeightedWithin(long maxWeight, Function<Out,java.lang.Long> costFn, java.time.Duration duration)
Chunk up this stream into groups of elements received within a time window, or limited by the weight of the elements, whatever happens first. Empty groups will not be emitted if no elements are received from upstream. The last group before end-of-stream will contain the buffered elements since the previously emitted group.'''Emits when''' the configured time elapses since the last group has been emitted or weight limit reached
'''Backpressures when''' downstream backpressures, and buffered group (+ pending element) weighs more than
maxWeight
'''Completes when''' upstream completes (emits last group)
'''Cancels when''' downstream completes
maxWeight
must be positive, andduration
must be greater than 0 seconds, otherwise IllegalArgumentException is thrown.
-
groupedWeightedWithin
public Source<java.util.List<Out>,Mat> groupedWeightedWithin(long maxWeight, int maxNumber, Function<Out,java.lang.Long> costFn, java.time.Duration duration)
Chunk up this stream into groups of elements received within a time window, or limited by the weight and number of the elements, whatever happens first. Empty groups will not be emitted if no elements are received from upstream. The last group before end-of-stream will contain the buffered elements since the previously emitted group.'''Emits when''' the configured time elapses since the last group has been emitted or weight limit reached
'''Backpressures when''' downstream backpressures, and buffered group (+ pending element) weighs more than
maxWeight
or has more thanmaxNumber
elements'''Completes when''' upstream completes (emits last group)
'''Cancels when''' downstream completes
maxWeight
must be positive,maxNumber
must be positive, andduration
must be greater than 0 seconds, otherwise IllegalArgumentException is thrown.
-
delay
public Source<Out,Mat> delay(scala.concurrent.duration.FiniteDuration of, DelayOverflowStrategy strategy)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Shifts elements emission in time by a specified amount. It allows to store elements in internal buffer while waiting for next element to be emitted. Depending on the definedpekko.stream.DelayOverflowStrategy
it might drop elements or backpressure the upstream if there is no space available in the buffer.Delay precision is 10ms to avoid unnecessary timer scheduling cycles
Internal buffer has default capacity 16. You can set buffer size by calling
withAttributes(inputBuffer)
'''Emits when''' there is a pending element in the buffer and configured time for this element elapsed * EmitEarly - strategy do not wait to emit element if buffer is full
'''Backpressures when''' depending on OverflowStrategy * Backpressure - backpressures when buffer is full * DropHead, DropTail, DropBuffer - never backpressures * Fail - fails the stream if buffer gets full
'''Completes when''' upstream completes and buffered elements has been drained
'''Cancels when''' downstream cancels
- Parameters:
of
- time to shift all messagesstrategy
- Strategy that is used when incoming elements cannot fit inside the buffer
-
delay
public Source<Out,Mat> delay(java.time.Duration of, DelayOverflowStrategy strategy)
Shifts elements emission in time by a specified amount. It allows to store elements in internal buffer while waiting for next element to be emitted. Depending on the definedpekko.stream.DelayOverflowStrategy
it might drop elements or backpressure the upstream if there is no space available in the buffer.Delay precision is 10ms to avoid unnecessary timer scheduling cycles
Internal buffer has default capacity 16. You can set buffer size by calling
withAttributes(inputBuffer)
'''Emits when''' there is a pending element in the buffer and configured time for this element elapsed * EmitEarly - strategy do not wait to emit element if buffer is full
'''Backpressures when''' depending on OverflowStrategy * Backpressure - backpressures when buffer is full * DropHead, DropTail, DropBuffer - never backpressures * Fail - fails the stream if buffer gets full
'''Completes when''' upstream completes and buffered elements has been drained
'''Cancels when''' downstream cancels
- Parameters:
of
- time to shift all messagesstrategy
- Strategy that is used when incoming elements cannot fit inside the buffer
-
delayWith
public Source<Out,Mat> delayWith(java.util.function.Supplier<DelayStrategy<Out>> delayStrategySupplier, DelayOverflowStrategy overFlowStrategy)
Shifts elements emission in time by an amount individually determined through delay strategy a specified amount. It allows to store elements in internal buffer while waiting for next element to be emitted. Depending on the definedpekko.stream.DelayOverflowStrategy
it might drop elements or backpressure the upstream if there is no space available in the buffer.It determines delay for each ongoing element invoking
DelayStrategy.nextDelay(elem: T): FiniteDuration
.Note that elements are not re-ordered: if an element is given a delay much shorter than its predecessor, it will still have to wait for the preceding element before being emitted. It is also important to notice that
DelayStrategy
can be stateful.Delay precision is 10ms to avoid unnecessary timer scheduling cycles.
Internal buffer has default capacity 16. You can set buffer size by calling
addAttributes(inputBuffer)
'''Emits when''' there is a pending element in the buffer and configured time for this element elapsed * EmitEarly - strategy do not wait to emit element if buffer is full
'''Backpressures when''' depending on OverflowStrategy * Backpressure - backpressures when buffer is full * DropHead, DropTail, DropBuffer - never backpressures * Fail - fails the stream if buffer gets full
'''Completes when''' upstream completes and buffered elements have been drained
'''Cancels when''' downstream cancels
- Parameters:
delayStrategySupplier
- creates newDelayStrategy
object for each materializationoverFlowStrategy
- Strategy that is used when incoming elements cannot fit inside the buffer
-
drop
public Source<Out,Mat> drop(long n)
Discard the given number of elements at the beginning of the stream. No elements will be dropped ifn
is zero or negative.'''Emits when''' the specified number of elements has been dropped already
'''Backpressures when''' the specified number of elements has been dropped and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
dropWithin
public Source<Out,Mat> dropWithin(scala.concurrent.duration.FiniteDuration duration)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Discard the elements received within the given duration at beginning of the stream.'''Emits when''' the specified time elapsed and a new upstream element arrives
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
dropWithin
public Source<Out,Mat> dropWithin(java.time.Duration duration)
Discard the elements received within the given duration at beginning of the stream.'''Emits when''' the specified time elapsed and a new upstream element arrives
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
takeWhile
public Source<Out,Mat> takeWhile(Predicate<Out> p, boolean inclusive)
Terminate processing (and cancel the upstream publisher) after predicate returns false for the first time, including the first failed element if inclusive is true Due to input buffering some elements may have been requested from upstream publishers that will then not be processed downstream of this step.The stream will be completed without producing any elements if predicate is false for the first stream element.
Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the predicate is true
'''Backpressures when''' downstream backpressures
'''Completes when''' predicate returned false (or 1 after predicate returns false if
inclusive
or upstream completes'''Cancels when''' predicate returned false or downstream cancels
See also
Source.limit
,Source.limitWeighted
-
takeWhile
public Source<Out,Mat> takeWhile(Predicate<Out> p)
Terminate processing (and cancel the upstream publisher) after predicate returns false for the first time. Due to input buffering some elements may have been requested from upstream publishers that will then not be processed downstream of this step.The stream will be completed without producing any elements if predicate is false for the first stream element.
'''Emits when''' the predicate is true
'''Backpressures when''' downstream backpressures
'''Completes when''' predicate returned false or upstream completes
'''Cancels when''' predicate returned false or downstream cancels
See also
Source.limit
,Source.limitWeighted
-
dropWhile
public Source<Out,Mat> dropWhile(Predicate<Out> p)
Discard elements at the beginning of the stream while predicate is true. No elements will be dropped after predicate first time returned false.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' predicate returned false and for all following stream elements
'''Backpressures when''' predicate returned false and downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
- Parameters:
p
- predicate is evaluated for each new element until first time returns false
-
take
public Source<Out,Mat> take(long n)
Terminate processing (and cancel the upstream publisher) after the given number of elements. Due to input buffering some elements may have been requested from upstream publishers that will then not be processed downstream of this step.The stream will be completed without producing any elements if
n
is zero or negative.'''Emits when''' the specified number of elements to take has not yet been reached
'''Backpressures when''' downstream backpressures
'''Completes when''' the defined number of elements has been taken or upstream completes
'''Cancels when''' the defined number of elements has been taken or downstream cancels
-
takeWithin
public Source<Out,Mat> takeWithin(scala.concurrent.duration.FiniteDuration duration)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Terminate processing (and cancel the upstream publisher) after the given duration. Due to input buffering some elements may have been requested from upstream publishers that will then not be processed downstream of this step.Note that this can be combined with
take(long)
to limit the number of elements within the duration.'''Emits when''' an upstream element arrives
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or timer fires
'''Cancels when''' downstream cancels or timer fires
-
takeWithin
public Source<Out,Mat> takeWithin(java.time.Duration duration)
Terminate processing (and cancel the upstream publisher) after the given duration. Due to input buffering some elements may have been requested from upstream publishers that will then not be processed downstream of this step.Note that this can be combined with
take(long)
to limit the number of elements within the duration.'''Emits when''' an upstream element arrives
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or timer fires
'''Cancels when''' downstream cancels or timer fires
-
conflateWithSeed
public <S> Source<S,Mat> conflateWithSeed(Function<Out,S> seed, Function2<S,Out,S> aggregate)
Allows a faster upstream to progress independently of a slower subscriber by conflating elements into a summary until the subscriber is ready to accept them. For example a conflate step might average incoming numbers if the upstream publisher is faster.This version of conflate allows to derive a seed from the first element and change the aggregated type to be different than the input type. See
Flow.conflate
for a simpler version that does not change types.This element only rolls up elements if the upstream is faster, but if the downstream is faster it will not duplicate elements.
Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' downstream stops backpressuring and there is a conflated element available
'''Backpressures when''' never
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
see also
Source.conflate
Source.batch
Source.batchWeighted
- Parameters:
seed
- Provides the first state for a conflated value using the first unconsumed element as a startaggregate
- Takes the currently aggregated value and the current pending element to produce a new aggregate
-
conflate
public Source<Out,Mat> conflate(Function2<Out,Out,Out> aggregate)
Allows a faster upstream to progress independently of a slower subscriber by conflating elements into a summary until the subscriber is ready to accept them. For example a conflate step might average incoming numbers if the upstream publisher is faster. This version of conflate does not change the output type of the stream. SeeSource.conflateWithSeed
for a more flexible version that can take a seed function and transform elements while rolling up.This element only rolls up elements if the upstream is faster, but if the downstream is faster it will not duplicate elements.
Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' downstream stops backpressuring and there is a conflated element available
'''Backpressures when''' never
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
see also
Source.conflateWithSeed
Source.batch
Source.batchWeighted
- Parameters:
aggregate
- Takes the currently aggregated value and the current pending element to produce a new aggregate
-
batch
public <S> Source<S,Mat> batch(long max, Function<Out,S> seed, Function2<S,Out,S> aggregate)
Allows a faster upstream to progress independently of a slower subscriber by aggregating elements into batches until the subscriber is ready to accept them. For example a batch step might store received elements in an array up to the allowed max limit if the upstream publisher is faster.This element only rolls up elements if the upstream is faster, but if the downstream is faster it will not duplicate elements.
Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' downstream stops backpressuring and there is an aggregated element available
'''Backpressures when''' there are
max
batched elements and 1 pending element and downstream backpressures'''Completes when''' upstream completes and there is no batched/pending element waiting
'''Cancels when''' downstream cancels
See also
Source.conflate
,Source.batchWeighted
- Parameters:
max
- maximum number of elements to batch before backpressuring upstream (must be positive non-zero)seed
- Provides the first state for a batched value using the first unconsumed element as a startaggregate
- Takes the currently batched value and the current pending element to produce a new aggregate
-
batchWeighted
public <S> Source<S,Mat> batchWeighted(long max, Function<Out,java.lang.Long> costFn, Function<Out,S> seed, Function2<S,Out,S> aggregate)
Allows a faster upstream to progress independently of a slower subscriber by aggregating elements into batches until the subscriber is ready to accept them. For example a batch step might concatenateByteString
elements up to the allowed max limit if the upstream publisher is faster.This element only rolls up elements if the upstream is faster, but if the downstream is faster it will not duplicate elements.
Batching will apply for all elements, even if a single element cost is greater than the total allowed limit. In this case, previous batched elements will be emitted, then the "heavy" element will be emitted (after being applied with the
seed
function) without batching further elements with it, and then the rest of the incoming elements are batched.'''Emits when''' downstream stops backpressuring and there is a batched element available
'''Backpressures when''' there are
max
weighted batched elements + 1 pending element and downstream backpressures'''Completes when''' upstream completes and there is no batched/pending element waiting
'''Cancels when''' downstream cancels
See also
Source.conflate
,Source.batch
- Parameters:
max
- maximum weight of elements to batch before backpressuring upstream (must be positive non-zero)costFn
- a function to compute a single element weightseed
- Provides the first state for a batched value using the first unconsumed element as a startaggregate
- Takes the currently batched value and the current pending element to produce a new batch
-
expand
public <U> Source<U,Mat> expand(Function<Out,java.util.Iterator<U>> expander)
Allows a faster downstream to progress independently of a slower publisher by extrapolating elements from an older element until new element comes from the upstream. For example an expand step might repeat the last element for the subscriber until it receives an update from upstream.This element will never "drop" upstream elements as all elements go through at least one extrapolation step. This means that if the upstream is actually faster than the upstream it will be backpressured by the downstream subscriber.
Expand does not support
pekko.stream.Supervision#restart
andpekko.stream.Supervision#resume
. Exceptions from theexpander
function will complete the stream with failure.See also
extrapolate(org.apache.pekko.japi.function.Function<Out, java.util.Iterator<Out>>)
for a version that always preserves the original element and allows for an initial "startup" element.'''Emits when''' downstream stops backpressuring
'''Backpressures when''' downstream backpressures or iterator runs empty
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
- Parameters:
expander
- Takes the current extrapolation state to produce an output element and the next extrapolation state.- See Also:
extrapolate(org.apache.pekko.japi.function.Function<Out, java.util.Iterator<Out>>)
-
extrapolate
public Source<Out,Mat> extrapolate(Function<Out,java.util.Iterator<Out>> extrapolator)
Allows a faster downstream to progress independent of a slower upstream.This is achieved by introducing "extrapolated" elements - based on those from upstream - whenever downstream signals demand.
Extrapolate does not support
pekko.stream.Supervision#restart
andpekko.stream.Supervision#resume
. Exceptions from theextrapolate
function will complete the stream with failure.See also
expand(org.apache.pekko.japi.function.Function<Out, java.util.Iterator<U>>)
for a version that can overwrite the original element.'''Emits when''' downstream stops backpressuring, AND EITHER upstream emits OR initial element is present OR
extrapolate
is non-empty and applicable'''Backpressures when''' downstream backpressures or current
extrapolate
runs empty'''Completes when''' upstream completes and current
extrapolate
runs empty'''Cancels when''' downstream cancels
- Parameters:
extrapolator
- Takes the current upstream element and provides a sequence of "extrapolated" elements based on the original, to be emitted in case downstream signals demand.- See Also:
expand(org.apache.pekko.japi.function.Function<Out, java.util.Iterator<U>>)
-
extrapolate
public Source<Out,Mat> extrapolate(Function<Out,java.util.Iterator<Out>> extrapolator, Out initial)
Allows a faster downstream to progress independent of a slower upstream.This is achieved by introducing "extrapolated" elements - based on those from upstream - whenever downstream signals demand.
Extrapolate does not support
pekko.stream.Supervision#restart
andpekko.stream.Supervision#resume
. Exceptions from theextrapolate
function will complete the stream with failure.See also
expand(org.apache.pekko.japi.function.Function<Out, java.util.Iterator<U>>)
for a version that can overwrite the original element.'''Emits when''' downstream stops backpressuring, AND EITHER upstream emits OR initial element is present OR
extrapolate
is non-empty and applicable'''Backpressures when''' downstream backpressures or current
extrapolate
runs empty'''Completes when''' upstream completes and current
extrapolate
runs empty'''Cancels when''' downstream cancels
- Parameters:
extrapolator
- takes the current upstream element and provides a sequence of "extrapolated" elements based on the original, to be emitted in case downstream signals demand.initial
- the initial element to be emitted, in case upstream is able to stall the entire stream.- See Also:
expand(org.apache.pekko.japi.function.Function<Out, java.util.Iterator<U>>)
-
buffer
public Source<Out,Mat> buffer(int size, OverflowStrategy overflowStrategy)
Adds a fixed size buffer in the flow that allows to store elements from a faster upstream until it becomes full. Depending on the definedpekko.stream.OverflowStrategy
it might drop elements or backpressure the upstream if there is no space available'''Emits when''' downstream stops backpressuring and there is a pending element in the buffer
'''Backpressures when''' downstream backpressures or depending on OverflowStrategy:
- Backpressure - backpressures when buffer is full
- DropHead, DropTail, DropBuffer - never backpressures
- Fail - fails the stream if buffer gets full
'''Completes when''' upstream completes and buffered elements has been drained
'''Cancels when''' downstream cancels
- Parameters:
size
- The size of the buffer in element countoverflowStrategy
- Strategy that is used when incoming elements cannot fit inside the buffer
-
prefixAndTail
public Source<Pair<java.util.List<Out>,Source<Out,NotUsed>>,Mat> prefixAndTail(int n)
Takes up ton
elements from the stream (less thann
if the upstream completes before emittingn
elements) and returns a pair containing a strict sequence of the taken element and a stream representing the remaining elements. If ''n'' is zero or negative, then this will return a pair of an empty collection and a stream containing the whole upstream unchanged.In case of an upstream error, depending on the current state - the master stream signals the error if less than
n
elements has been seen, and therefore the substream has not yet been emitted - the tail substream signals the error after the prefix and tail has been emitted by the main stream (at that point the main stream has already completed)'''Emits when''' the configured number of prefix elements are available. Emits this prefix, and the rest as a substream
'''Backpressures when''' downstream backpressures or substream backpressures
'''Completes when''' prefix elements has been consumed and substream has been consumed
'''Cancels when''' downstream cancels or substream cancels
-
flatMapPrefix
public <Out2,Mat2> Source<Out2,Mat> flatMapPrefix(int n, Function<java.lang.Iterable<Out>,Flow<Out,Out2,Mat2>> f)
Takes up ton
elements from the stream (less thann
only if the upstream completes before emittingn
elements), then applyf
on these elements in order to obtain a flow, this flow is then materialized and the rest of the input is processed by this flow (similar to via). This method returns a flow consuming the rest of the stream producing the materialized flow's output.'''Emits when''' the materialized flow emits. Notice the first
n
elements are buffered internally before materializing the flow and connecting it to the rest of the upstream - producing elements at its own discretion (might 'swallow' or multiply elements).'''Backpressures when''' downstream backpressures
'''Completes when''' the materialized flow completes. If upstream completes before producing
n
elements,f
will be applied with the provided elements, the resulting flow will be materialized and signalled for upstream completion, it can then complete or continue to emit elements at its own discretion.'''Cancels when''' the materialized flow cancels. Notice that when downstream cancels prior to prefix completion, the cancellation cause is stashed until prefix completion (or upstream completion) and then handed to the materialized flow.
- Parameters:
n
- the number of elements to accumulate before materializing the downstream flow.f
- a function that produces the downstream flow based on the upstream's prefix.
-
flatMapPrefixMat
public <Out2,Mat2,Mat3> Source<Out2,Mat3> flatMapPrefixMat(int n, Function<java.lang.Iterable<Out>,Flow<Out,Out2,Mat2>> f, Function2<Mat,java.util.concurrent.CompletionStage<Mat2>,Mat3> matF)
mat version offlatMapPrefix(int, org.apache.pekko.japi.function.Function<java.lang.Iterable<Out>, org.apache.pekko.stream.javadsl.Flow<Out, Out2, Mat2>>)
, this method gives access to a future materialized value of the downstream flow (as a completion stage). seeflatMapPrefix(int, org.apache.pekko.japi.function.Function<java.lang.Iterable<Out>, org.apache.pekko.stream.javadsl.Flow<Out, Out2, Mat2>>)
for details.
-
groupBy
public <K> SubSource<Out,Mat> groupBy(int maxSubstreams, Function<Out,K> f, boolean allowClosedSubstreamRecreation)
This operation demultiplexes the incoming stream into separate output streams, one for each element key. The key is computed for each element using the given function. When a new key is encountered for the first time a new substream is opened and subsequently fed with all elements belonging to that key.WARNING: If
allowClosedSubstreamRecreation
is set tofalse
(default behavior) the operator keeps track of all keys of streams that have already been closed. If you expect an infinite number of keys this can cause memory issues. Elements belonging to those keys are drained directly and not send to the substream.Note: If
allowClosedSubstreamRecreation
is set totrue
substream completion and incoming elements are subject to race-conditions. If elements arrive for a stream that is in the process of closing these elements might get lost.The object returned from this method is not a normal
Flow
, it is aSubFlow
. This means that after this operator all transformations are applied to all encountered substreams in the same fashion. Substream mode is exited either by closing the substream (i.e. connecting it to aSink
) or by merging the substreams back together; see theto
andmergeBack
methods onSubFlow
for more information.It is important to note that the substreams also propagate back-pressure as any other stream, which means that blocking one substream will block the
groupBy
operator itself—and thereby all substreams—once all internal or explicit buffers are filled.If the group by function
f
throws an exception and the supervision decision ispekko.stream.Supervision#stop
the stream and substreams will be completed with failure.If the group by function
f
throws an exception and the supervision decision ispekko.stream.Supervision#resume
orpekko.stream.Supervision#restart
the element is dropped and the stream and substreams continue.Function
f
MUST NOT returnnull
. This will throw exception and trigger supervision decision mechanism.'''Emits when''' an element for which the grouping function returns a group that has not yet been created. Emits the new group
'''Backpressures when''' there is an element pending for a group whose substream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels and all substreams cancel
- Parameters:
maxSubstreams
- configures the maximum number of substreams (keys) that are supported; if more distinct keys are encountered then the stream failsf
- computes the key for each elementallowClosedSubstreamRecreation
- enables recreation of already closed substreams if elements with their corresponding keys arrive after completion
-
groupBy
public <K> SubSource<Out,Mat> groupBy(int maxSubstreams, Function<Out,K> f)
This operation demultiplexes the incoming stream into separate output streams, one for each element key. The key is computed for each element using the given function. When a new key is encountered for the first time a new substream is opened and subsequently fed with all elements belonging to that key.The object returned from this method is not a normal
Flow
, it is aSubSource
. This means that after this operator all transformations are applied to all encountered substreams in the same fashion. Substream mode is exited either by closing the substream (i.e. connecting it to aSink
) or by merging the substreams back together; see theto
andmergeBack
methods onSubSource
for more information.It is important to note that the substreams also propagate back-pressure as any other stream, which means that blocking one substream will block the
groupBy
operator itself—and thereby all substreams—once all internal or explicit buffers are filled.If the group by function
f
throws an exception and the supervision decision ispekko.stream.Supervision#stop
the stream and substreams will be completed with failure.If the group by function
f
throws an exception and the supervision decision ispekko.stream.Supervision#resume
orpekko.stream.Supervision#restart
the element is dropped and the stream and substreams continue.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' an element for which the grouping function returns a group that has not yet been created. Emits the new group
'''Backpressures when''' there is an element pending for a group whose substream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels and all substreams cancel
- Parameters:
maxSubstreams
- configures the maximum number of substreams (keys) that are supported; if more distinct keys are encountered then the stream fails
-
splitWhen
public SubSource<Out,Mat> splitWhen(Predicate<Out> p)
This operation applies the given predicate to all incoming elements and emits them to a stream of output streams, always beginning a new one with the current element if the given predicate returns true for it. This means that for the following series of predicate values, three substreams will be produced with lengths 1, 2, and 3:false, // element goes into first substream true, false, // elements go into second substream true, false, false // elements go into third substream
In case the *first* element of the stream matches the predicate, the first substream emitted by splitWhen will start from that element. For example:
true, false, false // first substream starts from the split-by element true, false // subsequent substreams operate the same way
The object returned from this method is not a normal
Flow
, it is aSubSource
. This means that after this operator all transformations are applied to all encountered substreams in the same fashion. Substream mode is exited either by closing the substream (i.e. connecting it to aSink
) or by merging the substreams back together; see theto
andmergeBack
methods onSubSource
for more information.It is important to note that the substreams also propagate back-pressure as any other stream, which means that blocking one substream will block the
splitWhen
operator itself—and thereby all substreams—once all internal or explicit buffers are filled.If the split predicate
p
throws an exception and the supervision decision ispekko.stream.Supervision.Stop
the stream and substreams will be completed with failure.If the split predicate
p
throws an exception and the supervision decision ispekko.stream.Supervision.Resume
orpekko.stream.Supervision.Restart
the element is dropped and the stream and substreams continue.'''Emits when''' an element for which the provided predicate is true, opening and emitting a new substream for subsequent element
'''Backpressures when''' there is an element pending for the next substream, but the previous is not fully consumed yet, or the substream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels and substreams cancel
See also
Source.splitAfter
.
-
splitWhen
public SubSource<Out,Mat> splitWhen(SubstreamCancelStrategy substreamCancelStrategy, Predicate<Out> p)
Deprecated.Use .withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather than a SubstreamCancelStrategy. Since 1.1.0.This operation applies the given predicate to all incoming elements and emits them to a stream of output streams, always beginning a new one with the current element if the given predicate returns true for it.
-
splitAfter
public SubSource<Out,Mat> splitAfter(Predicate<Out> p)
This operation applies the given predicate to all incoming elements and emits them to a stream of output streams. It *ends* the current substream when the predicate is true. This means that for the following series of predicate values, three substreams will be produced with lengths 2, 2, and 3:false, true, // elements go into first substream false, true, // elements go into second substream false, false, true // elements go into third substream
The object returned from this method is not a normal
Flow
, it is aSubSource
. This means that after this operator all transformations are applied to all encountered substreams in the same fashion. Substream mode is exited either by closing the substream (i.e. connecting it to aSink
) or by merging the substreams back together; see theto
andmergeBack
methods onSubSource
for more information.It is important to note that the substreams also propagate back-pressure as any other stream, which means that blocking one substream will block the
splitAfter
operator itself—and thereby all substreams—once all internal or explicit buffers are filled.If the split predicate
p
throws an exception and the supervision decision ispekko.stream.Supervision.Stop
the stream and substreams will be completed with failure.If the split predicate
p
throws an exception and the supervision decision ispekko.stream.Supervision.Resume
orpekko.stream.Supervision.Restart
the element is dropped and the stream and substreams continue.'''Emits when''' an element passes through. When the provided predicate is true it emits the element and opens a new substream for subsequent element
'''Backpressures when''' there is an element pending for the next substream, but the previous is not fully consumed yet, or the substream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels and substreams cancel
See also
Source.splitWhen
.
-
splitAfter
public SubSource<Out,Mat> splitAfter(SubstreamCancelStrategy substreamCancelStrategy, Predicate<Out> p)
Deprecated.Use .withAttributes(ActorAttributes.supervisionStrategy(equivalentDecider)) rather than a SubstreamCancelStrategy. Since 1.1.0.This operation applies the given predicate to all incoming elements and emits them to a stream of output streams. It *ends* the current substream when the predicate is true.
-
flatMapConcat
public <T,M> Source<T,Mat> flatMapConcat(Function<Out,? extends Graph<SourceShape<T>,M>> f)
Transform each input element into aSource
of output elements that is then flattened into the output stream by concatenation, fully consuming one Source after the other.'''Emits when''' a currently consumed substream has an element available
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes and all consumed substreams complete
'''Cancels when''' downstream cancels
-
flatMapMerge
public <T,M> Source<T,Mat> flatMapMerge(int breadth, Function<Out,? extends Graph<SourceShape<T>,M>> f)
Transform each input element into aSource
of output elements that is then flattened into the output stream by merging, where at mostbreadth
substreams are being consumed at any given time.'''Emits when''' a currently consumed substream has an element available
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes and all consumed substreams complete
'''Cancels when''' downstream cancels
-
initialTimeout
public Source<Out,Mat> initialTimeout(scala.concurrent.duration.FiniteDuration timeout)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.If the first element has not passed through this operator before the provided timeout, the stream is failed with aInitialTimeoutException
.'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or fails if timeout elapses before first element arrives
'''Cancels when''' downstream cancels
-
initialTimeout
public Source<Out,Mat> initialTimeout(java.time.Duration timeout)
If the first element has not passed through this operator before the provided timeout, the stream is failed with aInitialTimeoutException
.'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or fails if timeout elapses before first element arrives
'''Cancels when''' downstream cancels
-
completionTimeout
public Source<Out,Mat> completionTimeout(scala.concurrent.duration.FiniteDuration timeout)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.If the completion of the stream does not happen until the provided timeout, the stream is failed with aCompletionTimeoutException
.'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or fails if timeout elapses before upstream completes
'''Cancels when''' downstream cancels
-
completionTimeout
public Source<Out,Mat> completionTimeout(java.time.Duration timeout)
If the completion of the stream does not happen until the provided timeout, the stream is failed with aCompletionTimeoutException
.'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or fails if timeout elapses before upstream completes
'''Cancels when''' downstream cancels
-
idleTimeout
public Source<Out,Mat> idleTimeout(scala.concurrent.duration.FiniteDuration timeout)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.If the time between two processed elements exceeds the provided timeout, the stream is failed with aStreamIdleTimeoutException
. The timeout is checked periodically, so the resolution of the check is one period (equals to timeout value).'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or fails if timeout elapses between two emitted elements
'''Cancels when''' downstream cancels
-
idleTimeout
public Source<Out,Mat> idleTimeout(java.time.Duration timeout)
If the time between two processed elements exceeds the provided timeout, the stream is failed with aStreamIdleTimeoutException
. The timeout is checked periodically, so the resolution of the check is one period (equals to timeout value).'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or fails if timeout elapses between two emitted elements
'''Cancels when''' downstream cancels
-
backpressureTimeout
public Source<Out,Mat> backpressureTimeout(scala.concurrent.duration.FiniteDuration timeout)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.If the time between the emission of an element and the following downstream demand exceeds the provided timeout, the stream is failed with aBackpressureTimeoutException
. The timeout is checked periodically, so the resolution of the check is one period (equals to timeout value).'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or fails if timeout elapses between element emission and downstream demand.
'''Cancels when''' downstream cancels
-
backpressureTimeout
public Source<Out,Mat> backpressureTimeout(java.time.Duration timeout)
If the time between the emission of an element and the following downstream demand exceeds the provided timeout, the stream is failed with aBackpressureTimeoutException
. The timeout is checked periodically, so the resolution of the check is one period (equals to timeout value).'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes or fails if timeout elapses between element emission and downstream demand.
'''Cancels when''' downstream cancels
-
keepAlive
public Source<Out,Mat> keepAlive(scala.concurrent.duration.FiniteDuration maxIdle, Creator<Out> injectedElem)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Injects additional elements if upstream does not emit for a configured amount of time. In other words, this operator attempts to maintains a base rate of emitted elements towards the downstream.If the downstream backpressures then no element is injected until downstream demand arrives. Injected elements do not accumulate during this period.
Upstream elements are always preferred over injected elements.
'''Emits when''' upstream emits an element or if the upstream was idle for the configured period
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
keepAlive
public Source<Out,Mat> keepAlive(java.time.Duration maxIdle, Creator<Out> injectedElem)
Injects additional elements if upstream does not emit for a configured amount of time. In other words, this operator attempts to maintains a base rate of emitted elements towards the downstream.If the downstream backpressures then no element is injected until downstream demand arrives. Injected elements do not accumulate during this period.
Upstream elements are always preferred over injected elements.
'''Emits when''' upstream emits an element or if the upstream was idle for the configured period
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
throttle
public Source<Out,Mat> throttle(int elements, java.time.Duration per)
Sends elements downstream with speed limited toelements/per
. In other words, this operator set the maximum rate for emitting messages. This operator works for streams where all elements have the same cost or length.Throttle implements the token bucket model. There is a bucket with a given token capacity (burst size). Tokens drops into the bucket at a given rate and can be
spared
for later use up to bucket capacity to allow some burstiness. Whenever stream wants to send an element, it takes as many tokens from the bucket as element costs. If there isn't any, throttle waits until the bucket accumulates enough tokens. Elements that costs more than the allowed burst will be delayed proportionally to their cost minus available tokens, meeting the target rate. Bucket is full when stream just materialized and started.The burst size is calculated based on the given rate (
cost/per
) as 0.1 * rate, for example: - rate < 20/second => burst size 1 - rate 20/second => burst size 2 - rate 100/second => burst size 10 - rate 200/second => burst size 20The throttle
mode
ispekko.stream.ThrottleMode.Shaping
, which makes pauses before emitting messages to meet throttle rate.'''Emits when''' upstream emits an element and configured time per each element elapsed
'''Backpressures when''' downstream backpressures or the incoming rate is higher than the speed limit
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
throttle
public Source<Out,Mat> throttle(int elements, scala.concurrent.duration.FiniteDuration per, int maximumBurst, ThrottleMode mode)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Sends elements downstream with speed limited toelements/per
. In other words, this operator set the maximum rate for emitting messages. This operator works for streams where all elements have the same cost or length.Throttle implements the token bucket model. There is a bucket with a given token capacity (burst size or maximumBurst). Tokens drops into the bucket at a given rate and can be
spared
for later use up to bucket capacity to allow some burstiness. Whenever stream wants to send an element, it takes as many tokens from the bucket as element costs. If there isn't any, throttle waits until the bucket accumulates enough tokens. Elements that costs more than the allowed burst will be delayed proportionally to their cost minus available tokens, meeting the target rate. Bucket is full when stream just materialized and started.Parameter
mode
manages behavior when upstream is faster than throttle rate: -pekko.stream.ThrottleMode.Shaping
makes pauses before emitting messages to meet throttle rate -pekko.stream.ThrottleMode.Enforcing
fails with exception when upstream is faster than throttle rateIt is recommended to use non-zero burst sizes as they improve both performance and throttling precision by allowing the implementation to avoid using the scheduler when input rates fall below the enforced limit and to reduce most of the inaccuracy caused by the scheduler resolution (which is in the range of milliseconds).
WARNING: Be aware that throttle is using scheduler to slow down the stream. This scheduler has minimal time of triggering next push. Consequently it will slow down the stream as it has minimal pause for emitting. This can happen in case burst is 0 and speed is higher than 30 events per second. You need to increase the
maximumBurst
if elements arrive with small interval (30 milliseconds or less). Use the overloadedthrottle
method withoutmaximumBurst
parameter to automatically calculate themaximumBurst
based on the given rate (cost/per
). In other words the throttler always enforces the rate limit whenmaximumBurst
parameter is given, but in certain cases (mostly due to limited scheduler resolution) it enforces a tighter bound than what was prescribed.'''Emits when''' upstream emits an element and configured time per each element elapsed
'''Backpressures when''' downstream backpressures or the incoming rate is higher than the speed limit
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
throttle
public Source<Out,Mat> throttle(int elements, java.time.Duration per, int maximumBurst, ThrottleMode mode)
Sends elements downstream with speed limited toelements/per
. In other words, this operator set the maximum rate for emitting messages. This operator works for streams where all elements have the same cost or length.Throttle implements the token bucket model. There is a bucket with a given token capacity (burst size or maximumBurst). Tokens drops into the bucket at a given rate and can be
spared
for later use up to bucket capacity to allow some burstiness. Whenever stream wants to send an element, it takes as many tokens from the bucket as element costs. If there isn't any, throttle waits until the bucket accumulates enough tokens. Elements that costs more than the allowed burst will be delayed proportionally to their cost minus available tokens, meeting the target rate. Bucket is full when stream just materialized and started.Parameter
mode
manages behavior when upstream is faster than throttle rate: -pekko.stream.ThrottleMode.Shaping
makes pauses before emitting messages to meet throttle rate -pekko.stream.ThrottleMode.Enforcing
fails with exception when upstream is faster than throttle rateIt is recommended to use non-zero burst sizes as they improve both performance and throttling precision by allowing the implementation to avoid using the scheduler when input rates fall below the enforced limit and to reduce most of the inaccuracy caused by the scheduler resolution (which is in the range of milliseconds).
WARNING: Be aware that throttle is using scheduler to slow down the stream. This scheduler has minimal time of triggering next push. Consequently it will slow down the stream as it has minimal pause for emitting. This can happen in case burst is 0 and speed is higher than 30 events per second. You need to increase the
maximumBurst
if elements arrive with small interval (30 milliseconds or less). Use the overloadedthrottle
method withoutmaximumBurst
parameter to automatically calculate themaximumBurst
based on the given rate (cost/per
). In other words the throttler always enforces the rate limit whenmaximumBurst
parameter is given, but in certain cases (mostly due to limited scheduler resolution) it enforces a tighter bound than what was prescribed.'''Emits when''' upstream emits an element and configured time per each element elapsed
'''Backpressures when''' downstream backpressures or the incoming rate is higher than the speed limit
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
throttle
public Source<Out,Mat> throttle(int cost, java.time.Duration per, Function<Out,java.lang.Integer> costCalculation)
Sends elements downstream with speed limited tocost/per
. Cost is calculating for each element individually by callingcalculateCost
function. This operator works for streams when elements have different cost(length). Streams ofByteString
for example.Throttle implements the token bucket model. There is a bucket with a given token capacity (burst size). Tokens drops into the bucket at a given rate and can be
spared
for later use up to bucket capacity to allow some burstiness. Whenever stream wants to send an element, it takes as many tokens from the bucket as element costs. If there isn't any, throttle waits until the bucket accumulates enough tokens. Elements that costs more than the allowed burst will be delayed proportionally to their cost minus available tokens, meeting the target rate. Bucket is full when stream just materialized and started.The burst size is calculated based on the given rate (
cost/per
) as 0.1 * rate, for example: - rate < 20/second => burst size 1 - rate 20/second => burst size 2 - rate 100/second => burst size 10 - rate 200/second => burst size 20The throttle
mode
ispekko.stream.ThrottleMode.Shaping
, which makes pauses before emitting messages to meet throttle rate.'''Emits when''' upstream emits an element and configured time per each element elapsed
'''Backpressures when''' downstream backpressures or the incoming rate is higher than the speed limit
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
throttle
public Source<Out,Mat> throttle(int cost, scala.concurrent.duration.FiniteDuration per, int maximumBurst, Function<Out,java.lang.Integer> costCalculation, ThrottleMode mode)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Sends elements downstream with speed limited tocost/per
. Cost is calculating for each element individually by callingcalculateCost
function. This operator works for streams when elements have different cost(length). Streams ofByteString
for example.Throttle implements the token bucket model. There is a bucket with a given token capacity (burst size or maximumBurst). Tokens drops into the bucket at a given rate and can be
spared
for later use up to bucket capacity to allow some burstiness. Whenever stream wants to send an element, it takes as many tokens from the bucket as element costs. If there isn't any, throttle waits until the bucket accumulates enough tokens. Elements that costs more than the allowed burst will be delayed proportionally to their cost minus available tokens, meeting the target rate. Bucket is full when stream just materialized and started.Parameter
mode
manages behavior when upstream is faster than throttle rate: -pekko.stream.ThrottleMode.Shaping
makes pauses before emitting messages to meet throttle rate -pekko.stream.ThrottleMode.Enforcing
fails with exception when upstream is faster than throttle rate. Enforcing cannot emit elements that cost more than the maximumBurstIt is recommended to use non-zero burst sizes as they improve both performance and throttling precision by allowing the implementation to avoid using the scheduler when input rates fall below the enforced limit and to reduce most of the inaccuracy caused by the scheduler resolution (which is in the range of milliseconds).
WARNING: Be aware that throttle is using scheduler to slow down the stream. This scheduler has minimal time of triggering next push. Consequently it will slow down the stream as it has minimal pause for emitting. This can happen in case burst is 0 and speed is higher than 30 events per second. You need to increase the
maximumBurst
if elements arrive with small interval (30 milliseconds or less). Use the overloadedthrottle
method withoutmaximumBurst
parameter to automatically calculate themaximumBurst
based on the given rate (cost/per
). In other words the throttler always enforces the rate limit whenmaximumBurst
parameter is given, but in certain cases (mostly due to limited scheduler resolution) it enforces a tighter bound than what was prescribed.'''Emits when''' upstream emits an element and configured time per each element elapsed
'''Backpressures when''' downstream backpressures or the incoming rate is higher than the speed limit
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
throttle
public Source<Out,Mat> throttle(int cost, java.time.Duration per, int maximumBurst, Function<Out,java.lang.Integer> costCalculation, ThrottleMode mode)
Sends elements downstream with speed limited tocost/per
. Cost is calculating for each element individually by callingcalculateCost
function. This operator works for streams when elements have different cost(length). Streams ofByteString
for example.Throttle implements the token bucket model. There is a bucket with a given token capacity (burst size or maximumBurst). Tokens drops into the bucket at a given rate and can be
spared
for later use up to bucket capacity to allow some burstiness. Whenever stream wants to send an element, it takes as many tokens from the bucket as element costs. If there isn't any, throttle waits until the bucket accumulates enough tokens. Elements that costs more than the allowed burst will be delayed proportionally to their cost minus available tokens, meeting the target rate. Bucket is full when stream just materialized and started.Parameter
mode
manages behavior when upstream is faster than throttle rate: -pekko.stream.ThrottleMode.Shaping
makes pauses before emitting messages to meet throttle rate -pekko.stream.ThrottleMode.Enforcing
fails with exception when upstream is faster than throttle rate. Enforcing cannot emit elements that cost more than the maximumBurstIt is recommended to use non-zero burst sizes as they improve both performance and throttling precision by allowing the implementation to avoid using the scheduler when input rates fall below the enforced limit and to reduce most of the inaccuracy caused by the scheduler resolution (which is in the range of milliseconds).
WARNING: Be aware that throttle is using scheduler to slow down the stream. This scheduler has minimal time of triggering next push. Consequently it will slow down the stream as it has minimal pause for emitting. This can happen in case burst is 0 and speed is higher than 30 events per second. You need to increase the
maximumBurst
if elements arrive with small interval (30 milliseconds or less). Use the overloadedthrottle
method withoutmaximumBurst
parameter to automatically calculate themaximumBurst
based on the given rate (cost/per
). In other words the throttler always enforces the rate limit whenmaximumBurst
parameter is given, but in certain cases (mostly due to limited scheduler resolution) it enforces a tighter bound than what was prescribed.'''Emits when''' upstream emits an element and configured time per each element elapsed
'''Backpressures when''' downstream backpressures or the incoming rate is higher than the speed limit
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
throttleEven
public Source<Out,Mat> throttleEven(int elements, scala.concurrent.duration.FiniteDuration per, ThrottleMode mode)
Deprecated.Use throttle without `maximumBurst` parameter instead. Since Akka 2.5.12.This is a simplified version of throttle that spreads events evenly across the given time interval.Use this operator when you need just slow down a stream without worrying about exact amount of time between events.
If you want to be sure that no time interval has no more than specified number of events you need to use
throttle(int,java.time.Duration)
with maximumBurst attribute.- See Also:
throttle(int, java.time.Duration)
-
throttleEven
public Source<Out,Mat> throttleEven(int elements, java.time.Duration per, ThrottleMode mode)
Deprecated.Use throttle without `maximumBurst` parameter instead. Since Akka 2.5.12.This is a simplified version of throttle that spreads events evenly across the given time interval.Use this operator when you need just slow down a stream without worrying about exact amount of time between events.
If you want to be sure that no time interval has no more than specified number of events you need to use
throttle(int,java.time.Duration)
with maximumBurst attribute.- See Also:
throttle(int, java.time.Duration)
-
throttleEven
public Source<Out,Mat> throttleEven(int cost, scala.concurrent.duration.FiniteDuration per, scala.Function1<Out,java.lang.Object> costCalculation, ThrottleMode mode)
Deprecated.Use throttle without `maximumBurst` parameter instead. Since Akka 2.5.12.This is a simplified version of throttle that spreads events evenly across the given time interval.Use this operator when you need just slow down a stream without worrying about exact amount of time between events.
If you want to be sure that no time interval has no more than specified number of events you need to use
throttle(int,java.time.Duration)
with maximumBurst attribute.- See Also:
throttle(int, java.time.Duration)
-
throttleEven
public Source<Out,Mat> throttleEven(int cost, java.time.Duration per, scala.Function1<Out,java.lang.Object> costCalculation, ThrottleMode mode)
Deprecated.Use throttle without `maximumBurst` parameter instead. Since Akka 2.5.12.This is a simplified version of throttle that spreads events evenly across the given time interval.Use this operator when you need just slow down a stream without worrying about exact amount of time between events.
If you want to be sure that no time interval has no more than specified number of events you need to use
throttle(int,java.time.Duration)
with maximumBurst attribute.- See Also:
throttle(int, java.time.Duration)
-
detach
public Source<Out,Mat> detach()
Detaches upstream demand from downstream demand without detaching the stream rates; in other words acts like a buffer of size 1.'''Emits when''' upstream emits an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
watchTermination
public <M> Source<Out,M> watchTermination(Function2<Mat,java.util.concurrent.CompletionStage<Done>,M> matF)
Materializes toFuture[Done]
that completes on getting termination message. The Future completes with success when received complete message from upstream or cancel from downstream. It fails with the same error when received error message from downstream.
-
monitor
public <M> Source<Out,M> monitor(Function2<Mat,FlowMonitor<Out>,M> combine)
Deprecated.Use monitor() or monitorMat(combine) instead. Since Akka 2.5.17.Materializes toFlowMonitor
that allows monitoring of the current flow. All events are propagated by the monitor unchanged. Note that the monitor inserts a memory barrier every time it processes an event, and may therefor affect performance. Thecombine
function is used to combine theFlowMonitor
with this flow's materialized value.
-
monitorMat
public <M> Source<Out,M> monitorMat(Function2<Mat,FlowMonitor<Out>,M> combine)
Materializes toFlowMonitor[Out]
that allows monitoring of the current flow. All events are propagated by the monitor unchanged. Note that the monitor inserts a memory barrier every time it processes an event, and may therefor affect performance. Thecombine
function is used to combine theFlowMonitor
with this flow's materialized value.
-
monitor
public Source<Out,Pair<Mat,FlowMonitor<Out>>> monitor()
Materializes toPair
, which is unlike most other operators (!), in which usually the default materialized value keeping semantics is to keep the left value (by passing> Keep.left()
to a*Mat
version of a method). This operator is an exception from that rule and keeps both values since dropping its sole purpose is to introduce that materialized value.The
FlowMonitor
allows monitoring of the current flow. All events are propagated by the monitor unchanged. Note that the monitor inserts a memory barrier every time it processes an event, and may therefor affect performance.
-
initialDelay
public Source<Out,Mat> initialDelay(scala.concurrent.duration.FiniteDuration delay)
Deprecated.Use the overloaded one which accepts java.time.Duration instead. Since Akka 2.5.12.Delays the initial element by the specified duration.'''Emits when''' upstream emits an element if the initial delay is already elapsed
'''Backpressures when''' downstream backpressures or initial delay is not yet elapsed
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
initialDelay
public Source<Out,Mat> initialDelay(java.time.Duration delay)
Delays the initial element by the specified duration.'''Emits when''' upstream emits an element if the initial delay is already elapsed
'''Backpressures when''' downstream backpressures or initial delay is not yet elapsed
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
withAttributes
public Source<Out,Mat> withAttributes(Attributes attr)
Replace the attributes of thisSource
with the given ones. If this Source is a composite of multiple graphs, new attributes on the composite will be less specific than attributes set directly on the individual graphs of the composite.- Specified by:
withAttributes
in interfaceGraph<Out,Mat>
-
addAttributes
public Source<Out,Mat> addAttributes(Attributes attr)
Add the given attributes to thisSource
. If the specific attribute was already present on this graph this means the added attribute will be more specific than the existing one. If this Source is a composite of multiple graphs, new attributes on the composite will be less specific than attributes set directly on the individual graphs of the composite.- Specified by:
addAttributes
in interfaceGraph<Out,Mat>
-
async
public Source<Out,Mat> async(java.lang.String dispatcher)
Put an asynchronous boundary around thisSource
-
async
public Source<Out,Mat> async(java.lang.String dispatcher, int inputBufferSize)
Put an asynchronous boundary around thisSource
-
log
public Source<Out,Mat> log(java.lang.String name, Function<Out,java.lang.Object> extract, LoggingAdapter log)
Logs elements flowing through the stream as well as completion and erroring.By default element and completion signals are logged on debug level, and errors are logged on Error level. This can be adjusted according to your needs by providing a custom
Attributes.LogLevels
attribute on the given Flow:The
extract
function will be applied to each element before logging, so it is possible to log only those fields of a complex object flowing through this element.Uses the given
LoggingAdapter
for logging.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
log
public Source<Out,Mat> log(java.lang.String name, Function<Out,java.lang.Object> extract)
Logs elements flowing through the stream as well as completion and erroring.By default element and completion signals are logged on debug level, and errors are logged on Error level. This can be adjusted according to your needs by providing a custom
Attributes.LogLevels
attribute on the given Flow:The
extract
function will be applied to each element before logging, so it is possible to log only those fields of a complex object flowing through this element.Uses an internally created
LoggingAdapter
which usesorg.apache.pekko.stream.Log
as it's source (use this class to configure slf4j loggers).'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
log
public Source<Out,Mat> log(java.lang.String name, LoggingAdapter log)
Logs elements flowing through the stream as well as completion and erroring.By default element and completion signals are logged on debug level, and errors are logged on Error level. This can be adjusted according to your needs by providing a custom
Attributes.LogLevels
attribute on the given Flow:Uses the given
LoggingAdapter
for logging.'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
log
public Source<Out,Mat> log(java.lang.String name)
Logs elements flowing through the stream as well as completion and erroring.By default element and completion signals are logged on debug level, and errors are logged on Error level. This can be adjusted according to your needs by providing a custom
Attributes.LogLevels
attribute on the given Flow:Uses an internally created
LoggingAdapter
which usesorg.apache.pekko.stream.Log
as it's source (use this class to configure slf4j loggers).'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
logWithMarker
public Source<Out,Mat> logWithMarker(java.lang.String name, Function<Out,LogMarker> marker, Function<Out,java.lang.Object> extract, MarkerLoggingAdapter log)
Logs elements flowing through the stream as well as completion and erroring.By default element and completion signals are logged on debug level, and errors are logged on Error level. This can be adjusted according to your needs by providing a custom
Attributes.LogLevels
attribute on the given Flow:The
extract
function will be applied to each element before logging, so it is possible to log only those fields of a complex object flowing through this element.Uses the given
MarkerLoggingAdapter
for logging.Adheres to the
ActorAttributes.SupervisionStrategy
attribute.'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
logWithMarker
public Source<Out,Mat> logWithMarker(java.lang.String name, Function<Out,LogMarker> marker, Function<Out,java.lang.Object> extract)
Logs elements flowing through the stream as well as completion and erroring.By default element and completion signals are logged on debug level, and errors are logged on Error level. This can be adjusted according to your needs by providing a custom
Attributes.LogLevels
attribute on the given Flow:The
extract
function will be applied to each element before logging, so it is possible to log only those fields of a complex object flowing through this element.Uses an internally created
MarkerLoggingAdapter
which usesorg.apache.pekko.stream.Log
as it's source (use this class to configure slf4j loggers).'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
logWithMarker
public Source<Out,Mat> logWithMarker(java.lang.String name, Function<Out,LogMarker> marker, MarkerLoggingAdapter log)
Logs elements flowing through the stream as well as completion and erroring.By default element and completion signals are logged on debug level, and errors are logged on Error level. This can be adjusted according to your needs by providing a custom
Attributes.LogLevels
attribute on the given Flow:Uses the given
MarkerLoggingAdapter
for logging.'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
logWithMarker
public Source<Out,Mat> logWithMarker(java.lang.String name, Function<Out,LogMarker> marker)
Logs elements flowing through the stream as well as completion and erroring.By default element and completion signals are logged on debug level, and errors are logged on Error level. This can be adjusted according to your needs by providing a custom
Attributes.LogLevels
attribute on the given Flow:Uses an internally created
MarkerLoggingAdapter
which usesorg.apache.pekko.stream.Log
as it's source (use this class to configure slf4j loggers).'''Emits when''' the mapping function returns an element
'''Backpressures when''' downstream backpressures
'''Completes when''' upstream completes
'''Cancels when''' downstream cancels
-
asSourceWithContext
public <Ctx> SourceWithContext<Out,Ctx,Mat> asSourceWithContext(Function<Out,Ctx> extractContext)
Transform this source whose element ise
into a source producing tuple
(e, f(e))
-
aggregateWithBoundary
public <Agg,Emit> Source<Emit,Mat> aggregateWithBoundary(java.util.function.Supplier<Agg> allocate, Function2<Agg,Out,Pair<Agg,java.lang.Object>> aggregate, Function<Agg,Emit> harvest, Pair<java.util.function.Predicate<Agg>,java.time.Duration> emitOnTimer)
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 generalgroupedWeightedWithin(long,org.apache.pekko.japi.function.Function<Out,java.lang.Long>,scala.concurrent.duration.FiniteDuration)
.'''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
- Parameters:
allocate
- allocate the initial data structure for aggregated elementsaggregate
- update the aggregated elements, return true if ready to emit after update.harvest
- this is invoked before emit within the current stage/operatoremitOnTimer
- decide whether the current aggregated elements can be emitted, the custom function is invoked on every interval
-
getAttributes
public Attributes getAttributes()
- Specified by:
getAttributes
in interfaceGraph<Out,Mat>
-
-