@Test public void testIsWorkToDo() { System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestContinuallyRunProcessorTask.class.getResource("/conf/nifi.properties").getFile()); final ProcessorNode procNode = Mockito.mock(ProcessorNode.class); Mockito.when(procNode.hasIncomingConnection()).thenReturn(false); assertTrue(ContinuallyRunProcessorTask.isWorkToDo(procNode)); final Connection selfLoopingConnection = Mockito.mock(Connection.class); when(selfLoopingConnection.getSource()).thenReturn(procNode); when(selfLoopingConnection.getDestination()).thenReturn(procNode); when(procNode.hasIncomingConnection()).thenReturn(true); when(procNode.getIncomingConnections()).thenReturn(Collections.singletonList(selfLoopingConnection)); assertTrue(ContinuallyRunProcessorTask.isWorkToDo(procNode)); final FlowFileQueue flowFileQueue = Mockito.mock(FlowFileQueue.class); when(flowFileQueue.isActiveQueueEmpty()).thenReturn(true); final FlowFileQueue nonEmptyQueue = Mockito.mock(FlowFileQueue.class); when(nonEmptyQueue.isActiveQueueEmpty()).thenReturn(false); when(selfLoopingConnection.getFlowFileQueue()).thenReturn(nonEmptyQueue); assertTrue(ContinuallyRunProcessorTask.isWorkToDo(procNode)); final Connection emptyConnection = Mockito.mock(Connection.class); when(emptyConnection.getSource()).thenReturn(Mockito.mock(ProcessorNode.class)); when(emptyConnection.getDestination()).thenReturn(procNode); when(emptyConnection.getFlowFileQueue()).thenReturn(flowFileQueue); when(procNode.getIncomingConnections()).thenReturn(Collections.singletonList(emptyConnection)); assertFalse(ContinuallyRunProcessorTask.isWorkToDo(procNode)); final Connection nonEmptyConnection = Mockito.mock(Connection.class); when(nonEmptyConnection.getSource()).thenReturn(Mockito.mock(ProcessorNode.class)); when(nonEmptyConnection.getDestination()).thenReturn(procNode); when(nonEmptyConnection.getFlowFileQueue()).thenReturn(nonEmptyQueue); when(procNode.getIncomingConnections()).thenReturn(Collections.singletonList(nonEmptyConnection)); assertTrue(ContinuallyRunProcessorTask.isWorkToDo(procNode)); }
static boolean isWorkToDo(final ProcessorNode procNode) { return procNode.isTriggerWhenEmpty() || !procNode.hasIncomingConnection() || !Connectables.hasNonLoopConnection(procNode) || Connectables.flowFilesQueued(procNode); }
ContinuallyRunProcessorTask implements Callable<Boolean> { static boolean isWorkToDo(final ProcessorNode procNode) { return procNode.isTriggerWhenEmpty() || !procNode.hasIncomingConnection() || !Connectables.hasNonLoopConnection(procNode) || Connectables.flowFilesQueued(procNode); } }
ContinuallyRunProcessorTask implements Callable<Boolean> { static boolean isWorkToDo(final ProcessorNode procNode) { return procNode.isTriggerWhenEmpty() || !procNode.hasIncomingConnection() || !Connectables.hasNonLoopConnection(procNode) || Connectables.flowFilesQueued(procNode); } ContinuallyRunProcessorTask(final SchedulingAgent schedulingAgent, final ProcessorNode procNode,
            final FlowController flowController, final ProcessContextFactory contextFactory, final ScheduleState scheduleState,
            final StandardProcessContext processContext); }
ContinuallyRunProcessorTask implements Callable<Boolean> { static boolean isWorkToDo(final ProcessorNode procNode) { return procNode.isTriggerWhenEmpty() || !procNode.hasIncomingConnection() || !Connectables.hasNonLoopConnection(procNode) || Connectables.flowFilesQueued(procNode); } ContinuallyRunProcessorTask(final SchedulingAgent schedulingAgent, final ProcessorNode procNode,
            final FlowController flowController, final ProcessContextFactory contextFactory, final ScheduleState scheduleState,
            final StandardProcessContext processContext); @Override Boolean call(); }
ContinuallyRunProcessorTask implements Callable<Boolean> { static boolean isWorkToDo(final ProcessorNode procNode) { return procNode.isTriggerWhenEmpty() || !procNode.hasIncomingConnection() || !Connectables.hasNonLoopConnection(procNode) || Connectables.flowFilesQueued(procNode); } ContinuallyRunProcessorTask(final SchedulingAgent schedulingAgent, final ProcessorNode procNode,
            final FlowController flowController, final ProcessContextFactory contextFactory, final ScheduleState scheduleState,
            final StandardProcessContext processContext); @Override Boolean call();  }