Merge pull request #6 from bolteu/task/RNTL-25522-non-wildcard-retained-keys-optimization

RNTL-25522: optimize non-wildcard retained topics handling

Author: keijokuusmik

persistence.js

@@ -81,11 +81,24 @@ async function getRetainedKeys (db, hasClusters) {
   return await db.hkeys(RETAINEDKEY)
 }
 
-async function getRetainedValue (db, topic, hasClusters) {
+async function getRetainedValueBuffer (db, topic, hasClusters) {
   if (hasClusters === true) {
-    return msgpack.decode(await db.getBuffer(retainedKey(topic)))
+    return db.getBuffer(retainedKey(topic))
+  }
+  return db.hgetBuffer(RETAINEDKEY, topic)
+}
+
+async function getRetainedValue (db, topic, hasClusters) {
+  return msgpack.decode(await getRetainedValueBuffer(db, topic, hasClusters))
+}
+
+async function * getRetainedValuesStream (db, topics, hasClusters) {
+  for (const topic of topics) {
+    const buffer = await getRetainedValueBuffer(db, topic, hasClusters)
+    if (buffer && buffer.length > 0) {
+      yield msgpack.decode(buffer)
+    }
   }
-  return msgpack.decode(await db.hgetBuffer(RETAINEDKEY, topic))
 }
 
 async function * createWillStream (db, brokers, maxWills) {
@@ -131,6 +144,9 @@ class RedisPersistence extends CachedPersistence {
 
     this.logger = opts.logger || new NoopLogger()
 
+    // By default retained values query optimization is disabled (-1)
+    this.retainedValuesQueryThreshold = opts.retained_values_query_threshold || -1
+
     if (this.sharedCacheRefreshIntervalSec) {
       this.once('ready', () => {
         setInterval(() => {
@@ -184,7 +200,15 @@ class RedisPersistence extends CachedPersistence {
     }
   }
 
+  hasWildcard (patterns) {
+    return patterns.some((pattern) => pattern.includes('+') || pattern.includes('#'))
+  }
+
   createRetainedStreamCombi (patterns) {
+    if (patterns.length <= this.retainedValuesQueryThreshold && !this.hasWildcard(patterns)) {
+      return Readable.from(getRetainedValuesStream(this._db, patterns, this.hasClusters))
+    }
+
     const qlobber = new QlobberTrue(qlobberOpts)
 
     for (const pattern of patterns) {