package org.ntlab.developrx;
import android.util.Pair;
import org.ntlab.developrx.utils.ThreadUtil;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.processors.PublishProcessor;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.subjects.PublishSubject;
/**
* Created by matsumoto_k on 2017/11/11.
*/
public class JavaRxProcess {
private PublishSubject<Integer> publishSubject = null;
/**
* Flowable (Reactive Streams対応)
*/
public void flowableReactiveStreams() {
Flowable<String> flowable = Flowable.create(emitter -> {
String[] datas = {"データ1", "データ2", "データ3", "データ4", "データ5"};
for (String data : datas) {
//購読解除されている場合は処理を止める
if (emitter.isCancelled()) {
return;
}
// データを通知する
emitter.onNext(data);
}
// 完了したことを通知
emitter.onComplete();
}, BackpressureStrategy.BUFFER);
flowable
// Subscriberの処理を別のスレッドで行うようにする
.observeOn(Schedulers.computation())
.subscribe(new Subscriber<String>() {
// データ数のリクエストおよび購読の解除を行うオブジェクト
Subscription subscription;
// 購読が開始された際の処理
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
// 受け取るデータ数をリクエストする
// Long.MAX_VALUEをrequestした場合、データ数を制限すること無くデータを通知する
this.subscription.request(1L);
}
// データを受け取った際の処理
@Override
public void onNext(String data) {
System.out.println(ThreadUtil.getCurrentThreadName() + ": " + data);
// 次に受け取るデータ数をリクエストする
this.subscription.request(1L);
}
// 完了を通知された際の処理
@Override
public void onComplete() {
System.out.println(ThreadUtil.getCurrentThreadName() + ": 完了しました");
}
// エラーを通知された際の処理
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
});
}
/**
* Observable (バックプレッシャー機能なし)
*/
public void observable() {
Observable<String> observable = Observable.create(emitter -> {
String[] datas = {"データ1", "データ2", "データ3", "データ4", "データ5"};
for (String data : datas) {
if (emitter.isDisposed()) {
return;
}
emitter.onNext(data);
}
emitter.onComplete();
});
observable.observeOn(Schedulers.computation())
.subscribe(
new Observer<String>() {
@Override
public void onSubscribe(Disposable disposable) {
// 何もしない
}
@Override
public void onNext(String item) {
System.out.println(ThreadUtil.getCurrentThreadName() + ": " + item);
}
@Override
public void onError(Throwable e) {
e.printStackTrace();
}
@Override
public void onComplete() {
System.out.println(ThreadUtil.getCurrentThreadName() + ": 完了しました");
}
}
);
}
/**
* 引数のデータを通知するFLowableを作成
* 引数は10個まで
*/
public void just() {
Flowable<String> flowable = Flowable.just("A", "B", "C", "D", "E", "F", "G", "H", "I", "J");
flowable.subscribe(new DebugSubscriber<>());
}
/**
* 引数のデータを通知するFlowableを作成
*/
public void fromArray() {
Flowable<String> flowable = Flowable.fromArray("A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K");
flowable.subscribe(new DebugSubscriber<>());
}
/**
* startから順にcount数のデータを通知する
*/
public void range() {
Flowable<Integer> flowable = Flowable.range(10, 3);
flowable.subscribe(new DebugSubscriber<>());
}
/**
* 指定時間ごとに通知を送る
*/
public void interval() {
Flowable<Long> flowable = Flowable.interval(1000L, TimeUnit.MILLISECONDS).take(5);
flowable.subscribe(data -> System.out.println(ThreadUtil.getCurrentThreadName() + ": " + System.currentTimeMillis() + ": " + "data = " + data));
}
/**
* 指定時間後に0を通知する
*/
public void timer() {
Flowable<Long> flowable = Flowable.timer(1000, TimeUnit.MILLISECONDS);
flowable.subscribe(
data -> System.out.println(ThreadUtil.getCurrentThreadName() + ": data = " + data),
error -> System.out.println("エラー = " + error),
() -> System.out.println("完了")
);
}
/**
* エラーのみを通知する
*/
public void error() {
Flowable.error(new Exception("例外発生")).subscribe(new DebugSubscriber<>());
}
/**
* データを変換して通知
*/
public void map() {
Flowable<String> flowable = Flowable.just("A", "B", "C", "D", "E")
.map(data -> data.toLowerCase());
flowable.subscribe(new DebugSubscriber<>());
}
/**
* からのFlowable/Observableを生成する
*/
public void empy() {
Flowable.empty().subscribe(new DebugSubscriber<>());
}
/**
* 受け取ったデータをFlowable/Observableに変換し、そのFlowable/Obsevableが持つデータを通知する
*/
public void flatmap1() {
Flowable<String> flowable = Flowable.just("A", "", "B", "", "C")
.flatMap(data -> {
if (data.isEmpty()) {
return Flowable.empty();
} else {
return Flowable.just(data.toUpperCase());
}
});
flowable.subscribe(new DebugSubscriber<>());
}
public void flatmap2() {
Flowable<String> flowable = Flowable.range(1, 3)
.flatMap(
data -> Flowable.interval(100L, TimeUnit.MILLISECONDS).take(3),
(sourceData, newData) -> "[" + sourceData + "] " + newData
);
flowable.subscribe(new DebugSubscriber<>());
}
public void meetingTest() {
PublishProcessor<Long> processor1 = PublishProcessor.create();
PublishProcessor<Long> processor2 = PublishProcessor.create();
Flowable<Pair<Long, Long>> flowable = PublishProcessor.combineLatest(processor1, processor2, (time1, time2) -> new Pair<Long, Long>(time1, time2));
flowable.subscribe(pair -> System.out.println(new Date(pair.first) + " : " + new Date(pair.second)));
new Thread(() -> {
try {
Thread.sleep(1000);
processor1.onNext(System.currentTimeMillis());
Thread.sleep(3000);
processor1.onNext(System.currentTimeMillis());
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
new Thread(() -> {
try {
Thread.sleep(2000);
processor2.onNext(System.currentTimeMillis());
Thread.sleep(1000);
processor2.onNext(System.currentTimeMillis());
} catch (InterruptedException e) {
e.printStackTrace();
}
}).start();
}
public PublishSubject<Integer> getPublishSubject() {
if (publishSubject == null) {
publishSubject = PublishSubject.create();
}
return publishSubject;
}
}
