ConnectionOperationQueueImpl.java 文件源码

@Override
@RestrictTo(RestrictTo.Scope.LIBRARY_GROUP)
public synchronized <T> Observable<T> queue(final Operation<T> operation) {
    if (!shouldRun) {
        return Observable.error(disconnectionException);
    }
    return Observable.create(new Action1<Emitter<T>>() {
        @Override
        public void call(Emitter<T> tEmitter) {
            final FIFORunnableEntry entry = new FIFORunnableEntry<>(operation, tEmitter);

            tEmitter.setCancellation(new Cancellable() {
                @Override
                public void cancel() throws Exception {
                    if (queue.remove(entry)) {
                        logOperationRemoved(operation);
                    }
                }
            });

            logOperationQueued(operation);
            queue.add(entry);
        }
    }, Emitter.BackpressureMode.NONE);
}