Hello, have anyone tried to implement something like RxJava PublishSubject with Coroutines? If so, c...

# coroutines

[Christopher Elías]

Hello, have anyone tried to implement something like RxJava PublishSubject with Coroutines? If so, could you please show some examples or gists?

[kenkyee]

Use StateFlow

[Christopher Elías]

Hi, do you have some sample? I mean, I found one with the following:

@Singleton
class PublishSubjectKtx @Inject constructor() {

    private val channel = BroadcastChannel<Any>(1)

    suspend fun publish(b: Any) {
        channel.send(b)
    }

    fun observe(): Flow<Any> =
        channel
            .asFlow()
}

The problem with replacing this with StateFlow is that it requires to have an initial value...

private val _stateFlow = MutableStateFlow<Any>(Any)

[kenkyee]

Use a sealed class with initial value and another one with a real value if you want that behavior...

[bezrukov]

Use SharedFlow, it does't require initial value.

Also StateFlow can't match PublishSubject behavior because it a) conflates emission b) replays last element when subscribing to it

[kenkyee]

Oops...Denis is right. Thought you wanted a BehaviorSubject 😔

[Christopher Elías]

Hello Denis, do you have some sample? For now the solution I wrote above is working great... but I would like to know how to apply your suggestion

[bezrukov]

something like

@Singleton
class PublishSubjectKtx @Inject constructor() {
    private val _flow = MutableSharedFlow<Any>(extraBufferCapacity = 1)

    suspend fun publish(b: Any) {
        _flow.emit(b)
    }
    fun observe(): Flow<Any> = _flow // you can add .asSharedFlow() for safety, so it will be readonly
}

[Christopher Elías]

Thanks Denis! I will try it

[ursus]

stateFlow.skip(1) ?