@Test public void testNoPruning() { long taskATimestamp = clock.nowMillis(); IScheduledTask a = makeTask("a", FINISHED); clock.advance(ONE_MS); long taskBTimestamp = clock.nowMillis(); IScheduledTask b = makeTask("b", LOST); expectNoImmediatePrune(ImmutableSet.of(a)); expectOneDelayedPrune(taskATimestamp); expectNoImmediatePrune(ImmutableSet.of(a, b)); expectOneDelayedPrune(taskBTimestamp); control.replay(); pruner.recordStateChange(TaskStateChange.initialized(a)); pruner.recordStateChange(TaskStateChange.initialized(b)); }
@Subscribe public void recordStateChange(TaskStateChange change) { if (Tasks.isTerminated(change.getNewState())) { long timeoutBasis = change.isTransition() ? clock.nowMillis() : Iterables.getLast(change.getTask().getTaskEvents()).getTimestamp(); registerInactiveTask( Tasks.getJob(change.getTask()), change.getTaskId(), calculateTimeout(timeoutBasis)); } }
TaskHistoryPruner implements EventSubscriber { @Subscribe public void recordStateChange(TaskStateChange change) { if (Tasks.isTerminated(change.getNewState())) { long timeoutBasis = change.isTransition() ? clock.nowMillis() : Iterables.getLast(change.getTask().getTaskEvents()).getTimestamp(); registerInactiveTask( Tasks.getJob(change.getTask()), change.getTaskId(), calculateTimeout(timeoutBasis)); } } }
TaskHistoryPruner implements EventSubscriber { @Subscribe public void recordStateChange(TaskStateChange change) { if (Tasks.isTerminated(change.getNewState())) { long timeoutBasis = change.isTransition() ? clock.nowMillis() : Iterables.getLast(change.getTask().getTaskEvents()).getTimestamp(); registerInactiveTask( Tasks.getJob(change.getTask()), change.getTaskId(), calculateTimeout(timeoutBasis)); } } @Inject  TaskHistoryPruner(
      @AsyncExecutor DelayExecutor executor,
      StateManager stateManager,
      Clock clock,
      HistoryPrunnerSettings settings,
      Storage storage,
      Lifecycle lifecycle); }
TaskHistoryPruner implements EventSubscriber { @Subscribe public void recordStateChange(TaskStateChange change) { if (Tasks.isTerminated(change.getNewState())) { long timeoutBasis = change.isTransition() ? clock.nowMillis() : Iterables.getLast(change.getTask().getTaskEvents()).getTimestamp(); registerInactiveTask( Tasks.getJob(change.getTask()), change.getTaskId(), calculateTimeout(timeoutBasis)); } } @Inject  TaskHistoryPruner(
      @AsyncExecutor DelayExecutor executor,
      StateManager stateManager,
      Clock clock,
      HistoryPrunnerSettings settings,
      Storage storage,
      Lifecycle lifecycle); @Subscribe void recordStateChange(TaskStateChange change); }
TaskHistoryPruner implements EventSubscriber { @Subscribe public void recordStateChange(TaskStateChange change) { if (Tasks.isTerminated(change.getNewState())) { long timeoutBasis = change.isTransition() ? clock.nowMillis() : Iterables.getLast(change.getTask().getTaskEvents()).getTimestamp(); registerInactiveTask( Tasks.getJob(change.getTask()), change.getTaskId(), calculateTimeout(timeoutBasis)); } } @Inject  TaskHistoryPruner(
      @AsyncExecutor DelayExecutor executor,
      StateManager stateManager,
      Clock clock,
      HistoryPrunnerSettings settings,
      Storage storage,
      Lifecycle lifecycle); @Subscribe void recordStateChange(TaskStateChange change);  }