Java 类io.reactivex.ObservableOperator 实例源码

项目：Learning-RxJava 文件：Ch9_8.java

public static <T> ObservableOperator<List<T>, T> myToList() {
        return observer -> new DisposableObserver<T>() {
            ArrayList<T> list = new ArrayList<>();

            @Override
            public void onNext(T value) {
//add to List, but don't pass anything downstream
                list.add(value);
            }

            @Override
            public void onError(Throwable t) {
                observer.onError(t);
            }

            @Override
            public void onComplete() {
                observer.onNext(list); //push List downstream
                observer.onComplete();
            }
        };
    }

项目：Learning-RxJava 文件：Ch9_7.java

public static <T> ObservableOperator<T, T> doOnEmpty(Action
                                                             action) {
    return new ObservableOperator<T, T>() {
        @Override
        public Observer<? super T> apply(Observer<? super T>
                                                 observer) throws Exception {
            return new DisposableObserver<T>() {
                boolean isEmpty = true;

                @Override
                public void onNext(T value) {
                    isEmpty = false;
                    observer.onNext(value);
                }

                @Override
                public void onError(Throwable t) {
                    observer.onError(t);
                }

                @Override
                public void onComplete() {
                    if (isEmpty) {
                        try {
                            action.run();
                        } catch (Exception e) {
                            onError(e);
                            return;
                        }
                    }
                    observer.onComplete();
                }
            };
        }
    };
}

项目：Java-EX 文件：RandomOperator.java

public static <T> ObservableOperator<T, T> observable() {
  return observable(DEFAULT_CACHE);
}

项目：Java-EX 文件：RandomOperator.java

public static <T> ObservableOperator<T, T> observable(int cacheSize) {
  return actual -> new RandomObserver<>(actual, cacheSize);
}

项目：Java-EX 文件：RandomOperator.java

public static <T> ObservableOperator<T, T> observable() {
  return observable(DEFAULT_CACHE);
}

项目：Java-EX 文件：RandomOperator.java

public static <T> ObservableOperator<T, T> observable(int cacheSize) {
  return actual -> new RandomObserver<>(actual, cacheSize);
}

项目：sqlbrite-sqlcipher 文件：SqlBrite.java

/**
 * Creates an {@linkplain ObservableOperator operator} which transforms a query returning a
 * single row to a {@code T} using {@code mapper}. Use with {@link Observable#lift}.
 * <p>
 * It is an error for a query to pass through this operator with more than 1 row in its result
 * set. Use {@code LIMIT 1} on the underlying SQL query to prevent this. Result sets with 0 rows
 * emit {@code defaultValue}.
 * <p>
 * This operator emits {@code defaultValue} if {@code null} is returned from {@link #run()}.
 *
 * @param mapper Maps the current {@link Cursor} row to {@code T}. May not return null.
 * @param defaultValue Value returned if result set is empty
 */
@SuppressWarnings("ConstantConditions") // Public API contract.
@CheckResult @NonNull
public static <T> ObservableOperator<T, Query> mapToOneOrDefault(
    @NonNull Function<Cursor, T> mapper, @NonNull T defaultValue) {
  if (defaultValue == null) throw new NullPointerException("defaultValue == null");
  return new QueryToOneOperator<>(mapper, defaultValue);
}

项目：sqlbrite-sqlcipher 文件：SqlBrite.java

/**
 * Creates an {@linkplain ObservableOperator operator} which transforms a query returning a
 * single row to a {@code Optional<T>} using {@code mapper}. Use with {@link Observable#lift}.
 * <p>
 * It is an error for a query to pass through this operator with more than 1 row in its result
 * set. Use {@code LIMIT 1} on the underlying SQL query to prevent this. Result sets with 0 rows
 * emit {@link Optional#empty() Optional.empty()}.
 * <p>
 * This operator ignores {@code null} cursors returned from {@link #run()}.
 *
 * @param mapper Maps the current {@link Cursor} row to {@code T}. May not return null.
 */
@RequiresApi(Build.VERSION_CODES.N) //
@CheckResult @NonNull //
public static <T> ObservableOperator<Optional<T>, Query> mapToOptional(
    @NonNull Function<Cursor, T> mapper) {
  return new QueryToOptionalOperator<>(mapper);
}

项目：sqlbrite 文件：SqlBrite.java

/**
 * Creates an {@linkplain ObservableOperator operator} which transforms a query returning a
 * single row to a {@code T} using {@code mapper}. Use with {@link Observable#lift}.
 * <p>
 * It is an error for a query to pass through this operator with more than 1 row in its result
 * set. Use {@code LIMIT 1} on the underlying SQL query to prevent this. Result sets with 0 rows
 * emit {@code defaultValue}.
 * <p>
 * This operator emits {@code defaultValue} if {@code null} is returned from {@link #run()}.
 *
 * @param mapper Maps the current {@link Cursor} row to {@code T}. May not return null.
 * @param defaultValue Value returned if result set is empty
 */
@SuppressWarnings("ConstantConditions") // Public API contract.
@CheckResult @NonNull
public static <T> ObservableOperator<T, Query> mapToOneOrDefault(
    @NonNull Function<Cursor, T> mapper, @NonNull T defaultValue) {
  if (defaultValue == null) throw new NullPointerException("defaultValue == null");
  return new QueryToOneOperator<>(mapper, defaultValue);
}

项目：sqlbrite 文件：SqlBrite.java

/**
 * Creates an {@linkplain ObservableOperator operator} which transforms a query returning a
 * single row to a {@code Optional<T>} using {@code mapper}. Use with {@link Observable#lift}.
 * <p>
 * It is an error for a query to pass through this operator with more than 1 row in its result
 * set. Use {@code LIMIT 1} on the underlying SQL query to prevent this. Result sets with 0 rows
 * emit {@link Optional#empty() Optional.empty()}.
 * <p>
 * This operator ignores {@code null} cursors returned from {@link #run()}.
 *
 * @param mapper Maps the current {@link Cursor} row to {@code T}. May not return null.
 */
@RequiresApi(Build.VERSION_CODES.N) //
@CheckResult @NonNull //
public static <T> ObservableOperator<Optional<T>, Query> mapToOptional(
    @NonNull Function<Cursor, T> mapper) {
  return new QueryToOptionalOperator<>(mapper);
}

项目：sqlbrite-sqlcipher 文件：SqlBrite.java

/**
 * Creates an {@linkplain ObservableOperator operator} which transforms a query returning a
 * single row to a {@code T} using {@code mapper}. Use with {@link Observable#lift}.
 * <p>
 * It is an error for a query to pass through this operator with more than 1 row in its result
 * set. Use {@code LIMIT 1} on the underlying SQL query to prevent this. Result sets with 0 rows
 * do not emit an item.
 * <p>
 * This operator ignores {@code null} cursors returned from {@link #run()}.
 *
 * @param mapper Maps the current {@link Cursor} row to {@code T}. May not return null.
 */
@CheckResult @NonNull //
public static <T> ObservableOperator<T, Query> mapToOne(@NonNull Function<Cursor, T> mapper) {
  return new QueryToOneOperator<>(mapper, null);
}

项目：sqlbrite-sqlcipher 文件：SqlBrite.java

/**
 * Creates an {@linkplain ObservableOperator operator} which transforms a query to a
 * {@code List<T>} using {@code mapper}. Use with {@link Observable#lift}.
 * <p>
 * Be careful using this operator as it will always consume the entire cursor and create objects
 * for each row, every time this observable emits a new query. On tables whose queries update
 * frequently or very large result sets this can result in the creation of many objects.
 * <p>
 * This operator ignores {@code null} cursors returned from {@link #run()}.
 *
 * @param mapper Maps the current {@link Cursor} row to {@code T}. May not return null.
 */
@CheckResult @NonNull
public static <T> ObservableOperator<List<T>, Query> mapToList(
    @NonNull Function<Cursor, T> mapper) {
  return new QueryToListOperator<>(mapper);
}

项目：sqlbrite 文件：SqlBrite.java

/**
 * Creates an {@linkplain ObservableOperator operator} which transforms a query returning a
 * single row to a {@code T} using {@code mapper}. Use with {@link Observable#lift}.
 * <p>
 * It is an error for a query to pass through this operator with more than 1 row in its result
 * set. Use {@code LIMIT 1} on the underlying SQL query to prevent this. Result sets with 0 rows
 * do not emit an item.
 * <p>
 * This operator ignores {@code null} cursors returned from {@link #run()}.
 *
 * @param mapper Maps the current {@link Cursor} row to {@code T}. May not return null.
 */
@CheckResult @NonNull //
public static <T> ObservableOperator<T, Query> mapToOne(@NonNull Function<Cursor, T> mapper) {
  return new QueryToOneOperator<>(mapper, null);
}

项目：sqlbrite 文件：SqlBrite.java

/**
 * Creates an {@linkplain ObservableOperator operator} which transforms a query to a
 * {@code List<T>} using {@code mapper}. Use with {@link Observable#lift}.
 * <p>
 * Be careful using this operator as it will always consume the entire cursor and create objects
 * for each row, every time this observable emits a new query. On tables whose queries update
 * frequently or very large result sets this can result in the creation of many objects.
 * <p>
 * This operator ignores {@code null} cursors returned from {@link #run()}.
 *
 * @param mapper Maps the current {@link Cursor} row to {@code T}. May not return null.
 */
@CheckResult @NonNull
public static <T> ObservableOperator<List<T>, Query> mapToList(
    @NonNull Function<Cursor, T> mapper) {
  return new QueryToListOperator<>(mapper);
}