Typed values¶

interface EtcdCodec<T> {
  fun encode(value: T): ByteSequence
  fun decode(bytes: ByteSequence): T
}

// The identity codec: the payload already is a ByteSequence, so nothing is marshalled.
val raw: EtcdCodec<ByteSequence> = ByteSequenceCodec

// UTF-8 text.
val text: EtcdCodec<String> = StringCodec

// kotlinx-serialization JSON for any @Serializable type. The reified factory resolves
// the serializer at the call site, so you never name it yourself.
val orders: EtcdCodec<Order> = jsonCodec<Order>()

@Serializable
data class Order(
  val id: String,
  val qty: Int,
)

val codec = jsonCodec<Order>()

// Encodes through the codec and composes the raw ByteSequence put, so it inherits
// that put's retry semantics.
client.putValue("/config/order", Order("A-1", 3), codec)

// Returns null when the key is absent — pick the fallback yourself.
val order: Order? = client.getValue("/config/order", codec)
logger.info { "Read back $order" }

TypedDistributedQueue(client, "/queues/orders", jsonCodec<Order>()).use { queue ->
  queue.enqueue(Order("A-1", 3))
  queue.enqueueAll([Order("A-2", 1), Order("A-3", 7)])

  // dequeue() hands back an Order, not a ByteSequence.
  val order: Order = queue.dequeue()
  logger.info { "Dequeued $order" }
}

val untyped: DistributedQueue

TypedDistributedQueue(client, "/queues/orders", jsonCodec<Order>()).use { queue ->
  // A decorator is a Closeable, not an EtcdConnector. The whole connector API is
  // reached through `untyped` — the very instance the decorator wraps.
  queue.untyped.addBackgroundExceptionListener { context, e ->
    logger.warn(e) { "Background failure in $context" }
  }

  if (!queue.untyped.isHealthy()) logger.warn { "Connection lost or closed" }

  queue.enqueue(Order("A-1", 3))
  queue.untyped.exceptions.forEach { e -> logger.error(e) { "Recorded" } }
}

// NodeCache is not a decorator: it is itself an EtcdConnector that happens to take a
// codec, so `exceptions` / `isHealthy()` sit directly on the instance — no `untyped`.
NodeCache(client, "/config/order", jsonCodec<Order>()).use { cache ->
  cache.addListener { event -> logger.info { "${event.type} -> ${event.value}" } }
  cache.start()

  // The live value, decoded on read; null while the key is absent.
  val current: Order? = cache.current
  logger.info { "Current $current, healthy=${cache.isHealthy()}" }
}

TypedPathChildrenCache(client, "/config/orders", jsonCodec<Order>()).use { cache ->
  cache.addListener { event -> logger.info { "${event.childName} -> ${event.data}" } }
  cache.start(buildInitial = true)

  cache.currentData.forEach { child -> logger.info { "${child.key} = ${child.value}" } }
  val one: Order? = cache.getCurrentData("A-1")
  logger.info { "One $one" }
}

// The published value is a String, so this needs a UTF-8 text codec.
TypedTransientKeyValue(client, "/nodes/n1", Order("A-1", 3), jsonCodec<Order>()).use { tkv ->
  tkv.addLeaseListener { event -> logger.info { "Lease: $event" } }

  // Read it back through the matching typed getValue.
  val published: Order? = client.getValue("/nodes/n1", jsonCodec<Order>())
  logger.info { "Published $published" }
}

val codec = jsonCodec<Order>()

// The typed payload layers over the opaque jsonPayload String, so the ServiceInstance
// wire format is unchanged. That is also why a binary codec is unsupported here.
val instance = serviceInstance("orders", Order("A-1", 3), codec) { apply { port = 8080 } }

ServiceDiscovery(client, "/discovery").use { discovery ->
  discovery.registerService(instance)

  discovery.queryForInstances("orders").forEach { found ->
    val order: Order = found.payload(codec)
    logger.info { "${found.name} carries $order" }
  }
}

// A codec is two functions and nothing more. Implement EtcdCodec directly when the
// payload is not JSON — a protobuf message, a packed binary struct, a bare Long.
object LongCodec : EtcdCodec<Long> {
  override fun encode(value: Long): ByteSequence = value.toString().asByteSequence

  override fun decode(bytes: ByteSequence): Long = bytes.asString.toLong()
}