SourceTriggeredStream - a stream of objects that come in at irregular times [1033671]

< > BotCompany Repo | #1033671 // SourceTriggeredStream - a stream of objects that come in at irregular times

JavaX fragment (include) [tags: use-pretranspiled]

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

  event newElement(A a) {
    lastElement = a;
    inc(elementCount);
  }
  
  void onNewElement(Runnable r) { onNewElement(runnableToIVF1(r)); }

  A get() { ret lastElement; }
  long elementCount() { ret elementCount!; }
  
  void end { ended.raise(); }
  bool ended() { ret ended.isUp(); }
  
  void catchError(Runnable r) {
    main catchError(ended, r);
  }
  
  // on new element, call consumer in same thread
  void directlyFeedInto(Consumer<A> consumer) {
    if (consumer != null)
      onNewElement(a -> consumer.accept(a));
  }
}

Travelled to 4 computer(s): bhatertpkbcr, ekrmjmnbrukm, mowyntqkapby, mqqgnosmbjvj

Snippet ID:

#1033671

Snippet name:

SourceTriggeredStream - a stream of objects that come in at irregular times

Eternal ID of this version:

#1033671/16

Text MD5:

f3ca9c6b871be04ad30efaeb429d73c2

Transpilation MD5:

533bae006213bc36a0eafffc077b14d6

Author:

stefan

Category:

javax / streams

Type:

JavaX fragment (include)

Public (visible to everyone):

Yes

Archived (hidden from active list):

Created/modified:

2021-12-28 21:29:07

Source code size:

703 bytes / 28 lines

Pitched / IR pitched:

No / No

Views / Downloads:

124 / 257

Version history:

15 change(s)

Referenced in:

[show references]

1	sclass SourceTriggeredStream<A> extends Meta {
2	DoneFlag ended = new(true);
3	volatile A lastElement;
4	new AtomicLong elementCount;
5
6	event newElement(A a) {
7	lastElement = a;
8	inc(elementCount);
9	}
10
11	void onNewElement(Runnable r) { onNewElement(runnableToIVF1(r)); }
12
13	A get() { ret lastElement; }
14	long elementCount() { ret elementCount!; }
15
16	void end { ended.raise(); }
17	bool ended() { ret ended.isUp(); }
18
19	void catchError(Runnable r) {
20	main catchError(ended, r);
21	}
22
23	// on new element, call consumer in same thread
24	void directlyFeedInto(Consumer<A> consumer) {
25	if (consumer != null)
26	onNewElement(a -> consumer.accept(a));
27	}
28	}