CheckpointActor { public void feedback(final FeedbackMessage message) throws ReadValueException, ChangeValueException, InvalidArgumentException, EntryScheduleException, EntryStorageAccessException { try { ISchedulerEntry entry = service.getEntryStorage().getEntry(message.getPrevCheckpointEntryId()); if (null != entry.getState().getValue(completedFieldName)) { return; } entry.pause(); try { entry.scheduleNext(System.currentTimeMillis() + COMPLETE_ENTRY_RESCHEDULE_DELAY); entry.getState().setValue(gotFeedbackFieldName, true); entry.getState().setValue(completedFieldName, true); } finally { entry.unpause(); } } catch (EntryNotFoundException | EntryPauseException ignore) { } } CheckpointActor(final IObject args); void enter(final EnteringMessage message); void feedback(final FeedbackMessage message); void start(final StartStopMessage message); void stop(final StartStopMessage message); void configure(final ConfigureMessage message);  }
@Test public void Should_processFeedbackMessages() throws Exception { long startTime = System.currentTimeMillis(); IObject args = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"), ("{" + "'connectionOptionsDependency':'the connection options'," + "'connectionPoolDependency':'the connection pool'," + "'collectionName':'" + collectionName + "'" + "}").replace('\'','"')); IObject entryState = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject")); CheckpointActor actor = new CheckpointActor(args); when(feedbackMessageMock.getPrevCheckpointEntryId()).thenReturn("prevId"); when(storageMock.getEntry("prevId")).thenReturn(entryMock[0]); when(entryMock[0].getState()).thenReturn(entryState); actor.feedback(feedbackMessageMock); ArgumentCaptor<Long> timeCaptor = ArgumentCaptor.forClass(long.class); assertNotNull(entryState.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "completed"))); verify(entryMock[0]).scheduleNext(timeCaptor.capture()); assertTrue(timeCaptor.getValue() >= (startTime + 1000)); }
@Test public void Should_notThrowWhenNoMatchingEntryFoundProcessingFeedbackMessage() throws Exception { IObject args = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"), ("{" + "'connectionOptionsDependency':'the connection options'," + "'connectionPoolDependency':'the connection pool'," + "'collectionName':'" + collectionName + "'" + "}").replace('\'','"')); CheckpointActor actor = new CheckpointActor(args); when(feedbackMessageMock.getPrevCheckpointEntryId()).thenReturn("prevId"); when(storageMock.getEntry("prevId")).thenThrow(EntryNotFoundException.class); actor.feedback(feedbackMessageMock); }
@Test public void Should_ignoreFeedbackMessagesIfTheEntryIsAlreadyCompleted() throws Exception { IObject args = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"), ("{" + "'connectionOptionsDependency':'the connection options'," + "'connectionPoolDependency':'the connection pool'," + "'collectionName':'" + collectionName + "'" + "}").replace('\'','"')); IObject entryState = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"), ("{'completed':true}").replace('\'','"')); CheckpointActor actor = new CheckpointActor(args); when(feedbackMessageMock.getPrevCheckpointEntryId()).thenReturn("prevId"); when(storageMock.getEntry("prevId")).thenReturn(entryMock[0]); when(entryMock[0].getState()).thenReturn(entryState); actor.feedback(feedbackMessageMock); verify(entryMock[0]).getState(); verifyNoMoreInteractions(entryMock[0]); }