io.reactivex.FlowableSubscriber#io.reactivex.exceptions.Exceptions源码实例Demo

@Override
public void onNext(Query<T> query) {
  try {
    // null cursor here is valid
    final Cursor cursor = query.rawQuery(true);
    final T item = query.map(cursor);
    if (!isDisposed()) {
      if (item != null) {
        downstream.onNext(item);
      } else if (defaultValue != null) {
        downstream.onNext(defaultValue);
      }
    }
  } catch (Throwable e) {
    Exceptions.throwIfFatal(e);
    onError(e);
  }
}
 

@Override
public void subscribe(@NonNull ObservableEmitter<T> subscriber) throws Exception {
    try {
        T data = execute();
        if (!subscriber.isDisposed()) {
            subscriber.onNext(data);
        }
    } catch (Throwable e) {
        HttpLog.e(e.getMessage());
        if (!subscriber.isDisposed()) {
            subscriber.onError(e);
        }
        Exceptions.throwIfFatal(e);
        //RxJavaPlugins.onError(e);
        return;
    }

    if (!subscriber.isDisposed()) {
        subscriber.onComplete();
    }
}
 

/**
 * Converts an {@link ApolloQueryWatcher} to an asynchronous Observable.
 *
 * @param watcher the ApolloQueryWatcher to convert.
 * @param <T>     the value type
 * @return the converted Observable
 * @throws NullPointerException if watcher == null
 */
@NotNull
@CheckReturnValue
public static <T> Observable<Response<T>> from(@NotNull final ApolloQueryWatcher<T> watcher) {
  checkNotNull(watcher, "watcher == null");
  return Observable.create(new ObservableOnSubscribe<Response<T>>() {
    @Override public void subscribe(final ObservableEmitter<Response<T>> emitter) throws Exception {
      cancelOnObservableDisposed(emitter, watcher);

      watcher.enqueueAndWatch(new ApolloCall.Callback<T>() {
        @Override public void onResponse(@NotNull Response<T> response) {
          if (!emitter.isDisposed()) {
            emitter.onNext(response);
          }
        }

        @Override public void onFailure(@NotNull ApolloException e) {
          Exceptions.throwIfFatal(e);
          if (!emitter.isDisposed()) {
            emitter.onError(e);
          }
        }
      });
    }
  });
}
 

private void processGetTransferToId(final String strAccount, final TextView textViewTo) {
    Flowable.just(strAccount)
            .subscribeOn(Schedulers.io())
            .map(accountName -> {
                account_object accountObject = BitsharesWalletWraper.getInstance().get_account_object(accountName);
                if (accountObject == null) {
                    throw new ErrorCodeException(ErrorCode.ERROR_NO_ACCOUNT_OBJECT, "it can't find the account");
                }

                return accountObject;
            }).observeOn(AndroidSchedulers.mainThread())
            .subscribe(accountObject -> {
                if (getActivity() != null && getActivity().isFinishing() == false) {
                    textViewTo.setText("#" + accountObject.id.get_instance());
                }
            }, throwable -> {
                if (throwable instanceof NetworkStatusException || throwable instanceof ErrorCodeException) {
                    if (getActivity() != null && getActivity().isFinishing() == false) {
                        textViewTo.setText("#none");
                    }
                } else {
                    throw Exceptions.propagate(throwable);
                }
            });
}
 

@Test
public void bodyThrowingInOnSuccessDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingMaybeObserver<String> observer = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.body().subscribe(new ForwardingObserver<String>(observer) {
        @Override
        public void onSuccess(String value) {
            throw e;
        }
    });

    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void responseThrowingInOnSuccessDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingMaybeObserver<Response<String>> observer = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.response().subscribe(new ForwardingObserver<Response<String>>(observer) {
        @Override
        public void onSuccess(Response<String> value) {
            throw e;
        }
    });

    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void resultThrowingInOnSuccessDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingMaybeObserver<Result<String>> observer = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.result().subscribe(new ForwardingObserver<Result<String>>(observer) {
        @Override
        public void onSuccess(Result<String> value) {
            throw e;
        }
    });

    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void throwingInOnCompleteDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> errorRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!errorRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable); // Don't swallow secondary errors!
            }
        }
    });

    RecordingCompletableObserver observer = observerRule.create();
    final RuntimeException e = new RuntimeException();
    service.completable().subscribe(new ForwardingCompletableObserver(observer) {
        @Override
        public void onComplete() {
            throw e;
        }
    });

    assertThat(errorRef.get()).isSameAs(e);
}
 

@Test
public void responseThrowingInOnCompleteDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingObserver<Response<String>> observer = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.response().subscribe(new ForwardingObserver<Response<String>>(observer) {
        @Override
        public void onComplete() {
            throw e;
        }
    });

    observer.assertAnyValue();
    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void resultThrowingInOnCompletedDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingObserver<Result<String>> observer = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.result().subscribe(new ForwardingObserver<Result<String>>(observer) {
        @Override
        public void onComplete() {
            throw e;
        }
    });

    observer.assertAnyValue();
    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void bodyThrowingInOnSuccessDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingSingleObserver<String> observer = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.body().subscribe(new ForwardingObserver<String>(observer) {
        @Override
        public void onSuccess(String value) {
            throw e;
        }
    });

    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void responseThrowingInOnSuccessDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingSingleObserver<Response<String>> observer = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.response().subscribe(new ForwardingObserver<Response<String>>(observer) {
        @Override
        public void onSuccess(Response<String> value) {
            throw e;
        }
    });

    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void resultThrowingInOnSuccessDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingSingleObserver<Result<String>> observer = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.result().subscribe(new ForwardingObserver<Result<String>>(observer) {
        @Override
        public void onSuccess(Result<String> value) {
            throw e;
        }
    });

    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void responseThrowingInOnCompleteDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.response().subscribe(new ForwardingSubscriber<Response<String>>(subscriber) {
        @Override
        public void onComplete() {
            throw e;
        }
    });

    subscriber.assertAnyValue();
    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Test
public void resultThrowingInOnCompletedDeliveredToPlugin() {
    server.enqueue(new MockResponse());

    final AtomicReference<Throwable> throwableRef = new AtomicReference<>();
    RxJavaPlugins.setErrorHandler(new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) throws Exception {
            if (!throwableRef.compareAndSet(null, throwable)) {
                throw Exceptions.propagate(throwable);
            }
        }
    });

    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
    final RuntimeException e = new RuntimeException();
    service.result().subscribe(new ForwardingSubscriber<Result<String>>(subscriber) {
        @Override
        public void onComplete() {
            throw e;
        }
    });

    subscriber.assertAnyValue();
    assertThat(throwableRef.get()).isSameAs(e);
}
 

@Override
public void onNext(final T t) {
    if (finished) {
        return;
    }
    queue.offer(t);
    final long waitTime;
    try {
        waitTime = timeout.apply(t);
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // we cancel upstream ourselves because the
        // error did not originate from source
        upstream.cancel();
        onError(e);
        return;
    }
    TimeoutAction<T> action = new TimeoutAction<T>(this, t);
    Disposable d = worker.schedule(action, waitTime, unit);
    DisposableHelper.set(scheduled, d);
    drain();
}
 

@Override
public void onComplete() {
    if (done) {
        return;
    }
    if (empty) {
        try {
            onEmpty.run();
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            onError(e);
            return;
        }
    }
    done = true;
    child.onComplete();
}
 

@Override
public void onComplete() {
    if (done) {
        return;
    }
    if (value != EMPTY) {
        T value2;
        try {
            value2 = function.apply(value);
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            parent.cancel();
            onError(e);
            return;
        }
        actual.onNext(value2);
    }
    done = true;
    actual.onComplete();
}
 

@Override
protected void subscribeActual(Subscriber<? super T> child) {

    Flowable<T> f;
    try {
        f = transform.apply(source);
    } catch (Exception e) {
        Exceptions.throwIfFatal(e);
        child.onSubscribe(SubscriptionHelper.CANCELLED);
        child.onError(e);
        return;
    }
    AtomicReference<Chain<T>> chainRef = new AtomicReference<Chain<T>>();
    DestinationSerializedSubject<T> destination = new DestinationSerializedSubject<T>(child,
            chainRef);
    Chain<T> chain = new Chain<T>(transform, destination, maxIterations, maxChained, tester);
    chainRef.set(chain);
    // destination is not initially subscribed to the chain but will be when
    // tester function result completes
    destination.subscribe(child);
    ChainedReplaySubject<T> sub = ChainedReplaySubject.create(destination, chain, tester);
    chain.initialize(sub);
    f.onTerminateDetach() //
            .subscribe(sub);
}
 

@Override
public void onNext(T t) {
    if (finished) {
        return;
    }
    queue.offer(t);
    Maybe<? extends T> maybe;
    try {
        maybe = valueToInsert.apply(t);
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // we cancel upstream ourselves because the
        // error did not originate from source
        upstream.cancel();
        onError(e);
        return;
    }
    ValueToInsertObserver<T> o = new ValueToInsertObserver<T>(this);
    if (DisposableHelper.set(valueToInsertObserver, o)) {
        // note that at this point we have to cover o being disposed
        // from another thread so the Observer class needs
        // to handle dispose being called before/during onSubscribe
        maybe.subscribe(o);
    }
    drain();
}
 

@Override
public void onNext(In t) {
    if (done) {
        return;
    }
    if (!createdState()) {
        return;
    }
    if (--count == 0) {
        requestsArrived = true;
        count = requestBatchSize;
    }
    try {
        drainCalled = false;
        state = ObjectHelper.requireNonNull(transition.apply(state, t, this),
                "intermediate state cannot be null");
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        onError(e);
        return;
    }
    if (!drainCalled) {
        drain();
    }
}
 

private boolean createdState() {
    if (state == null) {
        try {
            state = ObjectHelper.requireNonNull(initialState.call(),
                    "initial state cannot be null");
            return true;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            done = true;
            onError_(e);
            return false;
        }
    } else {
        return true;
    }
}
 

@Override
public void onComplete() {
    if (done) {
        return;
    }
    if (!createdState()) {
        return;
    }
    try {
        if (completionAction != null) {
            completionAction.accept(state, this);
        } else {
            onComplete_();
        }
        done = true;
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        onError(e);
        return;
    }
}
 

/**
 * Close GRPC connection.
 */
public void terminate() {
  if (!channel.isTerminated()) {
    channel.shutdownNow();
    boolean terminated = false;
    do {
      try {
        terminated = channel.awaitTermination(30, TimeUnit.SECONDS);
      } catch (InterruptedException e) {
        Exceptions.propagate(e);
      }
    } while (!terminated);
  }
}
 

@Override
public void onError(Throwable t) {
  try {
    this.onError.accept(t);
  } catch (Exception e) {
    Exceptions.propagate(e);
  }
}
 

@Override
public void onSubscribe(Subscription s) {
    if (SubscriptionHelper.setOnce(this, s)) {
        try {
            addObserver();
            downstream.onSubscribe(s);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            s.cancel();
            onError(ex);
        }
    }
}
 

@Override
public void onNext(T t) {
    if (isDisposed()) return;
    try {
        downstream.onNext(t);
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        get().cancel();
        onError(e);
    }
}
 

@Override
public void onError(Throwable t) {
    if (isDisposed()) {
        RxJavaPlugins.onError(t);
        return;
    }
    lazySet(SubscriptionHelper.CANCELLED);
    try {
        removeObserver();
        downstream.onError(t);
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        RxJavaPlugins.onError(new CompositeException(t, e));
    }
}
 

@Override
public void onComplete() {
    if (isDisposed()) return;
    lazySet(SubscriptionHelper.CANCELLED);
    try {
        removeObserver();
        downstream.onComplete();
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        RxJavaPlugins.onError(e);
    }
}
 

@Override
public void onSubscribe(Disposable d) {
    if (DisposableHelper.setOnce(this, d)) {
        try {
            addObserver();
            downstream.onSubscribe(d);
        } catch (Throwable ex) {
            Exceptions.throwIfFatal(ex);
            d.dispose();
            onError(ex);
        }
    }
}