Merge pull request #4 from SLNE-Development/copilot/add-spring-and-redis-connection

Add global Redis connection and Redis Streams support

Author: twisti-dev

README.md

@@ -36,6 +36,8 @@ runBlocking {
 - 🎯 Annotation-based event subscription
 - 🔧 Type-safe event handling with Kotlin Serialization
 - ⚡ High-performance event invocation using MethodHandles
+- 🔄 **Redis Streams support** for reliable event delivery with persistence
+- 🎛️ **Global Redis connection** management via RedisApi
 
 ## Requirements
 
@@ -129,6 +131,65 @@ Examples:
 - `redis://password@localhost:6379` - Local Redis with password
 - `redis://[email protected]:6379/0` - Remote Redis with password and database
 
+## Global Redis Connection (RedisApi)
+
+surf-redis provides a centralized way to manage Redis connections via the `RedisApi` singleton:
+
+```kotlin
+import de.slne.redis.RedisApi
+
+// Initialize global connection
+RedisApi.init(url = "redis://localhost:6379")
+
+// Alternative syntax
+RedisApi(url = "redis://localhost:6379").connect()
+
+// Check connection status
+if (RedisApi.isConnected()) {
+    println("Connected to: ${RedisApi.getUrl()}")
+}
+
+// Create connections
+val connection = RedisApi.createConnection()
+val pubSubConnection = RedisApi.createPubSubConnection()
+
+// Close connection when done
+RedisApi.disconnect()
+```
+
+The `RedisApi` automatically initializes with a default URL (`redis://localhost:6379`) if not explicitly configured.
+
+## Redis Streams
+
+For more reliable event delivery with message persistence, use `RedisStreamEventBus`:
+
+### Benefits of Redis Streams
+
+- 📦 **Message Persistence**: Events are stored and not lost if no consumer is online
+- 👥 **Consumer Groups**: Multiple instances can share the load
+- ✅ **Message Acknowledgment**: Ensures events are processed
+- 🔄 **Reprocessing**: Failed events can be reprocessed
+
+### Usage
+
+```kotlin
+import de.slne.redis.stream.RedisStreamEventBus
+
+// Create stream-based event bus
+val streamBus = RedisStreamEventBus(
+    streamName = "my-events",
+    consumerGroup = "my-app",
+    consumerName = "instance-1"
+)
+
+// Use exactly like RedisEventBus
+streamBus.registerListener(MyListener())
+streamBus.publish(MyEvent())
+streamBus.close()
+```
+
+Redis Streams provide stronger guarantees than pub/sub, making them ideal for critical events that must not be lost.
+
 ## How It Works
 
 1. **Event Publishing**: When you call `eventBus.publish(event)`, the event is serialized using Kotlin Serialization and published asynchronously to a Redis channel using coroutines.

build.gradle.kts

@@ -15,6 +15,9 @@ dependencies {
     // Lettuce Redis client
     implementation("io.lettuce:lettuce-core:6.3.0.RELEASE")
 
+    // Logging
+    implementation("org.slf4j:slf4j-api:2.0.9")
+    
     // Kotlin
     implementation(kotlin("stdlib"))
     implementation(kotlin("reflect"))

src/main/kotlin/de/slne/redis/RedisApi.kt

@@ -0,0 +1,122 @@
+package de.slne.redis
+
+import io.lettuce.core.RedisClient
+import io.lettuce.core.api.StatefulRedisConnection
+import io.lettuce.core.pubsub.StatefulRedisPubSubConnection
+
+/**
+ * Global Redis API for managing Redis connections.
+ * Provides a centralized way to initialize and access Redis connections.
+ * 
+ * Usage:
+ * ```
+ * RedisApi.init(url = "redis://localhost:6379")
+ * // or
+ * RedisApi(url = "redis://localhost:6379").connect()
+ * ```
+ */
+object RedisApi {
+    private var redisClient: RedisClient? = null
+    private var redisUrl: String = "redis://localhost:6379"
+    private var isInitialized = false
+    
+    /**
+     * Initialize the Redis API with a connection URL.
+     * This should be called once at application startup.
+     * 
+     * @param url The Redis connection URL (e.g., "redis://localhost:6379")
+     * @return This RedisApi instance for chaining
+     */
+    fun init(url: String = "redis://localhost:6379"): RedisApi {
+        if (isInitialized && redisUrl == url) {
+            return this
+        }
+        
+        // Close existing connection if any
+        if (isInitialized) {
+            disconnect()
+        }
+        
+        redisUrl = url
+        connect()
+        return this
+    }
+    
+    /**
+     * Alternative syntax for initialization.
+     * 
+     * @param url The Redis connection URL
+     * @return This RedisApi instance
+     */
+    operator fun invoke(url: String): RedisApi {
+        return init(url)
+    }
+    
+    /**
+     * Connect to Redis using the configured URL.
+     * Called automatically by init().
+     * 
+     * @return This RedisApi instance for chaining
+     */
+    fun connect(): RedisApi {
+        if (!isInitialized) {
+            redisClient = RedisClient.create(redisUrl)
+            isInitialized = true
+        }
+        return this
+    }
+    
+    /**
+     * Disconnect from Redis and clean up resources.
+     */
+    fun disconnect() {
+        redisClient?.shutdown()
+        redisClient = null
+        isInitialized = false
+    }
+    
+    /**
+     * Get the Redis client instance.
+     * Automatically initializes with default URL if not already initialized.
+     * 
+     * @return The RedisClient instance
+     */
+    fun getClient(): RedisClient {
+        if (!isInitialized) {
+            init()
+        }
+        return redisClient ?: throw IllegalStateException("Redis client not initialized")
+    }
+    
+    /**
+     * Get the configured Redis URL.
+     * 
+     * @return The Redis connection URL
+     */
+    fun getUrl(): String = redisUrl
+    
+    /**
+     * Check if Redis is initialized and connected.
+     * 
+     * @return True if initialized, false otherwise
+     */
+    fun isConnected(): Boolean = isInitialized && redisClient != null
+    
+    /**
+     * Create a new stateful connection for Redis commands.
+     * 
+     * @return A new StatefulRedisConnection
+     */
+    fun createConnection(): StatefulRedisConnection<String, String> {
+        return getClient().connect()
+    }
+    
+    /**
+     * Create a new stateful pub/sub connection for Redis pub/sub.
+     * 
+     * @return A new StatefulRedisPubSubConnection
+     */
+    fun createPubSubConnection(): StatefulRedisPubSubConnection<String, String> {
+        return getClient().connectPubSub()
+    }
+}

src/main/kotlin/de/slne/redis/example/FeatureDemo.kt

@@ -0,0 +1,81 @@
+package de.slne.redis.example
+
+import de.slne.redis.RedisApi
+import de.slne.redis.event.RedisEventBus
+import de.slne.redis.event.Subscribe
+import de.slne.redis.stream.RedisStreamEventBus
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+
+/**
+ * Example demonstrating the new features:
+ * 1. Global RedisApi connection
+ * 2. Redis Streams support
+ * 3. Direct usage without Spring
+ */
+
+fun main() = runBlocking {
+    println("=== surf-redis Feature Demo ===\n")
+    
+    // Feature 1: Global Redis Connection via RedisApi
+    println("1. Initializing global Redis connection via RedisApi")
+    RedisApi.init(url = "redis://localhost:6379")
+    println("   Redis connected: ${RedisApi.isConnected()}")
+    println("   Redis URL: ${RedisApi.getUrl()}\n")
+    
+    // Alternative syntax as requested
+    // RedisApi(url = "redis://localhost:6379").connect()
+    
+    // Feature 2: Traditional Event Bus (pub/sub)
+    println("2. Using traditional RedisEventBus (pub/sub)")
+    val eventBus = RedisEventBus("redis://localhost:6379")
+    val listener1 = DemoListener("Listener1")
+    eventBus.registerListener(listener1)
+    
+    eventBus.publish(PlayerJoinEvent("Alice", "uuid-1", "Server-1"))
+    delay(100) // Give time for async handling
+    println()
+    
+    // Feature 3: Redis Streams for reliable delivery
+    println("3. Using RedisStreamEventBus (Redis Streams)")
+    val streamBus = RedisStreamEventBus(
+        streamName = "demo-events",
+        consumerGroup = "demo-group",
+        consumerName = "demo-consumer-1"
+    )
+    val listener2 = DemoListener("StreamListener")
+    streamBus.registerListener(listener2)
+    
+    // Give time for consumer to start
+    delay(500)
+    
+    streamBus.publish(PlayerJoinEvent("Bob", "uuid-2", "Server-2"))
+    delay(100) // Give time for async handling
+    println()
+    
+    // Feature 4: Multiple events
+    println("4. Publishing multiple events")
+    eventBus.publish(ChatMessageEvent("Alice", "Hello everyone!", "Server-1"))
+    streamBus.publish(ChatMessageEvent("Bob", "Hi Alice!", "Server-2"))
+    delay(100)
+    println()
+    
+    // Clean up
+    println("5. Cleaning up...")
+    eventBus.close()
+    streamBus.close()
+    RedisApi.disconnect()
+    println("Demo completed successfully!")
+}
+
+class DemoListener(private val name: String) {
+    @Subscribe
+    fun onPlayerJoin(event: PlayerJoinEvent) {
+        println("   [$name] Player ${event.playerName} joined ${event.serverName}")
+    }
+    
+    @Subscribe
+    fun onChatMessage(event: ChatMessageEvent) {
+        println("   [$name] [${event.serverName}] ${event.playerName}: ${event.message}")
+    }
+}