sclass SourceTriggeredStream<A> extends Meta {
  DoneFlag ended = new(true);
  volatile A lastElement;
  new AtomicLong elementCount; 

  event newElement(A a) {
    lastElement = a;
    inc(elementCount);
  }

  A get() { ret lastElement; }
  long elementCount() { ret elementCount!; }
  
  void end { ended.raise(); }
  bool ended() { ret ended.isUp(); }
}