maven { url 'https://jitpack.io' }

compile 'com.github.senierr:SeTask:1.0.2'

Observable observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(Emitter<String> emitter) throws Exception {
        emitter.isCancel();                         // 判断订阅是否终止
        emitter.onProcess("value");                 // 触发处理回调，不会终止订阅
        emitter.onComplete();                       // 触发结束/完成回调，并终止订阅
        emitter.onError(new Exception("Error"));    // 触发异常回调，并终止订阅
    }
})
        .bindToObservatory(observatory)     // 绑定至观察站
        .subscribe(new Observer<String>() {
            @Override
            public void onProcess(String value) {
                // 处理回调
            }

            @Override
            public void onComplete() {
                // 结束/完成回调
            }

            @Override
            public void onError(Exception e) {
                // 异常回调
            }
        });

Observable.interval(delay, period, timeUnit)
        .bindToObservatory(observatory)
        .subscribe(new Observer<Long>() {
            @Override
            public void onProcess(Long value) {
                log("onProcess: " + value);
            }
        });

Observable.delay(delay, timeUnit)
        .bindToObservatory(observatory)
        .subscribe(new Observer<Long>() {
            @Override
            public void onComplete() {
                log("onComplete");
            }
        });

observable.cancel();

// 创建观察站
Observatory observatory = new Observatory();
// 绑定至观察站
observatory.bindToObservatory(observatory)
// 批量取消订阅
observatory.cancelAll();