Is it safe to read from an `InputStream` in a `flow { }` builder? Getting warnings from IntelliJ abo...

# coroutines

[Sam Garfinkel]

Is it safe to read from an InputStream in a flow { } builder? Getting warnings from IntelliJ about inappropriate blocking calls.

[streetsofboston]

If you have a blocking call inside your flow { … } builder, the collector may block as well, since its scope’s dispatcher will be used to run the flow’s builder code.

[gildor]

It will block consumer until you apply explicit dispatcher on flow using flowOn operator

[Sam Garfinkel]

So I want flowOn(<http://Dispatchers.IO|Dispatchers.IO>)?

[Sam Garfinkel]

I’m also using a STAX parser to read from an InputStream. I gather I should do the same with that to prevent it from blocking consumers?

[Sam Garfinkel]

I’m confused why this is an issue though. Flows are cold and unbuffered. Consumers execute the entire pipeline to get each value.

[streetsofboston]

But blocking code in the flow (producer) can cause the consumer to block, since they would use the same dispatcher. With the flowOn you can prevent that

[Sam Garfinkel]

In other words, the consumer’s pool might be able to do other work but instead is blocked by the flow operator? Does flowOn have a buffer to resolve this?

[streetsofboston]

No buffer. I uses suspend callbacks.

[streetsofboston]

Instead of being blocked, you collector will be suspended.

[Sam Garfinkel]

Ah I see

[Sam Garfinkel]

Ok thanks. Hopefully that will clear the warning.

[gildor]

One more way, is avoid calling blocking code directly from flow, instead you can wrap this blocking code to suspend function with dispatcher, so flow itself will not block, but it just a matter of style and particular case

[streetsofboston]

That’s why Flow don’t have backpressure issues like Rx does…

[Sam Garfinkel]

Does wrapping blocking calls in suspend properly prevent parent Coroutines from blocking?

[Sam Garfinkel]

Unless you are specifically forking with launch or async I didn’t think this was the case

[gildor]

If suspend function itself is not blocking, yes

[streetsofboston]

The warning/linting system may not be that smart. I can detect a withContext(<http://Dispatchers.IO|Dispatchers.IO>) and remove the warning, but I doubt it can figure out the flowOn(<http://Dispatchers.IO|Dispatchers.IO>) to remove the warnings….

[gildor]

Don't use launch, use withContexy

[Sam Garfinkel]

@streetsofboston Plus I believe withContext is illegal in a standard flow

[gildor]

Right

[streetsofboston]

You can’t emit from a different scope, but I think you can emit from a different context, correct?

[streetsofboston]

Ah…. I guess you can’t do that 🙂

[gildor]

withContext also create scope, so it's not allowed

[Sam Garfinkel]

channelFlow exists for this purpose but I think most uses of it are code smell

[streetsofboston]

It perfectly fine to use if you want to emit data from plain non-suspending callbacks

[gildor]

I think flowOn is just the easiest solution

[Sam Garfinkel]

This question has gone a bit off the rails but just to conclude, if you’re bridging callbacks callbackFlow is designed specifically for that purpose.

[Gabriel Feo]

Yes, because callbackFlow checks whether you called awaitClose in the end

[streetsofboston]

IIRC, that is the only difference between the channelFlow and callbackFlow, correct?

[Gabriel Feo]

AFAIK, yes

[gildor]

I don't think that usage callback/channel flow would be better in this case, it created for concurrent emit from different threads, you will also pay price for it, which doesn't make a lot of sense if just want emit from some blocking loop like reading from a stream