MessageProcessor implements ITask, IMessageProcessor, IManagedTask, IShutdownAwareTask { @Override public void process(final IObject theMessage, final IObject theContext) throws InvalidArgumentException, MessageProcessorProcessException { this.message = theMessage; this.context = theContext; try { this.response = IOC.resolve(IOC.resolve(IOC.getKeyForKeyStorage(), "info.smart_tools.smartactors.iobject.iobject.IObject")); rawEnvironment.setValue(configFieldName, config); rawEnvironment.setValue(sequenceFieldName, messageProcessingSequence); rawEnvironment.setValue(responseFieldName, response); rawEnvironment.setValue(messageFieldName, theMessage); rawEnvironment.setValue(contextFieldName, theContext); rawEnvironment.setValue(processorFieldName, this); currentEnvironment = rawEnvironment; List finalActionsList = (List) context.getValue(finalActionsFieldName); if (null == finalActionsList) { finalActionsList = new ArrayList(1); context.setValue(finalActionsFieldName, finalActionsList); } finalActionsList.add((IAction<IObject>) env -> { try { upCounter.down(); } catch (UpCounterCallbackExecutionException | IllegalUpCounterState e) { throw new ActionExecuteException(e); } }); upCounter.up(); notifiedOnShutdown = false; shutdownStatus = null; } catch (ReadValueException | ChangeValueException | InvalidArgumentException | ResolutionException | IllegalUpCounterState e) { throw new MessageProcessorProcessException(e); } enqueue(); } MessageProcessor(final IQueue<ITask> taskQueue, final IMessageProcessingSequence messageProcessingSequence, final IObject config); @Override void process(final IObject theMessage, final IObject theContext); @Override void pauseProcess(); @Override void continueProcess(final Throwable e); @Override void signal(final String signalName); @Override IObject getContext(); @Override IObject getResponse(); @Override IObject getMessage(); @Override IMessageProcessingSequence getSequence(); @Override IObject getEnvironment(); @Override void pushEnvironment(final IObject newEnvironment); @Override void resetEnvironment(); @Override void setConfig(final IObject config); @Override void execute(); @Override T getAs(final Class<T> clazz); @Override void notifyShuttingDown(); @Override void notifyIgnored(); @Override void setShutdownStatus(final Object status); @Override Object getShutdownStatus();  }
@Test public void Should_interruptThread_When_putToQueueIsInterrupted() throws Exception { IMessageReceiver messageReceiverMock1 = mock(IMessageReceiver.class); when(messageProcessingSequenceMock.getCurrentReceiver()).thenReturn(messageReceiverMock1); doThrow(new InterruptedException()).when(taskQueueMock).put(any()); MessageProcessor messageProcessor = new MessageProcessor(taskQueueMock, messageProcessingSequenceMock, configurationMock); when(IOC.resolve(KEY_FOR_NEW_IOBJECT)) .thenReturn(responseMock); messageProcessor.process(messageMock, contextMock); verify(taskQueueMock).put(same(messageProcessor)); assertTrue(Thread.interrupted()); }
@Test public void Should_notOverwriteExistFinalActionsList() throws Exception { IAction<IObject> actionStub = env -> {}; List<IAction<IObject>> list = new ArrayList<>(); list.add(actionStub); when(contextMock.getValue(eq(new FieldName("finalActions")))).thenReturn(list); MessageProcessor messageProcessor = new MessageProcessor(taskQueueMock, messageProcessingSequenceMock, configurationMock); messageProcessor.process(messageMock, contextMock); verify(contextMock, times(0)).setValue(eq(new FieldName("finalActions")), any()); assertSame(actionStub, list.get(0)); assertEquals(2, list.size()); }