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

"{ NameSpace: Xtreams }"

WriteStream subclass:#CollectWriteStream

	instanceVariableNames:'block cache'

	classVariableNames:''

	poolDictionaries:'XtreamsPool'

	category:'Xtreams-Transforms'

!

CollectWriteStream comment:'Converts written elements using the provided conversion block. Optimizes bulk writes by batching the conversion results, allowing the destination stream to perform bulk writes as well.

Instance Variables

	block	<BlockClosure> collecting block (same style as collect: blocks)

	cache	<SequenceableCollection> caches results after transformation so that they can be bulk written too

'

!

!CollectWriteStream class methodsFor:'instance creation'!

on: aWriteStream block: block

	^self new on: aWriteStream block: block

! !

!CollectWriteStream methodsFor:'accessing'!

block

	^block

!

insert: anInteger from: aSequenceableCollection at: startIndex

	| count amount written offset |

	count := 0.

	offset := startIndex.

	[count < anInteger] whileTrue:

		[amount := anInteger min: cache size.

		1 to: amount do: [:index | cache at: index put: (block value: (aSequenceableCollection at: offset+index))].

		written := [destination insert: anInteger from: cache. amount] on: Incomplete do: [ :ex | ex count ].

		count := count + written.

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

		offset := offset + amount ].

	^anInteger

!

put: anObject

	destination put: (block value: anObject).

	^anObject

!

write: anInteger from: aSequenceableCollection at: startIndex

	| count amount written offset |

	count := 0.

	offset := startIndex - 1.

	[count < anInteger] whileTrue:

		[amount := anInteger - count min: cache size.

		1 to: amount do: [:index | cache at: index put: (block value: (aSequenceableCollection at: offset + index))].

		written := [destination write: amount from: cache. amount] on: Incomplete do: [ :ex | ex count ].

		count := count + written.

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

		offset := offset + amount ].

	^anInteger

! !

!CollectWriteStream methodsFor:'initialize-release'!

close

	super close.

	cache recycle.

	cache := nil

!

contentsSpecies

	^cache species

!

contentsSpecies: aClass

	cache ifNotNil: [ cache recycle ].

	cache := aClass newRecycled: (

				(cache notNil and: [ cache size > 0 ])

					ifTrue: [cache size]

					ifFalse: [DefaultBufferSize])

!

on: aDestination block: aBlock

	self on: aDestination.

	block := aBlock.

	self contentsSpecies: aDestination contentsSpecies

! !

!CollectWriteStream methodsFor:'seeking'!

++ anInteger

	^destination ++ anInteger

!

-- anInteger

	^destination -- anInteger

!

length

	^destination length

!

position

	^destination position

!

position: anInteger

	^destination position: anInteger

! !

!CollectWriteStream methodsFor:'testing'!

isPositionable

	^destination isPositionable

! !

!CollectWriteStream class methodsFor:'documentation'!

version_HG

    ^ '$Changeset: <not expanded> $'

!

version_SVN

    ^ '$Id$'

! !