Implementing caching using Flow from Kotlin

Caching.kt

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

// Following the RxJava implementation here: https://blog.mindorks.com/implement-caching-in-android-using-rxjava-operators

data class Data(val source: String)

class MemoryDataSource {

    private var _data: Data = Data("Memory")

    val data: Flow<Data>
        get() = flow {
            emit(_data)
        }

    fun cacheInMemory(data: Data) {
        println("caching to memory from ${data.source}")
        _data = data.copy(source = "memory")
    }
}

class NetworkDataSource {

    val data: Flow<Data>
        get() = flow {
            val newData = Data("network")
            emit(newData)
        }
}

class DiskDataSource {

    private var _data: Data = Data("disk")

    val data: Flow<Data>
        get() = flow {
            emit(_data)
        }

    fun saveToDisk(data: Data) {
        println("saving to disk from ${data.source}")
        _data = data.copy(source = "disk")
    }
}

class DataSource {
    private val memoryDataSource: MemoryDataSource = MemoryDataSource()
    private val diskDataSource: DiskDataSource = DiskDataSource()
    private val networkDataSource: NetworkDataSource = NetworkDataSource()

    fun getDataFromMemory(): Flow<Data> = memoryDataSource.data
    fun getDataFromDisk(): Flow<Data> = diskDataSource.data.onEach {
        it.run {
            memoryDataSource.cacheInMemory(it)
        }
    }

    fun getDataFromNetwork(): Flow<Data> = networkDataSource.data.onEach {
        it.run {
            diskDataSource.saveToDisk(it)
            memoryDataSource.cacheInMemory(it)
        }
    }
}

runBlocking {
    val dataSource = DataSource()

    combineMerge(
        dataSource.getDataFromMemory(),
        dataSource.getDataFromDisk(),
        dataSource.getDataFromNetwork())
        .collectIndexed { index, it ->
            println("$it $index")
        }
}

fun <T> combineMerge(vararg flows: Flow<T>) = flow {
    flows.forEach { fflow ->
        fflow.collectIndexed {index, it ->
            println("$it $index")
            emit(it)
        }
    }
}