@Test public void canHandleJobStartAndStop() { StreamingJob job = mock(StreamingJob.class); TestJobController jobCtrl = setupMockJob(job); StreamingJob job2 = mock(StreamingJob.class); TestJobController jobCtrl2 = setupMockJob(job2); LaunchStreamingJobMessage launchMessage = StreamingJobTestUtil.createLaunchMessage(1); LaunchStreamingJobMessage launchMessage2 = StreamingJobTestUtil.createLaunchMessage(2); when(_mockJobFactory.createJob(launchMessage)) .thenReturn(job); when(_mockJobFactory.createJob(launchMessage2)) .thenReturn(job2); _childStreamingJobManager.handle(launchMessage); verify(job) .startJob(); _childStreamingJobManager.handle(launchMessage2); verify(job2) .startJob(); _childStreamingJobManager.handle(new StopStreamingJobMessage(1)); verify(job) .stopJob(); _childStreamingJobManager.handle(new StopStreamingJobMessage(2)); verify(job) .stopJob(); verify(_mockChannel, never()) .sendToMaster(any()); jobCtrl2.allowComplete(); verifyExitMessageSent(2, StreamingProcessExitReason.CANCELLED); verifyNoExitMessageSentForJob(1); jobCtrl.allowComplete(); verifyExitMessageSent(1, StreamingProcessExitReason.CANCELLED); }
public void handle(StreamingJobMessage message) { synchronized (_streamingJobs) { if (message instanceof LaunchStreamingJobMessage) { handleJobLaunch((LaunchStreamingJobMessage) message); } else if (message instanceof StopStreamingJobMessage) { handleJobStop((StopStreamingJobMessage) message); } else { throw new IllegalStateException("Unexpected message type: " + message); } } }
ChildStreamingJobManager { public void handle(StreamingJobMessage message) { synchronized (_streamingJobs) { if (message instanceof LaunchStreamingJobMessage) { handleJobLaunch((LaunchStreamingJobMessage) message); } else if (message instanceof StopStreamingJobMessage) { handleJobStop((StopStreamingJobMessage) message); } else { throw new IllegalStateException("Unexpected message type: " + message); } } } }
ChildStreamingJobManager { public void handle(StreamingJobMessage message) { synchronized (_streamingJobs) { if (message instanceof LaunchStreamingJobMessage) { handleJobLaunch((LaunchStreamingJobMessage) message); } else if (message instanceof StopStreamingJobMessage) { handleJobStop((StopStreamingJobMessage) message); } else { throw new IllegalStateException("Unexpected message type: " + message); } } } @Autowired  ChildStreamingJobManager(ChannelNode channelNode, StreamingJobFactory streamingJobFactory); }
ChildStreamingJobManager { public void handle(StreamingJobMessage message) { synchronized (_streamingJobs) { if (message instanceof LaunchStreamingJobMessage) { handleJobLaunch((LaunchStreamingJobMessage) message); } else if (message instanceof StopStreamingJobMessage) { handleJobStop((StopStreamingJobMessage) message); } else { throw new IllegalStateException("Unexpected message type: " + message); } } } @Autowired  ChildStreamingJobManager(ChannelNode channelNode, StreamingJobFactory streamingJobFactory); void handle(StreamingJobMessage message); }
ChildStreamingJobManager { public void handle(StreamingJobMessage message) { synchronized (_streamingJobs) { if (message instanceof LaunchStreamingJobMessage) { handleJobLaunch((LaunchStreamingJobMessage) message); } else if (message instanceof StopStreamingJobMessage) { handleJobStop((StopStreamingJobMessage) message); } else { throw new IllegalStateException("Unexpected message type: " + message); } } } @Autowired  ChildStreamingJobManager(ChannelNode channelNode, StreamingJobFactory streamingJobFactory); void handle(StreamingJobMessage message);  }