Packages

sealed abstract class Iterant[F[_], A] extends Product with Serializable

The Iterant is a type that describes lazy, possibly asynchronous streaming of elements using a pull-based protocol.

It is similar somewhat in spirit to Scala's own collection.immutable.Stream and with Java's Iterable, except that it is more composable and more flexible due to evaluation being controlled by an F[_] monadic type that you have to supply (like monix.eval.Task, monix.eval.Coeval or cats.effect.IO) which will control the evaluation. In other words, this Iterant type is capable of strict or lazy, synchronous or asynchronous evaluation.

Consumption of an Iterant happens typically in a loop where the current step represents either a signal that the stream is over, or a (head, rest) pair, very similar in spirit to Scala's standard List or Iterable.

The type is an ADT, meaning a composite of the following types:

  • Next which signals a single strict element, the head and a rest representing the rest of the stream
  • NextBatch is a variation on Next for signaling a whole batch of elements by means of a Batch, a type that's similar with Scala's Iterable, along with the rest of the stream.
  • NextCursor is a variation on Next for signaling a whole strict batch of elements as a traversable BatchCursor, a type that's similar with Scala's Iterator, along with the rest of the stream.
  • Suspend is for suspending the evaluation of a stream.
  • Concat represents the concatenation of two streams.
  • Scope is to specify the acquisition and release of resources. It is effectively the encoding of Bracket.
  • Halt represents an empty stream, signaling the end, either in success or in error.
  • Last represents a one-element stream, where Last(item) as an optimisation on Next(item, F.pure(Halt(None)), F.unit).

Parametric Polymorphism

The Iterant type accepts as type parameter an F monadic type that is used to control how evaluation happens. For example you can use monix.eval.Task, in which case the streaming can have asynchronous behavior, or you can use monix.eval.Coeval in which case it can behave like a normal, synchronous Iterable.

As restriction, this F[_] type used should be stack safe in map and flatMap, otherwise you might get stack-overflow exceptions. This is why in general the type class required for F is cats.effect.Sync.

When building instances, type F[_] which handles the evaluation needs to be specified upfront. Example:

import cats.effect.IO
import monix.eval.{Task, Coeval}

// Builds an Iterant powered by Monix's Task
Iterant[Task].of(1, 2, 3)

// Builds an Iterant powered by Monix's Coeval
Iterant[Coeval].of(1, 2, 3)

// Builds an Iterant powered by Cats's IO
Iterant[IO].of(1, 2, 3)

You'll usually pick between Task, Coeval or IO for your needs.

Attribution

This type was inspired by the Streaming type in the Typelevel Cats library (later moved to Dogs), originally committed in Cats by Erik Osheim. It was also inspired by other push-based streaming abstractions, like the Iteratee or IAsyncEnumerable.

F

is the data type that controls evaluation; note that it must be stack-safe in its map and flatMap operations

A

is the type of the elements produced by this Iterant

Self Type
Iterant[F, A]
Source
Iterant.scala
Linear Supertypes
Ordering
  1. Alphabetic
  2. By Inheritance
Inherited
  1. Iterant
  2. Serializable
  3. Product
  4. Equals
  5. AnyRef
  6. Any
  1. Hide All
  2. Show All
Visibility
  1. Public
  2. Protected

Abstract Value Members

  1. abstract def canEqual(that: Any): Boolean
    Definition Classes
    Equals
  2. abstract def productArity: Int
    Definition Classes
    Product
  3. abstract def productElement(n: Int): Any
    Definition Classes
    Product

Concrete Value Members

  1. final def !=(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  2. final def ##: Int
    Definition Classes
    AnyRef → Any
  3. final def ++[B >: A](rhs: => Iterant[F, B])(implicit F: Sync[F]): Iterant[F, B]

    Appends the given stream to the end of the source, effectively concatenating them.

    Appends the given stream to the end of the source, effectively concatenating them.

    Example:

    import monix.eval.Task
    
    // Yields 1, 2, 3, 4
    Iterant[Task].of(1, 2) ++ Iterant[Task].of(3, 4)
    rhs

    is the (right hand side) lazily evaluated iterant to concatenate at the end of this iterant.

  4. final def ++[B >: A](rhs: F[Iterant[F, B]])(implicit F: Sync[F]): Iterant[F, B]

    Appends a stream to the end of the source, effectively concatenating them.

    Appends a stream to the end of the source, effectively concatenating them.

    The right hand side is suspended in the F[_] data type, thus allowing for laziness.

    Example:

    import monix.eval.Task
    
    // Yields 1, 2, 3, 4
    Iterant[Task].of(1, 2) ++ Task.eval {
      Iterant[Task].of(3, 4)
    }
    rhs

    is the iterant to append at the end of our source.

  5. final def +:[B >: A](head: B)(implicit F: Applicative[F]): Iterant[F, B]

    Prepends an element to the iterant, returning a new iterant that will start with the given head and then continue with the source.

    Prepends an element to the iterant, returning a new iterant that will start with the given head and then continue with the source.

    Example:

    import monix.eval.Task
    
    // Yields 1, 2, 3, 4
    1 +: Iterant[Task].of(2, 3, 4)
    head

    is the element to prepend at the start of this iterant

  6. final def :+[B >: A](elem: B)(implicit F: Sync[F]): Iterant[F, B]

    Appends the right hand side element to the end of this iterant.

    Appends the right hand side element to the end of this iterant.

    Example:

    import monix.eval.Task
    
    // Yields 1, 2, 3, 4
    Iterant[Task].of(1, 2, 3) :+ 4
    elem

    is the element to append at the end

  7. final def ==(arg0: Any): Boolean
    Definition Classes
    AnyRef → Any
  8. final def asInstanceOf[T0]: T0
    Definition Classes
    Any
  9. final def attempt(implicit F: Sync[F]): Iterant[F, Either[Throwable, A]]

    Converts the source Iterant that emits A elements into an iterant that emits Either[Throwable, A], thus materializing whatever error that might interrupt the stream.

    Converts the source Iterant that emits A elements into an iterant that emits Either[Throwable, A], thus materializing whatever error that might interrupt the stream.

    Example:

    import monix.eval.Task
    import monix.execution.exceptions.DummyException
    
    // Yields Right(1), Right(2), Right(3)
    Iterant[Task].of(1, 2, 3).attempt
    
    // Yields Right(1), Right(2), Left(DummyException())
    (Iterant[Task].of(1, 2) ++
      Iterant[Task].raiseError[Int](DummyException("dummy"))).attempt
  10. def batched(count: Int)(implicit F: Sync[F]): Iterant[F, A]

    Optimizes the access to the source by periodically gathering items emitted into batches of the specified size and emitting NextBatch nodes.

    Optimizes the access to the source by periodically gathering items emitted into batches of the specified size and emitting NextBatch nodes.

    For this operation we have this law:

    source.batched(16) <-> source

    This means that the result will emit exactly what the source emits, however the underlying representation will be different, the emitted notes being of type NextBatch, wrapping arrays with the length equal to the given count.

    Very similar in behavior with bufferTumbling, however the batches are implicit, not explicit. Useful for optimization.

  11. final def bufferSliding(count: Int, skip: Int)(implicit F: Sync[F]): Iterant[F, Seq[A]]

    Returns an iterant that emits buffers of items it collects from the source iterant.

    Returns an iterant that emits buffers of items it collects from the source iterant. The resulting iterant emits buffers every skip items, each containing count items.

    If the source iterant completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

    For count and skip there are 3 possibilities:

    1. in case skip == count, then there are no items dropped and no overlap, the call being equivalent to buffer(count)
    2. in case skip < count, then overlap between buffers happens, with the number of elements being repeated being count - skip
    3. in case skip > count, then skip - count elements start getting dropped between windows

    Example:

    import monix.eval.Coeval
    
    val source = Iterant[Coeval].of(1, 2, 3, 4, 5, 6, 7)
    
    // Yields Seq(1, 2, 3), Seq(4, 5, 6), Seq(7)
    source.bufferSliding(3, 3)
    
    // Yields Seq(1, 2, 3), Seq(5, 6, 7)
    source.bufferSliding(3, 4)
    
    // Yields Seq(1, 2, 3), Seq(3, 4, 5), Seq(5, 6, 7)
    source.bufferSliding(3, 2)
    count

    the maximum size of each buffer before it should be emitted

    skip

    how many items emitted by the source iterant should be skipped before starting a new buffer. Note that when skip and count are equal, this is the same operation as bufferTumbling(count)

  12. def bufferTumbling(count: Int)(implicit F: Sync[F]): Iterant[F, Seq[A]]

    Periodically gather items emitted by an iterant into bundles and emit these bundles rather than emitting the items one at a time.

    Periodically gather items emitted by an iterant into bundles and emit these bundles rather than emitting the items one at a time. This version of buffer is emitting items once the internal buffer has reached the given count.

    If the source iterant completes, then the current buffer gets signaled downstream. If the source triggers an error then the current buffer is being dropped and the error gets propagated immediately.

    import monix.eval.Coeval
    
    // Yields Seq(1, 2, 3), Seq(4, 5, 6), Seq(7)
    Iterant[Coeval].of(1, 2, 3, 4, 5, 6, 7).bufferTumbling(3)
    count

    the maximum size of each buffer before it should be emitted

    See also

    bufferSliding for the more flexible version that allows to specify a skip argument.

  13. def clone(): AnyRef
    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.CloneNotSupportedException]) @native() @HotSpotIntrinsicCandidate()
  14. final def collect[B](pf: PartialFunction[A, B])(implicit F: Sync[F]): Iterant[F, B]

    Builds a new iterant by applying a partial function to all elements of the source on which the function is defined.

    Builds a new iterant by applying a partial function to all elements of the source on which the function is defined.

    Example:

    import monix.eval.Task
    
    // Yields 2, 4, 6
    Iterant[Task].of(1, 2, 3, 4, 5, 6)
      .map { x => Option(x).filter(_ % 2 == 0) }
      .collect { case Some(x) => x }
    B

    the element type of the returned iterant.

    pf

    the partial function that filters and maps the iterant

    returns

    a new iterant resulting from applying the partial function pf to each element on which it is defined and collecting the results. The order of the elements is preserved.

  15. final def completedL(implicit F: Sync[F]): F[Unit]

    Upon evaluation of the result, consumes this iterant to completion.

    Upon evaluation of the result, consumes this iterant to completion.

    Example:

    import cats.implicits._
    import monix.eval.Task
    
    // Whatever...
    val iterant = Iterant[Task].range(0, 10000)
    
    val onFinish: Task[Unit] =
      iterant.completedL >> Task.eval(println("Done!"))
  16. final def concat[B](implicit ev: <:<[A, Iterant[F, B]], F: Sync[F]): Iterant[F, B]

    Alias for concat.

  17. final def concatMap[B](f: (A) => Iterant[F, B])(implicit F: Sync[F]): Iterant[F, B]

    Alias for flatMap.

  18. final def consume(implicit F: Concurrent[F], cs: ContextShift[F]): Resource[F, Consumer[F, A]]

    Create a ConsumerF value that can be used to consume events from the channel.

    Create a ConsumerF value that can be used to consume events from the channel.

    The returned value is a Resource, because a consumer can be unsubscribed from the channel early, with its internal buffer being garbage collected and the finalizers of the source being triggered.

    import monix.eval.Task
    import monix.tail.Iterant.Consumer
    
    def sum(channel: Consumer[Task, Int], acc: Long = 0): Task[Long] =
      channel.pull.flatMap {
        case Right(a) =>
          sum(channel, acc + a)
        case Left(None) =>
          Task.pure(acc)
        case Left(Some(e)) =>
          Task.raiseError(e)
      }
    
    Iterant[Task].range(0, 10000).consume.use { consumer =>
      sum(consumer)
    }
    See also

    consumeWithConfig for fine tuning the internal buffer of the created consumer

  19. final def consumeWithConfig(config: Config)(implicit F: Concurrent[F], cs: ContextShift[F]): Resource[F, Consumer[F, A]]

    Version of consume that allows for fine tuning the underlying buffer used.

    Version of consume that allows for fine tuning the underlying buffer used.

    There are two parameters that can be configured:

    • the BufferCapacity, which can be Unbounded, for an unlimited internal buffer in case the consumer is definitely faster than the producer, or Bounded in case back-pressuring a slow consumer is desirable
    • the ChannelType.ConsumerSide, which specifies if this consumer will use multiple workers in parallel or not; this is an optimization, with the safe choice being MultiConsumer, which specifies that multiple workers can use the created consumer in parallel, pulling data from multiple threads at the same time; whereas SingleConsumer specifies that the data will be read sequentially by a single worker, not in parallel; this being a risky optimization
    config

    is the configuration object for fine tuning the behavior of the created consumer, see ConsumerF.Config1

    Annotations
    @UnsafeProtocol()
  20. final def countL(implicit F: Sync[F]): F[Long]

    Counts the total number of elements emitted by the source.

    Counts the total number of elements emitted by the source.

    Example:

    import cats.effect.IO
    
    // Yields 100
    Iterant[IO].range(0, 100).countL
    
    // Yields 1
    Iterant[IO].pure(1).countL
    
    // Yields 0
    Iterant[IO].empty[Int].countL
  21. final def distinctUntilChanged(implicit F: Sync[F], A: Eq[A]): Iterant[F, A]

    Suppress duplicate consecutive items emitted by the source.

    Suppress duplicate consecutive items emitted by the source.

    Example:

    import cats.implicits._
    import monix.eval.Coeval
    
    // Yields 1, 2, 1, 3, 2, 4
    Iterant[Coeval].of(1, 1, 1, 2, 2, 1, 1, 3, 3, 3, 2, 2, 4, 4, 4)
      .distinctUntilChanged

    Duplication is detected by using the equality relationship provided by the cats.Eq type class. This allows one to override the equality operation being used (e.g. maybe the default .equals is badly defined, or maybe you want reference equality, so depending on use case).

    Cats Eq and Scala Interop

    Monix prefers to work with cats.Eq for assessing the equality of elements that have an ordering defined, instead of scala.math.Equiv.

    We do this because Scala's Equiv has a default instance defined that's based on universal equality and that's a big problem, because when using the Eq type class, it is universal equality that we want to avoid and there have been countless of bugs in the ecosystem related to both universal equality and Equiv. Thankfully people are working to fix it.

    We also do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

    Defining Eq instance is easy and we can use universal equality in our definitions as well:

    import cats.Eq
    
    case class Address(host: String, port: Int)
    
    implicit val eqForAddress: Eq[Address] =
      Eq.fromUniversalEquals
    A

    is the cats.Eq instance that defines equality for A

  22. final def distinctUntilChangedByKey[K](key: (A) => K)(implicit F: Sync[F], K: Eq[K]): Iterant[F, A]

    Given a function that returns a key for each element emitted by the source, suppress consecutive duplicate items.

    Given a function that returns a key for each element emitted by the source, suppress consecutive duplicate items.

    Example:

    import cats.implicits._
    import monix.eval.Coeval
    
    // Yields 1, 2, 3, 4
    Iterant[Coeval].of(1, 3, 2, 4, 2, 3, 5, 7, 4)
      .distinctUntilChangedByKey(_ % 2)

    Duplication is detected by using the equality relationship provided by the cats.Eq type class. This allows one to override the equality operation being used (e.g. maybe the default .equals is badly defined, or maybe you want reference equality, so depending on use case).

    Cats Eq and Scala Interop

    Monix prefers to work with cats.Eq for assessing the equality of elements that have an ordering defined, instead of scala.math.Equiv.

    We do this because Scala's Equiv has a default instance defined that's based on universal equality and that's a big problem, because when using the Eq type class, it is universal equality that we want to avoid and there have been countless of bugs in the ecosystem related to both universal equality and Equiv. Thankfully people are working to fix it.

    We also do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

    Defining Eq instance is easy and we can use universal equality in our definitions as well:

    import cats.Eq
    
    case class Address(host: String, port: Int)
    
    implicit val eqForAddress: Eq[Address] =
      Eq.fromUniversalEquals
    key

    is a function that returns a K key for each element, a value that's then used to do the deduplication

    K

    is the cats.Eq instance that defines equality for the key type K

  23. final def drop(n: Int)(implicit F: Sync[F]): Iterant[F, A]

    Drops the first n elements (from the start).

    Drops the first n elements (from the start).

    Example:

    import monix.eval.Task
    
    // Yields 4, 5
    Iterant[Task].of(1, 2, 3, 4, 5).drop(3)
    n

    the number of elements to drop

    returns

    a new iterant that drops the first n elements emitted by the source

  24. final def dropLast(n: Int)(implicit F: Sync[F]): Iterant[F, A]

    Drops the last n elements (from the end).

    Drops the last n elements (from the end).

    Example:

    import monix.eval.Task
    
    // Yields 1, 2
    Iterant[Task].of(1, 2, 3, 4, 5).dropLast(3)
    n

    the number of elements to drop

    returns

    a new iterant that drops the last n elements emitted by the source

  25. final def dropWhile(p: (A) => Boolean)(implicit F: Sync[F]): Iterant[F, A]

    Drops the longest prefix of elements that satisfy the given predicate and returns a new iterant that emits the rest.

    Drops the longest prefix of elements that satisfy the given predicate and returns a new iterant that emits the rest.

    Example:

    import monix.eval.Task
    
    // Yields 4, 5
    Iterant[Task].of(1, 2, 3, 4, 5).dropWhile(_ < 4)
    p

    is the predicate used to test whether the current element should be dropped, if true, or to interrupt the dropping process, if false

    returns

    a new iterant that drops the elements of the source until the first time the given predicate returns false

  26. final def dropWhileWithIndex(p: (A, Int) => Boolean)(implicit F: Sync[F]): Iterant[F, A]

    Drops the longest prefix of elements that satisfy the given function and returns a new Iterant that emits the rest.

    Drops the longest prefix of elements that satisfy the given function and returns a new Iterant that emits the rest.

    In comparison with dropWhile, this version accepts a function that takes an additional parameter: the zero-based index of the element.

    Example:

    import monix.eval.Task
    
    // Yields 3, 4, 5
    Iterant[Task].of(1, 2, 3, 4, 5)
      .dropWhileWithIndex((value, index) => value >= index * 2)
    p

    is the predicate used to test whether the current element should be dropped, if true, or to interrupt the dropping process, if false

    returns

    a new iterant that drops the elements of the source until the first time the given predicate returns false

  27. final def dump(prefix: String, out: PrintStream = System.out)(implicit F: Sync[F]): Iterant[F, A]

    Dumps incoming events to standard output with provided prefix.

    Dumps incoming events to standard output with provided prefix.

    Utility that can be used for debugging purposes.

    Example:

    import monix.eval.Task
    import monix.execution.Scheduler.Implicits.global
    
    Iterant[Task].range(0, 4)
      .dump("O")
      .completedL
    
    // 0: O --> next-batch --> 0
    // 1: O --> next-batch --> 1
    // 2: O --> next-batch --> 2
    // 3: O --> next-batch --> 3
    // 4: O --> halt --> no error
  28. final def eq(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef
  29. def equals(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef → Any
  30. final def existsL(p: (A) => Boolean)(implicit F: Sync[F]): F[Boolean]

    Returns true in case the given predicate is satisfied by any of the emitted items, or false in case the end of the stream has been reached with no items satisfying the given predicate.

    Returns true in case the given predicate is satisfied by any of the emitted items, or false in case the end of the stream has been reached with no items satisfying the given predicate.

    Example:

    import monix.eval.Coeval
    
    val source = Iterant[Coeval].of(1, 2, 3, 4)
    
    // Yields true
    source.existsL(_ % 2 == 0)
    
    // Yields false
    source.existsL(_ % 7 == 0)
    p

    is a predicate function that's going to test each item emitted by the source until we get a positive match for one of them or until the stream ends

    returns

    true if any of the items satisfies the given predicate or false if none of them do

  31. final def filter(p: (A) => Boolean)(implicit F: Sync[F]): Iterant[F, A]

    Filters the iterant by the given predicate function, returning only those elements that match.

    Filters the iterant by the given predicate function, returning only those elements that match.

    Example:

    import monix.eval.Task
    
    // Yields 2, 4, 6
    Iterant[Task].of(1, 2, 3, 4, 5, 6).filter(_ % 2 == 0)
    p

    the predicate used to test elements.

    returns

    a new iterant consisting of all elements that satisfy the given predicate. The order of the elements is preserved.

  32. def findL(p: (A) => Boolean)(implicit F: Sync[F]): F[Option[A]]

    Given a predicate, finds the first item that satisfies it, returning Some(a) if available, or None otherwise.

    Given a predicate, finds the first item that satisfies it, returning Some(a) if available, or None otherwise.

    import monix.eval.Coeval
    
    // Yields Some(2)
    Iterant[Coeval].of(1, 2, 3, 4).findL(_ % 2 == 0)
    
    // Yields None
    Iterant[Coeval].of(1, 2, 3, 4).findL(_ > 10)

    The stream is traversed from beginning to end, the process being interrupted as soon as it finds one element matching the predicate, or until the stream ends.

    p

    is the function to test the elements of the source

    returns

    either Some(value) in case value is an element emitted by the source, found to satisfy the predicate, or None otherwise

  33. final def flatMap[B](f: (A) => Iterant[F, B])(implicit F: Sync[F]): Iterant[F, B]

    Applies the function to the elements of the source and concatenates the results.

    Applies the function to the elements of the source and concatenates the results.

    This operation is the monadic "bind", with all laws it entails.

    Also note that the implementation can use constant memory depending on usage, thus it can be used in tail recursive loops.

    Example:

    import monix.eval.Task
    
    // Effectively equivalent with .filter
    Iterant[Task].of(1, 2, 3, 4, 5, 6).flatMap { elem =>
      if (elem % 2 == 0)
        Iterant[Task].pure(elem)
      else
        Iterant[Task].empty[Int]
    }
    f

    is the function mapping elements from the source to iterants

  34. final def flatten[B](implicit ev: <:<[A, Iterant[F, B]], F: Sync[F]): Iterant[F, B]

    Given an Iterant that generates Iterant elements, concatenates all the generated iterants.

    Given an Iterant that generates Iterant elements, concatenates all the generated iterants.

    Equivalent with: source.flatMap(x => x)

  35. final def foldL(implicit F: Sync[F], A: Monoid[A]): F[A]

    Given evidence that type A has a cats.Monoid implementation, folds the stream with the provided monoid definition.

    Given evidence that type A has a cats.Monoid implementation, folds the stream with the provided monoid definition.

    For streams emitting numbers, this effectively sums them up. For strings, this concatenates them.

    Example:

    import cats.implicits._
    import monix.eval.Task
    
    // Yields 10
    Iterant[Task].of(1, 2, 3, 4).foldL
    
    // Yields "1234"
    Iterant[Task].of("1", "2", "3", "4").foldL

    Note, in case you don't have a Monoid instance in scope, but you feel like you should, try one of these imports:

    // everything
    import cats.implicits._
    // a la carte:
    import cats.instances.all._
    A

    is the cats.Monoid type class instance that's needed in scope for folding the source

    returns

    the result of combining all elements of the source, or the defined Monoid.empty element in case the stream is empty

  36. final def foldLeftL[S](seed: => S)(op: (S, A) => S)(implicit F: Sync[F]): F[S]

    Left associative fold using the function op.

    Left associative fold using the function op.

    On execution the stream will be traversed from left to right, and the given function will be called with the prior result, accumulating state until the end, when the summary is returned.

    Example:

    import monix.eval.Task
    
    // Yields 15 (1 + 2 + 3 + 4 + 5)
    Iterant[Task].of(1, 2, 3, 4, 5).foldLeftL(0)(_ + _)
    seed

    is the start value

    op

    is the binary operator

    returns

    the result of inserting op between consecutive elements of this iterant, going from left to right with the seed as the start value, or seed if the iterant is empty.

  37. final def foldRightL[B](b: F[B])(f: (A, F[B]) => F[B])(implicit F: Sync[F]): F[B]

    Lazily fold the stream to a single value from the right.

    Lazily fold the stream to a single value from the right.

    This is the common foldr operation from Haskell's Foldable, or foldRight from cats.Foldable, but with the difference that Iterant is a lazy data type and thus it has to operate in the F[_] context.

    Here's for example how existsL, forallL and ++ could be expressed in terms of foldRightL:

    import cats.implicits._
    import cats.effect.Sync
    
    def exists[F[_], A](fa: Iterant[F, A], p: A => Boolean)
      (implicit F: Sync[F]): F[Boolean] = {
    
      fa.foldRightL(F.pure(false)) { (a, next) =>
        if (p(a)) F.pure(true) else next
      }
    }
    
    def forall[F[_], A](fa: Iterant[F, A], p: A => Boolean)
      (implicit F: Sync[F]): F[Boolean] = {
    
      fa.foldRightL(F.pure(true)) { (a, next) =>
        if (!p(a)) F.pure(false) else next
      }
    }
    
    def concat[F[_], A](lh: Iterant[F, A], rh: Iterant[F, A])
      (implicit F: Sync[F]): Iterant[F, A] = {
    
      Iterant.suspend[F, A] {
        lh.foldRightL(F.pure(rh)) { (a, rest) =>
          F.pure(Iterant.nextS(a, rest))
        }
      }
    }

    In this example we are short-circuiting the processing in case we find the one element that we are looking for, otherwise we keep traversing the stream until the end, finally returning the default value in case we haven't found what we were looking for.

    b

    is the starting value; in case f is a binary operator, this is typically its left-identity (zero)

    f

    is the function to be called that folds the list, receiving the current element being iterated on (first param) and the (lazy) result from recursively combining the rest of the list (second param)

    See also

    foldWhileLeftL and foldWhileLeftEvalL

  38. final def foldWhileLeftEvalL[S](seed: F[S])(op: (S, A) => F[Either[S, S]])(implicit F: Sync[F]): F[S]

    Left associative fold using the function op that can be short-circuited.

    Left associative fold using the function op that can be short-circuited.

    On execution the stream will be traversed from left to right, and the given function will be called with the prior result, accumulating state either until the end, or until op returns a Right result, when the summary is returned.

    The results are returned in the F[_] functor context, meaning that we can have lazy or asynchronous processing and we can suspend side effects, depending on the F data type being used.

    Example using cats.effect.IO:

    import cats.implicits._
    import cats.effect.IO
    
    // Sums first 10 items
    Iterant[IO].range(0, 1000).foldWhileLeftEvalL(IO((0, 0))) {
      case ((sum, count), e) =>
        IO {
          val next = (sum + e, count + 1)
          if (count + 1 < 10) Left(next) else Right(next)
        }
    }
    
    // Implements exists(predicate)
    Iterant[IO].of(1, 2, 3, 4, 5).foldWhileLeftEvalL(IO(false)) {
      (default, e) =>
        IO { if (e == 3) Right(true) else Left(default) }
    }
    
    // Implements forall(predicate)
    Iterant[IO].of(1, 2, 3, 4, 5).foldWhileLeftEvalL(IO(true)) {
      (default, e) =>
        IO { if (e != 3) Right(false) else Left(default) }
    }
    seed

    is the start value

    op

    is the binary operator returning either Left, signaling that the state should be evolved or a Right, signaling that the process can be short-circuited and the result returned immediately

    returns

    the result of inserting op between consecutive elements of this iterant, going from left to right with the seed as the start value, or seed if the iterant is empty

    See also

    Iterant.foldWhileLeftL for the strict version.

  39. final def foldWhileLeftL[S](seed: => S)(op: (S, A) => Either[S, S])(implicit F: Sync[F]): F[S]

    Left associative fold using the function op that can be short-circuited.

    Left associative fold using the function op that can be short-circuited.

    On execution the stream will be traversed from left to right, and the given function will be called with the prior result, accumulating state either until the end, or until op returns a Right result, when the summary is returned.

    Example:

    import monix.eval.Task
    
    // Sums first 10 items
    Iterant[Task].range(0, 1000).foldWhileLeftL((0, 0)) {
      case ((sum, count), e) =>
        val next = (sum + e, count + 1)
        if (count + 1 < 10) Left(next) else Right(next)
    }
    
    // Implements exists(predicate)
    Iterant[Task].of(1, 2, 3, 4, 5).foldWhileLeftL(false) {
      (default, e) =>
        if (e == 3) Right(true) else Left(default)
    }
    
    // Implements forall(predicate)
    Iterant[Task].of(1, 2, 3, 4, 5).foldWhileLeftL(true) {
      (default, e) =>
        if (e != 3) Right(false) else Left(default)
    }
    seed

    is the start value

    op

    is the binary operator returning either Left, signaling that the state should be evolved or a Right, signaling that the process can be short-circuited and the result returned immediately

    returns

    the result of inserting op between consecutive elements of this iterant, going from left to right with the seed as the start value, or seed if the iterant is empty

    See also

    Iterant.foldWhileLeftL for the lazy, potentially asynchronous version.

  40. final def forallL(p: (A) => Boolean)(implicit F: Sync[F]): F[Boolean]

    Returns true in case the given predicate is satisfied by all of the emitted items, or false in case the given predicate fails for any of those items.

    Returns true in case the given predicate is satisfied by all of the emitted items, or false in case the given predicate fails for any of those items.

    Example:

    import monix.eval.Coeval
    
    val source = Iterant[Coeval].of(1, 2, 3, 4)
    
    // Yields false
    source.forallL(_ % 2 == 0)
    
    // Yields true
    source.existsL(_ < 10)
    p

    is a predicate function that's going to test each item emitted by the source until we get a negative match for one of them or until the stream ends

    returns

    true if all of the items satisfy the given predicate or false if any of them don't

  41. final def foreach(cb: (A) => Unit)(implicit F: Sync[F]): F[Unit]

    Consumes the source iterable, executing the given callback for each element.

    Consumes the source iterable, executing the given callback for each element.

    Example:

    import monix.eval.Task
    
    // Prints all elements, each one on a different line
    Iterant[Task].of(1, 2, 3).foreach { elem =>
      println("Elem: " + elem.toString)
    }
    cb

    is the callback to call for each element emitted by the source.

  42. final def getClass(): Class[_ <: AnyRef]
    Definition Classes
    AnyRef → Any
    Annotations
    @native() @HotSpotIntrinsicCandidate()
  43. final def guarantee(f: F[Unit])(implicit F: Sync[F]): Iterant[F, A]

    Given a routine make sure to execute it whenever the current stream reaches the end, successfully, in error, or canceled.

    Given a routine make sure to execute it whenever the current stream reaches the end, successfully, in error, or canceled.

    Implements cats.effect.Bracket.guarantee.

    Example:

    import monix.eval.Task
    
    def iterant: Iterant[Task, Int] =
      Iterant.delay(???)
    
    iterant.guarantee(Task.eval {
      println("Releasing resources!")
    })
    f

    is the function to execute on early stop

  44. final def guaranteeCase(f: (ExitCase[Throwable]) => F[Unit])(implicit F: Applicative[F]): Iterant[F, A]

    Returns a new iterant in which f is scheduled to be executed on halt or if canceled.

    Returns a new iterant in which f is scheduled to be executed on halt or if canceled.

    Implements cats.effect.Bracket.guaranteeCase.

    This would typically be used to ensure that a finalizer will run at the end of the stream.

    Example:

    import monix.eval.Task
    import cats.effect.ExitCase
    
    def iterant: Iterant[Task, Int] =
      Iterant.delay(???)
    
    iterant.guaranteeCase(err => Task.eval {
      err match {
        case ExitCase.Completed =>
          println("Completed successfully!")
        case ExitCase.Error(e) =>
          e.printStackTrace()
        case ExitCase.Canceled =>
          println("Was stopped early!")
      }
    })
    f

    is the finalizer to execute when streaming is terminated, by successful completion, error or cancellation

  45. def hashCode(): Int
    Definition Classes
    AnyRef → Any
    Annotations
    @native() @HotSpotIntrinsicCandidate()
  46. final def headOptionL(implicit F: Sync[F]): F[Option[A]]

    Optionally selects the first element.

    Optionally selects the first element.

    import monix.eval.Task
    
    // Yields Some(1)
    Iterant[Task].of(1, 2, 3, 4).headOptionL
    
    // Yields None
    Iterant[Task].empty[Int].headOptionL
    returns

    the first element of this iterant if it is nonempty, or None if it is empty, in the F context.

  47. final def interleave[B >: A](rhs: Iterant[F, B])(implicit F: Sync[F]): Iterant[F, B]

    Lazily interleaves two iterants together, starting with the first element from self.

    Lazily interleaves two iterants together, starting with the first element from self.

    The length of the result will be the shorter of the two arguments.

    Example:

    import monix.eval.Task
    
    val lh = Iterant[Task].of(11, 12)
    val rh = Iterant[Task].of(21, 22, 23)
    
    // Yields 11, 21, 12, 22
    lh.interleave(rh)
    rhs

    is the other iterant to interleave the source with (the right hand side)

  48. final def intersperse(start: A, separator: A, end: A)(implicit F: Sync[F]): Iterant[F, A]

    Creates a new stream from the source that will emit the start element followed by the upstream elements paired with the separator and lastly the end element.

    Creates a new stream from the source that will emit the start element followed by the upstream elements paired with the separator and lastly the end element.

    import monix.eval.Coeval
    
    // Yields '<', 'a', '-', 'b', '>'
    Iterant[Coeval].of('a', 'b').intersperse('<', '-', '>')
    start

    the first element emitted

    separator

    the separator

    end

    the last element emitted

  49. final def intersperse(separator: A)(implicit F: Sync[F]): Iterant[F, A]

    Creates a new stream from the source that will emit a specific separator between every pair of elements.

    Creates a new stream from the source that will emit a specific separator between every pair of elements.

    import monix.eval.Coeval
    
    // Yields 1, 0, 2, 0, 3
    Iterant[Coeval].of(1, 2, 3).intersperse(0)
    separator

    the separator

  50. final def isInstanceOf[T0]: Boolean
    Definition Classes
    Any
  51. final def lastOptionL(implicit F: Sync[F]): F[Option[A]]

    Optionally selects the last element.

    Optionally selects the last element.

    import monix.eval.Task
    
    // Yields Some(4)
    Iterant[Task].of(1, 2, 3, 4).lastOptionL
    
    // Yields None
    Iterant[Task].empty[Int].lastOptionL
    returns

    the last element of this iterant if it is nonempty, or None if it is empty, in the F context.

  52. final def map[B](f: (A) => B)(implicit F: Sync[F]): Iterant[F, B]

    Returns a new stream by mapping the supplied function over the elements of the source.

    Returns a new stream by mapping the supplied function over the elements of the source.

    import monix.eval.Task
    
    // Yields 2, 4, 6
    Iterant[Task].of(1, 2, 3).map(_ * 2)
    f

    is the mapping function that transforms the source

    returns

    a new iterant that's the result of mapping the given function over the source

  53. final def mapBatch[B](f: (A) => Batch[B])(implicit F: Sync[F]): Iterant[F, B]

    Returns a new stream by mapping the supplied function over the elements of the source yielding Iterant consisting of NextBatch nodes.

    Returns a new stream by mapping the supplied function over the elements of the source yielding Iterant consisting of NextBatch nodes.

    import monix.eval.Task
    import monix.tail.batches.Batch
    
    // Yields 1, 2, 3, 4, 5
    Iterant[Task].of(List(1, 2, 3), List(4), List(5)).mapBatch(Batch.fromSeq(_))
    // Yields 2, 4, 6
    Iterant[Task].of(1, 2, 3).mapBatch(x => Batch(x * 2))
    f

    is the mapping function that transforms the source into batches.

    returns

    a new iterant that's the result of mapping the given function over the source

  54. final def mapEval[B](f: (A) => F[B])(implicit F: Sync[F]): Iterant[F, B]

    Given a mapping function that returns a possibly lazy or asynchronous result, applies it over the elements emitted by the stream.

    Given a mapping function that returns a possibly lazy or asynchronous result, applies it over the elements emitted by the stream.

    import monix.eval.Task
    
    Iterant[Task].of(1, 2, 3, 4).mapEval { elem =>
      Task.eval {
        println("Received: " + elem.toString)
        elem * 2
      }
    }
    f

    is the mapping function that transforms the source

    returns

    a new iterant that's the result of mapping the given function over the source,

  55. final def mapK[G[_]](f: ~>[F, G])(implicit G: Sync[G]): Iterant[G, A]

    Given a functor transformation from F to G, lifts the source into an iterant that is going to use the resulting G for evaluation.

    Given a functor transformation from F to G, lifts the source into an iterant that is going to use the resulting G for evaluation.

    This can be used for replacing the underlying F type into something else. For example say we have an iterant that uses monix.eval.Coeval, but we want to convert it into one that uses monix.eval.Task for evaluation:

    import cats.~>
    import monix.eval._
    
    // Source is using Coeval for evaluation
    val source = Iterant[Coeval].of(1, 2, 3, 4)
    
    // Transformation to an Iterant of Task
    source.mapK(Coeval.liftTo[Task])

    This operator can be used for more than transforming the F type into something else.

    G

    is the data type that is going to drive the evaluation of the resulting iterant

    f

    is the functor transformation that's used to transform the source into an iterant that uses G for evaluation

  56. final def maxByL[K](key: (A) => K)(implicit F: Sync[F], K: Order[K]): F[Option[A]]

    Takes the elements of the source iterant and emits the element that has the maximum key value, where the key is generated by the given function.

    Takes the elements of the source iterant and emits the element that has the maximum key value, where the key is generated by the given function.

    Example:

    import cats.implicits._
    import monix.eval.Coeval
    
    case class Person(name: String, age: Int)
    
    // Yields Some(Person("Peter", 23))
    Iterant[Coeval].of(Person("Peter", 23), Person("May", 21))
      .maxByL(_.age)
    
    // Yields None
    Iterant[Coeval].empty[Person].maxByL(_.age)

    Cats Order and Scala Interop

    Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

    We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

    Building a cats.Order is easy to do if you already have a Scala Ordering instance:

    import cats.Order
    
    case class Person(name: String, age: Int)
    
    // Starting from a Scala Ordering
    implicit val scalaOrderingForPerson: Ordering[Person] =
      new Ordering[Person] {
        def compare(x: Person, y: Person): Int =
          x.age.compareTo(y.age) match {
            case 0 => x.name.compareTo(y.name)
            case o => o
          }
      }
    
    // Building a cats.Order from it
    implicit val catsOrderForPerson: Order[Person] =
      Order.fromOrdering

    You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

    val scalaOrdering = catsOrderForPerson.toOrdering
    key

    is the function that returns the key for which the given ordering is defined

    K

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the maximum element of the source stream, relative to its key generated by the given function and the given ordering

  57. final def maxL(implicit F: Sync[F], A: Order[A]): F[Option[A]]

    Given a cats.Order over the stream's elements, returns the maximum element in the stream.

    Given a cats.Order over the stream's elements, returns the maximum element in the stream.

    Example:

    import cats.implicits._
    import monix.eval.Coeval
    
    // Yields Some(20)
    Iterant[Coeval].of(1, 10, 7, 6, 8, 20, 3, 5).maxL
    
    // Yields None
    Iterant[Coeval].empty[Int].maxL

    Cats Order and Scala Interop

    Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

    We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

    Building a cats.Order is easy to do if you already have a Scala Ordering instance:

    import cats.Order
    
    case class Person(name: String, age: Int)
    
    // Starting from a Scala Ordering
    implicit val scalaOrderingForPerson: Ordering[Person] =
      new Ordering[Person] {
        def compare(x: Person, y: Person): Int =
          x.age.compareTo(y.age) match {
            case 0 => x.name.compareTo(y.name)
            case o => o
          }
      }
    
    // Building a cats.Order from it
    implicit val catsOrderForPerson: Order[Person] =
      Order.fromOrdering

    You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

    val scalaOrdering = catsOrderForPerson.toOrdering
    A

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the maximum element of the source stream, relative to the defined Order

  58. final def minByL[K](key: (A) => K)(implicit F: Sync[F], K: Order[K]): F[Option[A]]

    Takes the elements of the source iterant and emits the element that has the minimum key value, where the key is generated by the given function.

    Takes the elements of the source iterant and emits the element that has the minimum key value, where the key is generated by the given function.

    Example:

    import cats.implicits._
    import monix.eval.Coeval
    
    case class Person(name: String, age: Int)
    
    // Yields Some(Person("May", 21))
    Iterant[Coeval].of(Person("Peter", 23), Person("May", 21))
      .minByL(_.age)
    
    // Yields None
    Iterant[Coeval].empty[Person].minByL(_.age)

    Cats Order and Scala Interop

    Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

    We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

    Building a cats.Order is easy to do if you already have a Scala Ordering instance:

    import cats.Order
    
    case class Person(name: String, age: Int)
    
    // Starting from a Scala Ordering
    implicit val scalaOrderingForPerson: Ordering[Person] =
      new Ordering[Person] {
        def compare(x: Person, y: Person): Int =
          x.age.compareTo(y.age) match {
            case 0 => x.name.compareTo(y.name)
            case o => o
          }
      }
    
    // Building a cats.Order from it
    implicit val catsOrderForPerson: Order[Person] =
      Order.fromOrdering

    You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

    val scalaOrdering = catsOrderForPerson.toOrdering
    key

    is the function that returns the key for which the given ordering is defined

    K

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the minimum element of the source stream, relative to its key generated by the given function and the given ordering

  59. final def minL(implicit F: Sync[F], A: Order[A]): F[Option[A]]

    Given a cats.Order over the stream's elements, returns the minimum element in the stream.

    Given a cats.Order over the stream's elements, returns the minimum element in the stream.

    Example:

    import cats.implicits._
    import monix.eval.Coeval
    
    // Yields Some(3)
    Iterant[Coeval].of(10, 7, 6, 8, 20, 3, 5).minL
    
    // Yields None
    Iterant[Coeval].empty[Int].minL

    Cats Order and Scala Interop

    Monix prefers to work with cats.Order for assessing the order of elements that have an ordering defined, instead of scala.math.Ordering.

    We do this for consistency, as Monix is now building on top of Cats. This may change in the future, depending on what happens with typelevel/cats#2455.

    Building a cats.Order is easy to do if you already have a Scala Ordering instance:

    import cats.Order
    
    case class Person(name: String, age: Int)
    
    // Starting from a Scala Ordering
    implicit val scalaOrderingForPerson: Ordering[Person] =
      new Ordering[Person] {
        def compare(x: Person, y: Person): Int =
          x.age.compareTo(y.age) match {
            case 0 => x.name.compareTo(y.name)
            case o => o
          }
      }
    
    // Building a cats.Order from it
    implicit val catsOrderForPerson: Order[Person] =
      Order.fromOrdering

    You can also do that in reverse, so you can prefer cats.Order (due to Cats also exposing laws and tests for free) and build a Scala Ordering when needed:

    val scalaOrdering = catsOrderForPerson.toOrdering
    A

    is the cats.Order type class instance that's going to be used for comparing elements

    returns

    the minimum element of the source stream, relative to the defined Order

  60. final def ne(arg0: AnyRef): Boolean
    Definition Classes
    AnyRef
  61. final def notify(): Unit
    Definition Classes
    AnyRef
    Annotations
    @native() @HotSpotIntrinsicCandidate()
  62. final def notifyAll(): Unit
    Definition Classes
    AnyRef
    Annotations
    @native() @HotSpotIntrinsicCandidate()
  63. final def onErrorHandle[B >: A](f: (Throwable) => B)(implicit F: Sync[F]): Iterant[F, B]

    Returns an Iterant that mirrors the behavior of the source, unless the source is terminated with an error, in which case the streaming of events fallbacks to an iterant emitting a single element generated by the backup function.

    Returns an Iterant that mirrors the behavior of the source, unless the source is terminated with an error, in which case the streaming of events fallbacks to an iterant emitting a single element generated by the backup function.

    Example:

    import monix.eval.Task
    import monix.execution.exceptions.DummyException
    
    val prefix = Iterant[Task].of(1, 2, 3, 4)
    val suffix = Iterant[Task].raiseError[Int](DummyException("dummy"))
    val fa = prefix ++ suffix
    
    fa.onErrorHandle { _ => 5 }

    See onErrorRecover for the version that takes a partial function as a parameter.

    f

    is a function that matches errors with a backup element that is emitted when the source throws an error.

  64. final def onErrorHandleWith[B >: A](f: (Throwable) => Iterant[F, B])(implicit F: Sync[F]): Iterant[F, B]

    Returns an Iterant that mirrors the behavior of the source, unless the source is terminated with an error, in which case the streaming of events continues with the specified backup sequence generated by the given function.

    Returns an Iterant that mirrors the behavior of the source, unless the source is terminated with an error, in which case the streaming of events continues with the specified backup sequence generated by the given function.

    Example:

    import monix.eval.Task
    import monix.execution.exceptions.DummyException
    
    val prefix = Iterant[Task].of(1, 2, 3, 4)
    val suffix = Iterant[Task].raiseError[Int](DummyException("dummy"))
    val fa = prefix ++ suffix
    
    fa.onErrorHandleWith {
      case _: DummyException =>
        Iterant[Task].pure(5)
      case other =>
        Iterant[Task].raiseError[Int](other)
    }

    See onErrorRecoverWith for the version that takes a partial function as a parameter.

    f

    is a function that matches errors with a backup throwable that is subscribed when the source throws an error.

  65. final def onErrorIgnore(implicit F: Sync[F]): Iterant[F, A]

    Returns a new Iterant that mirrors the source, but ignores any errors in case they happen.

  66. final def onErrorRecover[B >: A](pf: PartialFunction[Throwable, B])(implicit F: Sync[F]): Iterant[F, B]

    Returns an Iterant that mirrors the behavior of the source, unless the source is terminated with an error, in which case the streaming of events fallbacks to an iterant emitting a single element generated by the backup function.

    Returns an Iterant that mirrors the behavior of the source, unless the source is terminated with an error, in which case the streaming of events fallbacks to an iterant emitting a single element generated by the backup function.

    The created Iterant mirrors the behavior of the source in case the source does not end with an error or if the thrown Throwable is not matched.

    Example:

    import monix.eval.Task
    import monix.execution.exceptions.DummyException
    
    val prefix = Iterant[Task].of(1, 2, 3, 4)
    val suffix = Iterant[Task].raiseError[Int](DummyException("dummy"))
    val fa = prefix ++ suffix
    
    fa.onErrorRecover {
      case _: DummyException => 5
    }

    See onErrorHandle for the version that takes a total function as a parameter.

    pf

    - a function that matches errors with a backup element that is emitted when the source throws an error.

  67. final def onErrorRecoverWith[B >: A](pf: PartialFunction[Throwable, Iterant[F, B]])(implicit F: Sync[F]): Iterant[F, B]

    Returns an Iterant that mirrors the behavior of the source, unless the source is terminated with an error, in which case the streaming of events continues with the specified backup sequence generated by the given partial function.

    Returns an Iterant that mirrors the behavior of the source, unless the source is terminated with an error, in which case the streaming of events continues with the specified backup sequence generated by the given partial function.

    The created Iterant mirrors the behavior of the source in case the source does not end with an error or if the thrown Throwable is not matched.

    Example:

    import monix.eval.Task
    import monix.execution.exceptions.DummyException
    
    val prefix = Iterant[Task].of(1, 2, 3, 4)
    val suffix = Iterant[Task].raiseError[Int](DummyException("dummy"))
    val fa = prefix ++ suffix
    
    fa.onErrorRecoverWith {
      case _: DummyException =>
        Iterant[Task].pure(5)
    }

    See onErrorHandleWith for the version that takes a total function as a parameter.

    pf

    is a function that matches errors with a backup throwable that is subscribed when the source throws an error.

  68. final def parZip[B](rhs: Iterant[F, B])(implicit F: Sync[F], P: Parallel[F]): Iterant[F, (A, B)]

    Lazily zip two iterants together, the elements of the emitted tuples being fetched in parallel.

    Lazily zip two iterants together, the elements of the emitted tuples being fetched in parallel.

    This is the parallel version of zip, the results are still ordered, but it can yield non-deterministic ordering of effects when fetching the elements of an emitted tuple.

    rhs

    is the other iterant to zip the source with (the right hand side)

  69. final def parZipMap[B, C](rhs: Iterant[F, B])(f: (A, B) => C)(implicit F: Sync[F], P: Parallel[F]): Iterant[F, C]

    Lazily zip two iterants together, in parallel, using the given function f to produce output values.

    Lazily zip two iterants together, in parallel, using the given function f to produce output values.

    This is like zipMap, except that the element pairs are processed in parallel (ordered results, but non-deterministic ordering of effects).

    rhs

    is the other iterant to zip the source with (the right hand side)

    f

    is the mapping function to transform the zipped (A, B) elements

  70. def productElementName(n: Int): String
    Definition Classes
    Product
  71. def productElementNames: Iterator[String]
    Definition Classes
    Product
  72. def productIterator: Iterator[Any]
    Definition Classes
    Product
  73. def productPrefix: String
    Definition Classes
    Product
  74. final def pushToChannel(channel: Producer[F, A])(implicit F: Sync[F]): F[Unit]

    Consumes the source by pushing it to the specified channel.

    Consumes the source by pushing it to the specified channel.

    channel

    is a ProducerF value that will be used for consuming the stream

  75. final def reduceL(op: (A, A) => A)(implicit F: Sync[F]): F[Option[A]]

    Reduces the elements of the source using the specified associative binary operator, going from left to right, start to finish.

    Reduces the elements of the source using the specified associative binary operator, going from left to right, start to finish.

    Example:

    import monix.eval.Coeval
    
    // Yields Some(10)
    Iterant[Coeval].of(1, 2, 3, 4).reduceL(_ + _)
    
    // Yields None
    Iterant[Coeval].empty[Int].reduceL(_ + _)
    op

    is an associative binary operation that's going to be used to reduce the source to a single value

    returns

    either Some(value) in case the stream is not empty, value being the result of inserting op between consecutive elements of this iterant, going from left to right, or None in case the stream is empty

  76. final def repeat(implicit F: Sync[F]): Iterant[F, A]

    Repeats the items emitted by the source continuously

    Repeats the items emitted by the source continuously

    It terminates either on error or if the source is empty.

    In case repetition on empty streams is desired, then combine with retryIfEmpty:

    import monix.eval.Coeval
    import scala.util.Random
    
    val stream = Iterant[Coeval].suspend(Coeval {
      val nr = Random.nextInt()
      if (nr % 10 != 0)
        Iterant[Coeval].empty[Int]
      else
        Iterant[Coeval].of(1, 2, 3)
    })
    
    // Will eventually repeat elements 1, 2, 3
    stream.retryIfEmpty(None).repeat
  77. final def retryIfEmpty(maxRetries: Option[Int])(implicit F: Sync[F]): Iterant[F, A]

    Retries processing the source stream after the search is detected as being empty.

    Retries processing the source stream after the search is detected as being empty.

    import monix.eval.Coeval
    import scala.util.Random
    
    val stream = Iterant[Coeval].suspend(Coeval {
      val nr = Random.nextInt()
      if (nr % 10 != 0)
        Iterant[Coeval].empty[Int]
      else
        Iterant[Coeval].of(1, 2, 3)
    })
    
    // Will eventually stream elements 1, 2, 3
    stream.retryIfEmpty(None)
    maxRetries

    is an optional integer specifying a maximum number of retries before it gives up and returns an empty stream

  78. final def scan[S](seed: => S)(op: (S, A) => S)(implicit F: Sync[F]): Iterant[F, S]

    Applies a binary operator to a start value and all elements of this Iterant, going left to right and returns a new Iterant that emits on each step the result of the applied function.

    Applies a binary operator to a start value and all elements of this Iterant, going left to right and returns a new Iterant that emits on each step the result of the applied function.

    Similar to foldLeftL, but emits the state on each step. Useful for modeling finite state machines.

    Example showing how state can be evolved and acted upon:

    import monix.eval.Task
    
    sealed trait State[+A] { def count: Int }
    case object Init extends State[Nothing] { def count = 0 }
    case class Current[A](current: A, count: Int) extends State[A]
    
    // Whatever...
    val source = Iterant[Task].range(0, 1000)
    
    val scanned = source.scan(Init : State[Int]) { (acc, a) =>
      acc match {
        case Init => Current(a, 1)
        case Current(_, count) => Current(a, count + 1)
      }
    }
    
    scanned
      .takeWhile(_.count < 10)
      .collect { case Current(a, _) => a }
    seed

    is the initial state

    op

    is the function that evolves the current state

    returns

    a new iterant that emits all intermediate states being resulted from applying function op

    See also

    scan0 for the version that emits seed element at the beginning

  79. final def scan0[S](seed: => S)(op: (S, A) => S)(implicit F: Sync[F]): Iterant[F, S]

    Applies a binary operator to a start value and all elements of this Iterant, going left to right and returns a new Iterant that emits on each step the result of the applied function.

    Applies a binary operator to a start value and all elements of this Iterant, going left to right and returns a new Iterant that emits on each step the result of the applied function.

    This is a version of scan that emits seed element at the beginning, similar to scanLeft on Scala collections.

  80. final def scanEval[S](seed: F[S])(op: (S, A) => F[S])(implicit F: Sync[F]): Iterant[F, S]

    Applies a binary operator to a start value and all elements of this Iterant, going left to right and returns a new Iterant that emits on each step the result of the applied function.

    Applies a binary operator to a start value and all elements of this Iterant, going left to right and returns a new Iterant that emits on each step the result of the applied function.

    Similar with scan, but this can suspend and evaluate side effects in the F[_] context, thus allowing for asynchronous data processing.

    Similar to foldLeftL and foldWhileLeftEvalL, but emits the state on each step. Useful for modeling finite state machines.

    Example showing how state can be evolved and acted upon:

    import monix.eval.Task
    
    sealed trait State[+A] { def count: Int }
    case object Init extends State[Nothing] { def count = 0 }
    case class Current[A](current: Option[A], count: Int)
      extends State[A]
    
    // Dummies
    case class Person(id: Int, name: String, age: Int)
    def requestPersonDetails(id: Int): Task[Option[Person]] = Task.delay(???)
    
    // Whatever
    val source = Iterant[Task].range(0, 1000)
    // Initial state
    val seed = Task.now(Init : State[Person])
    
    val scanned = source.scanEval(seed) { (state, id) =>
      requestPersonDetails(id).map { a =>
        state match {
          case Init =>
            Current(a, 1)
          case Current(_, count) =>
            Current(a, count + 1)
        }
      }
    }
    
    scanned
      .takeWhile(_.count < 10)
      .collect { case Current(Some(a), _) => a }
    seed

    is the initial state

    op

    is the function that evolves the current state

    returns

    a new iterant that emits all intermediate states being resulted from applying the given function

    See also

    scan for the version that does not require using F[_] in the provided operator

    scanEval0 for the version that emits seed element at the beginning

  81. final def scanEval0[S](seed: F[S])(op: (S, A) => F[S])(implicit F: Sync[F]): Iterant[F, S]

    Applies a binary operator to a start value and all elements of this Iterant, going left to right and returns a new Iterant that emits on each step the result of the applied function.

    Applies a binary operator to a start value and all elements of this Iterant, going left to right and returns a new Iterant that emits on each step the result of the applied function.

    This is a version of scanEval that emits seed element at the beginning, similar to scanLeft on Scala collections.

  82. final def scanMap[B](f: (A) => B)(implicit F: Sync[F], B: Monoid[B]): Iterant[F, B]

    Given a mapping function that returns a B type for which we have a cats.Monoid instance, returns a new stream that folds the incoming elements of the sources using the provided Monoid[B].combine, with the initial seed being the Monoid[B].empty value, emitting the generated values at each step.

    Given a mapping function that returns a B type for which we have a cats.Monoid instance, returns a new stream that folds the incoming elements of the sources using the provided Monoid[B].combine, with the initial seed being the Monoid[B].empty value, emitting the generated values at each step.

    Equivalent with scan applied with the given cats.Monoid, so given our f mapping function returns a B, this law holds:

    stream.scanMap(f) <-> stream.scan(Monoid[B].empty)(Monoid[B].combine)

    Example:

    import cats.implicits._
    import monix.eval.Task
    
    // Yields 2, 6, 12, 20, 30, 42
    Iterant[Task].of(1, 2, 3, 4, 5, 6).scanMap(x => x * 2)
    f

    is the mapping function applied to every incoming element of this Iterant before folding using Monoid[B].combine

    returns

    a new Iterant that emits all intermediate states being resulted from applying Monoid[B].combine function

    See also

    scanMap0 for the version that emits empty element at the beginning

  83. final def scanMap0[B](f: (A) => B)(implicit F: Sync[F], B: Monoid[B]): Iterant[F, B]

    Given a mapping function that returns a B type for which we have a cats.Monoid instance, returns a new stream that folds the incoming elements of the sources using the provided Monoid[B].combine, with the initial seed being the Monoid[B].empty value, emitting the generated values at each step.

    Given a mapping function that returns a B type for which we have a cats.Monoid instance, returns a new stream that folds the incoming elements of the sources using the provided Monoid[B].combine, with the initial seed being the Monoid[B].empty value, emitting the generated values at each step.

    This is a version of scanMap that emits seed element at the beginning.

  84. final def sumL(implicit F: Sync[F], A: Numeric[A]): F[A]

    Given evidence that type A has a scala.math.Numeric implementation, sums the stream of elements.

    Given evidence that type A has a scala.math.Numeric implementation, sums the stream of elements.

    An alternative to foldL which does not require any imports and works in cases cats.Monoid is not defined for values (e.g. A = Char)

  85. final def switchIfEmpty(backup: Iterant[F, A])(implicit F: Sync[F]): Iterant[F, A]

    In case this Iterant is empty, switch to the given backup.

  86. final def synchronized[T0](arg0: => T0): T0
    Definition Classes
    AnyRef
  87. final def tail(implicit F: Sync[F]): Iterant[F, A]

    Drops the first element of the source iterant, emitting the rest.

    Drops the first element of the source iterant, emitting the rest.

    Example:

    import monix.eval.Task
    
    // Yields 2, 3, 4
    Iterant[Task].of(1, 2, 3, 4).tail
    returns

    a new iterant that upon evaluation will emit all elements of the source, except for the head

  88. final def take(n: Int)(implicit F: Sync[F]): Iterant[F, A]

    Creates a new iterant that upon evaluation will select the first n elements from the source and then stop, in the order they are emitted by the source.

    Creates a new iterant that upon evaluation will select the first n elements from the source and then stop, in the order they are emitted by the source.

    Example:

    import monix.eval.Task
    
    // Yields 1, 2, 3
    Iterant[Task].of(1, 2, 3, 4, 5, 6).take(3)
    n

    is the number of elements to take from this iterant

    returns

    a new iterant instance that on evaluation will emit only the first n elements of this iterant

  89. final def takeEveryNth(n: Int)(implicit F: Sync[F]): Iterant[F, A]

    Takes every n-th element, dropping intermediary elements and returns a new iterant that emits those elements.

    Takes every n-th element, dropping intermediary elements and returns a new iterant that emits those elements.

    Example:

    import monix.eval.Task
    
    // Yields 2, 4, 6
    Iterant[Task].of(1, 2, 3, 4, 5, 6).takeEveryNth(2)
    
    // Yields 1, 2, 3, 4, 5, 6
    Iterant[Task].of(1, 2, 3, 4, 5, 6).takeEveryNth(1)
    n

    is the sequence number of an element to be taken (must be > 0)

    returns

    a new iterant instance that on evaluation will return only every n-th element of the source

  90. final def takeLast(n: Int)(implicit F: Sync[F]): Iterant[F, A]

    Creates a new iterable that only emits the last n elements emitted by the source.

    Creates a new iterable that only emits the last n elements emitted by the source.

    In case the source triggers an error, then the underlying buffer gets dropped and the error gets emitted immediately.

    Example:

    import monix.eval.Task
    
    // Yields 1, 2, 3
    Iterant[Task].of(1, 2, 3, 4, 5, 6).take(3)
    n

    is the number of elements to take from the end of the stream.

    returns

    a new iterant instance that on evaluation will emit the last n elements of the source

  91. final def takeWhile(p: (A) => Boolean)(implicit F: Sync[F]): Iterant[F, A]

    Takes longest prefix of elements that satisfy the given predicate and returns a new iterant that emits those elements.

    Takes longest prefix of elements that satisfy the given predicate and returns a new iterant that emits those elements.

    Example:

    import monix.eval.Task
    
    // Yields 1, 2, 3
    Iterant[Task].of(1, 2, 3, 4, 5, 6).takeWhile(_ < 4)
    p

    is the function that tests each element, stopping the streaming on the first false result

    returns

    a new iterant instance that on evaluation will all elements of the source for as long as the given predicate returns true, stopping upon the first false result

  92. final def takeWhileWithIndex(p: (A, Long) => Boolean)(implicit F: Sync[F]): Iterant[F, A]

    Takes longest prefix of elements zipped with their indices that satisfy the given predicate and returns a new iterant that emits those elements.

    Takes longest prefix of elements zipped with their indices that satisfy the given predicate and returns a new iterant that emits those elements.

    Example:

    import monix.eval.Task
    
    // Yields 1, 2
    Iterant[Task].of(1, 2, 3, 4, 5, 6).takeWhileWithIndex((_, idx) => idx != 2)
    p

    is the function that tests each element, stopping the streaming on the first false result

    returns

    a new iterant instance that on evaluation will all elements of the source for as long as the given predicate returns true, stopping upon the first false result

  93. final def toChannel(implicit F: Concurrent[F], cs: ContextShift[F]): Channel[F, A]

    Converts this Iterant to a monix.catnap.ChannelF.

  94. final def toListL(implicit F: Sync[F]): F[List[A]]

    Aggregates all elements in a List and preserves order.

    Aggregates all elements in a List and preserves order.

    Example:

    import monix.eval.Task
    
    // Yields List(1, 2, 3, 4)
    Iterant[Task].of(1, 2, 3, 4).toListL

    Note that this operation is dangerous, since if the iterant is infinite then this operation is non-terminating, the process probably blowing up with an out of memory error sooner or later.

  95. final def toReactivePublisher(implicit F: Effect[F]): Publisher[A]

    Converts this Iterant into an org.reactivestreams.Publisher.

    Converts this Iterant into an org.reactivestreams.Publisher.

    Meant for interoperability with other Reactive Streams implementations. Also useful because it turns the Iterant into another data type with a push-based communication protocol with back-pressure.

    Usage sample:

    import monix.eval.Task
    import monix.execution.rstreams.SingleAssignSubscription
    import org.reactivestreams.{Publisher, Subscriber, Subscription}
    
    def sum(source: Publisher[Int], requestSize: Int): Task[Long] =
      Task.create { (_, cb) =>
        val sub = SingleAssignSubscription()
    
        source.subscribe(new Subscriber[Int] {
          private[this] var requested = 0L
          private[this] var sum = 0L
    
          def onSubscribe(s: Subscription): Unit = {
            sub := s
            requested = requestSize
            s.request(requestSize)
          }
    
          def onNext(t: Int): Unit = {
            sum += t
            if (requestSize != Long.MaxValue) requested -= 1
    
            if (requested <= 0) {
              requested = requestSize
              sub.request(requestSize)
            }
          }
    
          def onError(t: Throwable): Unit =
            cb.onError(t)
          def onComplete(): Unit =
            cb.onSuccess(sum)
        })
    
        // Cancelable that can be used by Task
        sub
      }
    
    // Needed for `Effect[Task]`
    import monix.execution.Scheduler.Implicits.global
    
    val pub = Iterant[Task].of(1, 2, 3, 4).toReactivePublisher
    
    // Yields 10
    sum(pub, requestSize = 128)

    See the Reactive Streams for details.

  96. def toString(): String
    Definition Classes
    AnyRef → Any
  97. final def uncons(implicit F: Sync[F]): Iterant[F, (Option[A], Iterant[F, A])]

    Pull the first element out of this Iterant and return it and the rest.

    Pull the first element out of this Iterant and return it and the rest. If the returned Option is None, the remainder is always empty.

    The value returned is wrapped in Iterant to preserve resource safety, and consumption of the rest must not leak outside of use. The returned Iterant always contains a single element

    import cats._, cats.implicits._, cats.effect._
    
     def unconsFold[F[_]: Sync, A: Monoid](iterant: Iterant[F, A]): F[A] = {
      def go(iterant: Iterant[F, A], acc: A): Iterant[F, A] =
        iterant.uncons.flatMap {
          case (None, _) => Iterant.pure(acc)
          case (Some(a), rest) => go(rest, acc |+| a)
        }
    
      go(iterant, Monoid[A].empty).headOptionL.map(_.getOrElse(Monoid[A].empty))
    }
  98. final def unsafeFlatMap[B](f: (A) => Iterant[F, B])(implicit F: Sync[F]): Iterant[F, B]

    Applies the function to the elements of the source and concatenates the results.

    Applies the function to the elements of the source and concatenates the results.

    This variant of flatMap is not referentially transparent, because it tries to apply function f immediately, in case the Iterant is in a NextCursor or NextBatch state.

    To be used for optimizations, but keep in mind it's unsafe, as its application isn't referentially transparent.

    f

    is the function mapping elements from the source to iterants

  99. final def upcast[B >: A]: Iterant[F, B]

    Explicit covariance operator.

    Explicit covariance operator.

    The Iterant type isn't covariant in type param A, because covariance doesn't play well with a higher-kinded type like F[_]. So in case you have an Iterant[F, A], but need an Iterant[F, B], knowing that A extends B, then you can do an upcast.

    Example:

    import monix.eval.Task
    
    val source: Iterant[Task, List[Int]] = Iterant.suspend(Task[Iterant[Task, List[Int]]](???))
    
    // This will trigger an error because of the invariance:
    // val sequences: Iterant[Task, Seq[Int]] = source
    
    // But this will work just fine:
    val sequence: Iterant[Task, Seq[Int]] = source.upcast[Seq[Int]]
  100. final def wait(arg0: Long, arg1: Int): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException])
  101. final def wait(arg0: Long): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException]) @native()
  102. final def wait(): Unit
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.InterruptedException])
  103. final def withFilter(p: (A) => Boolean)(implicit F: Sync[F]): Iterant[F, A]

    Alias to filter to support syntax in for comprehension, i.e.

    Alias to filter to support syntax in for comprehension, i.e.

    Example:

    import monix.eval.Task
    
    case class Person(age: Int)
    
    val peopleSource: Iterant[Task, Person] =
      Iterant.range[Task](1, 100).map(Person.apply)
    
    for {
      adult <- peopleSource if adult.age >= 18
    } yield adult
  104. final def zip[B](rhs: Iterant[F, B])(implicit F: Sync[F]): Iterant[F, (A, B)]

    Lazily zip two iterants together.

    Lazily zip two iterants together.

    The length of the result will be the shorter of the two arguments.

    Example:

    import monix.eval.Task
    
    val lh = Iterant[Task].of(11, 12, 13, 14)
    val rh = Iterant[Task].of(21, 22, 23, 24, 25)
    
    // Yields (11, 21), (12, 22), (13, 23), (14, 24)
    lh.zip(rh)
    rhs

    is the other iterant to zip the source with (the right hand side)

  105. final def zipMap[B, C](rhs: Iterant[F, B])(f: (A, B) => C)(implicit F: Sync[F]): Iterant[F, C]

    Lazily zip two iterants together, using the given function f to produce output values.

    Lazily zip two iterants together, using the given function f to produce output values.

    The length of the result will be the shorter of the two arguments.

    Example:

    import monix.eval.Task
    
    val lh = Iterant[Task].of(11, 12, 13, 14)
    val rh = Iterant[Task].of(21, 22, 23, 24, 25)
    
    // Yields 32, 34, 36, 38
    lh.zipMap(rh) { (a, b) => a + b }
    rhs

    is the other iterant to zip the source with (the right hand side)

    f

    is the mapping function to transform the zipped (A, B) elements

  106. final def zipWithIndex(implicit F: Sync[F]): Iterant[F, (A, Long)]

    Zips the emitted elements of the source with their indices.

    Zips the emitted elements of the source with their indices.

    The length of the result will be the same as the source.

    Example:

    import monix.eval.Task
    
    val source = Iterant[Task].of("Sunday", "Monday", "Tuesday", "Wednesday")
    
    // Yields ("Sunday", 0), ("Monday", 1), ("Tuesday", 2), ("Wednesday", 3)
    source.zipWithIndex

Deprecated Value Members

  1. def finalize(): Unit
    Attributes
    protected[lang]
    Definition Classes
    AnyRef
    Annotations
    @throws(classOf[java.lang.Throwable]) @Deprecated
    Deprecated

Inherited from Serializable

Inherited from Product

Inherited from Equals

Inherited from AnyRef

Inherited from Any

Ungrouped