io.reactivex.rxjava3.core.ObservableSource#io.reactivex.rxjava3.annotations.Nullable源码实例Demo

/**
 * Creates an {@link ObservableTransformer} that will flatten the provided {@link Action} into the
 * stream as a {@link Completable} every time it receives an effect from the upstream effects
 * observable. This Completable will be subscribed on the specified {@link Scheduler}. This will
 * result in calling the provided Action on the specified scheduler every time an effect is
 * dispatched to the created effect transformer.
 *
 * @param doEffect the {@link Action} to be run every time the effect is requested
 * @param scheduler the {@link Scheduler} that the action should be run on
 * @param <F> the type of Effect this transformer handles
 * @param <E> these transformers are for effects that do not result in any events; however, they
 *     still need to share the same Event type
 * @return an {@link ObservableTransformer} that can be used with a {@link
 *     RxMobius.SubtypeEffectHandlerBuilder}.
 */
static <F, E> ObservableTransformer<F, E> fromAction(
    final Action doEffect, @Nullable final Scheduler scheduler) {
  return new ObservableTransformer<F, E>() {
    @Override
    public ObservableSource<E> apply(Observable<F> effectStream) {
      return effectStream
          .flatMapCompletable(
              new Function<F, CompletableSource>() {
                @Override
                public CompletableSource apply(F f) throws Exception {
                  return scheduler == null
                      ? Completable.fromAction(doEffect)
                      : Completable.fromAction(doEffect).subscribeOn(scheduler);
                }
              })
          .toObservable();
    }
  };
}
 

/**
 * Creates an {@link ObservableTransformer} that will flatten the provided {@link Consumer} into
 * the stream as a {@link Completable} every time it receives an effect from the upstream effects
 * observable. This will result in calling the consumer on the specified scheduler, and passing it
 * the requested effect object.
 *
 * @param doEffect the {@link Consumer} to be run every time the effect is requested
 * @param scheduler the {@link Scheduler} to be used when invoking the consumer
 * @param <F> the type of Effect this transformer handles
 * @param <E> these transformers are for effects that do not result in any events; however, they
 *     still need to share the same Event type
 * @return an {@link ObservableTransformer} that can be used with a {@link
 *     RxMobius.SubtypeEffectHandlerBuilder}.
 */
static <F, E> ObservableTransformer<F, E> fromConsumer(
    final Consumer<F> doEffect, @Nullable final Scheduler scheduler) {
  return new ObservableTransformer<F, E>() {
    @Override
    public ObservableSource<E> apply(Observable<F> effectStream) {
      return effectStream
          .flatMapCompletable(
              new Function<F, CompletableSource>() {
                @Override
                public CompletableSource apply(final F effect) {
                  Completable completable =
                      Completable.fromAction(
                          new Action() {
                            @Override
                            public void run() throws Throwable {
                              doEffect.accept(effect);
                            }
                          });
                  return scheduler == null ? completable : completable.subscribeOn(scheduler);
                }
              })
          .toObservable();
    }
  };
}
 

/**
 * Creates an {@link ObservableTransformer} that will flatten the provided {@link Function} into
 * the stream as an {@link Observable} every time it receives an effect from the upstream effects
 * observable. This will result in calling the function on the specified scheduler, and passing it
 * the requested effect object then emitting its returned value.
 *
 * @param function the {@link Function} to be invoked every time the effect is requested
 * @param scheduler the {@link Scheduler} to be used when invoking the function
 * @param <F> the type of Effect this transformer handles
 * @param <E> the type of Event this transformer emits
 * @return an {@link ObservableTransformer} that can be used with a {@link
 *     RxMobius.SubtypeEffectHandlerBuilder}.
 */
static <F, E> ObservableTransformer<F, E> fromFunction(
    final Function<F, E> function, @Nullable final Scheduler scheduler) {
  return new ObservableTransformer<F, E>() {
    @Override
    public ObservableSource<E> apply(Observable<F> effectStream) {
      return effectStream.flatMap(
          new Function<F, ObservableSource<E>>() {
            @Override
            public ObservableSource<E> apply(@NonNull F f) {
              Observable<E> eventObservable =
                  Observable.fromSupplier(
                      new Supplier<E>() {
                        @Override
                        public E get() throws Throwable {
                          return function.apply(f);
                        }
                      });
              return scheduler == null ? eventObservable : eventObservable.subscribeOn(scheduler);
            }
          });
    }
  };
}
 

@Override
@Nullable
@SuppressWarnings("unchecked")
public T getValue() {
    Node<T> h = head;

    for (;;) {
        Node<T> next = h.get();
        if (next == null) {
            break;
        }
        h = next;
    }

    return h.value;
}
 

@Override
@Nullable
@SuppressWarnings("unchecked")
public T getValue() {
    TimedNode<T> h = head;

    for (;;) {
        TimedNode<T> next = h.get();
        if (next == null) {
            break;
        }
        h = next;
    }

    long limit = scheduler.now(unit) - maxAge;
    if (h.time < limit) {
        return null;
    }

    return h.value;
}
 

@Override
@Nullable
@SuppressWarnings("unchecked")
public T getValue() {
    Node<T> h = head;

    for (;;) {
        Node<T> next = h.get();
        if (next == null) {
            break;
        }
        h = next;
    }

    return h.value;
}
 

@Override
@Nullable
@SuppressWarnings("unchecked")
public T getValue() {
    TimedNode<T> h = head;

    for (;;) {
        TimedNode<T> next = h.get();
        if (next == null) {
            break;
        }
        h = next;
    }

    long limit = scheduler.now(unit) - maxAge;
    if (h.time < limit) {
        return null;
    }

    return h.value;
}
 

@Override
@Nullable
@SuppressWarnings("unchecked")
public T getValue() {
    int s = size;
    if (s != 0) {
        return buffer.get(s - 1);
    }
    return null;
}
 

@Override
@Nullable
@SuppressWarnings("unchecked")
public T getValue() {
    int s = size;
    if (s != 0) {
        return buffer.get(s - 1);
    }
    return null;
}
 

/**
 * Returns a single value the Relay currently has or null if no such value exists.
 * <p>The method is thread-safe.
 */
@Nullable
public T getValue() {
    return value.get();
}
 

/**
 * Returns a single value the Relay currently has or null if no such value exists.
 * <p>The method is thread-safe.
 */
@Nullable
public T getValue() {
    return buffer.getValue();
}
 

@Nullable
T getValue();
 

private ReplayingShare(@Nullable T defaultValue) {
  this.defaultValue = defaultValue;
}
 

LastSeen(@Nullable T defaultValue) {
  this.defaultValue = defaultValue;
  value = defaultValue;
}
 

/**
 * Returns a single value the Relay currently has or null if no such value exists.
 * <p>The method is thread-safe.
 */
@Nullable
public T getValue() {
    return value.get();
}
 

/**
 * Returns a single value the Relay currently has or null if no such value exists.
 * <p>The method is thread-safe.
 */
@Nullable
public T getValue() {
    return buffer.getValue();
}
 

@Nullable
T getValue();