1 files changed, 46 insertions, 18 deletions

diff --git a/opendc-common/src/main/kotlin/org/opendc/common/utils/Redis.kt b/opendc-common/src/main/kotlin/org/opendc/common/utils/Redis.kt
index b659d40a..bf674a33 100644
--- a/opendc-common/src/main/kotlin/org/opendc/common/utils/Redis.kt
+++ b/opendc-common/src/main/kotlin/org/opendc/common/utils/Redis.kt
@@ -4,40 +4,68 @@ import com.fasterxml.jackson.dataformat.toml.TomlMapper
 import redis.clients.jedis.RedisClient
 import redis.clients.jedis.StreamEntryID
 import redis.clients.jedis.params.XReadParams
+import java.util.Map
 import java.util.Properties
 
+
 /**
  * This class represents the Redis server instance.
  * @author Mateusz Kwiatkowski
  * @see <a href=https://redis.io/docs/latest/>https://redis.io/docs/latest/</a>
  *
  * @see <a href=https://redis.io/docs/latest/develop/data-types/streams/>https://redis.io/docs/latest/develop/data-types/streams/</a>
+ *
+ * @see <a href=https://www.javadoc.io/doc/redis.clients/jedis/latest/index.html>https://www.javadoc.io/doc/redis.clients/jedis/latest/index.html</a>
  */
 
 @Suppress("DEPRECATION")
-public class Redis {
-    private var properties : Properties
+public class Redis private constructor() {
+    private var jedis : RedisClient? = null
+    private var properties : Properties? = null
 
     init {
-        properties  = TomlMapper().readerFor(Properties().javaClass)
-            .readValue(Kafka::class.java.getResource("/producer.toml"))
+        properties = TomlMapper().readerFor(Properties().javaClass).readValue(Kafka::class.java.getResource("/subscriber.toml"))
+        jedis = RedisClient.create("redis://localhost:${properties?.getProperty("port")}")
     }
 
-        public fun run() {
-            val jedis : RedisClient = RedisClient.create("redis://localhost:6379")
+    public companion object {
+        private var instance: Redis? = null
 
-            val res5 = jedis.xread(
-                XReadParams.xReadParams().block(300).count(100),
-                object : HashMap<String?, StreamEntryID?>() {
-                    init {
-                        put("${properties.getProperty("table")}", StreamEntryID())
+        public fun getInstance(): Redis? {
+            if (instance == null) {
+                instance = Redis()
+            }
+            return instance
+        }
+    }
+
+    public fun readOne() {
+        while(true) {
+            val messages = jedis?.xread(
+                XReadParams.xReadParams(),
+                Map.of("${properties?.getProperty("table")}", StreamEntryID()
+                ));
+
+            if (messages != null) {
+                for (stream in messages) {
+                    for (entry in stream.value) {
+                        if(entry.getFields().get("downtime") != null) {
+                            entry.getFields().get("downtime")?.toDouble()?.let {
+                                if (it > 0) {
+                                    println("ID: " + entry.getID())
+                                }
+                            }
+                        }
                     }
-                })
+                }
+            }
+        }
+            // https://redis.io/docs/latest/develop/use-cases/streaming/java-jedis/
+            // In Redis you can subscribe to updates to a stream.
+            // You should base your application off this.
+            // You can listen for new items with XREAD
 
-            // in Redis you can subscribe to updates to a stream.
-            // you should base your application off this.
-            // you can listen for new items with XREAD
-            println(res5)
-            jedis.close()
+            // do not close for now
+            //jedis.close()
         }
-}
\ No newline at end of file
+}