I've got a funny little coroutine puzzle: ``` val received = withTimeoutOrNull(Tv...

# coroutines

[reactormonk]

I've got a funny little coroutine puzzle:

val received = withTimeoutOrNull(TvOnTimeout) {
                        tvState.first()
                    }

                    irBlaster.send(commandToTest)

I wanna wait for the next tvState, but start to wait for that change before sending the command via irBlaster. How do I wire them up?

[Sergey Dmitriev]

You mean the next tvState that comes after received ?

[reactormonk]

the send should trigger a new tvState, and I wanna know if that happens within a certain timeout.

[Sergey Dmitriev]

Are you expecting any particular state that is different to received?
Because then you can do tvState.first { it == expectedState }.timeout()

[reactormonk]

Oh sweet, time to refactor some code

[reactormonk]

val tvState = cecFlows.tvState
                        .filter { it is TVState.Standby || it is TVState.TransientToStandby }

                    val received = withTimeoutOrNull(TvStandbyTimeout) {
                        tvState.first()
                    }

[reactormonk]

So yes, expecting some

[Sergey Dmitriev]

Then yes, it should work

[Sergey Dmitriev]

Should work btw even if expectedState == receivedState

[reactormonk]

Where is timeout from? Can't import it

[Sergey Dmitriev]

Ah first is a terminate operator so it’s either return the expected value or waits forever…
you can try

tvState.filter { it == expected }.timeout().first()

Or something similar

[reactormonk]

Don't want the exception though, just a null

[Sergey Dmitriev]

Then you can try to catch the TimeoutException and fallback to null
perhaps there is a better way though

[reactormonk]

Now to fix the original timing issue though, I wanna have the timeout waiting, launch the send, and continue waiting

[reactormonk]

val wait = Channel<Unit>(RENDEZVOUS)
                    
                    launch {
                        irBlaster.send(commandToTest)
                        wait.send(Unit)
                    }

                    wait.receive()
                    val received = withTimeoutOrNull(TvOnTimeout) {
                        cecFlows.tvState.first { it is TVState.Standby || it is TVState.TransientToStandby }
                    }

This should do

[reactormonk]

... nope

[Sam]

By any chance is tvState a StateFlow? If so, I think onSubscription has the behavior you're looking for:

tvState
  .onSubsbscription { irBlaster.send(commandToTest) }
  .first { it is TVState.Standby || it is TVState.TransientToStandby }

[Sam]

From the docs:

The action is called before any value is emitted from the upstream flow to this subscription but after the subscription is established. It is guaranteed that all emissions to the upstream flow that happen inside or immediately after this onSubscription action will be collected by this subscription.

Which sounds similar to

I wanna wait for the next tvState, but start to wait for that change before sending the command via irBlaster.

very nice