orion/purescript-csv-stream
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix: closer

Browse Source
This commit is contained in:
orion 2024-05-07 11:00:31 -05:00
parent f3e0a6095e
commit ad04aab031
Signed by: orion
GPG Key ID: 6D4165AE4C928719
5 changed files with 8208 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7941
index.js Executable file
View File

File diff suppressed because it is too large Load Diff

7
index.js.map Normal file
View File

File diff suppressed because one or more lines are too long

310
src/Node.Stream.Object.js
View File

@ -1,20 +1,54 @@
import Stream from "stream"; import Stream from "stream";
const DEBUG = process.env['NODEJS_OBJECT_STREAM_TRACE'] !== ''
/** @type {(s: string) => void} */
const log = m => DEBUG ? console.log(m) : undefined;
let chainCount = 0
let composeCount = 0
let neverCount = 0
let onceCount = 0
let bindCount = 0
let zipCount = 0
let mapCount = 0
let constCount = 0
let fromPromiseCount = 0
export class Never extends Stream.Readable { export class Never extends Stream.Readable {
constructor() { constructor() {
super({read: function() { super({objectMode: true})
this.id = neverCount++
}
_read() {
log(`Never {id: ${this.id}}#_read()`)
log(` this.push(null)`)
this.push(null) this.push(null)
}, objectMode: true})
} }
} }
/** @template T */ /** @template T */
export class Once extends Stream.Readable { export class Once extends Stream.Duplex {
/** @param {T} a */ /** @param {T} a */
constructor(a) { constructor(a) {
super({read: function() { }, objectMode: true}) super({objectMode: true, allowHalfOpen: false})
this.push(a) this.a = a
this.id = onceCount++
this.push(this.a)
this.push(null) this.push(null)
log(`Once {id: ${this.id}}#new()`)
log(` this.push(${a})`)
log(` this.push(null)`)
}
/** @type {Stream.Duplex['_write']} */
_write(_ck, _enc, cb) {
cb()
}
_read() {
log(`Once {id: ${this.id}}#_read()`)
} }
} }
@ -22,9 +56,17 @@ export class Once extends Stream.Readable {
export class Const extends Stream.Transform { export class Const extends Stream.Transform {
/** @param {T} a */ /** @param {T} a */
constructor(a) { constructor(a) {
super({transform: function(_c, _enc, cb) { super({objectMode: true})
cb(null, a) this.a = a
}, objectMode: true}) this.id = constCount++
}
/** @type {Stream.Transform['_transform']} */
_transform(_c, _enc, cb) {
log(`Const {id: ${this.id}}#_transform(${_c}, _, _)`)
log(` cb(${this.a})`)
this.push(this.a)
cb()
} }
} }
@ -32,16 +74,27 @@ export class Const extends Stream.Transform {
export class FromPromise extends Stream.Readable { export class FromPromise extends Stream.Readable {
/** @param {Promise<T>} p */ /** @param {Promise<T>} p */
constructor(p) { constructor(p) {
super({objectMode: true})
this.id = fromPromiseCount++
p p
.then(a => { .then(a => {
log(`FromPromise {id: ${this.id}}#new()`)
log(` ...p.then(...)`)
log(` this.push(${a})`)
log(` this.push(null)`)
this.push(a) this.push(a)
this.push(null) this.push(null)
}) })
.catch(e => { .catch(e => {
log(`FromPromise {id: ${this.id}}#new()`)
log(` ...p.catch(...)`)
log(` this.destroy(${e})`)
this.destroy(e) this.destroy(e)
}) })
}
super({read: function() {}, objectMode: true}) _read() {
log(`FromPromise {id: ${this.id}}#_read()`)
} }
} }
@ -50,39 +103,56 @@ export class Chain extends Stream.Readable {
/** @param {...Stream.Readable} streams */ /** @param {...Stream.Readable} streams */
constructor(...streams) { constructor(...streams) {
super({objectMode: true}) super({objectMode: true})
this.id = chainCount++
this.ix = -1 this.ix = -1
this.streams = streams this.streams = streams
/** @type {Stream.Readable | undefined} */
this.cur = undefined
this.next() this.next()
} }
next() { next() {
log(`Chain {id: ${this.id}}#next()`)
this.ix++ this.ix++
if (this.ix === this.streams.length) { if (this.ix === this.streams.length) {
return undefined log(` this.push(null)`)
this.push(null)
} else { } else {
this.cur = this.streams[this.ix] const cur = this.streams[this.ix]
this.cur.once('end', () => { cur.once('error', e => {
log(`Chain {id: ${this.id}}#next()`)
log(` cur.once('error', ...)`)
log(` this.destroy(${e})`)
this.destroy(e)
})
cur.once('end', () => {
log(`Chain {id: ${this.id}}#next()`)
log(` cur.once('end', ...)`)
log(` this.next()`)
this.next() this.next()
}) })
this.cur.on('data', ck => { cur.on('data', ck => {
log(`Chain {id: ${this.id}}#next()`)
log(` cur.on('data', ...)`)
log(` this.push(${ck})`)
const canPush = this.push(ck) const canPush = this.push(ck)
if (this.cur && !canPush) { if (cur && !canPush) {
this.cur.pause() log(` cur.pause()`)
cur.pause()
} }
}) })
} }
} }
_read() { _read() {
if (!this.cur) { log(`Chain {id: ${this.id}}#_read()`)
this.push(null) this.streams.forEach(s => {
} else if (this.cur.isPaused()) { if (s.isPaused()) {
this.cur.resume() log(` s.resume()`)
s.resume()
} }
})
} }
} }
@ -93,18 +163,18 @@ export class Chain extends Stream.Readable {
export class Map extends Stream.Transform { export class Map extends Stream.Transform {
/** @param {(t: T) => R} f */ /** @param {(t: T) => R} f */
constructor(f) { constructor(f) {
super({ super({objectMode: true})
transform: function (chunk, _, cb) { this.f = f
try { this.id = mapCount++
this.push(f(chunk))
cb();
} catch (e) {
// @ts-ignore
cb(e);
} }
},
objectMode: true /** @type {Stream.Transform['_transform']} */
}) _transform(ck, _, cb) {
log(`Map {id: ${this.id}}#_transform(${ck}, _, _)`)
const r = this.f(ck)
log(` const r = (${r})`)
log(` cb(null, ${r})`)
cb(null, r)
} }
} }
@ -118,25 +188,41 @@ export class Zip extends Stream.Readable {
/** @param {...Stream.Readable} streams */ /** @param {...Stream.Readable} streams */
constructor(...streams) { constructor(...streams) {
super({objectMode: true}) super({objectMode: true})
this.id = zipCount++
log(`Zip {id: ${this.id}}#new()`)
log(` this.streams = Array {streams: ${streams.length}}`)
this.streams = streams this.streams = streams
this.streams.forEach((s, ix) => { this.streams.forEach((s, ix) => {
log(` this.streams[${ix}].once('error', ...)`)
log(` this.streams[${ix}].once('end', ...)`)
log(` this.streams[${ix}].once('data', ...)`)
s.once('error', e => this.destroy(e)) s.once('error', e => this.destroy(e))
s.once('end', () => this.push(null)) s.once('end', () => this.push(null))
s.on('data', ck => { s.on('data', ck => {
const canPush = this.bufput(ix, ck) log(`Zip {id: ${this.id}}#new()`)
if (!canPush) { log(` this.streams[${ix}].once('data', ...)`)
this.streams.forEach(s => s.pause()) log(` this.bufput(${ix}, ${ck})`)
} log(` stream.pause()`)
this.bufput(ix, ck)
s.pause()
}) })
}) })
} }
/** @type {(ix: number, val: unknown) => boolean} */ /** @type {(ix: number, val: unknown) => boolean} */
bufput(ix, val) { bufput(ix, val) {
log(`Zip {id: ${this.id}}#bufput(${ix}, ${val})`)
const bufstr = JSON.stringify(this.buf.map(a => a === null ? 'null' : '..'))
log(` this.buf = ${bufstr}`)
this.buf[ix] = val this.buf[ix] = val
if (!this.isWaiting()) { if (!this.isWaiting()) {
log(` this.push(${bufstr})`)
const canPush = this.push(this.buf) const canPush = this.push(this.buf)
this.bufinit() this.bufinit()
if (canPush) {
log(` this.streams.forEach(s => s.resume())`)
this.streams.forEach(s => s.resume())
}
return canPush return canPush
} else { } else {
return true return true
@ -144,7 +230,9 @@ export class Zip extends Stream.Readable {
} }
bufinit() { bufinit() {
this.buf = this.streams.map(() => null) const nuls = this.streams.map(() => null)
log(` this.buf = ${JSON.stringify(nuls)}`)
this.buf = nuls
} }
isWaiting() { isWaiting() {
@ -152,6 +240,7 @@ export class Zip extends Stream.Readable {
} }
_read() { _read() {
log(`Zip {id: ${this.id}}#_read()`)
this.streams.forEach(s => { this.streams.forEach(s => {
if (s.isPaused()) { if (s.isPaused()) {
s.resume() s.resume()
@ -167,51 +256,100 @@ export class Compose extends Stream.Duplex {
*/ */
constructor(a, b) { constructor(a, b) {
super({objectMode: true}) super({objectMode: true})
this.id = composeCount++
this.a = a this.a = a
this.b = b this.b = b
log(`Compose {id: ${this.id}}#new()`)
log(` a.on('data', ...)`)
log(` a.once('end', ...)`)
log(` a.once('error', ...)`)
log(` a.pause()`)
log(` b.on('drain', ...)`)
log(` b.on('data', ...)`)
log(` b.on('error', ...)`)
log(` b.on('finish', ...)`)
this.a.once('end', () => {
log(`Compose {id: ${this.id}}#new()`)
log(` a.on('end', ...)`)
log(` b.end()`)
this.b.end()
})
this.a.on('data', ck => { this.a.on('data', ck => {
log(`Compose {id: ${this.id}}#new()`)
log(` a.on('data', ...)`)
log(` b.write(${ck})`)
const canWrite = this.b.write(ck) const canWrite = this.b.write(ck)
if (!canWrite) { if (!canWrite) {
log(` a.pause()`)
this.a.pause() this.a.pause()
} }
}) })
this.a.once('error', e => this.destroy(e))
this.a.once('end', () => this.push(null)) this.a.once('error', e => {
log(`Compose {id: ${this.id}}#new()`)
log(` a.once('error', ...)`)
log(` this.destroy(${e})`)
this.destroy(e)
this.b.destroy(e)
})
this.b.on('drain', () => { this.b.on('drain', () => {
if (this.a.isPaused()) { log(`Compose {id: ${this.id}}#new()`)
log(` b.on('drain', ...)`)
log(` this.a.resume()`)
this.a.resume() this.a.resume()
}
}) })
this.b.on('data', ck => { this.b.on('data', ck => {
log(`Compose {id: ${this.id}}#new()`)
log(` b.on('data', ...)`)
log(` this.push(${ck})`)
const canPush = this.push(ck) const canPush = this.push(ck)
if (!canPush) { if (!canPush) {
this.a.pause() log(` b.pause()`)
this.b.pause() this.b.pause()
} }
}) })
this.b.once('error', e => this.destroy(e))
this.b.once('end', () => this.emit('end')) this.b.once('error', e => {
this.b.once('finish', () => this.emit('finish')) log(`Compose {id: ${this.id}}#new()`)
log(` b.once('error', ...)`)
log(` this.destroy(${e})`)
this.destroy(e)
this.a.destroy(e)
})
this.b.once('finish', () => {
log(`Compose {id: ${this.id}}#new()`)
log(` b.once('finish', ...)`)
log(` this.emit('finish')`)
this.push(null)
this.end()
this.emit('finish')
})
} }
_read() { _read() {
if (this.a.isPaused()) { log(`Compose {id: ${this.id}}#_read()`)
this.a.resume()
}
if (this.b.isPaused()) { if (this.b.isPaused()) {
log(` b.resume()`)
this.b.resume() this.b.resume()
} }
} }
/** @type {Stream.Duplex['_write']} */ /** @type {Stream.Duplex['_write']} */
_write(ck, _enc, cb) { _write(ck, _enc, cb) {
log(`Compose {id: ${this.id}}#_write(${ck}, _, _)`)
if (this.a instanceof Stream.Readable) { if (this.a instanceof Stream.Readable) {
throw new Error('Cannot `write` to a Readable stream') throw new Error('Cannot `write` to a Readable stream')
} }
log(` this.a.write(${ck}, _, _)`)
this.a.write(ck, _enc, cb) this.a.write(ck, _enc, cb)
} }
} }
@ -224,46 +362,78 @@ export class Bind extends Stream.Duplex {
* @param {(t: T) => () => Stream.Readable} f * @param {(t: T) => () => Stream.Readable} f
*/ */
constructor(f) { constructor(f) {
super({objectMode: true}) super({objectMode: true, allowHalfOpen: true})
this.f = f this.f = f
this.id = bindCount++
/** @type {Stream.Readable | undefined } */ this.ix = 0
this.cur = undefined
/** @type {Array<Stream.Readable>} */ /** @type {Array<Stream.Readable>} */
this.streams = [] this.streams = []
/** @type {(() => void) | undefined} */
this.doneCb = undefined
this.paused = true
} }
initcur() { /** @type {NonNullable<Stream.Duplex['_final']>} */
if (!this.cur) { _final(cb) {
this.cur = this.streams[0] log(`Bind {id: ${this.id}}#_final(_)`)
this.doneCb = cb
} }
this.cur.on('data', ck => { init() {
log(`Bind {id: ${this.id}}#init()`)
const s = this.streams[this.ix]
if (this.paused || (!s && !this.doneCb)) {
this.paused = true
return
} else if (this.doneCb) {
log(` this.doneCb()`)
this.doneCb()
this.doneCb = undefined
return
}
log(` s.on('data', ...)`)
s.on('data', ck => {
log(`Bind {id: ${this.id}}#initcur()`)
log(` s.on('data', ...)`)
log(` this.push(${ck})`)
const canPush = this.push(ck) const canPush = this.push(ck)
if (!canPush && this.cur) { if (!canPush) {
this.cur.pause() s.pause()
} }
}) })
this.cur.on('end', () => { log(` s.once('end', ...)`)
this.streams.shift() s.once('end', () => {
if (this.streams.length > 0) { log(`Bind {id: ${this.id}}#initcur()`)
this.cur = this.streams[0] log(` s.once('end', ...)`)
this.initcur() log(` this.ix++`)
} else { log(` this.init()`)
this.cur = undefined this.ix++
} this.init()
}) })
} }
/** @type {Stream.Duplex['_write']} */ /** @type {Stream.Duplex['_write']} */
_write(ck, _, cb) { _write(ck, _, cb) {
log(`Bind {id: ${this.id}}#_write(${ck}, _, _)`)
try { try {
log(` this.streams = ${JSON.stringify(this.streams.map(_ => 'Readable'))}`)
this.streams.push(this.f(ck)()) this.streams.push(this.f(ck)())
this.initcur() if (this.paused) {
log(` this.init()`)
this.paused = false
this.init()
}
log(` cb()`)
cb() cb()
} catch(e) { } catch(e) {
log(` cb(${e})`)
// @ts-ignore // @ts-ignore
cb(e) cb(e)
} }
@ -271,9 +441,9 @@ export class Bind extends Stream.Duplex {
/** @type {Stream.Duplex['_read']} */ /** @type {Stream.Duplex['_read']} */
_read() { _read() {
if (this.cur && this.cur.isPaused()) { log(`Bind {id: ${this.id}}#_read()`)
this.cur.resume() this.streams.forEach(s =>
} s.isPaused() ? s.resume() : undefined)
} }
} }

9
src/Node.Stream.Object.purs
View File

@ -43,7 +43,7 @@ instance Apply (ObjectStream i) where
apply (ObjectStream iab) (ObjectStream ia) = wrap $ join $ pure applyImpl <*> iab <*> ia apply (ObjectStream iab) (ObjectStream ia) = wrap $ join $ pure applyImpl <*> iab <*> ia
instance Applicative (ObjectStream i) where instance Applicative (ObjectStream i) where
pure = wrap <<< constImpl pure = once
instance Monad (ObjectStream i) instance Monad (ObjectStream i)
@ -67,8 +67,8 @@ instance Profunctor ObjectStream where
sac <- pipeImpl sab sbc sac <- pipeImpl sab sbc
pipeImpl sac scd pipeImpl sac scd
-- | A stream that will emit the value `a` exactly once. -- | A stream that will emit the value `a` exactly once, and ignores any written chunks.
once :: forall a. a -> ObjectStream Unit a once :: forall i a. a -> ObjectStream i a
once = wrap <<< onceImpl once = wrap <<< onceImpl
-- | A stream that will immediately emit `close` and `end` events. -- | A stream that will immediately emit `close` and `end` events.
@ -141,9 +141,6 @@ run (ObjectStream s') = do
cancelError cancelError
cancelEnd cancelEnd
-- | Constructs a `Transform` stream that always invokes the callback with the provided value.
foreign import constImpl :: forall i a. a -> Effect (Stream i a)
-- | Constructs a Stream that re-emits the outputs from each stream, in order. -- | Constructs a Stream that re-emits the outputs from each stream, in order.
foreign import chainImpl :: forall a. Array (Stream Unit a) -> Effect (Stream Unit a) foreign import chainImpl :: forall a. Array (Stream Unit a) -> Effect (Stream Unit a)

5
test/Test/Node.Stream.Object.purs
View File

@ -45,6 +45,11 @@ spec =
it "creates a readable that emits each element" do it "creates a readable that emits each element" do
out <- Stream.run (Stream.fromFoldable [ 1, 2, 3 ]) out <- Stream.run (Stream.fromFoldable [ 1, 2, 3 ])
out `shouldEqual` [ 1, 2, 3 ] out `shouldEqual` [ 1, 2, 3 ]
it "bind maps each number" do
out <- Stream.run do
a <- Stream.fromFoldable [ 1, 2, 3 ]
pure $ a + 1
out `shouldEqual` [ 2, 3, 4 ]
it "bind fans out" do it "bind fans out" do
out <- Stream.run do out <- Stream.run do
a <- Stream.fromFoldable [ 1, 2, 3 ] a <- Stream.fromFoldable [ 1, 2, 3 ]