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

"{ NameSpace: Xtreams }"

ReadStream subclass:#SharedQueueReadStream
	instanceVariableNames:'contentsSpecies'
	classVariableNames:''
	poolDictionaries:''
	category:'Xtreams-Terminals'
!

SharedQueueReadStream comment:'Read stream on a SharedQueue. Primarily used for data transfer between processes.
{{{
	queue := SharedQueue new.
	in := queue reading.
	out := queue writing.
	received := Array new writing.
	done := Semaphore new.
	consumer :=
		[ | size |
			[       (size := in get) isZero
			] whileFalse: [ | word |
				word := ByteString new: size.
				in read: size into: word.
				received put: word ].
			done signal.
		] fork.
	#(one two three four) do: [ :word | out put: word size; write: word ].
	out put: 0.
	done wait.
	received conclusion
}}}

Instance Variables
	contentsSpecies <Class> species for collections of elements of this stream

'
!


!SharedQueueReadStream methodsFor:'accessing'!

get
	^source next
!

read: anInteger into: aSequenceableCollection at: startIndex
	startIndex to: startIndex + anInteger - 1 do: [:index | aSequenceableCollection at: index put: source next].
	^anInteger
! !

!SharedQueueReadStream methodsFor:'initialize-release'!

close
!

contentsSpecies
	^contentsSpecies
!

contentsSpecies: aClass
	contentsSpecies := aClass
!

on: aSource
	super on: aSource.
	contentsSpecies := Array
! !

!SharedQueueReadStream methodsFor:'private'!

streamingInsertInto: aWriteStream
	self shouldNotImplement
!

streamingWriteInto: aWriteStream
	self shouldNotImplement
! !

!SharedQueueReadStream class methodsFor:'documentation'!

version_HG

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