Fixed functionality of FilterStream (select: for streams).

[cslatevm.git] / src / core / stream.slate

prototypes ensureNamespace: #streams &delegate: True.

streams define: #Stream &parents: {Cloneable}.
"The shared protocol of all Stream objects, also providing a common
instantiation protocol."

s@(Stream traits) newOn: c
"Create a new stream of the same kind targeted on the given object."
[
  s clone on: c
].

s@(Stream traits) on: _
"(Re-)target the stream to some object. This modifies it in-place.
Overriding this will also customize newOn: for most uses."
[overrideThis].

s@(Stream traits) flush
"Do nothing by default."
[s].

s@(Stream traits) isAtEnd
"Answer whether the end of the Stream has been reached."
[overrideThis].

s@(Stream traits) hasAnEnd
"Answer whether the Stream has a known end, i.e. will it ever isAtEnd."
[overrideThis].

s@(Stream traits) elementType
"Returns a prototype for an appropriate element type for the Stream. This is
the most generic type."
[s contents elementType].

s@(Stream traits) collectionType
"Returns a prototype for an appropriate array type for the Stream. This is
the most generic type."
[Array].

s@(Stream traits) contents
"Answer what is conceptually being held or accumulated by the Stream."
[overrideThis].

Stream traits define: #Condition &parents: {SeriousCondition} &slots: {#stream}.
"Streaming-specific errors."

c@(Stream Condition traits) newFor: s
[c new `>> [stream := s. ]].

Stream traits define: #Exhaustion &parents: {Stream Condition. Error}.
"Stream or resource exhaustion, ie #isAtEnd Conditions."

e@(Stream Exhaustion traits) describeOn: out
[
  out ; 'The stream has reached its end, unhandled.'
].

s@(Stream traits) exhausted
[
  (s Exhaustion newFor: s) signal
].

s@(Stream traits) endCheck
[
  s isAtEnd ifTrue: [s exhausted]
].

Stream traits define: #Incomplete &parents: {Stream Condition} &slots: {
  #collection -> {}.
  #count -> 0.
  #start -> 0.
}.

i@(Stream Incomplete traits) on: c count: count at: start
[
  i new `>> [collection := c. count := count. start := start. ]
].

i@(Stream Incomplete traits) zero
[i on: Nil count: 0 at: Nil].

i@(Stream Incomplete traits) contents
[
  i collection isEmpty
    ifTrue: [i collection newSize: i count]
    ifFalse: [i collection copyFrom: i start below: i start + count]
].

streams define: #ReadStream &parents: {Stream}.
"Streams that read from some source."

s@(ReadStream traits) reader
"A ReadStream is already its own resource."
[s].

s@(ReadStream traits) next: n putInteractivelyInto: seq startingAt: start
"Returns the number of objects successfully read."
[| obj |
  0 below: n do:
    [| :index |
     (obj := s next) on: s Exhaustion do: [| :c | ^ index].
     seq at: start + index put: obj].
  n
].

s@(ReadStream traits) next: n putInto: seq startingAt: start
"Take the next N elements from the stream and write into the Sequence from
the given starting index."
[overrideThis].

s@(ReadStream traits) nextPutInto: seq startingAt: start
[
  s next: seq size putInto: seq startingAt: start
].

s@(ReadStream traits) nextPutInto: seq
[
  s next: seq size putInto: seq startingAt: 0
].

s@(ReadStream traits) next: n putInto: seq@(Sequence traits)
"Places the next N elements into a Sequence at the starting indices."
[
  s next: n putInto: seq startingAt: 0
].

s@(ReadStream traits) next: n putInto: seq@(ExtensibleArray traits)
"Places the next N elements at the end of a given ExtensibleArray."
[
  elems ::= s next: n.
  seq addAllLast: elems.
  elems size
].

s@(ReadStream traits) next: n
"Answer the next N elements."
[
  arr ::= s collectionType new &capacity: n.
  (read ::= s next: n putInto: arr startingAt: 0)
    = n ifTrue: [arr] ifFalse: [arr first: read]
].

s@(ReadStream traits) next
"Obtain and answer the next element from the Stream."
[
  (arr ::= s next: 1)
    isEmpty ifTrue: [s exhausted] ifFalse: [arr first]
].

s@(ReadStream traits) numAvailable
"Answer the number of elements that can be read (currently)."
[overrideThis].

s@(ReadStream traits) advanceToSatisfying: testBlock
"Advance the stream to the first object that satisfies the block test."
[
  [s isAtEnd \/ [testBlock applyWith: s next]] whileFalse  
].

s@(ReadStream traits) upToSatisfying: testBlock
"Answer all objects and advance up to one satisfying the block test."
[| elem |
  result ::= s collectionType new writer.
  [s isAtEnd \/ [testBlock applyWith: (elem := s next)]]
    whileFalse: [result nextPut: elem].
  result contents
].

s@(ReadStream traits) upTo: obj
"Answer all objects and advance up to one equal to the argument."
[s upToSatisfying: #(= obj) `er].

s@(ReadStream traits) upToAnyOf: c@(Collection traits)
"Answer all objects and advance up to one contained by the argument."
[s upToSatisfying: #(c includes: _) `er].

s@(ReadStream traits) upToEnd
"Supply all the elements and advance up to the end of the stream."
[
  [| :result |
   [s isAtEnd] whileFalse: [result nextPut: s next]] writingAs: s collectionType
].

s@(ReadStream traits) nextLine
"Fixme: this should be more sophisticated for cross platform support"
[ | result elem |
  result := s collectionType new writer.
  [s isAtEnd \/ [(elem := s next). elem = $\n]]
    whileFalse: [result nextPut: elem].
  s isAtEnd ifFalse: [result nextPut: elem].
  result contents
].

s@(ReadStream traits) do: block
"Call the block on all the elements in turn until the Stream is empty."
[
  [s isAtEnd] whileFalse: [block applyWith: s next].
  s
].

s@(ReadStream traits) do: block separatedBy: sepBlock
"Run the separator block between applying the block to each element."
[
  s isAtEnd
    ifTrue: [^ s]
    ifFalse: [block applyWith: s next].
  [s isAtEnd] whileFalse:
    [sepBlock do. block applyWith: s next].
  s
].

s1@(ReadStream traits) with: s2 do: block
"Call the block on all the elements in turn until the Stream is empty."
[
  [s1 isAtEnd \/ [s2 isAtEnd]] whileFalse:
    [block applyWith: s1 next with: s2 next].
  s1
].

s@(ReadStream traits) detect: succeed ifNone: fail
"Find an element satisfying a test. Conditionally execute a failure block."
[
  s do: [| :element | (succeed applyWith: element) ifTrue: [^ element]].
  fail do
].

s@(ReadStream traits) detect: succeed
[
  s detect: succeed ifNone: []
].

s@(ReadStream traits) anySatisfy: predicate
"Answer whether any elements cause the input block to be True."
[
  s do: [| :element | (predicate applyWith: element) ifTrue: [^ True]].
  False
].

s@(ReadStream traits) allSatisfy: predicate
"Answer whether all elements cause the input block to be True."
[
  s do: [| :element | (predicate applyWith: element) ifFalse: [^ False]].
  True
].

s@(ReadStream traits) noneSatisfy: predicate
"Answer whether none of s's elements cause the input block to be True."
[
  s do: [| :element | (predicate applyWith: element) ifTrue: [^ False]].
  True
].

streams define: #WriteStream &parents: {Stream}.
"Streams that write to some target."

s@(WriteStream traits) writer
"A WriteStream is already its own resource."
[s].

s@(WriteStream traits) next: n putInteger: val &littleEndian: le
[
  n isPositive ifTrue:
    [(le `defaultsTo: Platform Current endianness == #LittleEndian)
       ifTrue: [0 below: n do:
                  [| :byteIndex | s nextPut: ((val endianByteAt: byteIndex) as: s elementType)]]
       ifFalse: [n - 1 downTo: 0 do:
                   [| :byteIndex | s nextPut: ((val endianByteAt: byteIndex) as: s elementType)]]].
  s
].

s@(WriteStream traits) next: n putAll: seq startingAt: start
[overrideThis].

s@(WriteStream traits) nextPutAll: c
"Place the Collection's contents into the stream."
[
  s next: c size putAll: c startingAt: 0
].

s@(WriteStream traits) next: n put: obj
"Make the next N values the argument object."
[
  buffer ::= s collectionType newSize: n.
  buffer atAllPut: obj.
  s nextPutAll: buffer
].

s@(WriteStream traits) nextPutAll: seq from: start to: end
"Place a range of the Sequence's contents into the Stream."
[
  s next: end - start putAll: seq from: start
].

s@(WriteStream traits) next: n putAll: seq
[
  s next: n putAll: seq startingAt: 0
].

s@(WriteStream traits) nextPut: obj
"Place the given element on the Stream."
[
  buffer ::= s collectionType newSize: 1.
  buffer at: 0 put: obj.
  s next: 1 putAll: buffer startingAt: 0
].

s@(WriteStream traits) ; c
"Syntactic sugaring to make collection-insertion similar to concatenation."
[
  s nextPutAll: c.
  s
].

source@(Stream traits) >> sink@(WriteStream traits)
"Write the contents from source to target one element at a time.
The source can signal exhausted after an #isAtEnd returns true if
it doesn't have anything else left to write."
[
  [[source isAtEnd] whileFalse: 
     [sink nextPut: source next]] on: Stream Exhaustion do: [|:c| ^ sink].
  sink
].

source@(Stream traits) >> sink
[source >> sink writer].

sink@(Stream traits) << source
[source >> sink].

x@(Root traits) isSource
[#reader isFoundOn: {x}].

x@(Root traits) isSink
[#writer isFoundOn: {x}].

x@(Root traits) isTerminal
[x isSource \/ [x isSink]].

source@(Root traits) >> sink
[source isSource /\ [sink isSink] ifTrue: [source reader >> sink writer]].

sink@(Root traits) << source
[source isSource /\ [sink isSink] ifTrue: [source reader >> sink writer]].

source@(Stream traits) >>; sink
"Write the contents from source to target by taking each element as a
collection and concatenating their elements."
[
  [source isAtEnd]
    whileFalse: [sink ; source next].
  sink
].

sink@(Stream traits) <<; source
[source >>; sink].

source@(ReadStream traits) copyTo: sink &chunkSize: n
"Write the contents from source to target in big blocks."
[| position total |
  n `defaultsTo: 4096.
  buffer ::= source collectionType new &capacity: n.
  total := 0.
  [source isAtEnd] whileFalse:
    [position := [source next: n putInto: buffer]
       on: source Exhaustion do: #return `er.
     total += position.
     sink next: position putAll: buffer].
  total
].

streams define: #ReadWriteStream
           &parents: {ReadStream. WriteStream}.

s@(ReadWriteStream traits) iterator
"A ReadWriteStream is already its own resource."
[s].

streams define: #PeekableStream &parents: {Stream}.
"A mixin supporting protocols which rely on peek alone, but not full
positionability."

s@(PeekableStream traits) peek
[overrideThis].

s@(PeekableStream traits) nextWhile: testBlock
"Answer the next elements for which the test block is True. Note that this
relies on a Collection iterator Stream."
[
  [| :result |
   [s isAtEnd not /\ [testBlock applyWith: s peek]]
      whileTrue: [result nextPut: s next]] writingAs: s collectionType
].

s@(PeekableStream traits) nextUntil: testBlock
"Answer the next elements for which the test block is not True."
[
  s nextWhile: [| :each | (testBlock applyWith: each) not]
].

s@(PeekableStream traits) nextDelimitedBy: separatorBlock
"Answer the next elements between segments where the separator test is True."
[
  s skipWhile: separatorBlock.
  s nextUntil: separatorBlock
].

s@(PeekableStream traits) skipUntil: condition
"Move the position forward until the condition is satisfied."
[
  [s isAtEnd \/ [condition apply*, s peek]] whileFalse: [s next]
].

s@(PeekableStream traits) skipWhile: condition
"Move the position forward until the condition is no longer satisfied."
[
  [s isAtEnd \/ [(condition apply*, s peek) not]] whileFalse: [s next]
].

s@(PeekableStream traits) skipTo: obj
"Move the position forward until the object is found. This returns whether
it was found or the end was reached before then."
[
  s skipUntil: #(= obj) `er
].

streams define: #PositionableStream &parents: {PeekableStream}
  &slots: {#position -> 0. #readLimit -> 0}.
"PositionableStreams have an index and iterate over some sequenced collection,
but with a specific limit on the stream."

s@(PositionableStream traits) newOn: c from: start to: end
[
  s clone `>> [on: c. position := start. readLimit := end. ]
].

s@(PositionableStream traits) on: _
[s reset].

s@(PositionableStream traits) contents
"Answer the contents of the target up to the readLimit."
[overrideThis].

s@(PositionableStream traits) collectionType
"Answer the default collection prototype to dump contents into."
[String].

s@(PositionableStream traits) last
[overrideThis].

s@(PositionableStream traits) nextMatchAll: c
"Whether the next N objects in the Stream are in the other collection
(which generally should be a Sequence, ie have linear order)."
[
  pos ::= s position.
  c do: [| :each |
    s next = each ifFalse: [s position := pos. ^ False]].
  True
].

s@(PositionableStream traits) peek
"Returns the results of next without advancing the stream."
[
  s isAtEnd ifFalse:
    [obj ::= s next.
     s position -= 1.
     obj]
].

s@(PositionableStream traits) peek: n
"Answer the next N results without advancing the stream."
[
  origPosition ::= s position.
  result ::= s next: n.
  s position := origPosition.
  result
].

s@(PositionableStream traits) peekFor: obj
"Returns whether the object is next in the stream. Advances if true."
[
  s isAtEnd not
    /\ [obj = s next
          \/ [s position := s position - 1. False]]
].

s@(PositionableStream traits) peekForwardBy: offset
"Answers the element the given number of positions after the current position."
[overrideThis].

s@(PositionableStream traits) peekBackBy: offset
"Answers the element the given number of positions before the current
position."
[s peekForwardBy: offset negated].

s@(PositionableStream traits) firstPositionOfAnyOf: targets
"Answer the position of the Stream past the next occurrence of the targets,
or (one past) the end of the Stream if nothing was found."
[| firstMatch |
  start ::= s position.
  [s isAtEnd not /\ [(targets includes: s next) not]] whileTrue.
  firstMatch := s position.
  s isAtEnd /\ [s position := s position - 1. (targets includes: s next) not]
    ifTrue: [firstMatch += 1].
  s position := start.
  firstMatch
].

s@(PositionableStream traits) upToAnyOf: c
"Answer all objects up to the first occurrence of something in the collection,
or up to the end of the Stream if they are not found."
[
  endMatch ::= s firstPositionOfAnyOf: c.
  result ::= s next: endMatch - s position - 1.
  s position := endMatch.
  result
].

s@(PositionableStream traits) upToAll: pattern@(Sequence traits)
"Answer all the elements up to the occurrence of the pattern in the Stream,
but not including anything from the pattern."
[| end |
  patternSize ::= pattern size.
  start ::= s position.
  [s readLimit - s position < patternSize] "Not enough room left to match."
    whileFalse:
      [(s peek: patternSize) = pattern
         "The pattern has been found. Answer up to the pattern."
         ifTrue: [end := s position.
                  s position := start.
                  ^ (s next: end - start)]
         "Increment the position past the last start."
         ifFalse: [s position += 1]].
  s position := start.
  s upToEnd
].

s@(PositionableStream traits) upToEnd
"Answer all the elements up to the limit by a copy."
[overrideThis].

s@(PositionableStream traits) through: obj
"Answer the next elements up to and including the object given."
"This default implementation should be overridden for efficiency."
[(s upTo: obj) copyWith: obj].

s@(PositionableStream traits) throughAll: pattern
[
  result ::= (s collectionType new &capacity: 40) writer.
  result ; (s upToAll: pattern).
  s isAtEnd
    ifFalse: [result ; (s next: pattern size)].
  result contents
].

s@(PositionableStream traits) hasAnEnd [s readLimit < PositiveInfinity].

s@(PositionableStream traits) isAtEnd
[
  s position >= s readLimit
].

s@(PositionableStream traits) reset
[
  s position := 0.
  s
].

s@(PositionableStream traits) resetContents
[
  s position := 0.
  s readLimit := 0.
  s
].

s@(PositionableStream traits) setToEnd
"Position the Stream after the last readable element."
[
  s position := s readLimit.
  s
].

s@(PositionableStream traits) skip: n
[
  s position := s position + n.
  s
].

s@(PositionableStream traits) setFrom: start to: end
[
  s position := start.
  s readLimit := end + 1.
  s
].

s@(PositionableStream traits) beginsWith: seq
"Answer whether the Stream's next elements match the Sequence."
[
  s size >= seq size /\
    [oldPosition ::= s position.
     starters ::= s next: seq size.
     s position := oldPosition.
     starters = seq]
].

s@(PositionableStream traits) retract: n
"Retract N elements."
[
  s skip: n negated.
  s size.
  s readLimit := s readLimit - n.
  s
].

s@(PositionableStream traits) retract
"Retract one element."
[s retract: 1].

s@(PositionableStream traits) restoringPositionDo: block
[
  pos ::= s position.
  block ensure: [s position := pos].
].

streams define: #PositionableReadStream
  &parents: {PositionableStream. ReadStream}.
streams define: #PositionableWriteStream
  &parents: {PositionableStream. WriteStream}.
streams define: #PositionableReadWriteStream
  &parents: {PositionableStream. ReadStream. WriteStream}.

streams define: #StreamProcessor &parents: {ReadStream}
  &slots: {#source -> Stream clone}.

s@(StreamProcessor traits) on: source
[
  s source := source reader.
  s
].

s@(StreamProcessor traits) terminal [s source terminal].

s@(StreamProcessor traits) collectionType [s source collectionType].

s@(StreamProcessor traits) hasAnEnd [s source hasAnEnd].
s@(StreamProcessor traits) isAtEnd [s source isAtEnd].

source@(ReadStream traits) >> sink@(StreamProcessor traits)
"Overrides the basic >> method so that StreamProcessors can be composed
associatively in expressions for simple processing tasks."
[sink newOn: source].

streams define: #FilterStream &parents: {StreamProcessor} &slots: {
  #block -> [| :_ | True]
}.
"FilterStreams take source streams and apply a test block to each element,
only returning or acting on elements that satisfy the test."
"TODO: make a Write- variant"
"The result of sending #next is pre-computed, to determine isAtEnd while
not relying on the source stream being repositionable."

s@(Stream traits) select: block
[(FilterStream newOn: s) `>> [block := block. ]].

s@(Stream traits) reject: block
[(FilterStream newOn: s) `>> [block := [| :x | (block applyWith: x) not]. ]].

s@(FilterStream traits) next: n putInto: seq startingAt: start
[| numWritten |
  numWritten := 0.
  result ::= seq writer.
  result position := start.
  [s source isAtEnd \/ [numWritten >= seq size]] whileFalse:
    [s source skipUntil: [| :elem | s block applyWith: elem].
     [s source isAtEnd not /\ [numWritten < seq size /\ [s block applyWith: s source peek]]]
       whileTrue: [result nextPut: s source next. numWritten += 1]].
  numWritten
].

streams define: #CollectStream &parents: {StreamProcessor} &slots: {
  #block -> Method Identity "The default action, which is an identity."
}.
"CollectStreams take source streams and apply a block to each element,
returning the block's result for each."

s@(CollectStream traits) collectionType [Array].

s@(ReadStream traits) collect: block
[(CollectStream newOn: s) `>> [block := block. ]].

s@(CollectStream traits) numAvailable [s source numAvailable].

s@(CollectStream traits) next: n putInto: seq startingAt: start
[
  buffer ::= s source collectionType newSize: n.
  numRead ::= s source next: n putInto: buffer startingAt: 0.
  0 below: numRead do:
    [| :index | seq at: index + start put: (s block apply*, (buffer at: index))].
  numRead
].

s@(ReadStream traits) project: block
[s collect: [| :each | each -> (block applyWith: each)]].

streams define: #InjectStream &parents: {StreamProcessor}
&slots: {#accumulator -> [| :x :_ | x]. #currentValue -> Nil.}.

s@(Stream traits) inject: init into: accumulator
[(InjectStream newOn: s) `>> [currentValue: init. accumulator: accumulator. ]].

s@(InjectStream traits) next: n putInto: seq startingAt: start
[
  buffer ::= s collectionType newSize: n.
  numRead ::= s source next: n putInto: buffer startingAt: 0.
  0 below: numRead do:
    [| :index |
     buffer at: index put:
       (s currentValue := s accumulator apply*, s currentValue, (buffer at: index))].
  seq replaceFrom: start below: start + numRead with: buffer.
  numRead
].

s@(InjectStream traits) last
[| result |
  [s isAtEnd] whileFalse: [result := s next].
  result
].

source@(ReadStream traits) >> sink@(Method traits)
"Overrides the basic >> method so that >> can be composed associatively
in expressions for simple processing tasks."
[source collect: sink].

Stream traits define: #WrapperStream &parents: {ReadWriteStream} &slots: {#original -> Stream clone}.
"A WrapperStream wraps some stream and forwards operations to it where
applicable. This is an abstraction to be specialized."
"The stream to be wrapped."

w@(Stream WrapperStream traits) on: s
[
  w original: s.
  w
].

source@(ReadStream traits) >> sink@(Stream WrapperStream traits)
[sink newOn: source].

w@(Stream WrapperStream traits) terminal [w original terminal].
w@(Stream WrapperStream traits) hasAnEnd [w original hasAnEnd].
w@(Stream WrapperStream traits) isAtEnd [w original isAtEnd].
w@(Stream WrapperStream traits) next [w original next].
w@(Stream WrapperStream traits) next: n [w original next: n].
w@(Stream WrapperStream traits) next: n putInto: c [w original next: n putInto: c].
w@(Stream WrapperStream traits) next: n putInto: c startingAt: start [w original next: n putInto: c startingAt: start].
w@(Stream WrapperStream traits) next: n [w original next: n].
w@(Stream WrapperStream traits) nextPut: obj [w original nextPut: obj].
w@(Stream WrapperStream traits) nextPutAll: seq [w original nextPutAll: seq].
w@(Stream WrapperStream traits) next: n putAll: seq [w original next: n putAll: seq].
w@(Stream WrapperStream traits) next: n putAll: seq startingAt: start [w original next: n putAll: seq startingAt: start].
w@(Stream WrapperStream traits) resource [w original resource].
w@(Stream WrapperStream traits) resource: r [w original resource: r].
w@(Stream WrapperStream traits) flush [w original flush].
w@(Stream WrapperStream traits) position [w original position].
w@(Stream WrapperStream traits) position: n [w original position: n].
w@(Stream WrapperStream traits) contents [w original contents].
w@(Stream WrapperStream traits) elementType [w original elementType].
w@(Stream WrapperStream traits) collectionType [w original collectionType].

"This is a wrapper mixin for streams that calculates line numbers."
streams define: #LineNumberedReadStreamMixin &parents: {Mixin} &slots: {
  #eolPositions -> ExtensibleArray new. "The positions of the end of each line."
  #lastPosition -> 0.
"The position of the last character that EOL has been calculated for; we know
the line number for all characters before this position and nothing about
those after."
  #previousWasCR -> False.
"Whether the previous character was a CR, for CR-LF streams. CR-LF in
combination should only increment the line number by 1."
}.

s@(LineNumberedReadStreamMixin traits) on: r@(ReadStream traits)
[
  resend.
  s `>> [eolPositions: ({r position} as: ExtensibleArray).
         lastPosition: r position. previousWasCR: False. ]
].

s@(LineNumberedReadStreamMixin traits) newOn: s2@(LineNumberedReadStreamMixin traits)
"LineNumberedStreams should not wrap other ones of the same type."
[
  s2
].

s@(LineNumberedReadStreamMixin traits) lineNumber
[| index start stop |
  (pos ::= s position) >= s eolPositions last
    ifTrue: [^ s eolPositions size].
  start := 0.
  stop := s eolPositions size.
  [start + 1 < stop]
    whileTrue: [index := start + stop // 2.
                (s eolPositions at: index) <= pos
                  ifTrue: [start := index]
                  ifFalse: [stop := index]].
  start - 1
].

s@(LineNumberedReadStreamMixin traits) columnNumber
"Returns the index difference from the position to the last eolPosition."
[
  s position - s eolPositions last
].

s@(LineNumberedReadStreamMixin traits) atBeginningOfLine
[
  s columnNumber = 0
].

s@(LineNumberedReadStreamMixin traits) atBOL
[s atBeginningOfLine].

s@(LineNumberedReadStreamMixin traits) atEndOfLine
[
  s position = s eolPositions last
].

s@(LineNumberedReadStreamMixin traits) atEOL
[s atEndOfLine].

s@(LineNumberedReadStreamMixin traits) next
[
  char := resend.
  s position - 1 == s lastPosition
    ifTrue: [s lastPosition: s lastPosition + 1.
             char = String Character cr
               ifTrue: [s eolPositions add: s position.
                        s previousWasCR: True]
               ifFalse: [s previousWasCR not /\ [char = String Character lf]
                           ifTrue: [s eolPositions add: s position].
                         s previousWasCR: False]]
    ifFalse: [error: 'Got out of sync!'].
  char
].

s@(LineNumberedReadStreamMixin traits) nextLine
[
  result ::= s collectionType new writer.
  [| char eol |
   char := s next.
   (eol := s atEndOfLine) ifFalse: [result nextPut: char].
   eol \/ [s original isAtEnd]] whileFalse.
  result contents as: String
].

s@(LineNumberedReadStreamMixin traits) position: n
[
  n > s lastPosition
    ifTrue: 
      [| newPosition |
        newPosition: n.
        n: s lastPosition.
        resend.
        [s position < newPosition /\ [s isAtEnd not]]
          whileTrue: [s next]]
    ifFalse: [resend].
  s
].

streams define: #LineNumberedReadStream &parents: {LineNumberedReadStreamMixin. Stream WrapperStream}.

streams define: #LineNumberedWriteStreamMixin &parents: {Mixin} &slots: {#currentColumn -> 0. #currentLine -> 0}.
"This is a wrapper mixin for streams that keeps track of the cursor and
the number of lines printed."

s@(LineNumberedWriteStreamMixin traits) updateFrom: char
"Detect \n's and update internal state"
[
  char = $\n
    ifTrue: [
      s currentLine: s currentLine + 1.
      s currentColumn: 0]
    ifFalse: [
      char = $\t
        ifTrue: [s currentColumn: s currentColumn + 8]
        ifFalse: [s currentColumn: s currentColumn + 1]].
].

s@(LineNumberedWriteStreamMixin traits) padUntilColumn: col with: obj
"Detect \n's and update internal state in sync"
[
  s next: col - s currentColumn put: obj
].

s@(LineNumberedWriteStreamMixin traits) padUntilColumn: col
"Detect \n's and update internal state in sync"
[
  s next: col - s currentColumn put: $\s
].

s@(LineNumberedWriteStreamMixin traits) nextPut: char
"Detect \n's and update internal state in sync"
[
  s updateFrom: char.
  resend
].

s@(LineNumberedWriteStreamMixin traits) nextPutAll: seq
"Detect \n's and update internal state in sync"
[
  seq do: #(s updateFrom: _) `er.
  resend
].