SimpleTopology 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("SimpleTopology supports 1 entrance PI only. Number of entrance PIs is "+this.getEntranceProcessingItems().size()); SimpleEntranceProcessingItem entrancePi = (SimpleEntranceProcessingItem) this.getEntranceProcessingItems().toArray()[0]; entrancePi.getProcessor().onCreate(0); entrancePi.startSendingEvents(); } SimpleTopology(String name); void run();  }
@Test public void testRun() { topology.addEntranceProcessingItem(entrancePi); new NonStrictExpectations() { { entrancePi.getProcessor(); result=entranceProcessor; } }; new Expectations() { { entranceProcessor.onCreate(anyInt); entrancePi.startSendingEvents(); } }; topology.run(); }
@Test(expected=IllegalStateException.class) public void testRunWithoutEntrancePI() { topology.run(); }