PersistenceJobStateChangeListener implements JobStateChangeListener { @Override public void consumeMessage(final MessageEvent messageEvent) { try { JobMessage jobMessage = convertMessage(messageEvent); jobService.appendMessage(messageEvent.getJobId(), jobMessage); } catch(RuntimeException e) { LOG.error("Failed to persist job message (jobId="+messageEvent.getJobId()+"): "+messageEvent.getMessage(), e); } } PersistenceJobStateChangeListener(final JobService jobService); @Override void consumeStateChange(final StateChangeEvent event); @Override void consumeMessage(final MessageEvent messageEvent);  }
@Test public void shouldPersistMessage() throws Exception { MessageEvent messageEvent = newMessageEvent(jobRunnableMock, JOB_ID, INFO, "some message", Optional.empty()); OffsetDateTime timestamp = OffsetDateTime.ofInstant(Instant.ofEpochMilli(messageEvent.getTimestamp()), ZoneId.systemDefault()); subject.consumeMessage(messageEvent); verify(jobServiceMock).appendMessage(JOB_ID, JobMessage.jobMessage(INFO, "some message", timestamp)); }