1 streams define: #LimitedProcessor &parents: {DelegatingProcessor} &slots: {
5 s@(LimitedProcessor traits) for: n
7 s clone `>> [count := n. ]
10 s@(LimitedProcessor traits) isAtEnd
12 s count <= s count zero \/ [resend]
15 s@(LimitedProcessor traits) next: n putInto: seq startingAt: start
18 ifTrue: [s count -= n. resend]
19 ifFalse: [n := s count. s count := 0. resend]
22 streams define: #InOutProcessor &parents: {StreamProcessor} &slots: {
27 p@(InOutProcessor traits) on: s
29 resend `>> [inputBuffer := s source collectionType new. outputBuffer := s collectionType new. ]
32 s@(InOutProcessor traits) canProcessMore
34 s inputBuffer isNotEmpty \/ [s source isAtEnd not]
37 s@(InOutProcessor traits) next: n putInto: seq startingAt: start
39 [s outputBuffer size < n /\ [s canProcessMore]]
40 whileTrue: [s bufferInputTo: s requiredInSize. s process].
41 amt := n min: s outputBuffer size.
42 seq replaceFrom: start to: start + amt - 1 with: s outputBuffer.
43 s outputBuffer := s outputBuffer allButFirst: amt.
47 s@(InOutProcessor traits) isAtEnd
49 s outputBuffer isEmpty /\ [s inputBuffer isEmpty] /\ [resend]
52 s@(InOutProcessor traits) requiredInSize
53 "This is the amount we try to get into the in buffer every time. Useful
54 if you're looking for a search string of a specified size. Override in child classes."
59 "doesn't/can't guarantee that it's filled!"
60 s@(InOutProcessor traits) bufferInputTo: neededSize
61 [| nextBuffer amt needed |
62 s inputBuffer size >= neededSize ifFalse:
63 [needed := neededSize - s inputBuffer size.
64 nextBuffer := s source collectionType new &capacity: needed.
65 amt := s source next: needed putInto: nextBuffer.
66 s inputBuffer := s inputBuffer ; (nextBuffer first: amt)].
69 s@(InOutProcessor traits) process
74 s@(InOutProcessor traits) passInThrough
76 s write: s inputBuffer.
77 s inputBuffer := s source collectionType new.
80 s@(InOutProcessor traits) write: seq
82 s outputBuffer := s outputBuffer ; seq
85 s@(InOutProcessor traits) skip: amt
87 s inputBuffer := s inputBuffer allButFirst: amt
91 s@(InOutProcessor traits) guaranteedReadable
96 streams define: #SimpleReplacementStream &parents: {InOutProcessor} &slots: {
101 s@(SimpleReplacementStream traits) find: x replaceWith: y
103 s new `>> [search: x. replace: y. ]
106 s@(SimpleReplacementStream traits) requiredInSize
111 "fixme don't cons as much"
112 s@(SimpleReplacementStream traits) process
114 s search size < 1 ifTrue: [^ s passInThrough].
116 [s inputBuffer size >= s requiredInSize]
118 [(s inputBuffer first: s search size) = s search
119 ifTrue: [s write: s replace. s skip: s search size]
120 ifFalse: [s write: (s inputBuffer first: 1). s skip: 1 ]].
122 s source isAtEnd ifTrue: [s passInThrough]
125 s@(Sequence traits) find: find@(Sequence traits) replaceWith: replace@(Sequence traits)
128 s reader >> (SimpleReplacementStream find: find replaceWith: replace) >> out
132 streams define: #MultiProcessor &parents: {DelegatingProcessor} &slots: {#children}.
134 mp@(MultiProcessor traits) using: s
136 mp new `>> [children: s. ]
139 mp@(MultiProcessor traits) on: s
141 mp children isEmpty ifTrue: [^ (error: 'No children for processor')].
143 "fixme we should be able to write the following in a better way"
144 mp children first on: s.
145 (1 to: mp children size - 1 by: 1) do: [|:i| (mp children at: i) on: (mp children at: i - 1)].
146 mp source: mp children last.
151 d@(Dictionary traits) keysAndValuesCollect: block
153 d isEmpty ifTrue: [^ #{}].
154 result ::= ExtensibleArray new.
155 d keysAndValuesDo: [| :key :val | result add: (block applyWith: key with: val)].
159 s@(Sequence traits) sequentialFindAndReplace: dict@(Dictionary traits)
163 >> (MultiProcessor using:
164 (dict keysAndValuesCollect: [| :key :val | SimpleReplacementStream find: key replaceWith: val]))
168 streams define: #MultiReplacementStream &parents: {InOutProcessor} &slots: {
169 #dict -> Dictionary new.
173 s@(MultiReplacementStream traits) from: d@(Dictionary traits)
175 s new `>> [dict := d. ]
178 s@(MultiReplacementStream traits) requiredInSize
180 s cachedSize `defaultsTo: (s dict keys collect: #size `er) max
183 s@(MultiReplacementStream traits) process
185 s requiredInSize < 1 ifTrue: [^ s passInThrough].
186 [s inputBuffer size >= 1]
188 [hasMatched := False.
189 s dict keysAndValuesDo:
191 (s inputBuffer beginsWith: key)
195 hasMatched := True]].
197 [s write: s inputBuffer first.
201 s@(Sequence traits) findAndReplace: dict@(Dictionary traits)
204 s reader >> (MultiReplacementStream from: dict) >> out
208 "todo: look at using extensible arrays"
209 streams define: #LineProcessor &parents: {InOutProcessor} &slots: {#currentLine}.
211 s@(LineProcessor traits) collectionType [Array].
213 s@(LineProcessor traits) process
215 s currentLine `defaultsTo: (s inputBuffer newSize: 0).
216 (pos := s inputBuffer indexOfFirstSatisfying: [| :char | lastChar := char. char = $\n \/ [char = $\r]])
218 [s currentLine := s currentLine ; s inputBuffer.
219 s inputBuffer := s inputBuffer new.
221 (s currentLine := s currentLine ; (inputBuffer first: pos))
222 isEmpty ifFalse: [s currentLine last = $\r ifTrue: [s currentLine := s currentLine allButLast]].
223 s write: s currentLine.
224 s inputBuffer := inputBuffer allButFirst: pos.
226 s inputBuffer isEmpty not /\ [s inputBuffer first = $\n] /\ [lastChar = $\r]
227 ifTrue: [s inputBuffer := s inputBuffer allButFirst].
230 s@(LineProcessor traits) next: n putInto: seq startingAt: start
232 [s outputBuffer size < n /\ [s canProcessMore]] whileTrue:
233 [s bufferInputTo: s requiredInSize. s process].
234 amt := n min: s outputBuffer size.
235 seq replaceFrom: start to: start + amt - 1 with: s outputBuffer.
236 s outputBuffer := s outputBuffer allButFirst: amt.