PersistenceJobStateChangeListener implements JobStateChangeListener { @Override public void consumeStateChange(final StateChangeEvent event) { try { switch (event.getState()) { case START: break; case KEEP_ALIVE: jobService.keepAlive(event.getJobId()); break; case FAILED: final OffsetDateTime ts = ofInstant(ofEpochMilli(event.getTimestamp()), systemDefault()); jobService.appendMessage(event.getJobId(), jobMessage(ERROR, event.getMessage(), ts)); break; case RESTART: jobService.markRestarted(event.getJobId()); break; case DEAD: jobService.killJob(event.getJobId()); break; case SKIPPED: jobService.markSkipped(event.getJobId()); jobService.stopJob(event.getJobId()); break; case STOP: jobService.stopJob(event.getJobId()); break; } } catch(RuntimeException e) { LOG.error("Failed to persist job state change: jobId="+event.getJobId()+", state="+event.getState()+", message="+event.getMessage(), e); } } PersistenceJobStateChangeListener(final JobService jobService); @Override void consumeStateChange(final StateChangeEvent event); @Override void consumeMessage(final MessageEvent messageEvent);  }
@Test public void shouldPersistStillAliveEvent() throws Exception { subject.consumeStateChange(stateChangedEvent(KEEP_ALIVE)); verify(jobServiceMock).keepAlive(JOB_ID); }
@Test public void shouldPersistRestartEvent() throws Exception { subject.consumeStateChange(stateChangedEvent(RESTART)); verify(jobServiceMock).markRestarted(JOB_ID); }
@Test public void shouldPersistDeadEvent() throws Exception { JobDefinition mockDefinition = mock(JobDefinition.class); when(mockDefinition.jobType()).thenReturn(JOB_TYPE); when(jobRunnableMock.getJobDefinition()).thenReturn(mockDefinition); subject.consumeStateChange(stateChangedEvent(DEAD)); verify(jobServiceMock).killJob(JOB_ID); }
@Test public void shouldPersistStopEvent() throws Exception { subject.consumeStateChange(stateChangedEvent(STOP)); verify(jobServiceMock).stopJob(JOB_ID); }
@Test public void shouldPersistFailedEvent() throws Exception { final StateChangeEvent event = stateChangedEvent(FAILED); subject.consumeStateChange(event); final OffsetDateTime ts = ofInstant(ofEpochMilli(event.getTimestamp()), systemDefault()); verify(jobServiceMock).appendMessage(JOB_ID, jobMessage(ERROR, "", ts)); }
@Test public void shouldPersistSkippedEvent() throws Exception { subject.consumeStateChange(stateChangedEvent(SKIPPED)); verify(jobServiceMock).markSkipped(JOB_ID); }
@Test public void shouldNotThrowIfStateChangeFailsInDatabase() throws Exception { doThrow(new RuntimeException("Unexpected disturbance in the force")).when(jobServiceMock).stopJob(JOB_ID); subject.consumeStateChange(stateChangedEvent(STOP)); verify(jobServiceMock).stopJob(JOB_ID); }