java类io.reactivex.CompletableObserver的实例源码

private CompletableObserver getCompletableObserver() {
    return new CompletableObserver() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, " onSubscribe : " + d.isDisposed());
        }

        @Override
        public void onComplete() {
            textView.append(" onComplete");
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onComplete");
        }

        @Override
        public void onError(Throwable e) {
            textView.append(" onError : " + e.getMessage());
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onError : " + e.getMessage());
        }
    };
}

@Override
protected void subscribeActual(final CompletableObserver completableObserver) {
    verifyMainThread();
    final View view = viewWeakRef.get();
    if (view == null) {
        completableObserver.onComplete();
        return;
    }

    final ViewPropertyAnimatorCompat animator = ViewCompat.animate(view);

    completableObserver.onSubscribe(createAnimationDisposable(animator, onAnimationCancelAction));

    if (preTransformActions != null) {
        applyActions(preTransformActions, animator);
        animator.setDuration(NONE).setStartDelay(NONE)
                .withEndAction(() -> runAnimation(completableObserver, animator))
                .start();
    } else {
        runAnimation(completableObserver, animator);
    }
}

@Override public void onDeleteCheckPoint(int adapterPosition) {
  getCheckPointDataSource().deleteCheckPoint(allCheckPoints.get(adapterPosition))
      .subscribeOn(Schedulers.io())
      .observeOn(AndroidSchedulers.mainThread())
      .subscribeWith(new CompletableObserver() {
        @Override public void onSubscribe(Disposable d) {

        }

        @Override public void onComplete() {
          allCheckPoints.remove(adapterPosition);
          getView().removeMarker(adapterPosition);
          getView().notifyListAdapter();
        }

        @Override public void onError(Throwable e) {
          getView().showError("Delete Failed");
        }
      });
}

@Override
protected void subscribeActual(CompletableObserver s) {
    CompletableObserver observer;
    try {
        observer = ObjectHelper.requireNonNull(s, "Null Observer");
    } catch (Throwable e) {
        Exceptions.throwIfFatal(e);
        // can't call onError because no way to know if a Disposable has been set or not
        // can't call onSubscribe because the call might have set a Disposable already
        RxJavaPlugins.onError(e);

        NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
        npe.initCause(e);
        throw npe;
    }

    source.subscribe(new AttachCompletableObserver(observer, this.compositeDisposable));
}

@Override
protected void subscribeActual(CompletableObserver observer) {
    boolean b;

    try {
        b = condition.getAsBoolean();
    } catch (Throwable ex) {
        EmptyDisposable.error(ex, observer);
        return;
    }

    if (b) {
        then.subscribe(observer);
    } else {
        orElse.subscribe(observer);
    }
}

private CompletableObserver getCompletableObserver() {
    return new CompletableObserver() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, " onSubscribe : " + d.isDisposed());
        }

        @Override
        public void onComplete() {
            textView.append(" onComplete");
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onComplete");
        }

        @Override
        public void onError(Throwable e) {
            textView.append(" onError : " + e.getMessage());
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onError : " + e.getMessage());
        }
    };
}

private CompletableObserver getObserver() {
    return new CompletableObserver() {

        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, " onSubscribe : " + d.isDisposed());
        }

        @Override
        public void onError(Throwable e) {
            textView.append(" onError : " + e.getMessage());
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onError : " + e.getMessage());
        }

        @Override
        public void onComplete() {
            textView.append(" onComplete");
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onComplete");
        }
    };
}

@Override
public CompletableObserver apply(CompletableObserver observer) throws Exception {
    return new DisposableCompletableObserver() {
        @Override
        public void onComplete() {
            if (!isDisposed()) {
                observer.onError(throwable);
            }
        }

        @Override
        public void onError(Throwable ex) {
            if (!isDisposed()) {
                observer.onError(ex);
            }
        }
    };
}

protected CompletableObserver failOrCountDownCompletableObserver(CountDownLatch latch) {
    return new CompletableObserver() {

        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onComplete() {
            latch.countDown();
        }

        @Override
        public void onError(Throwable e) {
            fail(e);
        }

    };
}

/**
 * Performs the opposite translation from
 * {@link MaybeConsumers#fromCompletableObserver(CompletableObserver)}
 */
public static CompletableObserver toCompletableObserver(final MaybeConsumer<Success> c) {
    return new CompletableObserver() {
        @Override
        public void onSubscribe(@NonNull Disposable d) {
            // do nothing
        }

        @Override
        public void onComplete() {
            c.success(Success.SUCCESS);
        }

        @Override
        public void onError(@NonNull Throwable e) {
            c.fail(throwableToException(e));
        }
    };
}

public static CompletableObserver observe(String tag, String operation) {
    return new CompletableObserver() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onComplete() {

        }

        @Override
        public void onError(Throwable e) {
            complain(e, tag, operation);
        }
    };
}

@Test
public void decks_renamed() {
    List<Deck> renamedData = mUser.save().andThen((final CompletableObserver cs) -> {
        Deck deck = new Deck(mUser);
        deck.setName("ToRename");
        deck.setAccepted(true);
        deck.create().subscribe(cs);
    }).andThen((ObservableSource<List<Deck>>) observer ->
            mUser.fetchChildren(mUser.getChildReference(Deck.class), Deck.class)
                    .subscribe(observer)
    ).firstOrError().flatMapCompletable(data -> {
        assertTrue(data.size() == 1 && data.get(0).getName().equals("ToRename"));
        Deck fetchedDeck = data.get(0);
        fetchedDeck.setName("Renamed");
        return fetchedDeck.save();
    }).andThen((ObservableSource<List<Deck>>) observer ->
            mUser.fetchChildren(mUser.getChildReference(Deck.class), Deck.class)
                    .subscribe(observer)
    ).firstOrError().blockingGet();
    assertTrue(renamedData.size() == 1 && renamedData.get(0).getName()
            .equals("Renamed"));
}

@Test
public void decks_changedDeckType() {
    List<Deck> deckTypeData = mUser.save().andThen((final CompletableObserver cs) -> {
        Deck deck = new Deck(mUser);
        deck.setName("DeckType");
        deck.setAccepted(true);
        deck.create().subscribe(cs);
    }).andThen((ObservableSource<List<Deck>>) observer ->
            mUser.fetchChildren(mUser.getChildReference(Deck.class), Deck.class)
                    .subscribe(observer)
    ).firstOrError().flatMapCompletable(data -> {
        assertTrue(data.size() == 1 && data.get(0).getName().equals("DeckType"));
        Deck fetchedDeck = data.get(0);
        fetchedDeck.setDeckType(DeckType.SWISS.name());
        return fetchedDeck.save();
    }).andThen((ObservableSource<List<Deck>>) observer ->
            mUser.fetchChildren(mUser.getChildReference(Deck.class), Deck.class)
                    .subscribe(observer)
    ).firstOrError().blockingGet();
    assertTrue(deckTypeData.size() == 1 && deckTypeData.get(0).getDeckType()
            .equals(DeckType.SWISS.name()));
}

@Test
public void decks_createdAndDeleted() throws Exception {
    List<Deck> deletedDeck = mUser.save().andThen((final CompletableObserver cs) -> {
        Deck deck = new Deck(mUser);
        deck.setName("Created");
        deck.setAccepted(true);
        deck.create().subscribe(cs);
    }).andThen((ObservableSource<List<Deck>>) observer ->
            mUser.fetchChildren(mUser.getChildReference(Deck.class), Deck.class)
                    .subscribe(observer)
    ).firstOrError().flatMapCompletable(data -> {
        assertTrue(data.size() == 1 && data.get(0).getName().equals("Created"));
        return data.get(0).delete();
    }).andThen((ObservableSource<List<Deck>>) observer ->
            mUser.fetchChildren(mUser.getChildReference(Deck.class), Deck.class)
                    .subscribe(observer)
    ).firstOrError().blockingGet();
    assertTrue(deletedDeck.size() == 0);
}

@Test
public void cards_createdAndFetched() throws Exception {
    final Deck deck = new Deck(mUser);
    List<Card> cards = mUser.save().andThen((final CompletableObserver cs) -> {
        deck.setName("CreateCards");
        deck.setAccepted(true);
        deck.create().subscribe(cs);
    }).andThen((ObservableSource<List<Deck>>) observer ->
            mUser.fetchChildren(mUser.getChildReference(Deck.class), Deck.class)
                    .subscribe(observer)
    ).firstOrError().flatMapCompletable(data -> {
        assertTrue(data.size() == 1 && data.get(0).getName().equals("CreateCards"));
        Card newCard = new Card(data.get(0));
        newCard.setFront("frontSide");
        newCard.setBack("backSide");
        return newCard.create();
    }).andThen((ObservableSource<List<Card>>) observer ->
            deck.fetchChildren(deck.getChildReference(Card.class), Card.class)
                    .subscribe(observer)
    ).firstOrError().blockingGet();
    assertTrue(cards.size() == 1 && cards.get(0).getFront().equals("frontSide") &&
            cards.get(0).getBack().equals("backSide"));
}

private CompletableObserver getCompletableObserver() {
    return new CompletableObserver() {
        @Override
        public void onSubscribe(Disposable d) {
            Log.d(TAG, " onSubscribe : " + d.isDisposed());
        }

        @Override
        public void onComplete() {
            textView.append(" onComplete");
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onComplete");
        }

        @Override
        public void onError(Throwable e) {
            textView.append(" onError : " + e.getMessage());
            textView.append(AppConstant.LINE_SEPARATOR);
            Log.d(TAG, " onError : " + e.getMessage());
        }
    };
}

protected  CompletableObserver failOrCountDownCompletableObserver(CountDownLatch latch) {
    return new CompletableObserver() {

        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onComplete() {
            latch.countDown();
        }

        @Override
        public void onError(Throwable e) {
            fail(e);
        }

    };
}

@Test
public void shouldHonorDisposedWhenCallingOnComplete() throws Exception {
    // Given
    Disposable disposable = mock(Disposable.class);
    CompletableObserver childObserver = mock(CompletableObserver.class);
    CompletableObserver decoratedObserver = CircuitBreakerOperator.of(circuitBreaker).apply(childObserver);
    decoratedObserver.onSubscribe(disposable);

    // When
    ((Disposable) decoratedObserver).dispose();
    decoratedObserver.onComplete();

    // Then
    verify(childObserver, never()).onComplete();
    assertSingleSuccessfulCall();
}

@Test
public void shouldHonorDisposedWhenCallingOnError() throws Exception {
    // Given
    Disposable disposable = mock(Disposable.class);
    CompletableObserver childObserver = mock(CompletableObserver.class);
    CompletableObserver decoratedObserver = CircuitBreakerOperator.of(circuitBreaker).apply(childObserver);
    decoratedObserver.onSubscribe(disposable);

    // When
    ((Disposable) decoratedObserver).dispose();
    decoratedObserver.onError(new IllegalStateException());

    // Then
    verify(childObserver, never()).onError(any());
    assertSingleFailedCall();
}

@Test
public void shouldHonorDisposedWhenCallingOnComplete() throws Exception {
    // Given
    Disposable disposable = mock(Disposable.class);
    CompletableObserver childObserver = mock(CompletableObserver.class);
    CompletableObserver decoratedObserver = BulkheadOperator.of(bulkhead).apply(childObserver);
    decoratedObserver.onSubscribe(disposable);

    // When
    ((Disposable) decoratedObserver).dispose();
    decoratedObserver.onComplete();

    // Then
    verify(childObserver, never()).onComplete();
    assertThat(bulkhead.getMetrics().getAvailableConcurrentCalls()).isEqualTo(1);
}

@Test
public void shouldHonorDisposedWhenCallingOnError() throws Exception {
    // Given
    Disposable disposable = mock(Disposable.class);
    CompletableObserver childObserver = mock(CompletableObserver.class);
    CompletableObserver decoratedObserver = BulkheadOperator.of(bulkhead).apply(childObserver);
    decoratedObserver.onSubscribe(disposable);

    // When
    ((Disposable) decoratedObserver).dispose();
    decoratedObserver.onError(new IllegalStateException());

    // Then
    verify(childObserver, never()).onError(any());
    assertThat(bulkhead.getMetrics().getAvailableConcurrentCalls()).isEqualTo(1);
}

@Test
public void shouldNotReleaseBulkheadWhenWasDisposedAfterNotPermittedSubscribe() throws Exception {
    // Given
    Disposable disposable = mock(Disposable.class);
    CompletableObserver childObserver = mock(CompletableObserver.class);
    CompletableObserver decoratedObserver = BulkheadOperator.of(bulkhead).apply(childObserver);
    bulkhead.isCallPermitted();
    assertThat(bulkhead.getMetrics().getAvailableConcurrentCalls()).isEqualTo(0);
    decoratedObserver.onSubscribe(disposable);

    // When
    ((Disposable) decoratedObserver).dispose();

    // Then
    assertThat(bulkhead.getMetrics().getAvailableConcurrentCalls()).isEqualTo(0);
}

/**
 * Adapts an Vert.x {@code Handler<AsyncResult<T>>} to an RxJava2 {@link SingleObserver}.
 * <p>
 * The returned observer can be subscribed to an {@link Single#subscribe(SingleObserver)}.
 *
 * @param handler the handler to adapt
 * @return the observer
 */
public static <T> CompletableObserver toObserver(Handler<AsyncResult<T>> handler) {
  AtomicBoolean completed = new AtomicBoolean();
  return new CompletableObserver() {
    @Override
    public void onSubscribe(@NonNull Disposable d) {
    }
    @Override
    public void onComplete() {
      if (completed.compareAndSet(false, true)) {
        handler.handle(io.vertx.core.Future.succeededFuture());
      }
    }
    public void onSuccess() {
      if (completed.compareAndSet(false, true)) {
        handler.handle(io.vertx.core.Future.succeededFuture());
      }
    }
    @Override
    public void onError(Throwable error) {
      if (completed.compareAndSet(false, true)) {
        handler.handle(io.vertx.core.Future.failedFuture(error));
      }
    }
  };
}

@Override
protected void subscribeActual(final CompletableObserver completableObserver) {
    verifyMainThread();
    final UpdateListener updateListener = new UpdateListener(completableObserver, valueUpdateAction);
    final AnimationEndListener endListener = new AnimationEndListener(completableObserver, valueAnimator, animationCancelAction);
    final ValueAnimatorDisposable animatorDisposable = new ValueAnimatorDisposable(valueAnimator, updateListener, endListener);
    completableObserver.onSubscribe(animatorDisposable);

    valueAnimator.addUpdateListener(updateListener);
    valueAnimator.addListener(endListener);
    valueAnimator.start();
}

public static void main(String[] args) {
    // TODO Auto-generated method stub
    String[] fruits = { "mango", "pineapple", "apple", "mango", "papaya" };
    Observable.fromArray(fruits).ignoreElements().subscribe(
        new CompletableObserver() {
        long time=0;
        @Override
        public void onSubscribe(Disposable disposable) {
            // TODO Auto-generated method stub
            time=System.currentTimeMillis();
            System.out.println(disposable.isDisposed()+"\t"+time);


        }
        @Override
        public void onError(Throwable throwable) {
            // TODO Auto-generated method stub
            throwable.printStackTrace();
        }

        @Override
        public void onComplete() {
            // TODO Auto-generated method stub
            System.out.println("completed");
            long time_to_complete=System.currentTimeMillis()-time;
            System.out.println("process completedin: "+time_to_complete+"ms");

        }
    });



}

@Override
protected void subscribeActual(CompletableObserver observer) {
    Task<Void> task = run();
    CompletableTaskCallback callback = new CompletableTaskCallback(task, observer);
    observer.onSubscribe(callback);
    task.addOnCompleteListener(callback);
}

@Override
public void resetPassword(String email) {
    if (!mUserInfoValidator.isEmailValid(email)) {
        getView().showEmailError(R.string.email_error);
        return;
    }

    getView().showProgress(R.string.progress_message_authenticating);
    mUserManager.resetPassword(email).subscribe(new CompletableObserver() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onComplete() {
            getView().hideProgress();
            getView().navigateToMain();
        }

        @Override
        public void onError(Throwable e) {
            getView().hideProgress();
            getView().showSnackBar(e.getMessage());
        }
    });
}

@Override
public void signUp(String email, String password) {
    if (!mValidator.isEmailValid(email)) {
        getView().showEmailError(R.string.email_error);
        return;
    }

    if (!mValidator.isPasswordValid(password)) {
        getView().showPasswordError(R.string.password_error);
        return;
    }

    getView().showProgress(R.string.progress_message_loading);
    mUserManager.createUser(email, password).subscribe(new CompletableObserver() {
        @Override
        public void onComplete() {
            getView().hideProgress();
            getView().navigateToMain();
        }

        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onError(Throwable e) {
            getView().hideProgress();
            getView().showSnackBar(e.getMessage());
        }
    });
}

/**
 * MaybeConsumer<Success> is very similar to {@link CompletableObserver} in JavaRX.  Both are
 * looking for either a signal that a process has succeeded, or that it has failed with some
 * exception.  For the time period where we are still using both interfaces, we will find it
 * useful to be able to switch between them.
 *
 * @return a {@link MaybeConsumer<Success>} that pipes {@link MaybeConsumer#success(Object)}
 *         to {@link CompletableObserver#onComplete()}, and
 *         {@link MaybeConsumer#fail(Exception)} to
 *         {@link CompletableObserver#onError(Throwable)}
 */
public static MaybeConsumer<Success> fromCompletableObserver(final CompletableObserver o) {
    return new MaybeConsumer<Success>() {
        @Override
        public void success(Success value) {
            o.onComplete();
        }

        @Override
        public void fail(Exception e) {
            o.onError(e);
        }
    };
}

@Test
public void decks_createdAndFetched() throws Exception {
    List<Deck> data = mUser.save().andThen((final CompletableObserver cs) -> {
        Deck deck = new Deck(mUser);
        deck.setName("My Deck");
        deck.setAccepted(true);
        deck.create().subscribe(cs);
    }).andThen((ObservableSource<List<Deck>>) observer ->
            mUser.fetchChildren(mUser.getChildReference(Deck.class), Deck.class)
                    .subscribe(observer)
    ).firstOrError().blockingGet();
    assertTrue(data.size() == 1 && data.get(0).getName().equals("My Deck"));
}