io.reactivex.functions.Consumer#accept ( )源码实例Demo

@Test public void asConsumer() throws Exception {
  Preference<String> preference = rxPreferences.getString("foo");
  Consumer<? super String> consumer = preference.asConsumer();

  consumer.accept("bar");
  assertThat(preferences.getString("foo", null)).isEqualTo("bar");

  consumer.accept("baz");
  assertThat(preferences.getString("foo", null)).isEqualTo("baz");

  try {
    consumer.accept(null);
    fail("Disallow accepting null.");
  } catch (NullPointerException e) {
    assertThat(e).hasMessage("value == null");
  }
}
 

@Test
public void showFeedbackHandlerInvokesAppropriateViewMethods() throws Exception {
  TasksListViewActions view = mock(TasksListViewActions.class);
  Consumer<TasksListEffect.ShowFeedback> underTest = showFeedbackHandler(view);

  underTest.accept(showFeedback(FeedbackType.LOADING_ERROR).asShowFeedback());
  verify(view).showLoadingTasksError();

  reset(view);
  underTest.accept(showFeedback(FeedbackType.CLEARED_COMPLETED).asShowFeedback());
  verify(view).showCompletedTasksCleared();

  reset(view);
  underTest.accept(showFeedback(FeedbackType.MARKED_ACTIVE).asShowFeedback());
  verify(view).showTaskMarkedActive();

  reset(view);
  underTest.accept(showFeedback(FeedbackType.MARKED_COMPLETE).asShowFeedback());
  verify(view).showTaskMarkedComplete();

  reset(view);
  underTest.accept(showFeedback(FeedbackType.SAVED_SUCCESSFULLY).asShowFeedback());
  verify(view).showSuccessfullySavedMessage();
}
 

private static void sendThenDeleteCrashLogs(Consumer<List<CrashInfo>> consumer) throws Exception {
    File[] files = TombstoneManager.getAllTombstones();
    List<CrashInfo> crashes = new ArrayList<>();
    for (File f : files) {
        try {
            crashes.add(wrapCrashMessage(TombstoneParser.parse(f.getAbsolutePath(), null)));
        } catch (IOException e) {
            L.e(e);
        }
    }
    if (!crashes.isEmpty()) {
        L.d("Crash produce message when install, crash count:%s", crashes.size());
        consumer.accept(crashes);
        TombstoneManager.clearAllTombstones();
    }
}
 

@Override
public void onError(Throwable t) {
  if (!setDone()) {
    RxJavaPlugins.onError(t);
    return;
  }

  Objects.requireNonNull(t, "onError called with null");

  Consumer<? super Throwable> c;
  synchronized (this) {
    c = observableErrorHandler;
  }
  try {
    if (c != null) {
      c.accept(t);
    }
  } catch (Throwable t1) {
    Exceptions.throwIfFatal(t1);
    RxJavaPlugins.onError(t1);
  }
}
 

private void writeStreamEnd(AsyncResult<Void> result) {
  try {
    Action a;
    if (result.succeeded()) {
      synchronized (this) {
        a = writeStreamEndHandler;
      }
      if (a != null) {
        a.run();
      }
    } else {
      Consumer<? super Throwable> c;
      synchronized (this) {
        c = this.writeStreamEndErrorHandler;
      }
      if (c != null) {
        c.accept(result.cause());
      }
    }
  } catch (Throwable t) {
    Exceptions.throwIfFatal(t);
    RxJavaPlugins.onError(t);
  }
}
 

@Override
public void onError(Throwable t) {
  if (!setDone()) {
    RxJavaPlugins.onError(t);
    return;
  }

  Objects.requireNonNull(t, "onError called with null");

  Consumer<? super Throwable> c;
  synchronized (this) {
    c = flowableErrorHandler;
  }
  try {
    if (c != null) {
      c.accept(t);
    }
  } catch (Throwable t1) {
    Exceptions.throwIfFatal(t1);
    RxJavaPlugins.onError(t1);
  }
}
 

private void writeStreamEnd(AsyncResult<Void> result) {
  try {
    Action a;
    if (result.succeeded()) {
      synchronized (this) {
        a = writeStreamEndHandler;
      }
      if (a != null) {
        a.run();
      }
    } else {
      Consumer<? super Throwable> c;
      synchronized (this) {
        c = this.writeStreamEndErrorHandler;
      }
      if (c != null) {
        c.accept(result.cause());
      }
    }
  } catch (Throwable t) {
    Exceptions.throwIfFatal(t);
    RxJavaPlugins.onError(t);
  }
}
 

@Test
public void savingTasksAddsThemToBothLocalAndRemoteDataSources() throws Exception {
  FakeDataSource remote = new FakeDataSource();
  FakeDataSource local = new FakeDataSource();
  Consumer<SaveTask> saveTaskConsumer = saveTaskHandler(remote, local);
  saveTaskConsumer.accept(saveTask(TASK_1).asSaveTask());
  assertThat(remote.tasks, contains(TASK_1));
  assertThat(local.tasks, contains(TASK_1));
}
 

@Test
public void deletingTasksRemovesThemFromBothLocalAndRemoteDataSources() throws Exception {
  FakeDataSource remote = new FakeDataSource();
  FakeDataSource local = new FakeDataSource();
  remote.addTasks(TASK_1, TASK_2, TASK_3);
  local.addTasks(TASK_1, TASK_2, TASK_3);
  Consumer<TasksListEffect.DeleteTasks> saveTaskConsumer = deleteTasksHandler(remote, local);
  saveTaskConsumer.accept(deleteTasks(ImmutableList.of(TASK_1, TASK_3)).asDeleteTasks());

  assertThat(remote.tasks, allOf(not(hasItems(TASK_1)), not(hasItems(TASK_3))));
  assertThat(local.tasks, allOf(not(hasItems(TASK_1)), not(hasItems(TASK_3))));

  assertThat(remote.tasks, hasItems(TASK_2));
  assertThat(local.tasks, hasItems(TASK_2));
}
 

protected static <T> Consumer<T> wrapQueueConsumer(Consumer<T> action, IRxBusQueue isResumedProvider)
{
    return new Consumer<T>()
    {
        @Override
        public void accept(T t) throws Exception
        {
            if (RxUtil.safetyQueueCheck(t, isResumedProvider))
                action.accept(t);
        }
    };
}
 

public static <T> Consumer<T> wrapQueueAction(Consumer<T> action, IRxBusQueue isResumedProvider)
{
    return new Consumer<T>()
    {
        @Override
        public void accept(T t) throws Exception
        {
            if (RxUtil.safetyQueueCheck(t, isResumedProvider))
                action.accept(t);
        }
    };
}
 

public static <T> Consumer<T> wrapQueueAction(Consumer<T> consumer, IRxBusQueue isResumedProvider)
{
    return new Consumer<T>()
    {
        @Override
        public void accept(T t) throws Exception
        {
            if (RxUtil.safetyQueueCheck(t, isResumedProvider))
                consumer.accept(t);
        }
    };
}
 

private static void sendThenDeleteCrashLog(String logPath, String emergency, CrashConfig crashContext, Consumer<List<CrashInfo>> consumer) throws Exception {
    if (emergency != null || crashContext.immediate()) {// if emergency or immediate,output right now
        L.d("Crash produce message when emergency or immediate, crash count:%s, emergency:%s, logPath:%s", 1, emergency, logPath);
        consumer.accept(Collections.singletonList(wrapCrashMessage(TombstoneParser.parse(logPath, emergency))));
        TombstoneManager.deleteTombstone(logPath);
    }
}
 

private static Consumer<ErrorAndDuration> callActionExceptForLast(final Consumer<? super ErrorAndDuration> action) {
    return new Consumer<ErrorAndDuration>() {
        @Override
        public void accept(ErrorAndDuration e) throws Exception {
            if (e.durationMs() != NO_MORE_DELAYS) {
                action.accept(e);
            }
        }
    };
}
 

private void forAllObservables(Consumer<ManagedObservable<?>> action) {
  for (Map<String, ManagedObservable<?>> observableMap : groupMap.values()) {
    for (ManagedObservable<?> managedObservable : observableMap.values()) {
      try {
        action.accept(managedObservable);
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
  }
}
 

static Consumer<NavigateToTaskDetails> navigateToDetailsHandler(Consumer<Task> command) {
  return navigationEffect -> command.accept(navigationEffect.task());
}
 

public static <E> void forEach(java.lang.Iterable<E> iterable, Consumer<? super E> consumer) {
  Objects.requireNonNull(consumer);
  for (E item : iterable) {
    consumer.accept(item);
  }
}
 

public void ifPresent(Consumer<? super T> consumer) {
  if (value != null) {
    consumer.accept(value);
  }
}
 

public static <T> Predicate<T> check(Consumer<T> consumer) {
    return t -> {
        consumer.accept(t);
        return true;
    };
}
 

public static <T> Predicate<T> check(Consumer<T> consumer) {
    return t -> {
        consumer.accept(t);
        return true;
    };
}