Awaits for the specified number of consumers to be connected.

This is an utility to ensure that a certain number of consumers are connected before we start emitting events.

Closes the communication channel with a message that will be visible to all current and future consumers.

Note that if multiple halt events happen, then only the first one will be taken into account, all other halt messages are ignored.

Publishes an event on the channel.

Contract:

• in case the internal buffers are full, back-pressures until there's enough space for pushing the message, or until the channel was halted, whichever comes first
• returns true in case the message was pushed in the internal buffer or false in case the channel was halted and cannot receive any more events, in which case the producer's loop should terminate

Example:

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

def range[F[_]](channel: ProducerF[F, Int, Int], from: Int, until: Int)
  (implicit F: Async[F]): F[Unit] = {

  if (from < until) {
    if (from + 1 < until)
      channel.push(from).flatMap {
        case true =>
          // keep going
          range(channel, from + 1, until)
        case false =>
          // channel was halted by another producer loop, so stopping
          F.unit
      }
    else // we are done, publish the final event
      channel.halt(from + 1).as(())
  } else {
    F.unit // invalid range
  }
}

Publishes multiple events on the channel.

Contract:

• in case the internal buffers are full, back-pressures until there's enough space and the whole sequence was published, or until the channel was halted, in which case no further messages are allowed for being pushed
• returns true in case the whole sequence was pushed in the internal b or false in case the channel was halted and cannot receive any more events, in which case the producer's loop should terminate

Note: implementations may try to push events one by one. This is not an atomic operation. In case of concurrent producers, there's absolutely no guarantee for the order of messages coming from multiple producers. Also in case the channel is halted (and the resulting task returns false), the outcome can be that the sequence gets published partially.

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

def range[F[_]](channel: ProducerF[F, Int, Int], from: Int, until: Int)
  (implicit F: Async[F]): F[Unit] = {

  if (from < until) {
    val to = until - 1
    channel.pushMany(Range(from, to)).flatMap {
      case true =>
        channel.halt(to).as(()) // final event
      case false =>
        // channel was halted by a concurrent producer, so stop
        F.unit
    }
  } else {
    F.unit // invalid range
  }
}