"{ 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_SVN

    ^ '$Id$'

! !