ThreadsStream extends AbstractStream { @Override public synchronized void put(ContentEvent event) { this.put(event, this.getNextCounter()); } ThreadsStream(IProcessingItem sourcePi); void addDestination(StreamDestination destination); List<StreamDestination> getDestinations(); @Override synchronized void put(ContentEvent event);  }
@Test public void testPut() { new NonStrictExpectations() { { event.getKey(); result=eventKey; destination.getProcessingItem(); result=destPi; destination.getPartitioningScheme(); result=scheme; destination.getParallelism(); result=parallelism; } }; switch(this.scheme) { case SHUFFLE: case GROUP_BY_KEY: new Expectations() { { destPi.processEvent(event, anyInt); times=1; } }; break; case BROADCAST: new Expectations() { { destPi.processEvent(event, anyInt); times=parallelism; } }; break; } stream.put(event); }