adding an expire method to the buffer

Signed-off-by: Jesse Jaggars <jjaggars@redhat.com>

Author: jhjaggars

receptor/buffers/file.py

@@ -2,6 +2,7 @@
 import json
 import logging
 import os
+import time
 import uuid
 from collections import deque
 from concurrent.futures import ThreadPoolExecutor
@@ -61,27 +62,48 @@ def _read_manifest(self):
         except FileNotFoundError:
             return []
 
-    async def _get_file(self, path, handle_only=False, delete=True):
+    def _path_for_ident(self, ident):
+        return os.path.join(self._message_path, ident)
+
+    async def _get_file(self, ident, handle_only=False, delete=True):
         """
         Retrieves a file from disk. If handle_only is True then we will
         return the handle to the file and do nothing else. Otherwise the file
         is read into memory all at once and returned. If delete is True (the
         default) and handle_only is False (the default) then the underlying
         file will be removed as well.
         """
-        path = os.path.join(self._message_path, path)
+        path = self._path_for_ident(ident)
         fp = await self._loop.run_in_executor(pool, open, path, "rb")
         if handle_only:
             return fp
         bytes = await self._loop.run_in_executor(pool, lambda: fp.read())
         fp.close()
         if delete:
-            os.remove(path)
+            await self._loop.run_in_executor(pool, os.remove, path)
         return bytes
 
     def _write_file(self, data, ident):
         with open(os.path.join(self._message_path, ident), "wb") as fp:
             fp.write(data)
+
+    async def expire(self):
+        async with self._manifest_lock:
+            new_queue = asyncio.Queue()
+            while self.q.qsize() > 0:
+                ident = await self.q.get()
+                data = await self._get_file(ident, handle_only=True, delete=False)
+                msg = json.load(data)
+                if "expire_time" in msg and msg['expire_time'] < time.time():
+                    logger.info("Expiring message %s", ident)
+                    # TODO: Do something with expired message
+                    await self._loop.run_in_executor(pool, os.remove, self._path_for_ident(ident))
+                else:
+                    await new_queue.put(ident)
+            self.q = new_queue
+
+
+
 
 
 class FileBufferManager(BaseBufferManager):

receptor/receptor.py

@@ -47,12 +47,7 @@ async def watch_expire(self):
             current_manifest = self.get_connection_manifest()
             for connection in current_manifest:
                 buffer = self.buffer_mgr.get_buffer_for_node(connection["id"], self)
-                for ident, message in buffer:
-                    message_actual = json.loads(message)
-                    if "expire_time" in message_actual and message_actual['expire_time'] < time.time():
-                        buffer.read_message(ident, remove=True)
-                        logger.info("Expiring message {}:{}".format(ident, connection["id"]))
-                        # TODO: Do something with expired message
+                await buffer.expire()
                 if connection["last"] + 86400 < time.time():
                     logger.info("Expiring connection {}".format(connection["id"]))
                     write_manifest = copy.copy(current_manifest)