Packages

trait ConsumerF[F[_], E, A] extends Serializable

A simple interface that models the consumer side of a producer-consumer communication channel.

Currently exposed by ConcurrentChannel.consume.

F

is effect type used for processing tasks asynchronously

E

is the type for the completion event

A

is the type for the stream of events being consumed

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

Abstract Value Members

  1. abstract def pull: F[Either[E, A]]

    Pulls one message from the communication channel, when it becomes available.

    Pulls one message from the communication channel, when it becomes available.

    Example:

    import cats.implicits._
import cats.effect.Async

def sum[F[_]](channel: ConsumerF[F, Int, Int], acc: Long = 0)
  (implicit F: Async[F]): F[Long] = {

  channel.pull.flatMap {
    case Left(e) => F.pure(acc + e)
    case Right(i) => sum(channel, acc + i)
  }
}
    returns

    either Left(e), if the channel was closed with a final e completion event, or Right(a), representing a message that was pulled from the channel

  2. abstract def pullMany(minLength: Int, maxLength: Int): F[Either[E, Seq[A]]]

    Pulls a whole batch of messages from the channel, at least one, the returned sequence being no larger than the specified maxLength.

    Pulls a whole batch of messages from the channel, at least one, the returned sequence being no larger than the specified maxLength.

    import cats.implicits._
import cats.effect.Async

def sum[F[_]](channel: ConsumerF[F, Int, Int], acc: Long = 0)
  (implicit F: Async[F]): F[Long] = {

  channel.pullMany(1, 16).flatMap {
    case Left(e) => F.pure(acc + e)
    case Right(seq) => sum(channel, acc + seq.sum)
  }
}
    minLength

    is the minimum size of the returned sequence; for as long as the channel isn't halted, the returned task will back-pressure until the required number of events have been collected

    maxLength

    is the maximum size of the returned sequence; for fairness purposes (e.g. multiple workers consuming from the same ConsumerF), a smaller value is recommended, or otherwise Int.MaxValue can be used

    returns

    either Left(e), if the channel was closed with a final e completion event, or Right(seq), representing a non-empty sequence of messages pulled from the channel, but that is no larger than maxLength