RxJava/RxAndroid:ConnectableObservable & replay(long time, TimeUnit unit)

 

import android.support.v7.app.AppCompatActivity;
import android.os.Bundle;
import android.util.Log;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.observables.ConnectableObservable;
import io.reactivex.observers.DisposableObserver;

public class MainActivity extends AppCompatActivity {
    private String TAG = "输出";

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        ConnectableObservable<Long> mConnectableObservable = Observable
                .interval(1, TimeUnit.SECONDS)
                .replay(5, TimeUnit.SECONDS);

        mConnectableObservable
                .subscribe(new DisposableObserver<Long>() {
                    @Override
                    public void onNext(Long along) {
                        Log.d(TAG, "onNext1:" + along);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete1");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, e.toString());
                    }
                });

        mConnectableObservable.connect();

        mConnectableObservable
                .delaySubscription(5, TimeUnit.SECONDS)
                .subscribe(new DisposableObserver<Long>() {
                    @Override
                    public void onNext(Long along) {
                        Log.d(TAG, "onNext2:" + along);
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "onComplete2");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, e.toString());
                    }
                });
    }
}

output:

 

01-26 16:11:58.031 19725-19748/zhangphil.book D/输出: onNext1:0
01-26 16:11:59.030 19725-19748/zhangphil.book D/输出: onNext1:1
01-26 16:12:00.030 19725-19748/zhangphil.book D/输出: onNext1:2
01-26 16:12:01.030 19725-19748/zhangphil.book D/输出: onNext1:3
01-26 16:12:02.030 19725-19748/zhangphil.book D/输出: onNext1:4
01-26 16:12:02.032 19725-19749/zhangphil.book D/输出: onNext2:0
01-26 16:12:02.032 19725-19749/zhangphil.book D/输出: onNext2:1
01-26 16:12:02.032 19725-19749/zhangphil.book D/输出: onNext2:2
01-26 16:12:02.032 19725-19749/zhangphil.book D/输出: onNext2:3
01-26 16:12:02.032 19725-19749/zhangphil.book D/输出: onNext2:4
01-26 16:12:03.030 19725-19748/zhangphil.book D/输出: onNext1:5
01-26 16:12:03.030 19725-19748/zhangphil.book D/输出: onNext2:5
01-26 16:12:04.030 19725-19748/zhangphil.book D/输出: onNext1:6
01-26 16:12:04.030 19725-19748/zhangphil.book D/输出: onNext2:6
01-26 16:12:05.030 19725-19748/zhangphil.book D/输出: onNext1:7
01-26 16:12:05.030 19725-19748/zhangphil.book D/输出: onNext2:7
01-26 16:12:06.030 19725-19748/zhangphil.book D/输出: onNext1:8
01-26 16:12:06.030 19725-19748/zhangphil.book D/输出: onNext2:8
01-26 16:12:07.030 19725-19748/zhangphil.book D/输出: onNext1:9
01-26 16:12:07.030 19725-19748/zhangphil.book D/输出: onNext2:9
01-26 16:12:08.030 19725-19748/zhangphil.book D/输出: onNext1:10
01-26 16:12:08.030 19725-19748/zhangphil.book D/输出: onNext2:10
01-26 16:12:09.030 19725-19748/zhangphil.book D/输出: onNext1:11
01-26 16:12:09.031 19725-19748/zhangphil.book D/输出: onNext2:11
01-26 16:12:10.030 19725-19748/zhangphil.book D/输出: onNext1:12
01-26 16:12:10.031 19725-19748/zhangphil.book D/输出: onNext2:12


......