final class ConnectableSubscriber[-A] extends Subscriber[A]
Wraps a Subscriber into an implementation that abstains from emitting items until the call
to connect() happens. Prior to connect() you can enqueue
events for delivery once connect() happens, but before any items
emitted by onNext / onComplete and onError.
Example:
import monix.execution.Ack.Continue import monix.execution.Scheduler.Implicits.global val subscriber = new Subscriber[String] { val scheduler = global def onNext(a: String) = { println(s"Received: $$a") Continue } def onError(e: Throwable) = println(s"Error: $$e") def onComplete() = println("Completed!") } val out = ConnectableSubscriber(subscriber) // schedule onNext event, after connect() out.onNext("c") // schedule event "a" to be emitted first out.pushFirst("a") // schedule event "b" to be emitted second out.pushFirst("b") // underlying observer now gets events "a", "b", "c" in order out.connect()
Example of an observer ended in error:
val out2 = ConnectableSubscriber(subscriber) // schedule onNext event, after connect() out2.onNext("c") out2.pushFirst("a") // event "a" to be emitted first out2.pushFirst("b") // event "b" to be emitted second // schedule an onError sent downstream, once connect() // happens, but after "a" and "b" out2.pushError(new RuntimeException()) // underlying observer receives ... // onNext("a") -> onNext("b") -> onError(RuntimeException) out2.connect() // NOTE: that onNext("c") never happens
- Self Type
- ConnectableSubscriber[A]
- Source
- ConnectableSubscriber.scala
- Alphabetic
- By Inheritance
- ConnectableSubscriber
- Subscriber
- Observer
- Serializable
- AnyRef
- Any
- Hide All
- Show All
- Public
- Protected
Value Members
- final def !=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def ##: Int
- Definition Classes
- AnyRef → Any
- final def ==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def asInstanceOf[T0]: T0
- Definition Classes
- Any
- def clone(): AnyRef
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.CloneNotSupportedException]) @native() @HotSpotIntrinsicCandidate()
- def connect(): CancelableFuture[Ack]
Connects the underling observer to the upstream publisher.
Connects the underling observer to the upstream publisher.
This function should be idempotent. Calling it multiple times should have the same effect as calling it once.
- final def eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- def equals(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef → Any
- final def getClass(): Class[_ <: AnyRef]
- Definition Classes
- AnyRef → Any
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- def hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- final def isInstanceOf[T0]: Boolean
- Definition Classes
- Any
- final def ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- final def notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- final def notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native() @HotSpotIntrinsicCandidate()
- def onComplete(): Unit
The Subscriber.onComplete method that pushes the complete event to the underlying observer.
The Subscriber.onComplete method that pushes the complete event to the underlying observer.
It will wait for connect to happen and the queue of scheduled events to be drained.
- Definition Classes
- ConnectableSubscriber → Observer
- def onError(ex: Throwable): Unit
The Subscriber.onError method that pushes an error event to the underlying observer.
The Subscriber.onError method that pushes an error event to the underlying observer.
It will wait for connect to happen and the queue of scheduled events to be drained.
- Definition Classes
- ConnectableSubscriber → Observer
- def onNext(elem: A): Future[Ack]
The Subscriber.onNext method that pushes events to the underlying subscriber.
The Subscriber.onNext method that pushes events to the underlying subscriber.
It will back-pressure by means of its
Future[Ack]result until connect happens and the underlying queue of scheduled events have been drained.- Definition Classes
- ConnectableSubscriber → Observer
- def pushComplete(): Unit
Schedule a complete event when connect happens, but before any elements scheduled with pushFirst or pushFirstAll.
Schedule a complete event when connect happens, but before any elements scheduled with pushFirst or pushFirstAll.
After
pushCompleteno more pushFirst or onNext events are accepted. - def pushError(ex: Throwable): Unit
Schedule an error event when connect happens, but before any elements scheduled with pushFirst or pushFirstAll.
Schedule an error event when connect happens, but before any elements scheduled with pushFirst or pushFirstAll.
After
pushErrorno more pushFirst or onNext events are accepted. - def pushFirst(elem: A): Unit
Schedule one element to be pushed to the underlying subscriber when connect happens.
Schedule one element to be pushed to the underlying subscriber when connect happens.
The given elements are appended to a queue that will be drained on connect. Afterwards no more elements are allowed to be pushed in the queue.
These elements are streamed before any elements that will eventually get streamed with onNext, because of the applied back-pressure from
onNext. - def pushFirstAll[U <: A](xs: Iterable[U]): Unit
Schedule elements to be pushed to the underlying subscriber when connect happens.
Schedule elements to be pushed to the underlying subscriber when connect happens.
The given elements are appended to a queue that will be drained on connect. Afterwards no more elements are allowed to be pushed in the queue.
These elements are streamed before any elements that will eventually get streamed with onNext, because of the applied back-pressure from
onNext. - implicit val scheduler: Scheduler
- Definition Classes
- ConnectableSubscriber → Subscriber
- final def synchronized[T0](arg0: => T0): T0
- Definition Classes
- AnyRef
- def toString(): String
- Definition Classes
- AnyRef → Any
- final def wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException]) @native()
- final def wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
This is the API documentation for the Monix library.
Package Overview
monix.execution exposes lower level primitives for dealing with asynchronous execution:
Atomictypes, as alternative tojava.util.concurrent.atomicmonix.catnap exposes pure abstractions built on top of the Cats-Effect type classes:
monix.eval is for dealing with evaluation of results, thus exposing Task and Coeval.
monix.reactive exposes the
Observablepattern:Observableimplementationsmonix.tail exposes Iterant for purely functional pull based streaming:
BatchandBatchCursor, the alternatives to Scala'sIterableandIteratorrespectively that we are using within Iterant's encodingYou can control evaluation with type you choose - be it Task, Coeval, cats.effect.IO or your own as long as you provide correct cats-effect or cats typeclass instance.