Java 类io.reactivex.MaybeSource 实例源码

@Override
public Location loadInBackground() {
    mCancel = MaybeSubject.create();
    ArrayList<MaybeSource<Location>> sources;
    if (mPriorities != null) {
        sources = new ArrayList<>(mPriorities.length + 1);
        for (int priority : mPriorities) {
            sources.add(mProvider.getCurrentLocation(priority));
        }
    } else {
        sources = new ArrayList<>(2);
        sources.add(mProvider.getCurrentLocation(mPriority));
    }
    sources.add(mCancel);
    Location location = Maybe.amb(sources).blockingGet();
    mCancel = null;
    return location;
}

@Override
public MaybeSource<T> apply(@NonNull Maybe<B> upstream) {
  Maybe<Buffer> unwrapped = upstream.map(unwrap::apply);
  Maybe<T> unmarshalled = unwrapped.concatMap(buffer -> {
    if (buffer.length() > 0) {
      try {
        T obj = nonNull(mappedType) ? mapper.readValue(buffer.getBytes(), mappedType) :
          mapper.readValue(buffer.getBytes(), mappedTypeRef);
        return Maybe.just(obj);
      } catch (IOException e) {
        return Maybe.error(e);
      }
    } else {
      return Maybe.empty();
    }
  });
  return unmarshalled;
}

private Maybe<BookmarkData> getBookmarkDataObservable(Maybe<BufferedSource> source) {
  return source
      .flatMap(new Function<BufferedSource, MaybeSource<BookmarkData>>() {
        @Override
        public MaybeSource<BookmarkData> apply(BufferedSource bufferedSource) throws Exception {
          return mBookmarkImportExportModel.readBookmarks(bufferedSource).toMaybe();
        }
      })
      .subscribeOn(Schedulers.io());
}

@NonNull
@VisibleForTesting(otherwise = VisibleForTesting.PRIVATE)
Maybe<BufferedSource> parseUri(final Uri uri) {
  return Maybe.defer(new Callable<MaybeSource<BufferedSource>>() {
    @Override
    public MaybeSource<BufferedSource> call() throws Exception {
      ParcelFileDescriptor pfd = mAppContext.getContentResolver().openFileDescriptor(uri, "r");
      if (pfd != null) {
        FileDescriptor fd = pfd.getFileDescriptor();
        return Maybe.just(Okio.buffer(Okio.source(new FileInputStream(fd))));
      }
      return Maybe.empty();
    }
  });
}

@NonNull
@VisibleForTesting(otherwise = VisibleForTesting.PRIVATE)
Maybe<BufferedSource> parseExternalFile(final Uri uri) {
  return Maybe.defer(new Callable<MaybeSource<BufferedSource>>() {
    @Override
    public MaybeSource<BufferedSource> call() throws Exception {
      InputStream stream = mAppContext.getContentResolver().openInputStream(uri);
      if (stream != null) {
        return Maybe.just(Okio.buffer(Okio.source(stream)));
      }
      return Maybe.empty();
    }
  });
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    Maybe<T> transformedStream = upstream
            .cache() // Cache to replay emitted values to late subscriber
            .filter(filterIfDestroyedPredicate); // Filter to stop emitting items once LifecycleOwner is destroyed
    setReactiveType((R)transformedStream);

    return transformedStream;
}

@NonNull
private Function<StockUpdate, MaybeSource<? extends StockUpdate>> skipTweetsThatDoNotContainKeywords(String[] trackingKeywords) {
    return update -> Observable.fromArray(trackingKeywords)
            .filter(keyword -> update.getTwitterStatus().toLowerCase().contains(keyword.toLowerCase()))
            .map(keyword -> update)
            .firstElement();
}

@NonNull
private Function<StockUpdate, MaybeSource<? extends StockUpdate>> skipTweetsThatDoNotContainKeywords(String[] trackingKeywords) {
    return update -> Observable.fromArray(trackingKeywords)
            .filter(keyword -> update.getTwitterStatus().toLowerCase().contains(keyword.toLowerCase()))
            .map(keyword -> update)
            .firstElement();
}

/**
 * @param callable
 * @param <R>
 * @return
 */
@CheckReturnValue
@NonNull
public static <R> Maybe<R> maybe(@NonNull final Callable<Task<R>> callable) {
    return Single.fromCallable(callable).flatMapMaybe(
            new Function<Task<R>, MaybeSource<? extends R>>() {
        @Override
        public MaybeSource<? extends R> apply(Task<R> task) throws Exception {
            return maybe(task);
        }
    });
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    Maybe<T> tObservable = upstream
            .observeOn(AndroidSchedulers.mainThread());
    if (provider == null) {
        return tObservable;
    }
    return tObservable.compose(RxLifecycleDelegate.<T>bindLifecycle(provider));
}

private MaybeSource<GoosciSnapshotValue.SnapshotLabelValue.SensorSnapshot> makeSnapshot(
        String sensorId, SensorRegistry sensorRegistry) throws Exception {
    BehaviorSubject<ScalarReading> subject = mLatestValues.get(sensorId);
    if (subject == null) {
        return Maybe.empty();
    }
    final GoosciSensorSpec.SensorSpec spec = getSensorSpec(sensorId, sensorRegistry);
    return subject.firstElement().map(value -> generateSnapshot(spec, value));
}

/**
 * Accepts a {@link ScopeProvider} for scope resolution.
 *
 * @param provider the {@link ScopeProvider}.
 */
Scoper(final ScopeProvider provider) {
  this(Maybe.defer(new Callable<MaybeSource<?>>() {
    @Override public MaybeSource<?> call() throws Exception {
      return provider.requestScope();
    }
  }));
}

/**
 * Method which retrieve a list of DataSnapshot from multiple {@link DatabaseReference}.
 *
 * @param whereRefs array of {@link DatabaseReference references.}
 * @return a {@link Flowable} which emmit {@link DataSnapshot} from the given queries.
 */
@NonNull
public static Flowable<DataSnapshot> observeMultipleSingleValueEvent(@NonNull DatabaseReference... whereRefs) {
   return Maybe.merge(Flowable.fromArray(whereRefs)
   .map(new Function<DatabaseReference, MaybeSource<? extends DataSnapshot>>() {
         @Override
         public MaybeSource<? extends DataSnapshot> apply(@NonNull DatabaseReference databaseReference) throws
               Exception {
            return observeSingleValueEvent(databaseReference);
         }
      })
   );
}

@Override
public final MaybeSource<T> source() {
    return source;
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    return upstream.takeUntil(observable.firstElement());
}

MaybeIfThen(BooleanSupplier condition, MaybeSource<? extends T> then,
            MaybeSource<? extends T> orElse) {
    this.condition = condition;
    this.then = then;
    this.orElse = orElse;
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    return upstream;
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    return upstream.takeUntil(mObservable.firstElement());
}

AutoDisposeMaybe(MaybeSource<T> source, Maybe<?> scope) {
  this.source = source;
  this.scope = scope;
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    return observableCache.cacheMaybe(key, upstream);
}

@Override public MaybeSource<T> apply(Maybe<T> maybe) {
  return maybe.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread());
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    return upstream.takeUntil(observable.firstElement());
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    return ((Observable<T>) apply(upstream.toObservable())).singleElement();
}

@Override
public MaybeSource<T> apply(Maybe<T> upstream) {
    return ((Observable<T>) apply(upstream.toObservable())).singleElement();
}

/**
 * Entry point for auto-disposing streams from a {@link ScopeProvider}.
 * <p>
 * Example usage:
 * <pre><code>
 *   Observable.just(1)
 *        .as(AutoDispose.<Integer>autoDisposable(scope))
 *        .subscribe(...)
 * </code></pre>
 *
 * @param provider the target scope provider
 * @param <T> the stream type.
 * @return an {@link AutoDisposeConverter} to transform with operators like
 * {@link Observable#as(ObservableConverter)}
 */
public static <T> AutoDisposeConverter<T> autoDisposable(final ScopeProvider provider) {
  checkNotNull(provider, "provider == null");
  return autoDisposable(Maybe.defer(new Callable<MaybeSource<?>>() {
    @Override public MaybeSource<?> call() {
      return provider.requestScope();
    }
  }));
}

/**
 *
 * @param source the consumable MaybeSource
 * @param compositeDisposable the composite disposable
 */
public AttachDisposableMaybe(MaybeSource<T> source, CompositeDisposable compositeDisposable) {
    this.source = source;
    this.compositeDisposable = compositeDisposable;
}