Utilities to deal with streams
npm install @hazae41/cascadeNever let streams give you a headache again
``bash`
npm i @hazae41/cascade
It took me 1 year to manage JavaScript streams in such a way as to avoid headaches, fortunately you can skip this struggle and never get any headache (and even avoid Safari bugs!)
For basic scenarios, Cascade provides streams but with an accessible controller
`typescript
const encoder = new SuperTransformStream
async function onWrite(chunk: Uint8Array) {
/**
* No need to get the controller
*/
await stream.enqueue(encode(chunk))
}
example.readable
.pipeTo(encoder.substream.writable)
.then(() => console.log("Closed"))
.catch(e => console.error("Errored", e))
encoder.substream.readable
.pipeTo(example.writable)
.then(() => console.log("Closed"))
.catch(e => console.error("Errored", e))
/**
* You can close it at any time
*/
encoder.terminate()
`
For more complex scenarios, Cascade provides plexes, which are streams with events, and you can associate them to build complex and reliable structures
#### Simplex
A basic in-out stream with events
`tsx
const simplex = new Simplex
/**
* You can use pipes
*/
example.readable
.pipeTo(simplex.writable)
.then(() => console.log("Closed"))
.catch(e => console.error("Errored", e))
simplex.readable
.pipeTo(example.writable)
.then(() => console.log("Closed"))
.catch(e => console.error("Errored", e))
/**
* You can also use events
*/
const simplex = new Simplex
/**
* When the simplex starts
*/
async start() {
this.enqueue(new Uint8Array([0, 1, 2]))
},
/**
* When the simplex is closing
*/
async close() {
this.enqueue(new Uint8Array([7, 8, 9]))
},
/**
* When the simplex is erroring
*/
async error(error) {
console.log("Errored", error)
},
/**
* When the simplex receives chunks
*/
async write(chunk) {
/**
* Pass the chunk to the readable side
*/
this.enqueue(chunk)
},
})
/**
* You can use enqueue at any time
*/
simplex.enqueue(new Uint8Array([4, 5, 6]))
/**
* You can use error at any time
*/
simplex.error(new Error())
/**
* You can use close at any time
*/
simplex.close()
/**
* You can check if it's closed
*/
simplex.closed
/**
* You can check if it's errored
*/
simplex.errored
/**
* You can check if it's closed or errored
*/
simplex.stopped
`
#### Full-Duplex
A pair of simplexes that are closed independently
- When one side is errored, the other is automatically errored
- When one side is closed, the other is NOT automatically closed
Events
- error — called ONCE when input OR output are errored
- close — called ONCE when input AND output are closed
`tsx
class Crypter extends FullDuplex
constructor() {
super({
input: { message: m => this.#onInputMessage(m) },
output: { message: m => this.#onOutputMessage(m) },
close: () => this.#onClose(),
error: e => this.#onError(e)
})
}
async #onInputMessage(data: Uint8Array) {
this.input.enqueue(await encrypt(data))
}
async #onOutputMessage(data: Uint8Array) {
this.output.enqueue(await decrypt(data))
}
async #onError(reason?: unknown) {
console.error("Errored", reason)
}
async #onClose() {
console.log("Closed")
}
}
function crypt(subprotocol: FullDuplex
const crypter = new Crypter()
subprotocol.outer.readable.pipeTo(crypter.inner.writable).catch(() => { })
crypter.inner.readable.pipeTo(subprotocol.outer.writable).catch(() => { })
return crypter
}
`
#### Half-Duplex
A pair of simplexes that are closed together
- When one side is errored, the other is automatically errored
- When one side is closed, the other is automatically closed
Events
- error — called ONCE when input OR output are errored
- close — called ONCE when input OR output are closed
`tsx
class MySocket extends EventTarget {
readonly #duplex = new HalfDuplex
input: { message: m => this.#onMessage(m) },
error: e => this.#onError(e),
close: () => this.#onClose(),
})
get inner() {
return this.#duplex.inner
}
send(message: string) {
this.#duplex.output.enqueue(message)
}
error(reason?: unknown) {
this.#duplex.output.error(reason)
}
close() {
this.#duplex.output.close()
}
async #onMessage(data: string) {
this.dispatchEvent(new MessageEvent("message", { data }))
}
async #onError(reason?: unknown) {
this.dispatchEvent(new ErrorEvent("error", { error: reason }))
}
async #onClose() {
this.dispatchEvent(new Event("close"))
}
}
``