How to collect data from a stream in a closure without memory leaks/lifecycle?

I want to create a class which takes a stream as argument with an initial value and just execute logic everytime a value is emitted. I thought I could do this with MutableStateFlow, but that only will transmit after I called launchIn with a coroutine scope. The code is executed without coroutines and no lifecycle, so this is a no-go. Executing it with a GlobalScope scope will leak memory. The class needs to be deinitialized when there are no references to it anymore.

Essentially I am just looking for a CurrentValueSubject equivalent. I am looking for something like this:

fun listen(stream: StreamWithInitialValue<Boolean>) { 
    now = stream.currentValue

    someStrongReference = stream.sink { newBoolean -> 
        // ...
    }
}

Preferabele without any third party libraries.

from How to collect data from a stream in a closure without memory leaks/lifecycle?