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

"{ NameSpace: Xtreams }"

ReadStream subclass:#ExternalReadStream
	instanceVariableNames:'cache cachePosition cacheDataSize'
	classVariableNames:''
	poolDictionaries:'XtreamsPool'
	category:'Xtreams-Terminals'
!

ExternalReadStream comment:'Used to read from BlockableIOAccessors (e.g sockets or pipes). Elements are read-ahead but only what''s available. This is a binary stream (produces bytes/ByteArrays).

{{{
	[ :in :out |
		[       out writing write: ''Hello''; close.
			in reading read: 5
		] ensure: [ in close. out close ]
	] valueWithArguments: SocketAccessor openPair
}}}
{{{
	[ :in :out |
		[       out writing write: ''Hello''; close.
			in reading read: 5
		] ensure: [ in close. out close ]
	] valueWithArguments: OSSystemSupport concreteClass pipeAccessorClass openPair
}}}

Instance Variables
	cache   <ByteArray | ByteString> read-ahead buffer
	cachePosition   <SmallInteger> position in the buffer
	cacheDataSize   <SmallInteger> size of valid data in the buffer

'
!


!ExternalReadStream methodsFor:'accessing'!

get
	cachePosition < cacheDataSize ifFalse: [
		cachePosition := 0.
		cacheDataSize := (1 max: self bytesForRead) min: cache size.
		cacheDataSize := self readBytes: cacheDataSize into: cache startingAt: 1.
		cacheDataSize isZero ifTrue: [ self incompleteRead: Incomplete zero ] ].
	^cache at: (cachePosition := cachePosition + 1)
!

read: anInteger into: aSequenceableCollection at: startIndex

	| index toRead |
	anInteger isZero ifTrue: [^0].
	index := startIndex.
	toRead := anInteger.
	[toRead > 0] whileTrue: [ | amount |
		cachePosition < cacheDataSize ifFalse: [
			cachePosition := 0.
			cacheDataSize := (toRead max: self bytesForRead) min: cache size.
			cacheDataSize := self readBytes: cacheDataSize into: cache startingAt: 1.
			cacheDataSize isZero ifTrue: [ self incompleteRead: (Incomplete on: aSequenceableCollection count: anInteger - toRead at: startIndex) ] ].
		amount := (cacheDataSize - cachePosition) min: toRead.
		aSequenceableCollection replaceFrom: index to: index + amount - 1 with: cache startingAt: cachePosition + 1.
		cachePosition := cachePosition + amount.
		index := index + amount.
		toRead := toRead - amount ].
	^anInteger
! !

!ExternalReadStream methodsFor:'initialize-release'!

close
	super close.
	cache recycle
!

contentsSpecies
	^cache class
!

contentsSpecies: aClass
	cache ifNotNil: [ cache recycle ].
	cache := aClass newRecycled: DefaultBufferSize.
	cachePosition := 0.
	cacheDataSize := 0.
!

on: anAccessor
	super on: anAccessor.
	self contentsSpecies: ByteArray
! !

!ExternalReadStream methodsFor:'private'!

bytesForRead

	^source numAvailable
!

incompleteRead: incomplete

	incomplete raise
!

readBytes:count into:aByteBuffer startingAt:firstIndex

    ^source nextBytes:count into:aByteBuffer startingAt:firstIndex
! !

!ExternalReadStream class methodsFor:'documentation'!

version_HG

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