PipelinePersisterReactive.kt

val networkTrigger = Channel<Unit>(capacity = Channel.RENDEZVOUS)
val diskCommands = Channel<DiskCommand>(capacity = Channel.RENDEZVOUS)

launch {
            // trigger first load
            diskCommands.send(DiskCommand.ReadFirst)
        }
            
val networkFlow =  networkTrigger
 .consumeAsFlow()
 .filter { it }
 .flatMapLatest { fetcher.stream(request) }
 .doOnEach {  diskCommands.send(DiskCommand.Stop(stopAck)); write(it) }
 .map{ diskCommands.send(DiskCommand.Read(it))}

val diskFlow = flattenMerge(diskCommands.consumeAsFlow(), networkFlow)
  .flatmapLatest{ when (command) {....} //something from here can call networkTrigger(Unit) to kick off network request

return diskFlow //not sure what combine is needed for