"{ Package: 'stx:goodies/xtreams/terminals' }"

"{ NameSpace: Xtreams }"

ReadStream subclass:#BlockClosureGenerateStream
	instanceVariableNames:'process current contentsSpecies writing readingSemaphore
		writingSemaphore closed'
	classVariableNames:''
	poolDictionaries:''
	category:'Xtreams-Terminals'
!

BlockClosureGenerateStream comment:'Read stream on a one argument block, evaluates the block once for the life time of the stream. Once the block finishes execution, the stream is closed. The block is expected to write to the block argument, which will block exectuion until elements are read from the stream.
{{{
	"In this example, we have a -hard- loop, not using collection protocol, which will only run one element at a time."
	[:out | 1 to: 10 do: [:i | out put: i]] reading read: 5.
}}}
{{{
	"Fibonacci"
	[:out | | a b x |
	a := 0. b := 1.
	[out put: a.
	x := a.
	a := b.
	b := b + x] repeat] reading ++ 500; get.
}}}

Instance Variables
	current <Object>        the current element in the stream
	closed  <Boolean>       true if the block has finished execution
	contentsSpecies <Class> species for collections of elements of this stream
	process <Process>       the process executing the block
	readingSemaphore        <Semaphore>     attempts to read will wait on this
	writingSemaphore        <Semaphore>     attempts to write will wait on this, signaled when an attempt to read is performed

'
!


!BlockClosureGenerateStream methodsFor:'accessing'!

get
	writingSemaphore signal.
	closed ifTrue: [Incomplete zero raise].
	readingSemaphore wait.
	^current
!

read: anInteger into: aSequenceableCollection at: startIndex
	| count |
	count := 0.
	[[count < anInteger] whileTrue:
		[aSequenceableCollection at: startIndex + count put: self get.
		count := count + 1]]
			on: Incomplete do: [(Incomplete on: aSequenceableCollection count: count at: startIndex) raise].
	^anInteger
! !

!BlockClosureGenerateStream methodsFor:'initialize-release'!

close
	closed := true.
	process terminate.
	readingSemaphore signal
!

contentsSpecies
	^contentsSpecies
!

contentsSpecies: aClass
	contentsSpecies := aClass
!

on: aBlockClosure
	super on: aBlockClosure.
	contentsSpecies := Array.
	readingSemaphore := Semaphore new.
	writingSemaphore := Semaphore new.
	closed := false.
	process :=
		[aBlockClosure value: [:in |
			writingSemaphore wait.
			current := in.
			readingSemaphore signal] writing.
		closed := true] fork
! !

!BlockClosureGenerateStream class methodsFor:'documentation'!

version_HG

    ^ '$Changeset: <not expanded> $'
! !