Connectable Observable Operators

This section explains the ConnectableObservable subclass and its operators:

• ConnectableObservable.connect( ) — instructs a Connectable Observable to begin emitting items
• Observable.publish( ) and Observable.multicast( ) — represents an Observable as a Connectable Observable
• Observable.publishLast( ) — represent an Observable as a Connectable Observable that emits only the last item emitted by the source Observable
• Observable.replay( ) — ensures that all Observers see the same sequence of emitted items, even if they subscribe after the Observable begins emitting the items
• ConnectableObservable.refCount( ) — makes a Connectable Observable behave like an ordinary Observable

A Connectable Observable resembles an ordinary Observable, except that it does not begin emitting items when it is subscribed to, but only when its connect() method is called. In this way you can wait for all intended Observers to subscribe to the Observable before the Observable begins emitting items.

The following example code shows two Observers subscribing to the same Observable. In the first case, they subscribe to an ordinary Observable; in the second case, they subscribe to a Connectable Observable that only connects after both observers subscribe. Note the difference in the output:

Example #1:

def firstMillion  = Observable.range( 1, 1000000 ).sample(7, java.util.concurrent.TimeUnit.MILLISECONDS);

firstMillion.subscribe(
  { println("Observer #1:" + it); },    // onNext
  { println("Error encountered"); },    // onError
  { println("Sequence #1 complete"); }  // onCompleted
);

firstMillion.subscribe(
  { println("Observer #2:" + it); },    // onNext
  { println("Error encountered"); },    // onError
  { println("Sequence #2 complete"); }  // onCompleted
);

Observer #1:211128
Observer #1:411633
Observer #1:629605
Observer #1:841903
Sequence #1 complete
Observer #2:244776
Observer #2:431416
Observer #2:621647
Observer #2:826996
Sequence #2 complete

Example #2:

def firstMillion  = Observable.range( 1, 1000000 ).sample(7, java.util.concurrent.TimeUnit.MILLISECONDS).publish();

firstMillion.subscribe(
  { println("Observer #1:" + it); },    // onNext
  { println("Error encountered"); },    // onError
  { println("Sequence #1 complete"); }  // onCompleted
);

firstMillion.subscribe(
  { println("Observer #2:" + it); },    // onNext
  { println("Error encountered"); },    // onError
  { println("Sequence #2 complete"); }  // onCompleted
);

firstMillion.connect();

Observer #2:208683
Observer #1:208683
Observer #2:432509
Observer #1:432509
Observer #2:644270
Observer #1:644270
Observer #2:887885
Observer #1:887885
Sequence #2 complete
Sequence #1 complete

see also:

• javadoc: ConnectableObservable
• Introduction to Rx: Publish and Connect

---

ConnectableObservable.connect( )

instructs a Connectable Observable to begin emitting items

Call a Connectable Observable's connect( ) method to instruct it to begin emitting the items from its underlying Observable to its Observers.

The connect( ) method returns a Subscription. You can call that object's unsubscribe( ) method to instruct the Observable to stop emitting items to its Observers.

You can also use the connect( ) method to instruct an Observable to begin emitting items (or, to begin generating items that would be emitted) even before any Observer has subscribed to it.

see also

• javadoc: connect()
• Linq: Connect
• Introduction to Rx: Publish and Connect

---

Observable.publish( ) and Observable.multicast( )

represents an Observable as a Connectable Observable

To represent an Observable as a Connectable Observable, use the publish( ) or multicast() method.

see also:

• javadoc: multicast(subject)
• javadoc: publish()
• RxJS: multicast
• Linq: Multicast
• Linq: Publish
• Introduction to Rx: Publish and Connect
• Introduction to Rx: Multicast

---

Observable.publishLast( )

represent an Observable as a Connectable Observable that emits only the last item emitted by the source Observable

see also:

• RxJS: publishLast
• Linq: PublishLast
• Introduction to Rx: PublishLast

---

Observable.replay( )

ensures that all Observers see the same sequence of emitted items, even if they subscribe after the Observable begins emitting items

see also:

• javadoc: replay()
• RxJS: replay
• Linq: Replay
• Introduction to Rx: Replay

---

ConnectableObservable.refCount( )

makes a Connectable Observable behave like an ordinary Observable

You can represent a Connectable Observable so that it behaves much like an ordinary Observable by using the refCount( ) operator. This operator keeps track of how many Observers are subscribed to the resulting Observable and refrains from disconnecting from the source ConnectableObservable until all such Observables unsubscribe.

see also:

• RxJS: refCount
• Linq: RefCount
• Introduction to Rx: RefCount