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

"{ NameSpace: Xtreams }"

Object subclass:#Buffer

	instanceVariableNames:'cache readPosition writePosition dataLength'

	classVariableNames:''

	poolDictionaries:''

	category:'Xtreams-Core'

!

Buffer comment:'Buffer implements a buffering API over an in-memory cache. The cache will grow as required to fit new data in to the buffer. Two positions are kept, the read position and write position, allowing flexible usage of the buffer.

API:

	read:into:startingAt: -- reads data from the cache and moves readPosition forward

	write:into:startingAt: -- writes data into the cache and moves writePosition forward

	readSkip:/writeSkip: -- moves either readPosition or writePosition forward without changing the cache

	readPosition/readPosition:/writePosition/writePosition: -- accessors to the read/write positions

	clear -- remove all the data from the cache

	trim -- remove all the read data and all the empty space for writing in to from the cache

Instance Variables

	cache   <SequenceableCollection>        the cache for our buffer

	dataLength      <ArithmeticValue>       the amount of data in our cache, not the size of the cache

	readPosition    <ArithmeticValue>       the position within our data that we''re reading from (0..dataLength)

	writePosition   <ArithmeticValue>       the position within our data that we''re writing from (0..dataLength)

'

!

!Buffer class methodsFor:'instance creation'!

new: anInteger class: aClass

	^self on: (aClass newRecycled: anInteger)

!

on: aSequenceableCollection

	^self new on: aSequenceableCollection

! !

!Buffer methodsFor:'accessing'!

activeSize

	"The number of elements in the buffer that can be explored. This is <= cacheSize"

	^dataLength

!

cache

	^cache

!

cacheSize

	"The size of the cache, which is >= dataLength"

	^cache size

!

contentsFuture

	^cache copyFrom: readPosition + 1 to: readPosition + self readSize

!

contentsPast

	^cache copyFrom: 1 to: writePosition

!

inactiveSize

	"Free space to write in to"

	^cache size - dataLength

!

readSize

	"The number of elements available to read"

	^dataLength - readPosition

!

writeSize

	"The number of available slots to write in to, which might overwrite elements in dataLength; writeSize >= inactiveSize"

	^(cache size - writePosition + readPosition) min: cache size

! !

!Buffer methodsFor:'accessing - reading'!

get

	dataLength = readPosition ifTrue: [Incomplete zero raise].

	readPosition := readPosition + 1.

	^cache at: readPosition

!

read: anInteger into: aSequenceableCollection at: startIndex

	| count |

	count := anInteger min: (dataLength - readPosition).

	aSequenceableCollection replaceFrom: startIndex to: startIndex + count - 1 with: cache startingAt: readPosition + 1.

	readPosition := readPosition + count.

	count < anInteger ifTrue: [(Incomplete on: aSequenceableCollection count: count at: startIndex) raise].

	^anInteger

!

readPosition

	^readPosition

!

readPosition: aPosition

	readPosition := (aPosition max: 0) min: dataLength

!

readSkip: anInteger

	| old delta |

	old := readPosition.

	self readPosition: old + anInteger.

	delta := readPosition - old.

	^delta

! !

!Buffer methodsFor:'accessing - writing'!

insert: aStreamable

	^aStreamable streamingInsertInto: self

!

insert: anInteger from: aStreamable

	^aStreamable streamingInsert: anInteger into: self

!

insert: anInteger from: aSequenceableCollection at: startIndex

	dataLength + anInteger <= cache size ifFalse: [self growBy: anInteger].

	cache replaceFrom: writePosition + anInteger + 1 to: dataLength + anInteger with: cache startingAt: writePosition + 1.

	cache replaceFrom: writePosition + 1 to: writePosition + anInteger with: aSequenceableCollection startingAt: startIndex.

	writePosition := writePosition + anInteger.

	dataLength := dataLength + anInteger.

	^anInteger

!

put: anObject

	writePosition < cache size ifFalse: [self growBy: 1].

	cache at: writePosition + 1 put: anObject.

	writePosition := writePosition + 1.

	writePosition > dataLength ifTrue: [dataLength := writePosition].

	^anObject

!

write: aStreamable

	^aStreamable streamingWriteInto: self

!

write: anInteger from: aStreamable

	^aStreamable streamingWrite: anInteger into: self

!

write: anInteger from: aSequenceableCollection at: startIndex

	writePosition + anInteger <= cache size ifFalse: [self growBy: anInteger].

	self privateWrite: anInteger from: aSequenceableCollection at: startIndex.

	^anInteger

!

writePosition

	^writePosition

!

writePosition: aPosition

	^writePosition := (aPosition max: 0) min: dataLength

!

writeSkip: anInteger

	| old delta |

	old := writePosition.

	self writePosition: old + anInteger.

	delta := writePosition - old.

	^delta

! !

!Buffer methodsFor:'compressing'!

clear

	self clearCache.

	readPosition := writePosition := dataLength := 0

!

trim

	cache := cache copyFrom: readPosition + 1 to: writePosition.

	readPosition := 0.

	dataLength := writePosition := cache size

! !

!Buffer methodsFor:'initialize-release'!

close

	dataLength := writePosition

!

contentsSpecies

	^cache species

!

on: aSequenceableCollection

	cache := aSequenceableCollection.

	dataLength := 0.

	readPosition := 0.

	writePosition := 0

!

recycle

	cache recycle

! !

!Buffer methodsFor:'printing'!

printOn: aStream

	| stream |

	stream := String new writing.

	self streamingPrintOn: stream.

	aStream nextPutAll: stream conclusion

!

streamingPrintOn: aStream

	aStream write: self class name;

		space; write: 'data: '; print: dataLength;

		space; write: 'read: '; print: self readPosition;

		space; write: 'write: '; print: self writePosition

! !

!Buffer methodsFor:'private'!

clearCache

	cache := cache copyEmpty: 0

!

growBy: anInteger

	| replacement |

	replacement := cache copyEmpty: (cache size + anInteger) * 2.

	replacement replaceFrom: 1 to: cache size with: cache startingAt: 1.

	cache := replacement

!

privateWrite: anInteger from: aSequenceableCollection at: startIndex

	" Ensure we're only doing with (writePosition + anInteger) <= cache size "

	(writePosition + anInteger) <= cache size ifFalse: [self error: 'invalid privateWrite. Use the #write: protocol instead.'].

	cache replaceFrom: writePosition + 1 to: writePosition + anInteger with: aSequenceableCollection startingAt: startIndex.

	writePosition := writePosition + anInteger.

	writePosition > dataLength ifTrue: [dataLength := writePosition]

!

streamingInsert: anInteger into: aWriteStream

	| count |

	count := (writePosition - readPosition) min: anInteger.

	count <= 0 ifTrue: [^0].

	aWriteStream insert: count from: cache at: readPosition + 1.

	self readSkip: count

!

streamingInsertInto: aWriteStream

	| count |

	count := writePosition - readPosition.

	count <= 0 ifTrue: [^0].

	self streamingInsert: count into: aWriteStream.

	^count

!

streamingWrite: anInteger from: aReadStream

	"Implementing this method would allow the buffer to be treated like a stream sometimes. Currently this is not implemented, make a buffer stream on your buffer instead."

	self error: 'not yet implemented'

!

streamingWrite: anInteger into: aWriteStream

	| count |

	count := (writePosition - readPosition) min: anInteger.

	count <= 0 ifTrue: [^0].

	aWriteStream write: count from: cache at: readPosition + 1.

	self readSkip: count.

	^count

!

streamingWriteFrom: aReadStream

	"Implementing this method would allow the buffer to be treated like a stream sometimes. Currently this is not implemented, make a buffer stream on your buffer instead."

	self error: 'not yet implemented'

!

streamingWriteInto: aWriteStream

	| count |

	count := writePosition - readPosition.

	count <= 0 ifTrue: [^0].

	self streamingWrite: count into: aWriteStream.

	^count

! !

!Buffer methodsFor:'testing'!

hasDataToRead

	^readPosition < dataLength

!

hasFixedWriteSpace

	^false

!

hasSpaceToWrite

	^true

! !

!Buffer class methodsFor:'documentation'!

version_SVN

    ^ '$Id$'

! !