Merge pull request #2149 from certtools/enh-1827

FIX: Use redis BLMOVE instead of BRPOPLPUSH (deprecated)

CHANGELOG.md

@@ -33,6 +33,8 @@ CHANGELOG
   - Enhance behaviour if an unconfigured bot is started (PR#2054 by Sebastian Wagner).
   - Fix line recovery and message dumping of the `ParserBot` (PR#2192 by Sebastian Wagner).
     - Previously the dumped message was always the last message of a report if the report contained muliple lines leading to data-loss.
+- `intelmq.lib.pipeline`:
+  - Changed `BRPOPLPUSH` to `BLMOVE`, because `BRPOPLPUSH` has been marked as deprecated by redis in favor of `BLMOVE` (PR#2149 by Sebastian Waldbauer, fixes #1827)
 
 ### Development
 

intelmq/lib/pipeline.py

@@ -182,6 +182,7 @@ class Redis(Pipeline):
     destination_pipeline_db = 2
     source_pipeline_password = None
     destination_pipeline_password = None
+    _redis_server_version = None
 
     def load_configurations(self, queues_type):
         self.host = self.pipeline_args.get(f"{queues_type}_pipeline_host", "127.0.0.1")
@@ -213,6 +214,7 @@ def connect(self):
             kwargs['single_connection_client'] = True
 
         self.pipe = redis.Redis(db=self.db, password=self.password, **kwargs)
+        self._redis_server_version = tuple(int(x) for x in self.pipe.execute_command('INFO')['redis_version'].split('.'))
 
     def disconnect(self):
         pass
@@ -261,8 +263,10 @@ def _receive(self) -> bytes:
                 else:
                     break
             if not retval:
-                retval = self.pipe.brpoplpush(self.source_queue,
-                                              self.internal_queue, 0)
+                if self._redis_server_version >= (6, 2, 0):
+                    retval = self.pipe.blmove(self.source_queue, self.internal_queue, 0, 'RIGHT', 'LEFT')
+                else:
+                    retval = self.pipe.brpoplpush(self.source_queue, self.internal_queue, 0)
         except Exception as exc:
             raise exceptions.PipelineError(exc)
         else: