ThreadsTopology extends AbstractTopology { public void run() { if (this.getEntranceProcessingItems() == null) throw new IllegalStateException("You need to set entrance PI before running the topology."); if (this.getEntranceProcessingItems().size() != 1) throw new IllegalStateException("ThreadsTopology supports 1 entrance PI only. Number of entrance PIs is "+this.getEntranceProcessingItems().size()); this.setupProcessingItemInstances(); ThreadsEntranceProcessingItem entrancePi = (ThreadsEntranceProcessingItem) this.getEntranceProcessingItems().toArray()[0]; if (entrancePi == null) throw new IllegalStateException("You need to set entrance PI before running the topology."); entrancePi.getProcessor().onCreate(0); entrancePi.startSendingEvents(); } ThreadsTopology(String name); void run();  }
@Test public void testRun() { topology.addEntranceProcessingItem(entrancePi); new Expectations() { { entrancePi.getProcessor(); result=entranceProcessor; entranceProcessor.onCreate(anyInt); entrancePi.startSendingEvents(); } }; topology.run(); }
@Test(expected=IllegalStateException.class) public void testRunWithoutEntrancePI() { topology.run(); }