RxJava 勉強会 Week1

# 目次
- ReactiveX
- Sync vs Async
- RxJava
- Operators

***
# ReactiveX

*** 
# ReactiveX とは
- 非同期プログラムのためのライブラリ
- 複数のデータ（イベント）を中心に据える
- Observable Sequence を使う
- 低レイヤ（threading, synchronization, thread-safety, concurrency...）を隠蔽
- ReactiveX の Java 実装： RxJava

ref: http://reactivex.io/intro.html

***
# Sync vs Async

*** 
# 同期的なプログラミング
単一データを取得：`T getData()`

```java
// 単一データを同期的に取得して出力
Integer i = getData();
System.out.println(i);
```

***
# 同期的なプログラミング
複数データを取得：`Iterable<T> getData`

```java
// 複数データを同期的に取得して出力
Iterable<Integer> iter = getData();
for (Integer i : iter) {
    System.out.println(i);
}
```

***
# 問題
- getData() で処理がブロックする
- 複数データ：すべてのデータが揃わないと出力できない


***
# 非同期的なプログラミング
単一データ：`Future<T> getData()`

```java
Future<Integer> f = getData();
...(その他の処理)...
Integer i = f.get();
System.out.println(i);
```

***
# 非同期的なプログラミング
複数データの取得：`Observable<T> getData();`

```java
Observable<Integer> observable = getData();
observable.subscribe(i -> System.out.println(i));
...(その他の処理)...
```

***
# 利点
- getData() で処理がブロックしない
- 複数データ：すべてのデータが揃わなくても処理が進む

*** 
# ReactiveX とは
- 非同期プログラム × 複数のデータ
  - デザインパターンの _Observer Pattern_ を複数化したもの
  - Observables が中心
- 実装は何でも良い
  - シングルスレッド、マルチスレッド、Actor モデル、、、


***
#RxJava

***
#RxJava
- Reactive X の JVM 向けライブラリ
- Java で実装されている
- 1.x と 2.x で全然違う
  - 2.x を中心に解説

***
# 準備
- gradle に次を設定

```
compile "io.reactivex.rxjava2:rxjava:2.0.4"
```

***
# サンプルコード
- Pure Java 
- github: TODO

***
# Observable の生成
```java
// 配列を元にして Observable を生成
Integer[] numbers = {1, 2, 3, 4, 5};
Observable<Integer> observable 
    = Observable.fromArray(numbers);
```

⇒ Observable に factory メソッドがたくさんある

***
# subscribe

```java
// Observables を subscribe する
observable.subscribe(new Consumer<Integer>() {
    @Override 
    public void accept(Integer integer) throws Exception {
        System.out.println(integer);
    }
});
```

***
# 実行すると
```
1
2
3
4
5
```

***
# 単一 Item の Observable

```java
// 単一アイテムの Observable
Observable<Integer> observable = Observable.just(1);
```

⇒ 2.x では、Single を使った方がよい？（要確認）

***
# Java 8 のラムダを利用

```java
// Java 8 のラムダ構文
observable.subscribe(integer -> {
    System.out.println(integer);
});
```

メソッド参照をつかうとさらに簡単に

```java
observable.subscribe(System.out::println);
```


***
# 非同期処理
```java
// 5秒後に 1 を生成する Observable
Future<Integer> future = Executors.newSingleThreadExecutor().submit(() -> {
    // バックグラウンドスレッドで実行
    Thread.sleep(5000);
    return 1;
});

Observable<Integer> observable = Observable.fromFuture(future);
observable.subscribe(System.out::println); // 終了しないので注意
```

***
# Operators

***
# コレクション操作
- ReactiveX はコレクション操作が重要
 - ReactiveX ⇒ __Observables__ が中心
 - Observables ⇒ 非同期 × __複数データ__
 - 複数データ ⇒ __コレクション__
- 便利なコレクションメソッドを用意している

***
# 例：
> データ列の11~15個目の要素を "_N_ transformed" という形式で順に出力

*** 
# 同期的な手法
`Iterable<T>` の場合：

```java
Iterable<Integer> iter = getDataFromLocal();
iter.skip(10)
  .take(5) 
  .map(s -> s + "transformed") 
  .forEach(s -> System.out.println(s));
```

***
# 非同期的な手法
`Observable<T>` の場合：

```java
Observable<Integer> observable = getDataFromNetwork();
observable.skip(10)
  .take(5) 
  .map(i -> i + "transformed") 
  .subscribe(s -> System.out.println(s));
```

***
# Marble Diagrams
- メソッドの効果説明に使われる図

<img src="http://reactivex.io/assets/operators/legend.png">

ref: http://reactivex.io/documentation/observable.html

***
# Pull vs Push
- Pull
  - 同期的手法は Pull
  - データを消費する側が、データを __Pull__ する
-  Push
  - 非同期的手法は Push
  - データを生成する側が、データを __Push__ する

***
# すべてを Push に

ReactiveX 創始者 Erik Meijer:

> 世界の最先端にい続けたいのなら，“push”すること，つまりリアクティブであることが必要だ，と氏は言う。“pull”を使うというのは，基本的にはプログラムに，あらゆる種類のブロックポイントを加えるということであるから非効率的だ，と氏は言う。

ref: https://www.infoq.com/jp/news/2015/12/erik-meijer-hacker-way

***
# 次回
- Android で ReactiveX を使う？