Package org.apache.pekko.stream.scaladsl

Class Source<Out,​Mat>

  • All Implemented Interfaces:
    Graph<SourceShape<Out>,​Mat>, FlowOps<Out,​Mat>, FlowOpsMat<Out,​Mat>
    public final class Source<Out,​Mat>
extends java.lang.Object
implements FlowOpsMat<Out,​Mat>, Graph<SourceShape<Out>,​Mat>
    A Source is a set of stream processing steps that has one open output. It can comprise any number of internal sources and transformations that are wired together, or it can be an &ldquo;atomic&rdquo; source, e.g. from a collection or a file. Materialization turns a Source into a Reactive Streams Publisher (at least conceptually).

    • Constructor Detail

      • Source

        public Source​(org.apache.pekko.stream.impl.LinearTraversalBuilder traversalBuilder,
              SourceShape<Out> shape)

    • Method Detail

      • fromPublisher

        public static <T> Source<T,​NotUsed> fromPublisher​(org.reactivestreams.Publisher<T> publisher)
        Helper to create Source from Publisher.

        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 <T> Source<T,​NotUsed> fromIterator​(scala.Function0<scala.collection.Iterator<T>> f)
        Helper to create Source from Iterator. Example usage: Source.fromIterator(() => Iterator.from(0))

        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 the next() method. Elements are pulled out of the iterator in accordance with the demand coming from the downstream transformation steps.

      • fromJavaStream

        public static <T,​S extends java.util.stream.BaseStream<T,​S>> Source<T,​NotUsed> fromJavaStream​(scala.Function0<java.util.stream.BaseStream<T,​S>> stream)
        Creates a source that wraps a Java 8 Stream. 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 <T> Source<T,​NotUsed> cycle​(scala.Function0<scala.collection.Iterator<T>> f)
        Creates Source that will continually produce given elements in specified order.

        Starts 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<scala.Option<FOut>,​Mat> optionalVia​(Source<scala.Option<SOut>,​SMat> source,
                                                                                                                 Flow<SOut,​FOut,​FMat> viaFlow,
                                                                                                                 scala.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 element
        viaFlow - 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 Option which indicates whether the original source's element had viaFlow applied.
        Since:
        1.1.0

      • 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,​scala.concurrent.Future<M>> fromMaterializer​(scala.Function2<Materializer,​Attributes,​Source<T,​M>> factory)
        Defers the creation of a Source until materialization. The factory function exposes Materializer which is going to be used during materialization and Attributes of the Source returned by this method.

      • setup

        public static <T,​M> Source<T,​scala.concurrent.Future<M>> setup​(scala.Function2<ActorMaterializer,​Attributes,​Source<T,​M>> factory)
        Deprecated.
        Use 'fromMaterializer' instead. Since Akka 2.6.0.
        Defers the creation of a Source until materialization. The factory function exposes ActorMaterializer which is going to be used during materialization and Attributes of the Source returned by this method.

      • apply

        public static <T> Source<T,​NotUsed> apply​(scala.collection.immutable.Iterable<T> iterable)
        Helper to create Source from Iterable. Example usage: Source(Seq(1,2,3))

        Starts a new Source from the given Iterable. 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.

      • fromFuture

        public static <T> Source<T,​NotUsed> fromFuture​(scala.concurrent.Future<T> future)
        Deprecated.
        Use 'Source.future' instead. Since Akka 2.6.0.
        Starts a new Source from the given Future. The stream will consist of one element when the Future is completed with a successful value, which may happen before or after materializing the Flow. The stream terminates with a failure if the Future is completed with a failure.

      • fromCompletionStage

        public static <T> Source<T,​NotUsed> fromCompletionStage​(java.util.concurrent.CompletionStage<T> future)
        Deprecated.
        Use 'Source.completionStage' instead. Since Akka 2.6.0.
        Starts a new Source from the given Future. The stream will consist of one element when the Future is completed with a successful value, which may happen before or after materializing the Flow. The stream terminates with a failure if the Future is completed with a failure.

      • fromFutureSource

        public static <T,​M> Source<T,​scala.concurrent.Future<M>> fromFutureSource​(scala.concurrent.Future<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 the Future fails the stream is failed with the exception from the future. If downstream cancels before the stream completes the materialized Future will be failed with a StreamDetachedException

      • 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 scala-compat CompletionStage to future converter and 'Source.futureSource' instead. Since Akka 2.6.0.
        Streams the elements of an asynchronous source once its given completion operator completes. If the CompletionStage fails the stream is failed with the exception from the future. If downstream cancels before the stream completes the materialized Future will be failed with a StreamDetachedException

      • tick

        public static <T> Source<T,​Cancellable> tick​(scala.concurrent.duration.FiniteDuration initialDelay,
                                                   scala.concurrent.duration.FiniteDuration interval,
                                                   T 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 a Source with one element. Every connected Sink of this stream will see an individual stream consisting of one element.

      • repeat

        public static <T> Source<T,​NotUsed> repeat​(T element)
        Create a Source that will continually emit the given element.

      • unfold

        public static <S,​E> Source<E,​NotUsed> unfold​(S s,
                                                         scala.Function1<S,​scala.Option<scala.Tuple2<S,​E>>> f)
        Create a Source that will unfold a value of type S into a pair of the next state S and output elements of type E.

        For example, all the Fibonacci numbers under 10M: 

        
   Source.unfold(0 -> 1) {
    case (a, _) if a > 10000000 => None
    case (a, b) => Some((b -> (a + b)) -> a)
   }

      • unfoldAsync

        public static <S,​E> Source<E,​NotUsed> unfoldAsync​(S s,
                                                              scala.Function1<S,​scala.concurrent.Future<scala.Option<scala.Tuple2<S,​E>>>> f)
        Same as <S,E>unfold(S,scala.Function1<S,scala.Option<scala.Tuple2<S,E>>>), but uses an async function to generate the next state-element tuple.

        async fibonacci example: 

        
   Source.unfoldAsync(0 -> 1) {
    case (a, _) if a > 10000000 => Future.successful(None)
    case (a, b) => Future{
      Thread.sleep(1000)
      Some((b -> (a + b)) -> a)
    }
   }

      • iterate

        public static <T> Source<T,​NotUsed> iterate​(T seed,
                                                  scala.Function1<T,​java.lang.Object> p,
                                                  scala.Function1<T,​T> f)
        Creates a sequential Source by iterating with the given predicate and function, starting with the given seed value. If the predicate returns false for the seed, the Source completes with empty.

        Since:
        1.1.0
        See Also:
        <S,E>unfold(S,scala.Function1<S,scala.Option<scala.Tuple2<S,E>>>)

      • empty

        public static <T> Source<T,​NotUsed> empty()
        A Source with no elements, i.e. an empty stream that is completed immediately for every connected Sink.

      • maybe

        public static <T> Source<T,​scala.concurrent.Promise<scala.Option<T>>> maybe()
        Create a Source which materializes a Promise which controls what element will be emitted by the Source. If the materialized promise is completed with a Some, that value will be produced downstream, followed by completion. If the materialized promise is completed with a None, 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 None.

      • failed

        public static <T> Source<T,​NotUsed> failed​(java.lang.Throwable cause)
        Create a Source that immediately ends the stream with the cause error to every connected Sink.

      • lazily

        public static <T,​M> Source<T,​scala.concurrent.Future<M>> lazily​(scala.Function0<Source<T,​M>> create)
        Deprecated.
        Use 'Source.lazySource' instead. Since Akka 2.6.0.
        Creates a Source 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 the create factory is never called and the materialized Future is failed.

      • lazilyAsync

        public static <T> Source<T,​scala.concurrent.Future<NotUsed>> lazilyAsync​(scala.Function0<scala.concurrent.Future<T>> create)
        Deprecated.
        Use 'Source.lazyFuture' instead. Since Akka 2.6.0.
        Creates a Source 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 materialized Future 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 Future is successfully completed and then completes the stream. The stream fails if the Future is completed with a failure.

      • 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 given CompletionStage is successfully completed and then completes the stream. If the CompletionStage is completed with a failure the stream is failed.

        Here for Java interoperability, the normal use from Scala should be Source.future

      • futureSource

        public static <T,​M> Source<T,​scala.concurrent.Future<M>> futureSource​(scala.concurrent.Future<Source<T,​M>> futureSource)
        Turn a Future[Source] into a source that will emit the values of the source when the future completes successfully. If the Future is completed with a failure the stream is failed.

      • lazySingle

        public static <T> Source<T,​NotUsed> lazySingle​(scala.Function0<T> create)
        Defers invoking the create 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.

      • lazyFuture

        public static <T> Source<T,​NotUsed> lazyFuture​(scala.Function0<scala.concurrent.Future<T>> create)
        Defers invoking the create 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.

      • lazySource

        public static <T,​M> Source<T,​scala.concurrent.Future<M>> lazySource​(scala.Function0<Source<T,​M>> create)
        Defers invoking the create 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

      • lazyFutureSource

        public static <T,​M> Source<T,​scala.concurrent.Future<M>> lazyFutureSource​(scala.Function0<scala.concurrent.Future<Source<T,​M>>> create)
        Defers invoking the create 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 future completes successfully. Downstream completes when the created source completes and fails when the created source fails. If the future or the create 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 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

      • asSubscriber

        public static <T> Source<T,​org.reactivestreams.Subscriber<T>> asSubscriber()
        Creates a Source that is materialized as a Subscriber

      • actorRef

        public static <T> Source<T,​ActorRef> actorRef​(scala.PartialFunction<java.lang.Object,​CompletionStrategy> completionMatcher,
                                                    scala.PartialFunction<java.lang.Object,​java.lang.Throwable> failureMatcher,
                                                    int bufferSize,
                                                    OverflowStrategy overflowStrategy)
        Creates a Source that is materialized as an pekko.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. When bufferSize is 0 the overflowStrategy 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 extracted Throwable will be used to fail the stream. In case the Actor is still draining its internal buffer (after having received a message matched by completionMatcher) before signaling completion and it receives a message matched by failureMatcher, 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 stream
        failureMatcher - catches the failure message to fail the stream
        bufferSize - The size of the buffer in element count
        overflowStrategy - 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 instead. Since Akka 2.6.0.
        Creates a Source that is materialized as an pekko.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. When bufferSize is 0 the overflowStrategy does not matter.

        The stream can be completed successfully by sending the actor reference a pekko.actor.Status.Success. If the content is pekko.stream.CompletionStrategy.immediately the completion will be signaled immediately. Otherwise, if the content is pekko.stream.CompletionStrategy.draining (or anything else) already buffered elements will be sent out before signaling completion. Using pekko.actor.PoisonPill or pekko.actor.ActorSystem.stop to stop the actor and complete the stream is *not supported*.

        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 a pekko.actor.Status.Success) before signaling completion and it receives a pekko.actor.Status.Failure, the failure will be signaled downstream immediately (instead of the completion signal).

        The actor will be stopped when the stream is canceled from downstream, i.e. you can watch it to get notified when that happens.

        See also pekko.stream.scaladsl.Source.queue.

        Parameters:
        bufferSize - The size of the buffer in element count
        overflowStrategy - Strategy that is used when incoming elements cannot fit inside the buffer

      • actorRefWithBackpressure

        public static <T> Source<T,​ActorRef> actorRefWithBackpressure​(java.lang.Object ackMessage,
                                                                    scala.PartialFunction<java.lang.Object,​CompletionStrategy> completionMatcher,
                                                                    scala.PartialFunction<java.lang.Object,​java.lang.Throwable> failureMatcher)
        Creates a Source that is materialized as an pekko.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 extracted Throwable will be used to fail the stream. In case the Actor is still draining its internal buffer (after having received a message matched by completionMatcher) before signaling completion and it receives a message matched by failureMatcher, 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 instead. Since Akka 2.6.0.
        Creates a Source that is materialized as an pekko.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 is pekko.stream.CompletionStrategy.immediately the completion will be signaled immediately, otherwise if the content is pekko.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 a pekko.actor.Status.Success) before signaling completion and it receives a pekko.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.

      • combineMat

        public static <T,​U,​M1,​M2,​M> Source<U,​M> combineMat​(Source<T,​M1> first,
                                                                                 Source<T,​M2> second,
                                                                                 scala.Function1<java.lang.Object,​Graph<UniformFanInShape<T,​U>,​NotUsed>> fanInStrategy,
                                                                                 scala.Function2<M1,​M2,​M> matF)
        Combines several sources with fan-in strategy like Merge or Concat into a single Source with a materialized value.

      • zipN

        public static <T> Source<scala.collection.immutable.Seq<T>,​NotUsed> zipN​(scala.collection.immutable.Seq<Source<T,​?>> sources)
        Combine the elements of multiple streams into a stream of sequences.

      • zipWithN

        public static <T,​O> Source<O,​NotUsed> zipWithN​(scala.Function1<scala.collection.immutable.Seq<T>,​O> zipper,
                                                           scala.collection.immutable.Seq<Source<T,​?>> sources)
        Combine the elements of multiple streams into a stream of sequences using a combiner function.

      • queue

        public static <T> Source<T,​BoundedSourceQueue<T>> queue​(int bufferSize)
        Creates a Source that is materialized as an pekko.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 the Source is cancelled by the downstream. An element that was reported to be enqueued is not guaranteed to be processed by the rest of the stream. If the queue is failed by calling BoundedQueueSource.fail or the downstream cancels the stream, elements in the buffer are discarded.

        Acknowledgement of pushed elements is immediate. pekko.stream.BoundedSourceQueue.offer returns pekko.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 cancelled QueueOfferResult.Dropped element was dropped QueueOfferResult.QueueComplete the queue was completed with pekko.stream.BoundedSourceQueue.complete QueueOfferResult.Failure the queue was failed with pekko.stream.BoundedSourceQueue.fail or if the stream failed

        Parameters:
        bufferSize - size of the buffer in number of elements

      • create

        public static <T> Source<T,​NotUsed> create​(int bufferSize,
                                                 scala.Function1<BoundedSourceQueue<T>,​scala.runtime.BoxedUnit> producer)
        Creates a Source that will immediately execute the provided function producer with a BoundedSourceQueue when materialized. This allows defining element production logic at Source creation time.

        The function producer can push elements to the stream using the provided queue. The queue behaves the same as in Source.queue:
        - Elements are emitted when there is downstream demand, buffered otherwise
        - Elements are dropped if the buffer is full
        - Buffered elements are discarded if downstream terminates
        You should never block the producer thread, as it will block the stream from processing elements. If the function producer throws an exception, the queue will be failed and the exception will be propagated to the stream.

        Example usage: 

        
 Source.create[Int](10) { queue =>
   // This code is executed when the source is materialized
   queue.offer(1)
   queue.offer(2)
   queue.offer(3)
   queue.complete()
 }

        Parameters:
        bufferSize - the size of the buffer (number of elements)
        producer - function that receives the queue and defines how to produce data
        Returns:
        a Source that emits elements pushed to the queue
        Since:
        1.2.0

      • queue

        public static <T> Source<T,​SourceQueueWithComplete<T>> queue​(int bufferSize,
                                                                   OverflowStrategy overflowStrategy)
        Creates a Source that is materialized as an pekko.stream.scaladsl.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.scaladsl.SourceQueueWithComplete.offer returns Future[QueueOfferResult] which completes with QueueOfferResult.Enqueued if element was added to buffer or sent downstream. It completes with QueueOfferResult.Dropped if element was dropped. Can also complete with QueueOfferResult.Failure - when stream failed or QueueOfferResult.QueueClosed when downstream is completed.

        The strategy pekko.stream.OverflowStrategy.backpressure will not complete last offer():Future call when buffer is full.

        Instead of using the strategy pekko.stream.OverflowStrategy.dropNew it's recommended to use Source.queue(bufferSize) instead which returns a QueueOfferResult synchronously.

        You can watch accessibility of stream with pekko.stream.scaladsl.SourceQueueWithComplete.watchCompletion. It returns future that completes with success when the operator is completed or fails when the 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 count
        overflowStrategy - 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 a Source that is materialized as an pekko.stream.scaladsl.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.scaladsl.SourceQueueWithComplete.offer returns Future[QueueOfferResult] which completes with QueueOfferResult.Enqueued if element was added to buffer or sent downstream. It completes with QueueOfferResult.Dropped if element was dropped. Can also complete with QueueOfferResult.Failure - when stream failed or QueueOfferResult.QueueClosed when downstream is completed.

        The strategy pekko.stream.OverflowStrategy.backpressure will not complete maxConcurrentOffers number of offer():Future call when buffer is full.

        Instead of using the strategy pekko.stream.OverflowStrategy.dropNew it's recommended to use Source.queue(bufferSize) instead which returns a QueueOfferResult synchronously.

        You can watch accessibility of stream with pekko.stream.scaladsl.SourceQueueWithComplete.watchCompletion. It returns future that completes with success when the operator is completed or fails when the 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 count
        overflowStrategy - Strategy that is used when incoming elements cannot fit inside the buffer
        maxConcurrentOffers - maximum number of pending offers when buffer is full, should be greater than 0, not applicable when OverflowStrategy.dropNew is used

      • unfoldResource

        public static <T,​S> Source<T,​NotUsed> unfoldResource​(scala.Function0<S> create,
                                                                 scala.Function1<S,​scala.Option<T>> read,
                                                                 scala.Function1<S,​scala.runtime.BoxedUnit> close)
        Start a new Source 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 => Option(reader.readLine()),
   reader => reader.close())

        You can use the supervision strategy to handle exceptions for read function. All exceptions thrown by create or close will fail the stream.

        Restart supervision strategy will close and create blocking IO again. Default strategy is Stop which means that stream will be terminated on error in read 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 using ActorAttributes.

        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 when read returns None.
        close - - function that closes resource

      • unfoldResourceAsync

        public static <T,​S> Source<T,​NotUsed> unfoldResourceAsync​(scala.Function0<scala.concurrent.Future<S>> create,
                                                                      scala.Function1<S,​scala.concurrent.Future<scala.Option<T>>> read,
                                                                      scala.Function1<S,​scala.concurrent.Future<Done>> close)
        Start a new Source from some resource which can be opened, read and closed. It's similar to unfoldResource but takes functions that return Futures instead of plain values.

        You can use the supervision strategy to handle exceptions for read function or failures of produced Futures. All exceptions thrown by create or close as well as fails of returned futures will fail the stream.

        Restart supervision strategy will close and create resource. Default strategy is Stop which means that stream will be terminated on error in read 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 using ActorAttributes.

        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 when Future from read function returns None.
        close - - function that closes resource

      • mergePrioritizedN

        public static <T> Source<T,​NotUsed> mergePrioritizedN​(scala.collection.immutable.Seq<scala.Tuple2<Source<T,​?>,​java.lang.Object>> sourcesAndPriorities,
                                                            boolean eagerComplete)
        Merge multiple Sources. 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

      • 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 interface Graph<Out,​Mat>

      • 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.
        Specified by:
        shape in interface Graph<Out,​Mat>

      • toString

        public java.lang.String toString()
        Overrides:
        toString in class java.lang.Object

      • viaMat

        public <T,​Mat2,​Mat3> Source<T,​Mat3> viaMat​(Graph<FlowShape<Out,​T>,​Mat2> flow,
                                                             scala.Function2<Mat,​Mat2,​Mat3> combine)
        Description copied from interface: FlowOpsMat
        Transform this Flow by appending the given processing steps. 
        
     +---------------------------------+
     | Resulting Flow[In, T, M2]       |
     |                                 |
     |  +------+            +------+   |
     |  |      |            |      |   |
 In ~~> | this |  ~~Out~~>  | flow |  ~~> T
     |  |   Mat|            |     M|   |
     |  +------+            +------+   |
     +---------------------------------+
        The 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 and Keep.right combiners where appropriate instead of manually writing functions that pass through one of the values.

        Specified by:
        viaMat in interface FlowOpsMat<Out,​Mat>

      • to

        public <Mat2> RunnableGraph<Mat> to​(Graph<SinkShape<Out>,​Mat2> sink)
        Connect this pekko.stream.scaladsl.Source to a pekko.stream.scaladsl.Sink, concatenating the processing steps of both.
        Specified by:
        to in interface FlowOps<Out,​Mat>

      • toMat

        public <Mat2,​Mat3> RunnableGraph<Mat3> toMat​(Graph<SinkShape<Out>,​Mat2> sink,
                                                   scala.Function2<Mat,​Mat2,​Mat3> combine)
        Connect this pekko.stream.scaladsl.Source to a pekko.stream.scaladsl.Sink, concatenating the processing steps of both.
        Specified by:
        toMat in interface FlowOpsMat<Out,​Mat>

      • mapMaterializedValue

        public <Mat2> Source<Out,​Mat2> mapMaterializedValue​(scala.Function1<Mat,​Mat2> f)
        Transform only the materialized value of this Source, leaving all other properties as they were.
        Specified by:
        mapMaterializedValue in interface FlowOpsMat<Out,​Mat>

      • preMaterialize

        public scala.Tuple2<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.

      • materializeIntoSource

        public <Mat2> Source<Mat2,​scala.concurrent.Future<NotUsed>> materializeIntoSource​(Graph<SinkShape<Out>,​scala.concurrent.Future<Mat2>> sink)
        Materializes this Source using the Sink, immediately returning the values via the provided Sink as a new Source.

        Parameters:
        sink - A sink which needs to materialize into a Future, typically one that collects values such as Sink.head or Sink.seq
        Returns:
        A new Source that contains the results of the provided Source's elements run with the Sink
        Since:
        1.2.0

      • run

        public scala.concurrent.Future<Done> run​(Materializer materializer)
        Connect this Source to the Sink.ignore and run it. Elements from the stream will be consumed and discarded.

        Note that the ActorSystem can be used as the implicit materializer parameter to use the pekko.stream.SystemMaterializer for running the stream.

      • runWith

        public <Mat2> Mat2 runWith​(Graph<SinkShape<Out>,​Mat2> sink,
                           Materializer materializer)
        Connect this Source to a Sink and run it. The returned value is the materialized value of the Sink, e.g. the Publisher of a pekko.stream.scaladsl.Sink#publisher.

        Note that the ActorSystem can be used as the implicit materializer parameter to use the pekko.stream.SystemMaterializer for running the stream.

      • runFold

        public <U> scala.concurrent.Future<U> runFold​(U zero,
                                              scala.Function2<U,​Out,​U> f,
                                              Materializer materializer)
        Shortcut for running this Source with a fold function. The given function is invoked for every received element, giving it its previous output (or the given zero value) and the element as input. The returned Future will be completed with value of the final function evaluation when the input stream ends, or completed with Failure if there is a failure signaled in the stream.

        Note that the ActorSystem can be used as the implicit materializer parameter to use the pekko.stream.SystemMaterializer for running the stream.

      • runFoldAsync

        public <U> scala.concurrent.Future<U> runFoldAsync​(U zero,
                                                   scala.Function2<U,​Out,​scala.concurrent.Future<U>> f,
                                                   Materializer materializer)
        Shortcut for running this Source with a foldAsync function. The given function is invoked for every received element, giving it its previous output (or the given zero value) and the element as input. The returned Future will be completed with value of the final function evaluation when the input stream ends, or completed with Failure if there is a failure signaled in the stream.

        Note that the ActorSystem can be used as the implicit materializer parameter to use the pekko.stream.SystemMaterializer for running the stream.

      • runReduce

        public <U> scala.concurrent.Future<U> runReduce​(scala.Function2<U,​U,​U> f,
                                                Materializer materializer)
        Shortcut for running this Source with a reduce function. The given function is invoked for every received element, giving it its previous output (from the second element) and the element as input. The returned Future will be completed with value of the final function evaluation when the input stream ends, or completed with Failure if there is a failure 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 ActorSystem can be used as the implicit materializer parameter to use the pekko.stream.SystemMaterializer for running the stream.

      • runForeach

        public scala.concurrent.Future<Done> runForeach​(scala.Function1<Out,​scala.runtime.BoxedUnit> f,
                                                Materializer materializer)
        Shortcut for running this Source with a foreach procedure. The given procedure is invoked for each received element. The returned Future will be completed with Success when reaching the normal end of the stream, or completed with Failure if there is a failure signaled in the stream.

        Note that the ActorSystem can be used as the implicit materializer parameter to use the pekko.stream.SystemMaterializer for running the stream.

      • withAttributes

        public Source<Out,​Mat> withAttributes​(Attributes attr)
        Replace the attributes of this Source 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 interface FlowOps<Out,​Mat>
        Specified by:
        withAttributes in interface Graph<Out,​Mat>

      • addAttributes

        public Source<Out,​Mat> addAttributes​(Attributes attr)
        Add the given attributes to this Source. If the specific attribute was already on this source it will replace the previous value. If this Source is a composite of multiple graphs, the added attributes will be on the composite and therefore less specific than attributes set directly on the individual graphs of the composite.
        Specified by:
        addAttributes in interface FlowOps<Out,​Mat>
        Specified by:
        addAttributes in interface Graph<Out,​Mat>

      • async

        public Source<Out,​Mat> async​(java.lang.String dispatcher)
        Put an asynchronous boundary around this Graph

        Specified by:
        async in interface Graph<Out,​Mat>
        Parameters:
        dispatcher - Run the graph on this dispatcher

      • async

        public Source<Out,​Mat> async​(java.lang.String dispatcher,
                                   int inputBufferSize)
        Put an asynchronous boundary around this Graph

        Specified by:
        async in interface Graph<Out,​Mat>
        Parameters:
        dispatcher - Run the graph on this dispatcher
        inputBufferSize - Set the input buffer to this size for the graph

      • asJava

        public Source<Out,​Mat> asJava()
        Converts this Scala DSL element to it's Java DSL counterpart.

      • combine

        public <T,​U> Source<U,​NotUsed> combine​(Source<T,​?> first,
                                                   Source<T,​?> second,
                                                   scala.collection.immutable.Seq<Source<T,​?>> rest,
                                                   scala.Function1<java.lang.Object,​Graph<UniformFanInShape<T,​U>,​NotUsed>> strategy)
        Deprecated.
        Use `Source.combine` on companion object instead. Since Akka 2.5.5.
        Combines several sources with fan-in strategy like Merge or Concat and returns Source.

      • asSourceWithContext

        public <Ctx> SourceWithContext<Out,​Ctx,​Mat> asSourceWithContext​(scala.Function1<Out,​Ctx> f)
        Transform this source whose element is e into a source producing tuple (e, f(e))