94 replies

Stream queued data control

Is there a way to be notified when a Stream internal queue is full/not full ?
I'm using Stream.async , but i need to pause/resume when the underlying source tries to push data to the stream when emiting data would push to much data to the queue.

example :

Stream.async((emit) => {
      this.source.onData((chunk, isLast) => {
        const copiedChunk = (new Uint8Array(chunk)).slice()
        emit(Effect.succeed(Chunk.of(copiedChunk)))
        if (isLast) {
          emit(Effect.fail(Option.none()))
        }
      })
// need to call source.pause() when queue is full
// need to call source.resume() when queue is not full anymore
    })

Stream.async((emit) => {
      this.source.onData((chunk, isLast) => {
        const copiedChunk = (new Uint8Array(chunk)).slice()
        emit(Effect.succeed(Chunk.of(copiedChunk)))
        if (isLast) {
          emit(Effect.fail(Option.none()))
        }
      })
// need to call source.pause() when queue is full
// need to call source.resume() when queue is not full anymore
    })

Effect Community•2y ago•

94 replies

ecyrbe

Stream queued data control

Stream.async((emit) => {
      this.source.onData((chunk, isLast) => {
        const copiedChunk = (new Uint8Array(chunk)).slice()
        emit(Effect.succeed(Chunk.of(copiedChunk)))
        if (isLast) {
          emit(Effect.fail(Option.none()))
        }
      })
// need to call source.pause() when queue is full
// need to call source.resume() when queue is not full anymore
    })

Stream.async((emit) => {
      this.source.onData((chunk, isLast) => {
        const copiedChunk = (new Uint8Array(chunk)).slice()
        emit(Effect.succeed(Chunk.of(copiedChunk)))
        if (isLast) {
          emit(Effect.fail(Option.none()))
        }
      })
// need to call source.pause() when queue is full
// need to call source.resume() when queue is not full anymore
    })

Stream queued data control

Similar Threads

Stream queued data control

Similar Threads

Similar Threads

Similar Threads