HotObservable.java 文件源码

public void call(final Subscriber<? super T> subscriber) {
    final SubscriptionLog subscriptionLog = new SubscriptionLog(scheduler.now());
    observable.subscriptions.add(subscriptionLog);
    final int subscriptionIndex = observable.getSubscriptions().size() - 1;

    subscribers.add(subscriber);

    subscriber.add((Subscriptions.create(new Action0() {
        @Override
        public void call() {
            // on unsubscribe
            observable.subscriptions.set(
                    subscriptionIndex,
                    new SubscriptionLog(subscriptionLog.subscribe, scheduler.now())
            );
            subscribers.remove(subscriber);
        }
    })));
}