TaskFactoryUtil { public static TaskFactory getTaskFactory(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc) { if (appDesc instanceof TaskApplicationDescriptorImpl) { return ((TaskApplicationDescriptorImpl) appDesc).getTaskFactory(); } else if (appDesc instanceof StreamApplicationDescriptorImpl) { return (StreamOperatorTaskFactory) () -> new StreamOperatorTask( ((StreamApplicationDescriptorImpl) appDesc).getOperatorSpecGraph()); } throw new IllegalArgumentException(String.format("ApplicationDescriptorImpl has to be either TaskApplicationDescriptorImpl or " + "StreamApplicationDescriptorImpl. class %s is not supported", appDesc.getClass().getName())); }  static TaskFactory getTaskFactory(ApplicationDescriptorImpl<? extends ApplicationDescriptor> appDesc); static TaskFactory getTaskFactory(String taskClassName); static TaskFactory finalizeTaskFactory(TaskFactory factory, ExecutorService taskThreadPool);  }
@Test public void testStreamTaskClass() { TaskFactory retFactory = TaskFactoryUtil.getTaskFactory(MockStreamTask.class.getName()); assertTrue(retFactory instanceof StreamTaskFactory); assertTrue(((StreamTaskFactory) retFactory).createInstance() instanceof MockStreamTask); try { TaskFactoryUtil.getTaskFactory("no.such.class"); fail("Should have failed w/ no.such.class"); } catch (ConfigException cfe) { } }
@Test public void testAsyncStreamTask() { TaskFactory retFactory = TaskFactoryUtil.getTaskFactory(MockAsyncStreamTask.class.getName()); assertTrue(retFactory instanceof AsyncStreamTaskFactory); assertTrue(((AsyncStreamTaskFactory) retFactory).createInstance() instanceof MockAsyncStreamTask); try { TaskFactoryUtil.getTaskFactory("no.such.class"); fail("Should have failed w/ no.such.class"); } catch (ConfigException cfe) { } }
@Test public void testGetTaskFactoryWithStreamAppDescriptor() { StreamApplicationDescriptorImpl mockStreamApp = mock(StreamApplicationDescriptorImpl.class); OperatorSpecGraph mockSpecGraph = mock(OperatorSpecGraph.class); when(mockStreamApp.getOperatorSpecGraph()).thenReturn(mockSpecGraph); TaskFactory streamTaskFactory = TaskFactoryUtil.getTaskFactory(mockStreamApp); assertTrue(streamTaskFactory instanceof StreamOperatorTaskFactory); AsyncStreamTask streamTask = ((StreamOperatorTaskFactory) streamTaskFactory).createInstance(); assertTrue(streamTask instanceof StreamOperatorTask); verify(mockSpecGraph).clone(); }
@Test public void testGetTaskFactoryWithTaskAppDescriptor() { TaskApplicationDescriptorImpl mockTaskApp = mock(TaskApplicationDescriptorImpl.class); TaskFactory mockTaskFactory = mock(TaskFactory.class); when(mockTaskApp.getTaskFactory()).thenReturn(mockTaskFactory); TaskFactory taskFactory = TaskFactoryUtil.getTaskFactory(mockTaskApp); assertEquals(mockTaskFactory, taskFactory); }
@Test(expected = IllegalArgumentException.class) public void testGetTaskFactoryWithInvalidAddDescriptorImpl() { ApplicationDescriptorImpl mockInvalidApp = mock(ApplicationDescriptorImpl.class); TaskFactoryUtil.getTaskFactory(mockInvalidApp); }