@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() {
			{
				
				// TODO: restrict the range of counter value
				destPi.processEvent(event, anyInt); times=1;
			}
		};
		break;
	case BROADCAST:
		new Expectations() {
			{
				// TODO: restrict the range of counter value
				destPi.processEvent(event, anyInt); times=parallelism;
			}
		};
		break;
	}
	stream.put(event);
}