RxManager.java 文件源码

private Observable<ManagedState> getState(final Id id) {
    return Observable.create(new Observable.OnSubscribe<ManagedState>() {
        @Override
        public void call(final Subscriber<? super ManagedState> subscriber) {
            final ManagedState state = state(id, true);

            subscriber.add(BooleanSubscription.create(new Action0() {
                @Override
                public void call() {
                    state.removeSubscriber(subscriber);
                }
            }));

            int publishCount = state.publishCount;
            state.addSubscriber(subscriber);
            // check to avoid double-publishing
            if (publishCount == state.publishCount) {
                subscriber.onNext(state);
            }
        }
    });
}