Code Recipes Blog

In this post I will explain with example the difference between autoConnect and connect.

In terms of similarities both are used with ConnectableObservable which is an Hot Observable. Both tells the Observable when to start emitting items.

The difference being when to tell the Observable, at what point to start emitting items.

For example, if you want to tell the Observable to start emitting items after 2 Observers are subscribed, in advance or at the time of creating the Hot Observable itself. You need to use “autoConnect”. Below is the snippet of it.

    Observable<Long> coldObservable = Observable.interval(1, TimeUnit.SECONDS);
    Observable<Long> hotObservable = coldObservable.publish().autoConnect(2);

In the above code, at second line we are creating an hot Observable named “hotObservable” from “coldObservable” instance and at the same time telling the hotObservable to start emitting items the moment
two Observers subscribe to the “hotObservable”.

If you don’t know in advance or at the time of creating the hot Observable itself, after how many Observers subscription to start emitting items then you have to use “connect”. Below is the snippet of it.

1 Observable<Long> observable = Observable.interval(1, TimeUnit.SECONDS);
2 ConnectableObservable<Long> connectableObservable = observable.publish();
3 Disposable disposable1 = connectableObservable.subscribe((t) -> System.out.println("Observer1: " + t));
4 Thread.sleep(10000);
5 Disposable disposable2 = connectableObservable.subscribe((t) -> System.out.println("Observer2: " + t));
6 connectableObservable.connect();

In the above code, at line 2 we are creating a hot observable from observable instance. At this point we dont know when to tell the Observable to emitting items. So after two Observers subscribe, at line 6
we manually tell connectableObservable to start emitting items by calling “connect” method.

Below is the complete code for your reference.

<h1>Main Class</h1>
<pre><code>
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.observables.ConnectableObservable;

public class Example15 {
	public static void main(String[] args) throws Exception {
		Example15 example = new Example15();
		System.out.println("------------Connect Example---------------");
		example.connectExample();
		System.out.println("------------AutoConnect Example---------------");
		example.autoConnectExample();
	}
	
	public void connectExample() throws Exception {
		Observable<Long> observable = Observable.interval(1, TimeUnit.SECONDS);
		ConnectableObservable<Long> connectableObservable = observable.publish();
		Disposable disposable1 = connectableObservable.subscribe((t) -> System.out.println("Observer1: " + t));
		Thread.sleep(10000);
		Disposable disposable2 = connectableObservable.subscribe((t) -> System.out.println("Observer2: " + t));
		connectableObservable.connect();
		Thread.sleep(10000);
		Disposable disposable3 = connectableObservable.subscribe((t) -> System.out.println("Observer3: " + t));
		Thread.sleep(10000);
		disposable1.dispose();
		disposable2.dispose();
		disposable3.dispose();
	}
	
	public void autoConnectExample() throws Exception {
		Observable<Long> coldObservable = Observable.interval(1, TimeUnit.SECONDS);
		Observable<Long> hotObservable = coldObservable.publish().autoConnect(2);
		Disposable disposable1 = hotObservable.subscribe((t) -> System.out.println("Observer1: " + t));
		Thread.sleep(10000);
		Disposable disposable2 = hotObservable.subscribe((t) -> System.out.println("Observer2: " + t));
		Thread.sleep(10000);
		Disposable disposable3 = hotObservable.subscribe((t) -> System.out.println("Observer3: " + t));
		Thread.sleep(10000);
		disposable1.dispose();
		disposable2.dispose();
		disposable3.dispose();
	}
}
</code></pre>