How can we subscribe channel with coroutine?

[wangmir]

Same as title,
I'm looking for handling subscribe channel with using coroutine because I wanna reveal this data with using SharedFlow to outside.
But I can't find any document which describe the method to do this.. can u help me?

[mp911de]

Subscription and pub/sub consumption are different topics. For subscribing, you use the regular (p)subscribe() methods passing on channel/pattern names. For listening, I recommend using the reactive API where you can obtain a Flow object from a message Flux.

[wangmir]

Thanks for advice, I tested ur advice with test code, and it works perfectly,

This is my code example for other people

class FakeEventReceiver(private val redisUrl: String) {

    val pubSubFlow = MutableSharedFlow<Pair<String, String>>()

    private val client by lazy {
        RedisClient.create(redisUrl)
    }

    private val pubSubConnection by lazy {
        client.connectPubSub()
    }

    suspend fun startSubscribe() {
        pubSubConnection.reactive().subscribe("pubsub_event").subscribe()

        pubSubConnection.reactive().observeChannels().asFlow().onEach { message ->
            pubSubFlow.emit(Pair(message.channel, message.message))
            log.info(marker, "Pubsub event to flow, $message")
        }.launchIn(CoroutineScope(Dispatchers.IO))
    }

    companion object {
        val log = KotlinLogging.logger { }
        val marker = MarkerFactory.getMarker("EventReceiver")
    }
}