官术网_书友最值得收藏!

  • Learning RxJava
  • Thomas Nield
  • 594字
  • 2021-07-02 22:22:52

Cold Observables

Cold Observables are much like a music CD that can be replayed to each listener, so each person can hear all the tracks at any time. In the same manner, cold Observables will replay the emissions to each Observer, ensuring that all Observers get all the data. Most data-driven Observables are cold, and this includes the Observable.just() and Observable.fromIterable() factories.

In the following example, we have two Observers subscribed to one Observable. The Observable will first play all the emissions to the first Observer and then call onComplete(). Then, it will play all the emissions again to the second Observer and call onComplete(). They both receive the same datasets by getting two separate streams each, which is typical behavior for a cold Observable:

      import io.reactivex.Observable;

public class Launcher {
public static void main(String[] args) {

Observable<String> source =
Observable.just("Alpha","Beta","Gamma","Delta","Epsilon");

//first observer
source.subscribe(s -> System.out.println("Observer 1 Received:
"
+ s));

//second observer
source.subscribe(s -> System.out.println("Observer 2 Received:
"
+ s));

}
}

The output is as follows:

    Observer 1 Received: Alpha
Observer 1 Received: Beta
Observer 1 Received: Gamma
Observer 1 Received: Delta
Observer 1 Received: Epsilon
Observer 2 Received: Alpha
Observer 2 Received: Beta
Observer 2 Received: Gamma
Observer 2 Received: Delta
Observer 2 Received: Epsilon

Even if the second Observer transforms its emissions with operators, it will still get its own stream of emissions. Using operators such as map() and filter() against a cold Observable will still maintain the cold nature of the yielded Observables:

    import io.reactivex.Observable;

public class Launcher {
public static void main(String[] args) {

Observable<String> source =
Observable.just("Alpha","Beta","Gamma","Delta","Epsilon");

//first observer
source.subscribe(s -> System.out.println("Observer 1 Received:
"
+ s));

//second observer
source.map(String::length).filter(i -> i >= 5)
.subscribe(s -> System.out.println("Observer 2 Received: " +
s));

}
}

The output is as follows:

    Observer 1 Received: Alpha
Observer 1 Received: Beta
Observer 1 Received: Gamma
Observer 1 Received: Delta
Observer 1 Received: Epsilon
Observer 2 Received: 5
Observer 2 Received: 5
Observer 2 Received: 5
Observer 2 Received: 7

As stated earlier, Observable sources that emit finite datasets are usually cold.

Here is a more real-world example: Dave Moten's RxJava-JDBC (https://github.com/davidmoten/rxjava-jdbc) allows you to create cold Observables built off of SQL database queries. We will not digress into this library for too long, but if you want to query a SQLite database, for instance, include the SQLite JDBC driver and RxJava-JDBC libraries in your project. You can then query a database table reactively, as shown in the following code snippet:

    import com.github.davidmoten.rx.jdbc.ConnectionProviderFromUrl;
import com.github.davidmoten.rx.jdbc.Database;
import rx.Observable;
import java.sql.Connection;

public class Launcher {
public static void main(String[] args) {
Connection conn =
new ConnectionProviderFromUrl("jdbc:sqlite:/home/thomas
/rexon_metals.db"
).get();

Database db = Database.from(conn);

Observable<String> customerNames =
db.select("SELECT NAME FROM CUSTOMER")
.getAs(String.class);

customerNames.subscribe(s -> System.out.println(s));
}
}

The output is as follows:

    LITE Industrial
Rex Tooling Inc
Re-Barre Construction
Prairie Construction
Marsh Lane Metal Works

This SQL-driven Observable is cold. Many Observables emitting from finite data sources such as databases, text files, or JSON are cold. It is still important to note how the source Observable is architected. RxJava-JDBC will run the query each time for each Observer. This means that if the data changes in between two subscriptions, the second Observer will get different emissions than the first one. But the Observable is still cold since it is replaying the query even if the resulting data changes from the underlying tables.

Again, cold Observables will, in some shape or form, repeat the operation to generate these emissions to each Observer. Next, we will cover hot Observables that resemble events more than data.

主站蜘蛛池模板: 黔东| 海兴县| 庆阳市| 汉寿县| 右玉县| 齐河县| 浦江县| 呼图壁县| 华池县| 遂昌县| 芒康县| 叶城县| 江都市| 台中市| 彝良县| 汝城县| 玛沁县| 当阳市| 基隆市| 张北县| 凤阳县| 获嘉县| 永福县| 南江县| 冀州市| 高邑县| 鄂伦春自治旗| 安仁县| 东安县| 紫金县| 富宁县| 雅安市| 合山市| 金溪县| 和龙市| 江都市| 安新县| 咸阳市| 铜陵市| 木里| 望奎县|