CheckpointActor { public void enter(final EnteringMessage message) throws ReadValueException, InvalidArgumentException, ResolutionException, ChangeValueException, SendingMessageException, SerializeException { IObject originalCheckpointStatus = message.getCheckpointStatus(); if (null != originalCheckpointStatus) { ISchedulerEntry presentEntry = storageObserver .getPresentEntry(originalCheckpointStatus.getValue(checkpointEntryIdFieldName).toString()); if (null != presentEntry) { sendFeedback(originalCheckpointStatus, message.getCheckpointId(), presentEntry.getId()); message.getProcessor().getSequence().end(); return; } } IObject entryArguments = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject")); entryArguments.setValue(schedulingFieldName, message.getSchedulingConfiguration()); entryArguments.setValue(messageFieldName, cloneMessage(message.getMessage())); if (null != originalCheckpointStatus) { entryArguments.setValue(prevCheckpointEntryIdFieldName, originalCheckpointStatus.getValue(responsibleCheckpointIdFieldName)); entryArguments.setValue(prevCheckpointIdFieldName, originalCheckpointStatus.getValue(checkpointEntryIdFieldName)); } entryArguments.setValue(recoverFieldName, message.getRecoverConfiguration()); entryArguments.setValue(actionFieldName, CHECKPOINT_ACTION); entryArguments.setValue(processorFieldName, message.getProcessor()); ISchedulerEntry entry = IOC.resolve(Keys.getOrAdd("new scheduler entry"), entryArguments, service.getEntryStorage()); IObject newCheckpointStatus = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject")); newCheckpointStatus.setValue(responsibleCheckpointIdFieldName, message.getCheckpointId()); newCheckpointStatus.setValue(checkpointEntryIdFieldName, entry.getId()); message.setCheckpointStatus(newCheckpointStatus); if (null != originalCheckpointStatus) { sendFeedback(originalCheckpointStatus, message.getCheckpointId(), entry.getId()); } } 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_createNewEntryForNewEnteringMessage() 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); IObject message = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"), "{'theUniqueAndVeryImportantField':'42'}".replace('\'','"')); when(enteringMessageMock.getCheckpointId()).thenReturn("thisCp"); when(enteringMessageMock.getCheckpointStatus()).thenReturn(null); when(enteringMessageMock.getMessage()).thenReturn(message); when(enteringMessageMock.getProcessor()).thenReturn(messageProcessorMock); when(enteringMessageMock.getRecoverConfiguration()).thenReturn(mock(IObject.class)); when(enteringMessageMock.getSchedulingConfiguration()).thenReturn(mock(IObject.class)); when(newEntryStrategyMock.resolve(objectsArgumentCaptor.capture())).thenReturn(entryMock[0]); when(entryMock[0].getId()).thenReturn("entry0"); actor.enter(enteringMessageMock); verify(newEntryStrategyMock).resolve(any(), any()); assertSame(storageMock, objectsArgumentCaptor.getAllValues().get(1)); IObject entryArgs = (IObject) objectsArgumentCaptor.getAllValues().get(0); assertSame(enteringMessageMock.getRecoverConfiguration(), entryArgs.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "recover"))); assertSame(enteringMessageMock.getSchedulingConfiguration(), entryArgs.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "scheduling"))); assertEquals("checkpoint scheduler action", entryArgs.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "action"))); assertEquals("42", ((IObject) entryArgs.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "message"))) .getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "theUniqueAndVeryImportantField"))); assertNotSame(message, entryArgs.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "message"))); assertSame(messageProcessorMock, entryArgs.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "processor"))); verify(enteringMessageMock).setCheckpointStatus(iObjectArgumentCaptor.capture()); IObject newCS = iObjectArgumentCaptor.getValue(); assertEquals("thisCp", newCS.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "responsibleCheckpointId"))); assertEquals("entry0", newCS.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "checkpointEntryId"))); }
@Test public void Should_interruptMessageProcessingWhenEntryIsPresent() 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); IObject status = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"), ("{" + "'responsibleCheckpointId':'prevCp'," + "'checkpointEntryId':'prevId'" + "}").replace('\'','"')); ISchedulerEntry oldEntry = entryMock[2]; when(oldEntry.getId()).thenReturn("newIdPresent"); when(oldEntry.getState()).thenReturn(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"), ("{" + "'prevCheckpointId':'prevCp'," + "'prevCheckpointEntryId':'prevId'" + "}").replace('\'','"'))); observer.onUpdateEntry(oldEntry); when(enteringMessageMock.getCheckpointId()).thenReturn("thisCp"); when(enteringMessageMock.getCheckpointStatus()).thenReturn(status); when(enteringMessageMock.getMessage()).thenReturn(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"))); when(enteringMessageMock.getProcessor()).thenReturn(messageProcessorMock); when(enteringMessageMock.getRecoverConfiguration()).thenReturn(mock(IObject.class)); when(enteringMessageMock.getSchedulingConfiguration()).thenReturn(mock(IObject.class)); when(newEntryStrategyMock.resolve(objectsArgumentCaptor.capture())).thenReturn(entryMock[0]); when(entryMock[0].getId()).thenReturn("entry0"); actor.enter(enteringMessageMock); verify(messageProcessingSequenceMock).end(); }
@Test public void Should_sendFeedbackMessage() 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); IObject status = IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"), ("{" + "'responsibleCheckpointId':'prevCp'," + "'checkpointEntryId':'prevId'" + "}").replace('\'','"')); when(enteringMessageMock.getCheckpointId()).thenReturn("thisCp"); when(enteringMessageMock.getCheckpointStatus()).thenReturn(status); when(enteringMessageMock.getMessage()).thenReturn(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.iobject.IObject"))); when(enteringMessageMock.getProcessor()).thenReturn(messageProcessorMock); when(enteringMessageMock.getRecoverConfiguration()).thenReturn(mock(IObject.class)); when(enteringMessageMock.getSchedulingConfiguration()).thenReturn(mock(IObject.class)); when(newEntryStrategyMock.resolve(objectsArgumentCaptor.capture())).thenReturn(entryMock[0]); when(entryMock[0].getId()).thenReturn("entry0"); actor.enter(enteringMessageMock); verify(messageProcessingSequenceMock, never()).end(); verify(messageBusHandlerMock).handle(iObjectArgumentCaptor.capture(), eq("checkpoint_feedback_chain__0")); IObject fbMessage = iObjectArgumentCaptor.getValue(); assertEquals("thisCp", fbMessage.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "responsibleCheckpointId"))); assertEquals("entry0", fbMessage.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "checkpointEntryId"))); assertEquals("prevCp", fbMessage.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "prevCheckpointId"))); assertEquals("prevId", fbMessage.getValue(IOC.resolve(Keys.getOrAdd("info.smart_tools.smartactors.iobject.ifield_name.IFieldName"), "prevCheckpointEntryId"))); }