Merge pull request #43 from kube-HPC/streaming_memory_leak

change needed since working with flows. Memory leak fix
kube-HPC · Dec 29, 2020 · dd1b0d3 · dd1b0d3
2 parents 333c484 + 117e633
commit dd1b0d3
Showing 1 changed file with 11 additions and 12 deletions.
diff --git a/hkube_python_wrapper/communication/zmq/streaming/producer/MessageQueue.py b/hkube_python_wrapper/communication/zmq/streaming/producer/MessageQueue.py
@@ -46,19 +46,18 @@ def pop(self, consumerType):
             index = nextItemIndex + 1
             self.indexPerConsumer[consumerType] = index
             self.sent[consumerType] += 1
-            if (index == 1):
-                anyZero = False
-                for value in self.indexPerConsumer.values():
-                    if (value == 0):
-                        anyZero = True
-                        break
+            anyZero = False
+            for value in self.indexPerConsumer.values():
+                if (value == 0):
+                    anyZero = True
+                    break
 
-                if not (anyZero):
-                    self.queue.pop(0)
-                    _, _, msg = out
-                    self.sizeSum -= len(msg)
-                    for key in self.indexPerConsumer.keys():
-                        self.indexPerConsumer[key] = self.indexPerConsumer[key] - 1
+            if not (anyZero):
+                self.queue.pop(0)
+                _, _, msg = out
+                self.sizeSum -= len(msg)
+                for key in self.indexPerConsumer.keys():
+                    self.indexPerConsumer[key] = self.indexPerConsumer[key] - 1
             return out
         return None