add combinlateset

Author: Carson-Ho

app/src/main/java/scut/carson_ho/rxjava_operators/MainActivity.java

@@ -4,8 +4,9 @@
 import android.support.v7.app.AppCompatActivity;
 import android.util.Log;
 
+import java.util.concurrent.TimeUnit;
+
 import io.reactivex.Observable;
-import io.reactivex.annotations.NonNull;
 import io.reactivex.functions.BiFunction;
 import io.reactivex.functions.Consumer;
 
@@ -19,23 +20,62 @@ protected void onCreate(Bundle savedInstanceState) {
         super.onCreate(savedInstanceState);
         setContentView(R.layout.activity_main);
 
-        Observable.just(1,2,3,4)
-                .reduce(new BiFunction<Integer, Integer, Integer>() {
-                    // 在该复写方法中复写聚合的逻辑
-                    @Override
-                    public Integer apply(@NonNull Integer s1, @NonNull Integer s2) throws Exception {
-                        Log.e(TAG, "本次计算的数据是： "+s1 +" 乘 "+ s2);
-                        return s1 * s2;
-                        // 本次聚合的逻辑是：全部数据相乘起来
-                        // 原理：第1次取前2个数据相乘，之后每次获取到的数据 = 返回的数据x原始下1个数据每
-                    }
-                }).subscribe(new Consumer<Integer>() {
-            @Override
-            public void accept(@NonNull Integer s) throws Exception {
-                Log.e(TAG, "最终计算的结果是： "+s);
-
-            }
-        });
+
+
+            Observable.combineLatest(
+                    Observable.intervalRange(0, 3, 0, 2, TimeUnit.SECONDS), // 第1个发送数据事件的Observable
+                    Observable.intervalRange(0, 3, 0, 1, TimeUnit.SECONDS), // 第2个发送数据事件的Observable：从0开始发送、共发送3个数据、第1次事件延迟发送时间 = 1s、间隔时间 = 1s
+                    new BiFunction<Long, Long, Long>() {
+                @Override
+                public Long apply(Long o1, Long o2) throws Exception {
+                    // o1 = 第1个Observable发送的最新（最后）1个数据
+                    // o2 = 第2个Observable发送的每1个数据
+                    Log.e(TAG, "合并的数据是： "+ o1 + " "+ o2);
+                    return o1 + o2;
+                    // 合并的逻辑 = 相加
+                    // 即第1个Observable发送的最后1个数据 与 第2个Observable发送的每1个数据进行相加
+                }
+            }).subscribe(new Consumer<Long>() {
+                @Override
+                public void accept(Long s) throws Exception {
+                    Log.e(TAG, "合并的结果是： "+s);
+                }
+            });
+
+//
+//        Observable.concatArrayDelayError(
+//                Observable.create(new ObservableOnSubscribe<Integer>() {
+//                    @Override
+//                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
+//
+//                        emitter.onNext(1);
+//                        emitter.onNext(2);
+//                        emitter.onNext(3);
+//                        emitter.onError(new NullPointerException()); // 发送Error事件，因为使用了concatDelayError，所以第2个Observable将会发送事件，等发送完毕后，再发送错误事件
+//                        emitter.onComplete();
+//                    }
+//                }),
+//                Observable.just(4, 5, 6))
+//                .subscribe(new Observer<Integer>() {
+//                    @Override
+//                    public void onSubscribe(Disposable d) {
+//
+//                    }
+//                    @Override
+//                    public void onNext(Integer value) {
+//                        Log.d(TAG, "接收到了事件"+ value  );
+//                    }
+//
+//                    @Override
+//                    public void onError(Throwable e) {
+//                        Log.d(TAG, "对Error事件作出响应");
+//                    }
+//
+//                    @Override
+//                    public void onComplete() {
+//                        Log.d(TAG, "对Complete事件作出响应");
+//                    }
+//                });
 
         // 在一个被观察者发送事件前，追加发送被观察者 & 发送数据
         // 注：追加数据顺序 = 后调用先追加