Maybe I shouldn t even be using PublishSubject here not sure

Question

Guillermo Alcantara · Answer

```what are you trying to do Can t you do Single fromCallable lt code gt subscribeOn Schedulers computation Sorry for the markdown ```

Gabe Kauffman · Answer

It s basically a volume indicator I have a very fast stream of raw audio data in the form of bytes I do a CPU intensive calculation to turn those bytes into a Double of the volume level and then I adjust the UI accordingly

Gabe Kauffman · Answer

I have this one callback `renderSample ` which is providing me with raw audio data 1000s of times per second

Gabe Kauffman · Answer

If your suggestion is to put `Single fromCallable lt code gt subscribeOn Schedulers computation ` in the callback that won t work AFAIK since I need to define the observable outside of the callback and then only pass data through the observable in the callback

Guillermo Alcantara · Answer

check for something like Flowable range 0 N map renderSample subscribeOn Schedulers computation

Guillermo Alcantara · Answer

Sounds like you might need some back pressure handling

Gabe Kauffman · Answer

I can t call renderSample on my own it s a callback from a third party library I do turn it into a flowable later on for the backpressure handling

Gabe Kauffman · Answer

All I can do is define what is to be done when the library sends samples through renderSample

Guillermo Alcantara · Answer

I m confused then You can enqueue things to a subject Then whatever subscribes that s where you subscribeOn computation

Gabe Kauffman · Answer

If subscribingOn computation downstream would affect which thread the computation is done here then that would work

StavFX · Answer

In the current form of the code there s no use of observe subscribeOn that will make `calculateAudioVolume` run on a different thread than the one `renderSample` is being called on What you can do is have a subject of ByteBuffer instead of Double then you can handle `audioSample` on another thread

Gabe Kauffman · Answer

Ahhhhhhhh thank you so much That feels like a step in the right direction And then I can do something like ```val observable = publishSubject hide observeOn Schedulers computation flatMap Observable just AudioProcessingHelper calculateAudioVolume it ```

StavFX · Answer

yup I would get rid of `hide ` since the other operators will do that for you anyway And anytime you see ` flatMap Observable just foo ` you can use ` map foo ` instead Also now you ll have to pay attention to whether or not you can actually still trust the content of ByteBuffer to be usable after renderSample has completed or does the SDK recycle it making it not thread safe

Gabe Kauffman · Answer

Brilliant thank you thank you thank you