ListDataStream implements DataStream<T> { @Override public void append(T t) { values.add(t); subscribers.forEach(s -> s.insert(t)); }  @SafeVarargs static ListDataStream<T> create(T... ts); @Override void append(T t); @Override void subscribe(DataProcessor subscriber);  }
@Test public void testAppend() { Counter counter = new Counter(); ListDataStream<Integer> integers = ListDataStream.create(); integers.subscribe(counter); assertEquals(0, counter.count); integers.append(10); assertEquals(1, counter.count); integers.append(20); integers.append(30); assertEquals(3, counter.count); }