OnetimeWhileSubscribed

OnetimeWhileSubscribed.kt

// Designed and developed by skydoves (Jaewoong Eum)
public class OnetimeWhileSubscribed(
  private val stopTimeout: Long,
  private val replayExpiration: Long = Long.MAX_VALUE,
) : SharingStarted {

  private val hasCollected: MutableStateFlow<Boolean> = MutableStateFlow(false)

  init {
    require(stopTimeout >= 0) { "stopTimeout($stopTimeout ms) cannot be negative" }
    require(replayExpiration >= 0) { "replayExpiration($replayExpiration ms) cannot be negative" }
  }

  override fun command(subscriptionCount: StateFlow<Int>): Flow<SharingCommand> =
    combine(hasCollected, subscriptionCount) { collected, counts ->
      collected to counts
    }
      .transformLatest { pair ->
        val (collected, count) = pair
        if (count > 0 && !collected) {
          emit(SharingCommand.START)
          hasCollected.value = true
        } else {
          delay(stopTimeout)
          if (replayExpiration > 0) {
            emit(SharingCommand.STOP)
            delay(replayExpiration)
          }
          emit(SharingCommand.STOP_AND_RESET_REPLAY_CACHE)
        }
      }
      .dropWhile {
        it != SharingCommand.START
      } // don't emit any STOP/RESET_BUFFER to start with, only START
      .distinctUntilChanged() // just in case somebody forgets it, don't leak our multiple sending of START
}