@Test public void canHandleGeneralJobException() { StreamingJob job = mock(StreamingJob.class); TestJobController jobCtrl = setupMockJob(job); LaunchStreamingJobMessage launchMessage = StreamingJobTestUtil.createLaunchMessage(1); when(_mockJobFactory.createJob(launchMessage)) .thenReturn(job); _childStreamingJobManager.handle(launchMessage); verify(job) .startJob(); verifyNoExitMessageSentForJob(1); jobCtrl.causeException(new IllegalStateException("Intentional")); verifyExitMessageSent(1, StreamingProcessExitReason.UNEXPECTED_ERROR); }
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);  }