ThreadsProcessingItem extends AbstractProcessingItem { public void processEvent(ContentEvent event, int counter) { if (this.piInstances == null || this.piInstances.size() < this.getParallelism()) throw new IllegalStateException("ThreadsWorkerProcessingItem(s) need to be setup before process any event (i.e. in ThreadsTopology.start())."); ThreadsProcessingItemInstance piInstance = this.piInstances.get(counter); ThreadsEventRunnable runnable = new ThreadsEventRunnable(piInstance, event); ThreadsEngine.getThreadWithIndex(piInstance.getThreadIndex()).submit(runnable); } ThreadsProcessingItem(Processor processor, int parallelismHint); List<ThreadsProcessingItemInstance> getProcessingItemInstances(); void processEvent(ContentEvent event, int counter); void setupInstances();  }
@Test(expected=IllegalStateException.class) public void testProcessEventError() { pi.processEvent(event, counter); }
@Test public void testProcessEvent() { new Expectations() { { for (int i=0; i<parallelism; i++) { processor.newProcessor(processor); result=processor; processor.onCreate(anyInt); } } }; pi.setupInstances(); instance = pi.getProcessingItemInstances().get(counter); new NonStrictExpectations() { { ThreadsEngine.getThreadWithIndex(anyInt); result=threadPool; } }; new Expectations() { { task = new ThreadsEventRunnable(instance, event); threadPool.submit(task); } }; pi.processEvent(event, counter); }