Stream processing hangs after second item due to queue issue

Not sure how best to describe this issue, maybe just looking for some general directions to look at. I have a stream created from an unbounded queue using

Stream.fromQueue

Stream.fromQueue

, and am running an effect for each item in the stream using

Stream.tap

Stream.tap

. The effect that runs for each item in the stream may or may not trigger a

Queue.offerAll

Queue.offerAll

. Everything seems to work on the first item added to the queue, but after the 2nd item is added to the queue (verifying that

offer

offer

returned

true

true

), everything hangs.

I've forked the

Stream.runDrain

Stream.runDrain

call, as well as the

Queue.offerAll

Queue.offerAll

call. I also created an independent Fiber that runs once per second to look at the queue size, and when it gets to the part where it hangs, the queue size stays stuck at

-1

-1

. As per the docs, it's true that there's a Fiber waiting for elements to be added to the queue, so my question is, what could be causing the stream to not continue?

Effect Community•12mo ago

Johnil

Stream processing hangs after second item due to queue issue

Not sure how best to describe this issue, maybe just looking for some general directions to look at. I have a stream created from an unbounded queue using

Stream.fromQueue

Stream.fromQueue

, and am running an effect for each item in the stream using

Stream.tap

Stream.tap

. The effect that runs for each item in the stream may or may not trigger a

Queue.offerAll

Queue.offerAll

. Everything seems to work on the first item added to the queue, but after the 2nd item is added to the queue (verifying that

offer

offer

returned

true

true

), everything hangs.

I've forked the

Stream.runDrain

Stream.runDrain

call, as well as the

Queue.offerAll

Queue.offerAll

call. I also created an independent Fiber that runs once per second to look at the queue size, and when it gets to the part where it hangs, the queue size stays stuck at

-1

-1

. As per the docs, it's true that there's a Fiber waiting for elements to be added to the queue, so my question is, what could be causing the stream to not continue?

Stream processing hangs after second item due to queue issue

Similar Threads

Stream processing hangs after second item due to queue issue

Similar Threads

Similar Threads

Similar Threads