Flows¶

callbackFlow {
  val listener = SomeListener { event -> trySendBlocking(event) }
  addSomeListener(listener)
  awaitClose { removeSomeListener(listener) }
}.buffer(capacity = Channel.UNLIMITED)

// A bounded capacity opts INTO backpressure: once 64 elements are pending, the
// watcher's own dispatcher thread parks — which also stalls recovery and resync
// for this watch. Only bound it when a stalled watch beats an unbounded buffer.
client
  .watchEventsAsFlow("/config", watchOption { isPrefix(true) }, capacity = 64)
  .collect { event -> logger.info { "${event.eventType} ${event.keyAsString}" } }

sealed interface WatchFlowEvent {
  data class Response(val response: WatchResponse) : WatchFlowEvent
  data class Recovery(val event: WatchRecoveryEvent) : WatchFlowEvent
}

client
  .watchAsFlow("/config", watchOption { isPrefix(true) })
  .collect { element ->
    when (element) {
      is WatchFlowEvent.Response -> {
        element.response.events.forEach { event ->
          logger.info { "${event.eventType} ${event.keyAsString} = ${event.valueAsString}" }
        }
      }

      // Recovery arrives in-band and in order, so a collector keeping derived
      // state learns about a gap before the events that follow it.
      is WatchFlowEvent.Recovery -> {
        logger.warn { "Watch recovery: ${element.event}" }
      }
    }
  }

// resyncWith runs after a compaction: re-read your world and return the revision
// to re-anchor at. Events between the compaction point and the anchor are gone —
// this hook is your only chance to reconcile.
val resyncWith = {
  val response = client.getResponse("/config", getOption { isPrefix(true) })
  response.kvs.forEach { kv -> logger.info { "Resync: ${kv.key.asString}" } }
  response.header.revision + 1
}

client
  .watchAsFlow("/config", watchOption { isPrefix(true) }, resyncWith = resyncWith)
  .collect { element ->
    if (element is WatchFlowEvent.Recovery && element.event is WatchRecoveryEvent.Resynced)
      logger.warn { "Reconciled after a compaction" }
  }

// Flattens responses into events and drops recovery transitions. Fine for a
// stateless collector; wrong for one that keeps derived state.
client
  .watchEventsAsFlow("/config", watchOption { isPrefix(true) })
  .collect { event ->
    logger.info { "${event.eventType} ${event.keyAsString}" }
  }

sealed interface LeadershipEvent {
  data class Elected(val leaderName: String) : LeadershipEvent
  data object Vacated : LeadershipEvent
  data class WatchFailed(val cause: Throwable?) : LeadershipEvent
}

// Observing, not participating: the current leader is emitted immediately on
// collect, so a late subscriber is never left blind waiting for a hand-off.
client
  .leadershipAsFlow("/elections/orders")
  .collect { event ->
    when (event) {
      is LeadershipEvent.Elected -> logger.info { "Leader is ${event.leaderName}" }
      is LeadershipEvent.Vacated -> logger.warn { "No leader right now" }
      is LeadershipEvent.WatchFailed -> logger.error(event.cause) { "Observation stopped" }
    }
  }

PathChildrenCache(client, "/services").use { cache ->
  // Subscribe BEFORE starting the cache: events fired while nothing collects are
  // not buffered, so a late collector misses INITIALIZED.
  scope.launch {
    cache.eventsAsFlow().collect { event ->
      logger.info { "${event.type} ${event.childName}" }
    }
  }
  scope.launch {
    cache.recoveryEventsAsFlow().collect { event ->
      logger.warn { "Cache watch recovery: $event" }
    }
  }
  cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT)
}

NodeCache(client, "/config/mode", StringCodec).use { cache ->
  scope.launch {
    cache.eventsAsFlow().collect { event ->
      logger.info { "${event.type}: ${event.value}" }
    }
  }
  cache.start()
}

data class ServiceCacheEvent(
  val eventType: WatchEvent.EventType,
  val isAdd: Boolean,
  val serviceName: String,
  val serviceInstance: ServiceInstance?,
)

ServiceCache(client, "/services/orders/names", "worker").use { cache ->
  scope.launch {
    cache.eventsAsFlow().collect { event ->
      val verb = if (event.isAdd) "joined" else "left"
      logger.info { "${event.serviceInstance?.id} $verb ${event.serviceName}" }
    }
  }
  cache.start()
}

data class LockLostEvent(val cause: Throwable?)
data class PermitLostEvent(val cause: Throwable?)

DistributedMutex(client, "/locks/orders").use { mutex ->
  scope.launch {
    // Unbuffered by design: the listener fires on jetcd's lease-callback thread,
    // which must never block, so this flow takes no capacity argument.
    mutex.lockLostAsFlow().collect { event ->
      logger.error(event.cause) { "Lost the lock; abandoning work" }
    }
  }
}

DistributedSemaphore(client, "/semaphores/uploads", permits = 3).use { semaphore ->
  scope.launch {
    semaphore.permitLostAsFlow().collect { event ->
      logger.error(event.cause) { "Lost a permit" }
    }
  }
}

TransientKeyValue(client, "/nodes/worker-1", "alive").use { kv ->
  scope.launch {
    kv.leaseEventsAsFlow().collect { event ->
      when (event) {
        is LeaseEvent.Suspended -> logger.warn { "Keep-alive faltering: ${event.cause}" }
        is LeaseEvent.Expired -> logger.error { "Key ownership was lost" }
        is LeaseEvent.Restored -> logger.info { "Healed onto lease ${event.newLeaseId}" }
        is LeaseEvent.Failed -> logger.error(event.cause) { "Healing abandoned" }
      }
    }
  }
}

PathChildrenCache(client, "/services").use { cache ->
  scope.launch {
    // Emits the current state on collect, then every transition. Conflated: a slow
    // collector sees the latest state rather than every intermediate one.
    cache.connectionStateAsFlow().collect { state ->
      if (state == ConnectionState.LOST) logger.error { "Cache is no longer trustworthy" }
    }
  }
  scope.launch {
    // The push counterpart to the pull-only `exceptions` list.
    cache.backgroundExceptionsAsFlow().collect { failure ->
      logger.error(failure.throwable) { "Background failure in ${failure.context}" }
    }
  }
}

PathChildrenCache(client, "/services").use { cache ->
  // One parent job per subscription set: cancelling `collectors` unsubscribes
  // every flow — each awaitClose removes its listener or closes its watcher.
  val collectors =
    scope.launch {
      launch {
        cache.eventsAsFlow().collect { event ->
          logger.info { "Cache: ${event.type} ${event.childName}" }
        }
      }
      launch {
        cache.connectionStateAsFlow().collect { state ->
          logger.info { "Connection state: $state" }
        }
      }
      launch {
        client.leadershipAsFlow("/elections/orders").collect { event ->
          logger.info { "Leadership: $event" }
        }
      }
    }

  cache.start(buildInitial = true)
  collectors.cancel()
}