Packages

object Consumer extends Serializable

The companion object of Consumer, defines consumer builders.

Source
Consumer.scala
Linear Supertypes
Serializable, AnyRef, Any
Ordering
  1. Alphabetic
  2. By Inheritance
Inherited
  1. Consumer
  2. Serializable
  3. AnyRef
  4. Any
  1. Hide All
  2. Show All
Visibility
  1. Public
  2. Protected

Type Members

  1. trait Sync[-In, +R] extends Consumer[In, R]

    Defines a synchronous Consumer that builds synchronous subscribers.

Value Members

  1. def cancel[A]: Sync[A, Unit]

    A consumer that immediately cancels its upstream after subscription.

  2. implicit def catsContravariant[C]: Contravariant[[α$0$]Consumer[α$0$, C]]

    cats.Contravariant instance for Consumer.

  3. implicit val catsProfunctor: Profunctor[Consumer]

    cats.arrow.Profunctor instance for Consumer.

  4. def complete[A]: Sync[A, Unit]

    A simple consumer that consumes all elements of the stream and then signals its completion.

  5. def create[In, Out](f: (Scheduler, Cancelable, Callback[Throwable, Out]) => Observer[In]): Consumer[In, Out]

    Creates a Consumer out of the given function.

    Creates a Consumer out of the given function.

    The function returns an Observer and takes as input:

    • a Scheduler for any asynchronous execution needs the returned observer might have
    • a Cancelable that can be used for concurrently canceling the stream (in addition to being able to return Stop from onNext)
    • a Callback that must be called to signal the final result, after the observer finished processing the stream, or an error if the processing finished in error
    f

    is the input function with an injected Scheduler, Cancelable, Callback and that returns an Observer

  6. def firstNotification[A]: Sync[A, Notification[A]]

    A consumer that will produce a Notification of the first value received (onNext, onComplete or onError), after which the streaming gets cancelled.

    A consumer that will produce a Notification of the first value received (onNext, onComplete or onError), after which the streaming gets cancelled.

    • OnNext will be signaled on the first onNext event if it happens and the streaming will be stopped by Stop.
    • OnComplete will be signaled if the stream was empty and thus completed without any onNext.
    • OnError will be signaled if the stream was completed in error before the first onNext happened.
  7. def foldLeft[S, A](initial: => S)(f: (S, A) => S): Sync[A, S]

    Given a fold function and an initial state value, applies the fold function to every element of the stream and finally signaling the accumulated value.

    Given a fold function and an initial state value, applies the fold function to every element of the stream and finally signaling the accumulated value.

    initial

    is a lazy value that will be fed at first in the fold function as the initial state.

    f

    is the function that calculates a new state on each emitted value by the stream, for accumulating state

  8. def foldLeftEval[F[_], S, A](initial: => S)(f: (S, A) => F[S])(implicit F: TaskLike[F]): Consumer[A, S]

    Given a fold function and an initial state value, applies the fold function to every element of the stream and finally signaling the accumulated value.

    Given a fold function and an initial state value, applies the fold function to every element of the stream and finally signaling the accumulated value.

    The given fold function returns an F[A] value, where F is any data type that implements cats.effect.Effect (e.g. Task, Coeval), thus able to do asynchronous processing, with ordering of calls being guaranteed.

    initial

    is a lazy value that will be fed at first in the fold function as the initial state.

    f

    is the function that calculates a new state on each emitted value by the stream, for accumulating state, returning a F[A] capable of lazy or asynchronous execution.

  9. def foldLeftTask[S, A](initial: => S)(f: (S, A) => Task[S]): Consumer[A, S]

    Given a fold function and an initial state value, applies the fold function to every element of the stream and finally signaling the accumulated value.

    Given a fold function and an initial state value, applies the fold function to every element of the stream and finally signaling the accumulated value.

    The given fold function returns a Task that can execute an asynchronous operation, with ordering of calls being guaranteed.

    initial

    is a lazy value that will be fed at first in the fold function as the initial state.

    f

    is the function that calculates a new state on each emitted value by the stream, for accumulating state, returning a Task capable of asynchronous execution.

  10. def foreach[A](cb: (A) => Unit): Sync[A, Unit]

    Builds a consumer that will consume the stream, applying the given function to each element and then finally signaling its completion.

    Builds a consumer that will consume the stream, applying the given function to each element and then finally signaling its completion.

    cb

    is the function that will be called for each element

  11. def foreachEval[F[_], A](cb: (A) => F[Unit])(implicit F: TaskLike[F]): Consumer[A, Unit]

    Builds a consumer that will consume the stream, applying the given function to each element and then finally signaling its completion.

    Builds a consumer that will consume the stream, applying the given function to each element and then finally signaling its completion.

    The given callback function returns a F[A] value that can execute an asynchronous operation, with ordering of calls being guaranteed, given that the F[_] data type is any type that implements cats.effect.Effect (e.g. Task, IO).

    cb

    is the function that will be called for each element

  12. def foreachParallel[A](parallelism: Int)(cb: (A) => Unit): Consumer[A, Unit]

    Builds a consumer that will consume the stream, applying the given function to each element, in parallel, then finally signaling its completion.

    Builds a consumer that will consume the stream, applying the given function to each element, in parallel, then finally signaling its completion.

    parallelism

    is the maximum number of (logical) threads to use

    cb

    is the function that will be called for each element

  13. def foreachParallelTask[A](parallelism: Int)(cb: (A) => Task[Unit]): Consumer[A, Unit]

    Builds a consumer that will consume the stream, applying the given function to each element, in parallel, then finally signaling its completion.

    Builds a consumer that will consume the stream, applying the given function to each element, in parallel, then finally signaling its completion.

    The given callback function returns a Task that can execute an asynchronous operation, with ordering of calls being guaranteed per subscriber.

    parallelism

    is the maximum number of (logical) threads to use

    cb

    is the function that will be called for each element

  14. def foreachTask[A](cb: (A) => Task[Unit]): Consumer[A, Unit]

    Builds a consumer that will consume the stream, applying the given function to each element and then finally signaling its completion.

    Builds a consumer that will consume the stream, applying the given function to each element and then finally signaling its completion.

    The given callback function returns a Task that can execute an asynchronous operation, with ordering of calls being guaranteed.

    cb

    is the function that will be called for each element

  15. def fromObserver[In](f: (Scheduler) => Observer[In]): Consumer[In, Unit]

    Given a function taking a Scheduler and returning an Observer, builds a consumer from it.

    Given a function taking a Scheduler and returning an Observer, builds a consumer from it.

    You can use the Scheduler as the execution context, for working with Future, for forcing asynchronous boundaries or for executing tasks with a delay.

  16. def head[A]: Sync[A, A]

    A consumer that will produce the first streamed value on onNext after which the streaming gets cancelled.

    A consumer that will produce the first streamed value on onNext after which the streaming gets cancelled.

    In case the stream is empty and so no onNext happen before onComplete, then the a NoSuchElementException will get triggered.

  17. def headOption[A]: Sync[A, Option[A]]

    A consumer that will produce the first streamed value on onNext after which the streaming gets cancelled.

    A consumer that will produce the first streamed value on onNext after which the streaming gets cancelled.

    In case the stream is empty and so no onNext happen before onComplete, then the a NoSuchElementException will get triggered.

  18. def loadBalance[A, R](consumers: Consumer[A, R]*): Consumer[A, List[R]]

    Creates a consumer that, when consuming the stream, will start multiple subscribers corresponding and distribute the load between them.

    Creates a consumer that, when consuming the stream, will start multiple subscribers corresponding and distribute the load between them.

    Once each subscriber emits a final result, this consumer will return a list of aggregated results.

    Has the following rules:

    • items are pushed on free subscribers, respecting their contract, each item being pushed to the first available subscriber in the queue
    • in case no free subscribers are available, then the source gets back-pressured until free subscribers are available
    • in case of onComplete or onError, all subscribers that are still active will receive the event
    • the onSuccess callback of individual subscribers is aggregated in a list buffer and once the aggregate contains results from all subscribers, the load-balancing consumer will emit the aggregate
    • the onError callback triggered by individual subscribers will signal that error upstream and cancel the streaming for every other subscriber
    • in case any of the subscribers cancels its subscription (either returning Stop in onNext or canceling its assigned cancelable), it gets excluded from the pool of active subscribers, but the other active subscribers will still receive notifications
    • if all subscribers canceled (either by returning Stop or by canceling their assignable cancelable reference), then streaming stops as well

    In other words the Task, created by applying this consumer to an observable, will complete once all the subscribers emit a result or as soon as an error happens.

    consumers

    is a list of consumers that will initialize the subscribers that will process events in parallel, with the parallelism factor being equal to the number of consumers specified in this list.

    returns

    a list of aggregated results that were computed by all of the subscribers as their result

  19. def loadBalance[A, R](parallelism: Int, consumer: Consumer[A, R]): Consumer[A, List[R]]

    Creates a consumer that, when consuming the stream, will start multiple subscribers corresponding and distribute the load between them.

    Creates a consumer that, when consuming the stream, will start multiple subscribers corresponding and distribute the load between them.

    Once each subscriber emits a final result, this consumer will return a list of aggregated results.

    Has the following rules:

    • items are pushed on free subscribers, respecting their contract, each item being pushed to the first available subscriber in the queue
    • in case no free subscribers are available, then the source gets back-pressured until free subscribers are available
    • in case of onComplete or onError, all subscribers that are still active will receive the event
    • the onSuccess callback of individual subscribers is aggregated in a list buffer and once the aggregate contains results from all subscribers, the load-balancing consumer will emit the aggregate
    • the onError callback triggered by individual subscribers will signal that error upstream and cancel the streaming for every other subscriber
    • in case any of the subscribers cancels its subscription (either returning Stop in onNext or canceling its assigned cancelable), it gets excluded from the pool of active subscribers, but the other active subscribers will still receive notifications
    • if all subscribers canceled (either by returning Stop or by canceling their assignable cancelable reference), then streaming stops as well

    In other words the Task, created by applying this consumer to an observable, will complete once all the subscribers emit a result or as soon as an error happens.

    parallelism

    is the number of subscribers that will get initialized to process incoming events in parallel.

    consumer

    is the subscriber factory that will initialize all needed subscribers, in number equal to the specified parallelism and thus that will be fed in parallel

    returns

    a list of aggregated results that were computed by all of the subscribers as their result

  20. def raiseError[In, R](ex: Throwable): Sync[In, R]

    A consumer that triggers an error and immediately cancels its upstream after subscription.

  21. def toList[A]: Consumer[A, List[A]]

    A consumer that will return a collection of all the streamed values as a Scala List.

    A consumer that will return a collection of all the streamed values as a Scala List.

    Note that this method is functionally the same as the Observable.toListL

    WARNING: For infinite streams the process will eventually blow up with an out of memory error.