Java 类io.reactivex.Notification 实例源码

private void calcNext() {
  if (completed) {
    return;
  }
  if (next == null) {
    Notification<T> take = uncheck(queue::take);
    if (take.isOnNext()) {
      next = take;
    } else if (take.isOnError()) {
      completed = true;
      throw new RuntimeException(take.getError());
    } else {
      completed = true;
    }
  }
}

private void calcNext() {
  if (completed) {
    return;
  }
  if (next == null) {
    subscription.request(1);
    Notification<T> take = uncheck(() -> queue.take());
    if (take.isOnNext()) {
      next = take;
    } else if (take.isOnError()) {
      completed = true;
      throw new RuntimeException(take.getError());
    } else {
      completed = true;
    }
  }
}

private static <T> Flowable<Tx<T>> inTransaction(CallableBuilder b,
        Function<Single<Connection>, Flowable<Notification<T>>> f) {
    return Flowable.defer(() -> {
        AtomicReference<Connection> con = new AtomicReference<Connection>();
        // set the atomic reference when transactedConnection emits
        Single<Connection> transactedConnection = b.connection //
                .map(c -> Util.toTransactedConnection(con, c));
        return f.apply(transactedConnection) //
                .<Tx<T>>flatMap(n -> Tx.toTx(n, con.get(), b.db)) //
                .doOnNext(tx -> {
                    if (tx.isComplete()) {
                        ((TxImpl<T>) tx).connection().commit();
                    }
                });
    });
}

static Flowable<Notification<TupleN<Object>>> createWithNParameters( //
        Single<Connection> connection, //
        String sql, //
        Flowable<List<Object>> parameterGroups, //
        List<ParameterPlaceholder> parameterPlaceholders, //
        List<Class<?>> outClasses) {
    return connection //
            .toFlowable() //
            .flatMap( //
                    con -> createWithParameters( //
                            con, //
                            sql, //
                            parameterGroups, //
                            parameterPlaceholders, //
                            (stmt, parameters) -> createWithNParameters(stmt, parameters, parameterPlaceholders,
                                    outClasses)));
}

private static <T1> Flowable<Notification<CallableResultSet1<T1>>> createWithOneResultSet(Connection con,
        String sql, Flowable<List<Object>> parameterGroups, List<ParameterPlaceholder> parameterPlaceholders,
        Function<? super ResultSet, ? extends T1> f1, int fetchSize) {
    log.debug("Update.create {}", sql);
    Callable<NamedCallableStatement> resourceFactory = () -> Util.prepareCall(con, sql, parameterPlaceholders);
    final Function<NamedCallableStatement, Flowable<Notification<CallableResultSet1<T1>>>> flowableFactory = //
            stmt -> parameterGroups //
                    .flatMap(parameters -> {
                        List<Object> outputValues = executeAndReturnOutputValues(parameterPlaceholders, stmt,
                                parameters);
                        Flowable<T1> flowable1 = createFlowable(stmt, f1);
                        return Single.just(new CallableResultSet1<T1>(outputValues, flowable1)).toFlowable();
                    }) //
                    .materialize() //
                    .doOnComplete(() -> Util.commit(stmt.stmt)) //
                    .doOnError(e -> Util.rollback(stmt.stmt));
    Consumer<NamedCallableStatement> disposer = Util::closeCallableStatementAndConnection;
    return Flowable.using(resourceFactory, flowableFactory, disposer, true);
}

private static <T1, T2> Flowable<Notification<CallableResultSet2<T1, T2>>> createWithTwoResultSets(Connection con,
        String sql, Flowable<List<Object>> parameterGroups, List<ParameterPlaceholder> parameterPlaceholders,
        Function<? super ResultSet, ? extends T1> f1, Function<? super ResultSet, ? extends T2> f2, int fetchSize) {
    Callable<NamedCallableStatement> resourceFactory = () -> Util.prepareCall(con, sql, parameterPlaceholders);
    final Function<NamedCallableStatement, Flowable<Notification<CallableResultSet2<T1, T2>>>> flowableFactory = //
            stmt -> parameterGroups //
                    .flatMap(parameters -> {
                        List<Object> outputValues = executeAndReturnOutputValues(parameterPlaceholders, stmt,
                                parameters);
                        final Flowable<T1> flowable1 = createFlowable(stmt, f1);
                        stmt.stmt.getMoreResults(Statement.KEEP_CURRENT_RESULT);
                        final Flowable<T2> flowable2 = createFlowable(stmt, f2);
                        return Single.just(new CallableResultSet2<T1, T2>(outputValues, flowable1, flowable2))
                                .toFlowable();
                    }) //
                    .materialize() //
                    .doOnComplete(() -> Util.commit(stmt.stmt)) //
                    .doOnError(e -> Util.rollback(stmt.stmt));
    Consumer<NamedCallableStatement> disposer = Util::closeCallableStatementAndConnection;
    return Flowable.using(resourceFactory, flowableFactory, disposer, true);
}

private static <T1, T2, T3> Flowable<Notification<CallableResultSet3<T1, T2, T3>>> createWithThreeResultSets(
        Connection con, String sql, Flowable<List<Object>> parameterGroups,
        List<ParameterPlaceholder> parameterPlaceholders, Function<? super ResultSet, ? extends T1> f1,
        Function<? super ResultSet, ? extends T2> f2, Function<? super ResultSet, ? extends T3> f3, int fetchSize) {
    Callable<NamedCallableStatement> resourceFactory = () -> Util.prepareCall(con, sql, parameterPlaceholders);
    final Function<NamedCallableStatement, Flowable<Notification<CallableResultSet3<T1, T2, T3>>>> flowableFactory = //
            stmt -> parameterGroups //
                    .flatMap(parameters -> {
                        List<Object> outputValues = executeAndReturnOutputValues(parameterPlaceholders, stmt,
                                parameters);
                        final Flowable<T1> flowable1 = createFlowable(stmt, f1);
                        stmt.stmt.getMoreResults(Statement.KEEP_CURRENT_RESULT);
                        final Flowable<T2> flowable2 = createFlowable(stmt, f2);
                        stmt.stmt.getMoreResults(Statement.KEEP_CURRENT_RESULT);
                        final Flowable<T3> flowable3 = createFlowable(stmt, f3);
                        return Single.just(
                                new CallableResultSet3<T1, T2, T3>(outputValues, flowable1, flowable2, flowable3))
                                .toFlowable();
                    }) //
                    .materialize() //
                    .doOnComplete(() -> Util.commit(stmt.stmt)) //
                    .doOnError(e -> Util.rollback(stmt.stmt));
    Consumer<NamedCallableStatement> disposer = Util::closeCallableStatementAndConnection;
    return Flowable.using(resourceFactory, flowableFactory, disposer, true);
}

private static Flowable<Notification<CallableResultSetN>> createWithNResultSets(Connection con, String sql,
        Flowable<List<Object>> parameterGroups, List<ParameterPlaceholder> parameterPlaceholders,
        List<Function<? super ResultSet, ?>> functions, int fetchSize) {
    Callable<NamedCallableStatement> resourceFactory = () -> Util.prepareCall(con, sql, parameterPlaceholders);
    final Function<NamedCallableStatement, Flowable<Notification<CallableResultSetN>>> flowableFactory = //
            stmt -> parameterGroups //
                    .flatMap(parameters -> {
                        List<Object> outputValues = executeAndReturnOutputValues(parameterPlaceholders, stmt,
                                parameters);
                        List<Flowable<?>> flowables = Lists.newArrayList();
                        int i = 0;
                        do {
                            Function<? super ResultSet, ?> f = functions.get(i);
                            flowables.add(createFlowable(stmt, f));
                            i++;
                        } while (stmt.stmt.getMoreResults(Statement.KEEP_CURRENT_RESULT));
                        return Single.just(new CallableResultSetN(outputValues, flowables)).toFlowable();
                    }) //
                    .materialize() //
                    .doOnComplete(() -> Util.commit(stmt.stmt)) //
                    .doOnError(e -> Util.rollback(stmt.stmt));
    Consumer<NamedCallableStatement> disposer = Util::closeCallableStatementAndConnection;
    return Flowable.using(resourceFactory, flowableFactory, disposer, true);
}

@Override
protected void match() throws Exception {
    Object[] notifications = new Object[this.observers.size()];
    int j = 0;
    int completedCount = 0;
    for (JoinObserver1<? extends Object> jo : this.observers) {
        if (jo.queue().isEmpty()) {
            return;
        }
        Notification<? extends Object> n = jo.queue().peek();
        if (n.isOnComplete()) {
            completedCount++;
        }
        notifications[j] = n.getValue();
        j++;
    }

    if (completedCount == j) {
        onCompleted.run();
    } else {
        dequeue();
        onNext.accept(notifications);
    }
}

@Override
protected void match() throws Exception {
    if (!jo1.queue().isEmpty()
            && !jo2.queue().isEmpty()
            && !jo3.queue().isEmpty()
            && !jo4.queue().isEmpty()) {
        Notification<T1> n1 = jo1.queue().peek();
        Notification<T2> n2 = jo2.queue().peek();
        Notification<T3> n3 = jo3.queue().peek();
        Notification<T4> n4 = jo4.queue().peek();

        if (n1.isOnComplete()
                || n2.isOnComplete()
                || n3.isOnComplete()
                || n4.isOnComplete()) {
            onCompleted.run();
        } else {
            dequeue();
            onNext.accept(n1.getValue(), n2.getValue(), n3.getValue(), n4.getValue());
        }
    }
}

@Override
protected void match() throws Exception {
    if (!first.queue().isEmpty()
            && !second.queue().isEmpty()
            && !third.queue().isEmpty()) {
        Notification<T1> n1 = first.queue().peek();
        Notification<T2> n2 = second.queue().peek();
        Notification<T3> n3 = third.queue().peek();

        if (n1.isOnComplete() || n2.isOnComplete() || n3.isOnComplete()) {
            onCompleted.run();
        } else {
            dequeue();
            onNext.accept(n1.getValue(), n2.getValue(), n3.getValue());
        }
    }
}

private void calcNext() {
  if (completed) {
    return;
  }
  if (next == null) {
    Notification<T> take = uncheck(queue::take);
    if (take.isOnNext()) {
      next = take;
    } else if (take.isOnError()) {
      completed = true;
      throw new RuntimeException(take.getError());
    } else {
      completed = true;
    }
  }
}

private void calcNext() {
  if (completed) {
    return;
  }
  if (next == null) {
    subscription.request(1);
    Notification<T> take = uncheck(() -> queue.take());
    if (take.isOnNext()) {
      next = take;
    } else if (take.isOnError()) {
      completed = true;
      throw new RuntimeException(take.getError());
    } else {
      completed = true;
    }
  }
}

@Test public void completeCompletesInner() {
  Observable<Message> messages = Observable.just(new Message("Bob", "Hello"));
  final AtomicInteger seen = new AtomicInteger();
  WindowIfChanged.create(messages, userSelector)
      .switchMap(
          new Function<GroupedObservable<String, Message>, Observable<Notification<String>>>() {
            @Override public Observable<Notification<String>> apply(
                GroupedObservable<String, Message> group) {
              final int count = seen.incrementAndGet();
              return group.map(new Function<Message, String>() {
                @Override public String apply(Message message) throws Exception {
                  return count + " " + message;
                }
              }).materialize();
            }
          })
      .test()
      .assertValues( //
          Notification.createOnNext("1 Bob Hello"), //
          Notification.<String>createOnComplete()) //
      .assertComplete();
}

@Test public void errorCompletesInner() {
  RuntimeException error = new RuntimeException("boom!");
  Observable<Message> messages = Observable.just( //
      Notification.createOnNext(new Message("Bob", "Hello")),
      Notification.createOnError(error)
  ).dematerialize();
  final AtomicInteger seen = new AtomicInteger();
  WindowIfChanged.create(messages, userSelector)
      .switchMap(
          new Function<GroupedObservable<String, Message>, Observable<Notification<String>>>() {
            @Override public Observable<Notification<String>> apply(
                GroupedObservable<String, Message> group) {
              final int count = seen.incrementAndGet();
              return group.map(new Function<Message, String>() {
                @Override public String apply(Message message) throws Exception {
                  return count + " " + message;
                }
              }).materialize();
            }
          })
      .test()
      .assertValues( //
          Notification.createOnNext("1 Bob Hello"), //
          Notification.<String>createOnComplete()) //
      .assertError(error);
}

private Notification<T> takeNotification() {
  Notification<T> notification = events.pollFirst();
  if (notification == null) {
    throw new AssertionError("No event found!");
  }
  return notification;
}

public T takeValue() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnNext())
      .as("Expected onNext event but was " + notification)
      .isTrue();
  return notification.getValue();
}

public Throwable takeError() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnError())
      .as("Expected onError event but was " + notification)
      .isTrue();
  return notification.getError();
}

public void assertComplete() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnComplete())
      .as("Expected onCompleted event but was " + notification)
      .isTrue();
  assertNoEvents();
}

private Notification<T> takeNotification() {
  Notification<T> notification = events.pollFirst();
  if (notification == null) {
    throw new AssertionError("No event found!");
  }
  return notification;
}

public T takeValue() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnNext())
      .as("Expected onNext event but was " + notification)
      .isTrue();
  return notification.getValue();
}

public Throwable takeError() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnError())
      .as("Expected onError event but was " + notification)
      .isTrue();
  return notification.getError();
}

public void assertComplete() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnComplete())
      .as("Expected onCompleted event but was " + notification)
      .isTrue();
  assertNoEvents();
}

private Notification<?> takeNotification() {
  Notification<?> notification = events.pollFirst();
  if (notification == null) {
    throw new AssertionError("No event found!");
  }
  return notification;
}

public Throwable takeError() {
  Notification<?> notification = takeNotification();
  assertThat(notification.isOnError())
      .as("Expected onError event but was " + notification)
      .isTrue();
  return notification.getError();
}

public void assertComplete() {
  Notification<?> notification = takeNotification();
  assertThat(notification.isOnComplete())
      .as("Expected onCompleted event but was " + notification)
      .isTrue();
  assertNoEvents();
}

private Notification<T> takeNotification() {
  Notification<T> notification = events.pollFirst();
  if (notification == null) {
    throw new AssertionError("No event found!");
  }
  return notification;
}

public T takeValue() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnNext())
      .as("Expected onNext event but was " + notification)
      .isTrue();
  return notification.getValue();
}

public Throwable takeError() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnError())
      .as("Expected onError event but was " + notification)
      .isTrue();
  return notification.getError();
}

private Notification<T> takeNotification() {
  Notification<T> notification = events.pollFirst();
  if (notification == null) {
    throw new AssertionError("No event found!");
  }
  return notification;
}

public T takeValue() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnNext())
      .as("Expected onNext event but was " + notification)
      .isTrue();
  return notification.getValue();
}

public Throwable takeError() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnError())
      .as("Expected onError event but was " + notification)
      .isTrue();
  return notification.getError();
}

private Notification<T> takeNotification() {
  Notification<T> notification = events.pollFirst();
  if (notification == null) {
    throw new AssertionError("No event found!");
  }
  return notification;
}

public T takeValue() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnNext())
      .as("Expected onNext event but was " + notification)
      .isTrue();
  return notification.getValue();
}

public Throwable takeError() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnError())
      .as("Expected onError event but was " + notification)
      .isTrue();
  return notification.getError();
}

public void assertComplete() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnComplete())
      .as("Expected onCompleted event but was " + notification)
      .isTrue();
  assertNoEvents();
}

private Notification<T> takeNotification() {
  Notification<T> notification = events.pollFirst();
  if (notification == null) {
    throw new AssertionError("No event found!");
  }
  return notification;
}

public T takeValue() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnNext())
      .as("Expected onNext event but was " + notification)
      .isTrue();
  return notification.getValue();
}

public Throwable takeError() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnError())
      .as("Expected onError event but was " + notification)
      .isTrue();
  return notification.getError();
}

public void assertComplete() {
  Notification<T> notification = takeNotification();
  assertThat(notification.isOnComplete())
      .as("Expected onCompleted event but was " + notification)
      .isTrue();
  assertNoEvents();
}