# How CoSky's Event Listener System Works for Instance Changes: Redis Pub/Sub Deep Dive > Discover how CoSky's event listener system uses Redis Pub/Sub to stream instance changes. Understand the mechanics of Cosky's reactive approach for instance event handling. - Repository: [Ahoo Wang/cosky](https://github.com/ahoo-wang/cosky) - Tags: deep-dive - Published: 2026-02-23 --- **CoSky implements a reactive event listener system for instance changes through a generic `EventListenerContainer` interface that uses Redis Pub/Sub patterns to stream `InstanceChangedEvent` objects to consumers.** The **ahoo-wang/cosky** repository provides a Spring Cloud-compatible service discovery platform that uses Redis as its backing store. Understanding how CoSky's event listener system handles instance changes is crucial for building reactive service mesh applications that respond to dynamic service registration and deregistration in real-time. ## Core Architecture of the Event Listener System ### The Generic Event Listener Contract At the foundation of CoSky's event system lies the `EventListenerContainer` interface defined in [[`cosky-core/src/main/kotlin/me/ahoo/cosky/core/EventListenerContainer.kt`](https://github.com/ahoo-wang/cosky/blob/main/cosky-core/src/main/kotlin/me/ahoo/cosky/core/EventListenerContainer.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-core/src/main/kotlin/me/ahoo/cosky/core/EventListenerContainer.kt). This generic contract abstracts any asynchronous event source: ```kotlin interface EventListenerContainer : AutoCloseable { fun receive(topic: T): Flux } ``` The interface uses **Project Reactor's `Flux`** to provide a non-blocking stream of events. The type parameter `T` represents the topic identifier, while `E` constrains the event payload. By extending `AutoCloseable`, the container ensures proper resource cleanup when the application shuts down. ### Domain-Specific Abstraction for Service Discovery Building upon the generic contract, CoSky defines `InstanceEventListenerContainer` in [[`cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/InstanceEventListenerContainer.kt`](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/InstanceEventListenerContainer.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/InstanceEventListenerContainer.kt): ```kotlin interface InstanceEventListenerContainer : EventListenerContainer ``` This specialization fixes the topic type to `NamespacedServiceId` (a data class combining **namespace** and **serviceId**) and the event type to `InstanceChangedEvent`. This type-safe approach ensures that consumers receive strongly-typed events specific to service instance lifecycle changes. ## Redis Implementation Details ### RedisInstanceEventListenerContainer Implementation The concrete implementation resides in [[`cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/redis/RedisInstanceEventListenerContainer.kt`](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/redis/RedisInstanceEventListenerContainer.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/redis/RedisInstanceEventListenerContainer.kt). This class bridges the domain-specific interface with Spring Data Redis's reactive messaging capabilities: ```kotlin class RedisInstanceEventListenerContainer( delegate: ReactiveRedisMessageListenerContainer, private val serviceTopology: ServiceTopology = ServiceTopology.NO_OP ) : InstanceEventListenerContainer, RedisEventListenerContainer(delegate) { override fun receiveEvent(topic: NamespacedServiceId): Flux { val pattern = if (topic.serviceId.isNotBlank()) { DiscoveryKeyGenerator.getInstanceKeyPatternOfService(topic.namespace, topic.serviceId) } else { DiscoveryKeyGenerator.getInstanceKeyPatternOfNamespace(topic.namespace) } return delegate.receive(PatternTopic.of(pattern)) .map { asEvent(it) } .doOnSubscribe { if (topic.serviceId.isNotBlank()) { @Suppress("CallingSubscribeInNonBlockingScope") serviceTopology.addTopology(topic.namespace, topic.serviceId).subscribe() } } } } ``` The constructor accepts a `ReactiveRedisMessageListenerContainer` as its delegate and an optional `ServiceTopology` instance for load-balancer integration. ### Topic Pattern Construction and Message Conversion The `receiveEvent` method dynamically constructs Redis key patterns using `DiscoveryKeyGenerator`. When a specific `serviceId` is provided, it generates a pattern matching that service's instances; otherwise, it matches all services within the namespace. Message conversion occurs in the private `asEvent` method: ```kotlin private fun asEvent(message: ReactiveSubscription.Message): InstanceChangedEvent { val namespace = DiscoveryKeyGenerator.getNamespaceOfKey(message.channel) val instanceId = DiscoveryKeyGenerator.getInstanceIdOfKey(namespace, message.channel) val instance: Instance = instanceId.asInstance() val serviceId = instance.serviceId val namespacedServiceId = NamespacedServiceId(namespace, serviceId) return InstanceChangedEvent( namespacedServiceId, message.message.asServiceChangedEvent(), instance ) } ``` This method extracts the namespace and instance ID from the Redis channel name, reconstructs the `Instance` object, parses the payload into a `ServiceChangedEvent`, and assembles the final `InstanceChangedEvent`. ### Service Topology Integration A critical side effect of subscription occurs in the `doOnSubscribe` operator. When listening to a specific service, the container invokes `serviceTopology.addTopology(namespace, serviceId)` to register the service in the internal topology map. This registration ensures that CoSky's load balancers (such as `RandomLoadBalancer` or `TreeChooser`) maintain an up-to-date view of available instances. ## Event Types and Supporting Infrastructure ### NamespacedServiceId and InstanceChangedEvent The [[`NamespacedServiceId.kt`](https://github.com/ahoo-wang/cosky/blob/main/NamespacedServiceId.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/NamespacedServiceId.kt) file defines the topic identifier as a simple data class holding `namespace` and `serviceId` strings. The event payload, defined in [[`InstanceChangedEvent.kt`](https://github.com/ahoo-wang/cosky/blob/main/InstanceChangedEvent.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/InstanceChangedEvent.kt), wraps three critical pieces of information: - The `NamespacedServiceId` identifying which service changed - The `ServiceChangedEvent` indicating the type of change (registration, deregistration, or metadata update) - The `Instance` object containing the full service instance details ### DiscoveryKeyGenerator Utility Located in [[`DiscoveryKeyGenerator.kt`](https://github.com/ahoo-wang/cosky/blob/main/DiscoveryKeyGenerator.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/DiscoveryKeyGenerator.kt), this utility class provides static methods for mapping domain objects to Redis keys and patterns. Key methods include: - `getInstanceKeyPatternOfService(namespace, serviceId)` – Creates a Redis pattern for a specific service - `getInstanceKeyPatternOfNamespace(namespace)` – Creates a pattern for all services in a namespace - `getNamespaceOfKey(channel)` – Extracts namespace from a Redis channel name - `getInstanceIdOfKey(namespace, channel)` – Parses the instance ID from the channel ## Practical Usage and Integration ### Bootstrap Configuration Spring Boot auto-configuration creates the necessary beans during application startup. The [[`CoSkyDiscoveryAutoConfiguration.kt`](https://github.com/ahoo-wang/cosky/blob/main/CoSkyDiscoveryAutoConfiguration.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-spring-cloud-starter-discovery/src/main/kotlin/me/ahoo/cosky/discovery/spring/cloud/discovery/CoSkyDiscoveryAutoConfiguration.kt) registers a `RedisInstanceEventListenerContainer` bean that injects the `ReactiveRedisMessageListenerContainer` provided by Spring Data Redis. ### Consuming Instance Change Events Components react to instance changes by injecting `InstanceEventListenerContainer` and subscribing to the returned `Flux`. Here is a complete example demonstrating how to watch for new instance registrations: ```kotlin @Component class InstanceWatcher( private val listenerContainer: InstanceEventListenerContainer ) { fun watch(namespace: String, serviceId: String): Disposable { val topic = NamespacedServiceId(namespace, serviceId) return listenerContainer.receive(topic) .filter { it.changedEvent.isRegister } // Only registration events .doOnNext { event -> println("New instance registered: ${event.instance.instanceId}") println("Service: ${event.namespacedServiceId.serviceId}") println("Event type: ${event.changedEvent}") } .subscribe() } } ``` The `receive` method returns a cold `Flux`, meaning the actual Redis subscription only occurs when a consumer subscribes to the stream. This lazy initialization conserves resources until the application explicitly requests event notifications. ## Summary - **CoSky's event listener system** uses a three-layer architecture: generic `EventListenerContainer`, domain-specific `InstanceEventListenerContainer`, and the Redis-backed `RedisInstanceEventListenerContainer`. - **Redis Pub/Sub patterns** enable real-time streaming of instance changes through the `ReactiveRedisMessageListenerContainer` delegate. - **Dynamic topic construction** via `DiscoveryKeyGenerator` allows listening to either specific services or entire namespaces using key patterns. - **Automatic topology registration** ensures load balancers receive immediate updates when new service listeners are attached. - **Reactive Streams integration** provides backpressure-aware event consumption through Project Reactor's `Flux` API. ## Frequently Asked Questions ### What is the primary interface for listening to instance changes in CoSky? The primary interface is `InstanceEventListenerContainer`, defined in [[`InstanceEventListenerContainer.kt`](https://github.com/ahoo-wang/cosky/blob/main/InstanceEventListenerContainer.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/InstanceEventListenerContainer.kt). It extends the generic `EventListenerContainer` interface and specifies `NamespacedServiceId` as the topic type and `InstanceChangedEvent` as the event payload type. Applications inject this interface to receive reactive streams of service instance changes. ### How does CoSky convert Redis messages into domain events? Conversion occurs in the `asEvent` method of `RedisInstanceEventListenerContainer`. This method parses the Redis channel name using `DiscoveryKeyGenerator` to extract the namespace and instance ID, reconstructs the `Instance` object from the channel metadata, and parses the message payload into a `ServiceChangedEvent`. These components are then assembled into an `InstanceChangedEvent` object that the application consumes. ### What role does ServiceTopology play in the event listener system? `ServiceTopology`, defined in [[`ServiceTopology.kt`](https://github.com/ahoo-wang/cosky/blob/main/ServiceTopology.kt)](https://github.com/ahoo-wang/cosky/blob/main/cosky-discovery/src/main/kotlin/me/ahoo/cosky/discovery/ServiceTopology.kt), maintains an internal graph of service relationships for load balancing. When `RedisInstanceEventListenerContainer.receiveEvent()` is called with a specific service ID, it triggers `serviceTopology.addTopology(namespace, serviceId)` during subscription. This registration ensures that CoSky's load balancers (such as `RandomLoadBalancer`) can route requests to newly discovered instances immediately. ### Can I listen to all services within a namespace rather than a specific service? Yes. When constructing the `NamespacedServiceId` topic, pass an empty or blank `serviceId`. The `receiveEvent` method in `RedisInstanceEventListenerContainer` checks `topic.serviceId.isNotBlank()` and, if false, uses `DiscoveryKeyGenerator.getInstanceKeyPatternOfNamespace(topic.namespace)` to create a Redis pattern that matches instance changes for all services within that namespace. Note that topology registration is skipped when listening at the namespace level.