Add support for Spring Data and update Lettuce support

build.gradle

@@ -18,6 +18,9 @@ buildscript {
         gradleVersionsVersion = "0.47.0"
         jedisVersion = "4.4.3"
         lettuceVersion = "6.2.6.RELEASE"
+        springDataVersion = "2.7.18"
+        testcontainersVersion= "1.19.3"
+        testcontainersRedisVersion= "2.0.1"
     }
 
     configurations.classpath {
@@ -61,12 +64,18 @@ dependencies {
     testImplementation "org.amshove.kluent:kluent:$kluentVersion"
     testImplementation "org.junit.jupiter:junit-jupiter:$junitVersion"
     testImplementation "org.skyscreamer:jsonassert:$jsonAssertVersion"
+    testImplementation "org.testcontainers:junit-jupiter:$testcontainersVersion"
+    testImplementation "com.redis:testcontainers-redis:$testcontainersRedisVersion"
+
     testImplementation "redis.clients:jedis:$jedisVersion"
+    testImplementation "io.lettuce:lettuce-core:$lettuceVersion"
+    testImplementation "org.springframework.boot:spring-boot-starter-data-redis:$springDataVersion"
 
     testRuntimeOnly "net.bytebuddy:byte-buddy:$byteBuddyVersion"
 
     compileOnly "redis.clients:jedis:$jedisVersion"
     compileOnly "io.lettuce:lettuce-core:$lettuceVersion"
+    compileOnly "org.springframework.boot:spring-boot-starter-data-redis:$springDataVersion"
 }
 
 java {

src/main/kotlin/de/smartsquare/socketio/emitter/LettucePublisher.kt

@@ -1,9 +1,9 @@
 package de.smartsquare.socketio.emitter
 
-import io.lettuce.core.api.StatefulRedisConnection
+import io.lettuce.core.api.sync.RedisCommands
 
-class LettucePublisher(private val connection: StatefulRedisConnection<Any, Any>) : RedisPublisher {
+class LettucePublisher(private val commands: RedisCommands<String, String>) : RedisPublisher {
     override fun publish(channel: String, message: ByteArray) {
-        connection.use { it.sync().publish(channel, message.decodeToString()) }
+        commands.publish(channel, message.decodeToString())
     }
 }

src/main/kotlin/de/smartsquare/socketio/emitter/SpringDataPublisher.kt

@@ -0,0 +1,9 @@
+package de.smartsquare.socketio.emitter
+
+import org.springframework.data.redis.core.RedisTemplate
+
+class SpringDataPublisher(private val redisTemplate: RedisTemplate<String, String>) : RedisPublisher {
+    override fun publish(channel: String, message: ByteArray) {
+        redisTemplate.execute({ it.publish(channel.toByteArray(), message) }, true)
+    }
+}

src/test/kotlin/de/smartsquare/socketio/emitter/JedisPublisherTest.kt

@@ -0,0 +1,65 @@
+package de.smartsquare.socketio.emitter
+
+import com.redis.testcontainers.RedisContainer
+import org.amshove.kluent.shouldBeEqualTo
+import org.amshove.kluent.shouldBeTrue
+import org.amshove.kluent.shouldContain
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import redis.clients.jedis.JedisPool
+import redis.clients.jedis.JedisPubSub
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+@Testcontainers
+class JedisPublisherTest {
+
+    @Container
+    private val redis = RedisContainer("redis:6-alpine")
+
+    private lateinit var pool: JedisPool
+
+    @BeforeEach
+    fun setUp() {
+        pool = JedisPool(redis.redisURI)
+    }
+
+    @AfterEach
+    fun tearDown() {
+        pool.close()
+    }
+
+    @Test
+    fun `publish string message`() {
+        val publisher = Emitter(JedisPublisher(pool))
+
+        val countDownLatch = CountDownLatch(1)
+
+        val listener = object : JedisPubSub() {
+            override fun onMessage(channel: String, message: String) {
+                channel shouldBeEqualTo "socket.io#/#"
+                message shouldContain "test 123"
+
+                countDownLatch.countDown()
+            }
+        }
+
+        val executor = Executors.newSingleThreadExecutor()
+        val jedis = pool.resource
+
+        try {
+            executor.submit { jedis.subscribe(listener, "socket.io#/#") }
+
+            publisher.broadcast("topic", "test 123")
+
+            countDownLatch.await(5, TimeUnit.SECONDS).shouldBeTrue()
+        } finally {
+            listener.unsubscribe()
+            executor.shutdown()
+        }
+    }
+}

src/test/kotlin/de/smartsquare/socketio/emitter/LettucePublisherTest.kt

@@ -0,0 +1,60 @@
+package de.smartsquare.socketio.emitter
+
+import com.redis.testcontainers.RedisContainer
+import io.lettuce.core.RedisClient
+import io.lettuce.core.pubsub.RedisPubSubAdapter
+import org.amshove.kluent.shouldBeEqualTo
+import org.amshove.kluent.shouldBeTrue
+import org.amshove.kluent.shouldContain
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+@Testcontainers
+class LettucePublisherTest {
+
+    @Container
+    private val redis = RedisContainer("redis:6-alpine")
+
+    private lateinit var client: RedisClient
+
+    @BeforeEach
+    fun setUp() {
+        client = RedisClient.create(redis.redisURI)
+    }
+
+    @AfterEach
+    fun tearDown() {
+        client.close()
+    }
+
+    @Test
+    fun `publish string message`() {
+        val connection = client.connect().sync()
+        val publisher = Emitter(LettucePublisher(connection))
+
+        val countDownLatch = CountDownLatch(1)
+
+        val listener = object : RedisPubSubAdapter<String, String>() {
+            override fun message(channel: String, message: String) {
+                channel shouldBeEqualTo "socket.io#/#"
+                message shouldContain "test 123"
+
+                countDownLatch.countDown()
+            }
+        }
+
+        client.connectPubSub().apply {
+            addListener(listener)
+            sync().subscribe("socket.io#/#")
+        }
+
+        publisher.broadcast("topic", "test 123")
+
+        countDownLatch.await(5, TimeUnit.SECONDS).shouldBeTrue()
+    }
+}

src/test/kotlin/de/smartsquare/socketio/emitter/SpringDataPublisherTest.kt

@@ -0,0 +1,65 @@
+package de.smartsquare.socketio.emitter
+
+import com.redis.testcontainers.RedisContainer
+import org.amshove.kluent.shouldBeEqualTo
+import org.amshove.kluent.shouldBeTrue
+import org.amshove.kluent.shouldContain
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.springframework.data.redis.connection.Message
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory
+import org.springframework.data.redis.core.StringRedisTemplate
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+
+@Testcontainers
+class SpringDataPublisherTest {
+
+    @Container
+    private val redis = RedisContainer("redis:6-alpine")
+
+    private lateinit var lettuceConnectionFactory: LettuceConnectionFactory
+    private lateinit var template: StringRedisTemplate
+
+    @BeforeEach
+    fun setUp() {
+        lettuceConnectionFactory = LettuceConnectionFactory(redis.host, redis.firstMappedPort).apply {
+            afterPropertiesSet()
+        }
+
+        template = StringRedisTemplate().apply {
+            connectionFactory = lettuceConnectionFactory
+            afterPropertiesSet()
+        }
+    }
+
+    @AfterEach
+    fun tearDown() {
+        lettuceConnectionFactory.destroy()
+    }
+
+    @Test
+    fun `publish string message`() {
+        val publisher = Emitter(SpringDataPublisher(template))
+
+        val countDownLatch = CountDownLatch(1)
+
+        val messageListener = { message: Message, _: ByteArray? ->
+            message.channel.decodeToString() shouldBeEqualTo "socket.io#/#"
+            message.body.decodeToString() shouldContain "test 123"
+
+            countDownLatch.countDown()
+        }
+
+        template.requiredConnectionFactory.connection.also {
+            it.subscribe(messageListener, "socket.io#/#".toByteArray())
+        }
+
+        publisher.broadcast("topic", "test 123")
+
+        countDownLatch.await(5, TimeUnit.SECONDS).shouldBeTrue()
+    }
+}