Hi I did search in this channel and google but not found SO

Question

Hi I did search in this channel and google but not found SO my question is ```kotlin var subject = PublishSubject create lt Location gt subject buffer 1L SECONDS subscribe list gt list lastOrNull let ``` The subscribe s on Next is called every 1 second and it process only the first one This is helpful that I have too many log and I just want print every seconds How I do it in coroutines Thanks

Dominaezzz · Answer

So you have a stream of values and you only want the first item of a 1 second window

fangzhzh · Answer

yes

Dominaezzz · Answer

``` flowOf Location conflate delayEach 1000 collect lastItem gt Do stuff ```

Dominaezzz · Answer

Hold on that s wrong It gives last instead of first

Dominaezzz · Answer

``` flowOf Location debounce 1000 collect firstItem gt Do stuff ```

Dominaezzz · Answer

That s better

Dominaezzz · Answer

``` var subject = PublishSubject create lt Location gt subject asFlow debounce 1000 collect firstItem gt Do stuff ``` This is immediately usable slightly smiling face

fangzhzh · Answer

Thanks I did use PublishSubject before but I don t like the dispose

fangzhzh · Answer

In the case of PublishSubject asFlow case `subject onNext location ` should works is it correct

fangzhzh · Answer

The location is feed by other place like `LocationListener`

fangzhzh · Answer

< Dominaezzz> would you mind taking another look thank you

Dominaezzz · Answer

If I m being honest I don t know too much about Rx

Dominaezzz · Answer

One sec I ll do some googling

Dominaezzz · Answer

Ah You need a `BroadcastChannel`

Dominaezzz · Answer

``` val subject = BroadcastChannel This is a subscription subject asFlow debounce 1000 collect firstItem gt Do stuff This is `onNext` subject send location ```

Dominaezzz · Answer

Why don t you like the dispose

fangzhzh · Answer

boilerplate

fangzhzh · Answer

Cool The broadcastChannel is exactly what I m looking for Thank you

Dominaezzz · Answer

Don t forget to call `subject close ` once if you re done publishing sweat smile

fangzhzh · Answer

sweat smile Okay so they are some in the sense of need to manual release

Dominaezzz · Answer

It s not strictly required but it s good practice if you re able to do it

gildor · Answer

You don t need broadcast channel for this it s just implementation detail of source what is your source of events Do you have sample of you code which you want improve Also debounce is wrong operator for this of you have stream of events which emit more often than your denounce delay it will never emit

gildor · Answer

So could you show example of your existing code and semantics which you want essentially you want not more than 1 event per N seconds right

fangzhzh · Answer

< gildor> sorry was in a meeting ```private val sensorListener = object SensorEventListener override fun onAccuracyChanged sensor Sensor accuracy Int override fun onSensorChanged event SensorEvent event let localScope launch subject send it ```

fangzhzh · Answer

```localScope launch subject asFlow debounce 1000 collect event gt logger debug ELDServiceDaemon accelerometer sensor $ event values get 0 + $ event values get 1 $ event values get 2 ```

fangzhzh · Answer

sensorListener is the source of event Because sensor events are way too many more than ten per second I want to throttle it and process one per second

gildor · Answer

What is `subject` Looks like a channel

gildor · Answer

see debounce is wrong choice in this case if your subject asFlow emits alwas faster than 1 element per 1000 you will get no events

gildor · Answer

You need `sample 1000 ` instead

gildor · Answer

But it also not the same as your Rx sample which not drop events but buffer them and emit as collection

gildor · Answer

This is why I asked about required semantics

fangzhzh · Answer

joy Yes I got no event so I changed it back to rxjava way

fangzhzh · Answer

cool let me try sample

gildor · Answer

gt back to rxjava way Could you elaborate do you use both flow and rxjava

fangzhzh · Answer

```

fangzhzh · Answer

rxjava way ``` private val subject = PublishSubject create lt SensorEvent gt disposable add subject buffer 1L TimeUnit SECONDS subscribe events gt logger debug ELDServiceDaemon accelerometer sensor $ event 0 values get 0 + $ event 0 values get 1 $ event 0 values get 2 override fun onSensorChanged event SensorEvent subject onNext it ```

gildor · Answer

But it s exactly the same it debounce and drop events

fangzhzh · Answer

flow way ``` private val flowSubject = BroadcastChannel lt SensorEvent gt 1000 localScope launch flowSubject asFlow sample 1000 collect event gt logger debug ELDServiceDaemon flowSubject accelerometer sensor $ event values get 0 + $ event values get 1 $ event values get 2 localScope launch flowSubject send it ```

gildor · Answer

also you don t need BroadcastChannel

fangzhzh · Answer

yes I failed to explain my semantic I don t need save all events as array It s good to drop others

gildor · Answer

better to wrap callback to Flow or to Observable depends on what is better for you directly it will be more safe no need to close and expose global state

gildor · Answer

okay your Rx snippet is also wrong you will get no events if source is faster than debounce

fangzhzh · Answer

Yep I forget replace the debouce what I used is buffer

fangzhzh · Answer

updated

fangzhzh · Answer

```better to wrap callback to Flow or to Observable depends on what is better for you directly it will be more safe no need to close and expose global state``` regarding this could you pls elaberate more or a link to an article

gildor · Answer

Rx also has `sample ` operator

gildor · Answer

You can check sample in documentation of callbackFlow <https kotlin github io kotlinx coroutines kotlinx coroutines core kotlinx coroutines flow callback flow html>

gildor · Answer

it shows how to wrap callback to flow using callbackFlow builder

fangzhzh · Answer

THANKS

fangzhzh · Answer

sample is way better than buffer in my case

fangzhzh · Answer

Thanks < gildor> < Dominaezzz> it works now