@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); }
@Override public void put(ContentEvent event) { this.put(event, this.getNextCounter()); }
SimpleStream extends AbstractStream { @Override public void put(ContentEvent event) { this.put(event, this.getNextCounter()); } }
SimpleStream extends AbstractStream { @Override public void put(ContentEvent event) { this.put(event, this.getNextCounter()); } SimpleStream(IProcessingItem sourcePi); }
SimpleStream extends AbstractStream { @Override public void put(ContentEvent event) { this.put(event, this.getNextCounter()); } SimpleStream(IProcessingItem sourcePi); @Override void put(ContentEvent event); void addDestination(StreamDestination destination); }
SimpleStream extends AbstractStream { @Override public void put(ContentEvent event) { this.put(event, this.getNextCounter()); } SimpleStream(IProcessingItem sourcePi); @Override void put(ContentEvent event); void addDestination(StreamDestination destination);  }