### Question:
MovingAverage implements Processor<Tuple, Tuple> { @Override public Stream<Tuple> process(Stream<Tuple> inputStream) { return inputStream.map(tuple -> { return tuple.getDouble("measurement"); }) .buffer(5) .map(data -> tuple().of("average", avg(data))); }  @Override Stream<Tuple> process(Stream<Tuple> inputStream);  }### Answer:
@Test public void simple() { Environment.initializeIfEmpty(); final Broadcaster<Object> broadcaster = SerializedBroadcaster.create(); Processor processor = new MovingAverage(); Stream<?> outputStream = processor.process(broadcaster); outputStream.consume(new Consumer<Object>() { @Override public void accept(Object o) { System.out.println("processed : " + o); } }); List<Tuple> inputData = new ArrayList<Tuple>(); for (int i = 0; i < 10; i++) { inputData.add(TupleBuilder.tuple().of("id", i, "measurement", new Double(i+10))); } for (Tuple tuple: inputData) { broadcaster.onNext(tuple); } }