TappyStatusServiceImpl.java 文件源码

public TappyStatusServiceImpl(Scheduler ioScheduler) {
    this.ioScheduler = ioScheduler;
    tappyStatusObservable = Observable.create(new Observable.OnSubscribe<Map<String,Integer>>() {

        @Override
        public void call(final Subscriber<? super Map<String,Integer>> subscriber) {
            final TappyStatusListener listener = new TappyStatusListener() {
                public void onUpdatedTappyStatusMap(Map<String,Integer> statusMap) {
                    subscriber.onNext(statusMap);
                }
            };

            TappyStatusServiceImpl.this.registerListener(listener);

            subscriber.add(Subscriptions.create(new Action0() {
                @Override
                public void call() {
                    TappyStatusServiceImpl.this.unregisterListener(listener);
                }
            }));

            subscriber.onNext(getCurrentStatusMapSync());
        }
    }).subscribeOn(ioScheduler);
}