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

!

!Block methodsFor:'xtreams'!

streamingReadMatching: aStream inclusive: tail
        ^Xtreams::TestReadSubstream on: aStream test: self inclusive: tail
! !
!Block methodsFor:'xtreams'!

streamingWriteMatching: aStream inclusive: tail
        ^Xtreams::TestWriteSubstream on: aStream test: self inclusive: tail
! !
!Object methodsFor:'xtreams'!

streamingReadMatching: aStream inclusive: tail
	^[:each | each == self] streamingReadMatching: aStream inclusive: tail
! !
!Object methodsFor:'xtreams'!

streamingWriteMatching: aStream inclusive: tail
	^[:each | each == self] streamingWriteMatching: aStream inclusive: tail
! !
!SequenceableCollection methodsFor:'xtreams'!

streamingMatchPrefixFunction
	"Compute the array that tells us how far we need to back up when a match fails. This is using the Knuth-Morris-Pratt matching algorithm"

	| backtrack partialMatch |
	backtrack := Array new: self size + 1. 
	backtrack at: 1 put: 1; at: 2 put: 1.
	partialMatch := 1.

	2 to: self size do: [:index |
		| current |
		current := self at: index.

		"If there's a mismatch, back up to the previous partial match and see if the match can continue from there. If not, repeat until we find a match or hit the beginning"
		[partialMatch > 1 and: [(self at: partialMatch) ~= current]]
			whileTrue: [partialMatch := backtrack at: partialMatch - 1].

		(self at: partialMatch) = current
			ifTrue: [partialMatch := partialMatch + 1].
		backtrack at: index put: partialMatch].

	^backtrack
! !
!SequenceableCollection methodsFor:'xtreams'!

streamingReadMatching: aStream inclusive: tail
        "Return a stream that produces substreams that separate by @pattern and include the pattern in the result if @tail is true.
         It uses the Knuth-Morris-Pratt algorithm, from Cormen et al.'s Algorithms, page 871. See also Gusfield's Alorithms on Strings, Trees and Sequences, page 23. This algorithm has the advantage in stream matching that it reads every character in the the text to be matched against exactly once, and never backtracks."

        | backtrack |
        backtrack := self streamingMatchPrefixFunction.
        ^Xtreams::MatchReadSubstream on: aStream pattern: self backtrack: backtrack inclusive: tail
! !
!SequenceableCollection methodsFor:'xtreams'!

streamingWriteMatching: aStream inclusive: tail
        "Return a stream that produces substreams that separate by @pattern and include the pattern in the result if @tail is true.
         It uses the Knuth-Morris-Pratt algorithm, from Cormen et al.'s Algorithms, page 871. See also Gusfield's Alorithms on Strings, Trees and Sequences, page 23. This algorithm has the advantage in stream matching that it reads every character in the the text to be matched against exactly once, and never backtracks."

        | backtrack |
        backtrack := self streamingMatchPrefixFunction.
        ^Xtreams::MatchWriteSubstream on: aStream pattern: self backtrack: backtrack inclusive: tail
! !
!Xtreams::ReadStream methodsFor:'substreaming'!

, aReadStream
	"Return a read stream that combines self and @aReadStream into a single stream.
	""
		((1 to: 5) reading, (6 to: 10) reading) rest
	""
		| files |
		files := '/pub/vw7.8' asFilename reading.
		[ | fn | fn := files get. fn isDirectory ifTrue: [ files := fn reading, files ]. fn ] reading rest
	"
	^(Array with: self with: aReadStream) reading stitching
! !
!Xtreams::ReadStream methodsFor:'substreaming'!

closing: aBlock
	^(PositionReadSubstream on: self)
		closeBlock: aBlock;
		yourself
! !
!Xtreams::ReadStream methodsFor:'substreaming'!

ending: aMatchable
	"Creates a substream that will end when aMatchable finds a match in the content passing through. aMatchable is either
		* a block that is evaluated with each element; the stream ends when the block returns true
		* a collection that is matched against the last elements read, the stream ends when the collection matches
		* any other object, the stream matches when an equal object is read from the stream"
	"	aMatchable	<BlockClosure | Collection | Object>
		^<TransformReadStream>
	""
		('abcdefghijklmnopqrstuvxyz' reading ending: $j) rest.
	""
		('abcdefghijklmnopqrstuvxyz' reading ending: 'mno') rest
	""
		('abcdefghijklmnopqrstuvxyz' reading ending: [ :e | 'gmt' includes: e ]) rest
	"
	^self ending: aMatchable inclusive: false
! !
!Xtreams::ReadStream methodsFor:'substreaming'!

ending: aMatchable inclusive: inclusive
	"Creates a substream that will end when aMatchable finds a match in the content passing through. aMatchable is either
		* a block that is evaluated with each element - the stream ends when the block returns true
		* a collection that is matched against the last elements read - the stream ends when the collection matches
		* any other object - the stream matches when an equal object is read from the stream
	The inclusive parameter determins if the elements matching the end condition should be included in the substream contents or not."
	"	aMatchable	<BlockClosure | Collection | Object>	the substream ending criteria
		inclusive	<Boolean> should the content matching the end condition be included in the substream
		^<TransformReadStream>
	""
		('abcdefghijklmnopqrstuvxyz' reading ending: $j inclusive: true) rest.
	""
		('abcdefghijklmnopqrstuvxyz' reading ending: 'mno' inclusive: true) rest
	""
		('abcdefghijklmnopqrstuvxyz' reading ending: [ :e | 'gmt' includes: e ] inclusive: true) rest
	"
	^aMatchable streamingReadMatching: self inclusive: inclusive
! !
!Xtreams::ReadStream methodsFor:'substreaming'!

limiting: limit
	"Create a substream that will allow at most @limit number of elements to be read from the source."
	"	limit		<Integer>	maximum number of elements that can be read from the source
		^<LimitReadStream>"
	"
		('abcdefghi' reading limiting: 5) rest
	"
	^LimitReadSubstream on: self limit: limit
! !
!Xtreams::ReadStream methodsFor:'substreaming'!

slicing
	"From a readable stream, return a readable stream that acts as a prototype factory for the readable stream."
	"	^<ReadStream>"
	"
		((1 to: 100) reading limiting: 10) slicing do: [:substream | Transcript cr; print: substream rest]
	"
	| substream |
	substream := nil.
	^[substream == nil ifFalse:
		[substream substreamClosed ifFalse: [substream close].
		substream subseekend.
		substream sourceAtEnd ifTrue: [Incomplete zero raise]].
	substream := self copy]
		reading
			closeBlock: [source close];
			yourself
! !
!Xtreams::ReadStream methodsFor:'substreaming'!

stitching
	"From a stream that returns streams (either read or write streams), stitch them together sequencially such that they appear to be one contiguous stream."
	"^ <StitchReadStream>"
	"
		| data current |
		data := (1 to: 100) reading.
		current := nil.
		[	(current notNil and: [ current position < 10 ]) ifTrue: [ Incomplete zero raise ].
			current := data limiting: 10
		] reading stitching rest
	"
	| first |
	first := self get.
	first isReadable ifTrue: [^StitchReadStream on: self first: first].
	first isWritable ifTrue: [^StitchWriteStream on: self first: first].
	^self error: 'Cannot read or write to this stream, what is it?'
! !
!Xtreams::WriteStream methodsFor:'substreaming'!

closing: aBlock
	^(PositionWriteSubstream on: self)
		closeBlock: aBlock;
		yourself
! !
!Xtreams::WriteStream methodsFor:'substreaming'!

ending: aMatchable
	"Creates a substream that will end when aMatchable finds a match in the content passing through. aMatchable is either
		* a block that is evaluated with each element - the stream ends when the block returns true
		* a collection that is matched against the last elements written - the stream ends when the collection matches
		* any other object - the stream ends when an equal object is written into the stream"
	"	aMatchable	<BlockClosure | Collection | Object> the substream ending criteria
		^<TransformWriteStream>
	""	
		| stream slice |
		stream := String new writing.
		slice := stream ending: $j.
		[ slice write: 'abcdefghijklmnopqrstuvxyz' ] on: Incomplete do: [].
		stream conclusion
	""
		| stream slice |
		stream := String new writing.
		slice := stream ending: 'mno'.
		[ slice write: 'abcdefghijklmnopqrstuvxyz' ] on: Incomplete do: [].
		stream conclusion
	""
		| stream slice |
		stream := String new writing.
		slice := stream ending: [ :e | 'gmt' includes: e ].
		[ slice write: 'abcdefghijklmnopqrstuvxyz' ] on: Incomplete do: [].
		stream conclusion
	"
	^self ending: aMatchable inclusive: false
! !
!Xtreams::WriteStream methodsFor:'substreaming'!

ending: aMatchable inclusive: inclusive
	"Creates a substream that will end when aMatchable finds a match in the content passing through. aMatchable is either
		* a block that is evaluated with each element - the stream ends when the block returns true
		* a collection that is matched against the last elements written - the stream ends when the collection matches
		* any other object - the stream ends when an equal object is written into the stream"
	"	aMatchable	<BlockClosure | Collection | Object> the substream ending criteria
		inclusive <Boolean> should the matched elements be included in the stream contents or not
		^<TransformWriteStream>
	""
		| stream slice |
		stream := String new writing.
		slice := stream ending: $j inclusive: true.
		[ slice write: 'abcdefghijklmnopqrstuvxyz' ] on: Incomplete do: [].
		stream conclusion
	""
		| stream slice |
		stream := String new writing.
		slice := stream ending: 'mno' inclusive: true.
		[ slice write: 'abcdefghijklmnopqrstuvxyz' ] on: Incomplete do: [].
		stream conclusion
	""
		| stream slice |
		stream := String new writing.
		slice := stream ending: [ :e | 'gmt' includes: e ] inclusive: true.
		[ slice write: 'abcdefghijklmnopqrstuvxyz' ] on: Incomplete do: [].
		stream conclusion
	"
	^aMatchable streamingWriteMatching: self inclusive: inclusive
! !
!Xtreams::WriteStream methodsFor:'substreaming'!

limiting: limit
	"Create a substream that will allow at most @limit number of elements written into the destination."
	"	limit	<Integer>	maximum number of elements that can be written into destination
		^<LimitWriteStream>"
	"
		| stream slice |
		stream := String new writing.
		slice := stream limiting: 5.
		[ slice write: 'abcdefghi' ] on: Incomplete do: [].
		stream conclusion
	"

	^LimitWriteSubstream on: self limit: limit
! !
!Xtreams::WriteStream methodsFor:'substreaming'!

slicing
	"From a writable stream, return a readable stream that acts as a prototype factory for the writable stream."
	"	^<ReadStream>"
	"(destination limiting: 10) slicing"

	| substream |
	substream := nil.
	^[substream == nil ifFalse:
		[substream substreamClosed ifFalse: [substream close].
		substream subseekend.
		substream destinationAtEnd ifTrue: [Incomplete zero raise]].
		substream := self copy]
		reading
			closeBlock: [destination close];
			yourself
! !
!Xtreams::WriteStream methodsFor:'substreaming'!

stitching
	^self error: 'You can only stitch a read stream, however that read stream can return write streams and in so doing, you will create a stitching write stream.'
! !