io.reactivex.CompletableOnSubscribe#io.reactivex.CompletableEmitter源码实例Demo

public static Completable completable(final OnSubscribeAction<Void> subscribe){
    return Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(final CompletableEmitter emitter) throws Exception {
            RxActionDelegate<Void> delegate = new RxActionDelegate<>(new ActionDelegate<Void>() {
                @Override
                public void onSuccess(Void result) {
                    emitter.onComplete();
                }

                @Override
                public void onError(Exception e) {
                    emitter.onError(e);
                }
            });
            emitter.setDisposable(delegate);
            subscribe.subscribe(delegate);
        }
    });
}
 

/**
 * Given an operation that takes a {@link MaybeConsumer<Success>}, create a JavaRX {@link
 * Completable} that succeeds iff the operation does.
 *
 * <p>Example:
 *
 * <pre>
 *     // update the experiment, and then log that it was successful
 *     DataController dc = getDataController();
 *     MaybeConsumers.buildCompleteable(mc -> dc.updateExperiment(e.getExperimentId(), mc))
 *                   .subscribe(() -> log("successfully updated!"));
 * </pre>
 */
public static Completable buildCompleteable(
    io.reactivex.functions.Consumer<MaybeConsumer<Success>> c) {
  return Completable.create(
      new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter emitter) throws Exception {
          c.accept(
              new MaybeConsumer<Success>() {
                @Override
                public void success(Success value) {
                  emitter.onComplete();
                }

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

@Test
public void whenTheConstructorIsCalledWithAValidEmitterThenGetOnErrorReturnsTheEmitter() {

    // Given
    final CompletableEmitter emitter = Mockito
            .mock(CompletableEmitter.class);
    final Throwable ex = Mockito.mock(Throwable.class);
    final CompletableEmitterMqttActionListener listener = new CompletableEmitterMqttActionListener(
            emitter) {

        @Override
        public void onSuccess(final IMqttToken asyncActionToken) {
            // Not invoked
        }
    };

    // When
    final OnError onError = listener.getOnError();
    onError.onError(ex);

    // Then
    Mockito.verify(emitter).onError(ex);
}
 

/**
 * Converts an {@link ApolloPrefetch} to a synchronous Completable
 *
 * @param prefetch the ApolloPrefetch to convert
 * @return the converted Completable
 * @throws NullPointerException if prefetch == null
 */
@NotNull
@CheckReturnValue
public static Completable from(@NotNull final ApolloPrefetch prefetch) {
  checkNotNull(prefetch, "prefetch == null");

  return Completable.create(new CompletableOnSubscribe() {
    @Override public void subscribe(final CompletableEmitter emitter) {
      cancelOnCompletableDisposed(emitter, prefetch);
      prefetch.enqueue(new ApolloPrefetch.Callback() {
        @Override public void onSuccess() {
          if (!emitter.isDisposed()) {
            emitter.onComplete();
          }
        }

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

@Override @NonNull public Completable observeClear() {
  return Completable.create(new CompletableOnSubscribe() {
    @Override public void subscribe(final CompletableEmitter emitter) throws Exception {
      runInWriteLock(readWriteLock, new ThrowingRunnable() {
        @Override public void run() throws Exception {
          if (file.exists() && !file.delete()) {
            throw new IOException("Clear operation on store failed.");
          } else {
            emitter.onComplete();
          }

          updateSubject.onNext(ValueUpdate.<T>empty());
        }
      });
    }
  });
}
 

@Test
public void testCompletableObservable() {

    Completable.create(new CompletableOnSubscribe() {
        @Override
        public void subscribe(CompletableEmitter emitter) {
            emitter.onComplete();
        }
    })
            .subscribeOn(Schedulers.io())
            .subscribe(new CompletableObserver() {
                @Override
                public void onSubscribe(Disposable d) {

                }

                @Override
                public void onComplete() {
                    System.out.println("onComplete is called");
                }

                @Override
                public void onError(Throwable e) {
                    System.out.println("onError is called" + e.getMessage());
                }
            });
}
 

@Test
public void whenOnSuccessIsCalledThenObserverOnNextAndOnCompletedAreCalled()
        throws Exception {
    final CompletableEmitter observer = Mockito
            .mock(CompletableEmitter.class);
    final ConnectActionListener listener = new ConnectFactory.ConnectActionListener(
            observer);
    final IMqttToken asyncActionToken = Mockito.mock(IMqttToken.class);
    listener.onSuccess(asyncActionToken);
    Mockito.verify(observer).onComplete();
}
 

@Test
public void whenOnSuccessIsCalledThenObserverOnNextAndOnCompletedAreCalled()
        throws Exception {
    final CompletableEmitter observer = Mockito
            .mock(CompletableEmitter.class);
    final UnsubscribeActionListener listener = new UnsubscribeFactory.UnsubscribeActionListener(
            observer);
    final IMqttToken asyncActionToken = Mockito.mock(IMqttToken.class);
    listener.onSuccess(asyncActionToken);
    Mockito.verify(observer).onComplete();
}
 

@Test
public void whenOnSuccessIsCalledThenObserverOnNextAndOnCompletedAreCalled()
        throws Exception {
    final CompletableEmitter observer = Mockito
            .mock(CompletableEmitter.class);
    final DisconnectActionListener listener = new DisconnectFactory.DisconnectActionListener(
            observer);
    final IMqttToken asyncActionToken = Mockito.mock(IMqttToken.class);
    listener.onSuccess(asyncActionToken);
    Mockito.verify(observer).onComplete();
}
 

@Override
public void subscribe(CompletableEmitter emitter) {
    if (!emitter.isDisposed()) {
        instance.signOut();
        emitter.onComplete();
    }
}
 

/**
 * Load image from Uri with RxJava2
 *
 * @param sourceUri Image Uri
 *
 * @see #load(Uri)
 *
 * @return Completable of loading image
 */
public Completable loadAsCompletable(final Uri sourceUri, final boolean useThumbnail,
    final RectF initialFrameRect) {
  return Completable.create(new CompletableOnSubscribe() {

    @Override public void subscribe(@NonNull final CompletableEmitter emitter) throws Exception {

      mInitialFrameRect = initialFrameRect;
      mSourceUri = sourceUri;

      if (useThumbnail) {
        applyThumbnail(sourceUri);
      }

      final Bitmap sampled = getImage(sourceUri);

      mHandler.post(new Runnable() {
        @Override public void run() {
          mAngle = mExifRotation;
          setImageDrawableInternal(new BitmapDrawable(getResources(), sampled));
          emitter.onComplete();
        }
      });
    }
  }).doOnSubscribe(new Consumer<Disposable>() {
    @Override public void accept(@NonNull Disposable disposable) throws Exception {
      mIsLoading.set(true);
    }
  }).doFinally(new Action() {
    @Override public void run() throws Exception {
      mIsLoading.set(false);
    }
  });
}
 

@Override
public void subscribe(@NonNull CompletableEmitter emitter) throws Exception {
    try {
        preparedOperation.executeAsBlocking();
        emitter.onComplete();
    } catch (Exception e) {
        emitter.onError(e);
    }
}
 

AppInitHelper(RxActivity activity, CompletableEmitter emitter)
{
    super(activity, emitter);
}
 

AppInitHelperBase(RxActivity activity, CompletableEmitter emitter)
{
    _activity = activity;
    _settings = UserSettings.getSettings(activity);
    _initFinished = emitter;
}
 

public ConnectActionListener(final CompletableEmitter emitter) {
    super(emitter);
}
 

public UnsubscribeActionListener(final CompletableEmitter emitter) {
    super(emitter);
}
 

public CompletableEmitterMqttActionListener(
        final CompletableEmitter emitter) {
    this.emitter = Objects.requireNonNull(emitter);
}
 

public DisconnectActionListener(final CompletableEmitter emitter) {
    super(emitter);
}
 

private static void cancelOnCompletableDisposed(CompletableEmitter emitter, final Cancelable cancelable) {
  emitter.setDisposable(getRx2Disposable(cancelable));
}