ykonsole2/ykonsole-core/src/main/kotlin/net/aiterp/git/ykonsole2/domain/runtime/FlowBus.kt

package net.aiterp.git.ykonsole2.domain.runtime

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

interface FlowBus<T : Any> {
  /**
   * Return this as shared flow
   */
  fun asSharedFlow(): SharedFlow<T>

  /**
   * Subscribe and read flow of [T].
   */
  suspend fun collect(forceAll: Boolean = false, action: suspend (value: T) -> Unit)

  /**
   * Emit an event to all subscribers.
   */
  suspend fun emit(data: T)

  /**
   * Emit [data] outside a coroutine.
   */
  fun emitBlocking(data: T) = runBlocking { emit(data) }
}

private class FlowBusImpl<T : Any> : FlowBus<T> {
  private val internal = MutableSharedFlow<T>()
  private val shared = internal.asSharedFlow()

  override fun asSharedFlow() = shared
  override suspend fun collect(forceAll: Boolean, action: suspend (value: T) -> Unit) = if (forceAll) {
    shared.buffer(capacity = 3).collect(action)
  } else shared.collectLatest(action)
  override suspend fun emit(data: T) = internal.emit(data)
}

fun CommandBus(): CommandBus = FlowBusImpl()
typealias CommandBus = FlowBus<Command>

fun EventBus(): EventBus = FlowBusImpl()
typealias EventBus = FlowBus<Event>