Converted MsgTrack to ObjectStoreMixin

aprsd/main.py

@@ -146,8 +146,7 @@ def signal_handler(sig, frame):
             ),
         )
         time.sleep(1.5)
-        tracker = messaging.MsgTrack()
-        tracker.save()
+        messaging.MsgTrack().save()
         packets.WatchList().save()
         packets.SeenList().save()
         LOG.info(stats.APRSDStats())
@@ -480,13 +479,13 @@ def server(
     packets.PacketList(config=config)
     if flush:
         LOG.debug("Deleting saved MsgTrack.")
-        messaging.MsgTrack().flush()
+        messaging.MsgTrack(config=config).flush()
         packets.WatchList(config=config)
         packets.SeenList(config=config)
     else:
         # Try and load saved MsgTrack list
         LOG.debug("Loading saved MsgTrack object.")
-        messaging.MsgTrack().load()
+        messaging.MsgTrack(config=config).load()
         packets.WatchList(config=config).load()
         packets.SeenList(config=config).load()
 

aprsd/messaging.py

@@ -2,16 +2,11 @@ import abc
 import datetime
 import logging
 from multiprocessing import RawValue
-import os
-import pathlib
-import pickle
 import re
 import threading
 import time
 
-from aprsd import client
-from aprsd import config as aprsd_config
-from aprsd import packets, stats, threads
+from aprsd import client, objectstore, packets, stats, threads
 
 
 LOG = logging.getLogger("APRSD")
@@ -21,7 +16,7 @@ LOG = logging.getLogger("APRSD")
 NULL_MESSAGE = -1
 
 
-class MsgTrack:
+class MsgTrack(objectstore.ObjectStoreMixin):
     """Class to keep track of outstanding text messages.
 
     This is a thread safe class that keeps track of active
@@ -38,7 +33,7 @@ class MsgTrack:
     _start_time = None
     lock = None
 
-    track = {}
+    data = {}
     total_messages_tracked = 0
 
     def __new__(cls, *args, **kwargs):
@@ -47,93 +42,65 @@ class MsgTrack:
             cls._instance.track = {}
             cls._instance._start_time = datetime.datetime.now()
             cls._instance.lock = threading.Lock()
+            cls._instance.config = kwargs["config"]
+            cls._instance._init_store()
         return cls._instance
 
     def __getitem__(self, name):
         with self.lock:
-            return self.track[name]
+            return self.data[name]
 
     def __iter__(self):
         with self.lock:
-            return iter(self.track)
+            return iter(self.data)
 
     def keys(self):
         with self.lock:
-            return self.track.keys()
+            return self.data.keys()
 
     def items(self):
         with self.lock:
-            return self.track.items()
+            return self.data.items()
 
     def values(self):
         with self.lock:
-            return self.track.values()
+            return self.data.values()
 
     def __len__(self):
         with self.lock:
-            return len(self.track)
+            return len(self.data)
 
     def __str__(self):
         with self.lock:
             result = "{"
-            for key in self.track.keys():
-                result += f"{key}: {str(self.track[key])}, "
+            for key in self.data.keys():
+                result += f"{key}: {str(self.data[key])}, "
             result += "}"
             return result
 
     def add(self, msg):
         with self.lock:
             key = int(msg.id)
-            self.track[key] = msg
+            self.data[key] = msg
             stats.APRSDStats().msgs_tracked_inc()
             self.total_messages_tracked += 1
 
     def get(self, id):
         with self.lock:
-            if id in self.track:
-                return self.track[id]
+            if id in self.data:
+                return self.data[id]
 
     def remove(self, id):
         with self.lock:
             key = int(id)
-            if key in self.track.keys():
-                del self.track[key]
-
-    def save(self):
-        """Save any queued to disk?"""
-        LOG.debug(f"Save tracker to disk? {len(self)}")
-        if len(self) > 0:
-            LOG.info(f"Saving {len(self)} tracking messages to disk")
-            pickle.dump(self.dump(), open(aprsd_config.DEFAULT_SAVE_FILE, "wb+"))
-        else:
-            LOG.debug(
-                "Nothing to save, flushing old save file '{}'".format(
-                    aprsd_config.DEFAULT_SAVE_FILE,
-                ),
-            )
-            self.flush()
-
-    def dump(self):
-        dump = {}
-        with self.lock:
-            for key in self.track.keys():
-                dump[key] = self.track[key]
-
-        return dump
-
-    def load(self):
-        if os.path.exists(aprsd_config.DEFAULT_SAVE_FILE):
-            raw = pickle.load(open(aprsd_config.DEFAULT_SAVE_FILE, "rb"))
-            if raw:
-                self.track = raw
-                LOG.debug("Loaded MsgTrack dict from disk.")
-                LOG.debug(self)
+            if key in self.data.keys():
+                del self.data[key]
 
     def restart(self):
         """Walk the list of messages and restart them if any."""
 
-        for key in self.track.keys():
-            msg = self.track[key]
+        for key in self.data.keys():
+            msg = self.data[key]
             if msg.last_send_attempt < msg.retry_count:
                 msg.send()
 
@@ -145,14 +112,14 @@ class MsgTrack:
         """Walk the list of delayed messages and restart them if any."""
         if not count:
             # Send all the delayed messages
-            for key in self.track.keys():
-                msg = self.track[key]
+            for key in self.data.keys():
+                msg = self.data[key]
                 if msg.last_send_attempt == msg.retry_count:
                     self._resend(msg)
         else:
             # They want to resend <count> delayed messages
             tmp = sorted(
-                self.track.items(),
+                self.data.items(),
                 reverse=most_recent,
                 key=lambda x: x[1].last_send_time,
             )
@@ -160,13 +127,6 @@ class MsgTrack:
             for (_key, msg) in msg_list:
                 self._resend(msg)
 
-    def flush(self):
-        """Nuke the old pickle file that stored the old results from last aprsd run."""
-        if os.path.exists(aprsd_config.DEFAULT_SAVE_FILE):
-            pathlib.Path(aprsd_config.DEFAULT_SAVE_FILE).unlink()
-        with self.lock:
-            self.track = {}
-
 
 class MessageCounter:
     """

aprsd/objectstore.py

@@ -53,7 +53,6 @@ class ObjectStoreMixin:
     def _save_filename(self):
         save_location = self._save_location()
 
-        LOG.debug(f"{self.__class__.__name__}::Using save location {save_location}")
         return "{}/{}.p".format(
             save_location,
             self.__class__.__name__.lower(),
@@ -65,15 +64,12 @@ class ObjectStoreMixin:
             for key in self.data.keys():
                 dump[key] = self.data[key]
 
-        LOG.debug(f"{self.__class__.__name__}:: DUMP")
-        LOG.debug(dump)
-
         return dump
 
     def save(self):
         """Save any queued to disk?"""
         if len(self) > 0:
-            LOG.info(f"{self.__class__.__name__}::Saving {len(self)} entries to disk")
+            LOG.info(f"{self.__class__.__name__}::Saving {len(self)} entries to disk at {self._save_location()}")
             pickle.dump(self._dump(), open(self._save_filename(), "wb+"))
         else:
             LOG.debug(

tests/test_messaging.py

@@ -6,9 +6,14 @@ from aprsd import messaging
 
 
 class TestMessageTrack(unittest.TestCase):
+
+    def setUp(self) -> None:
+        config = {}
+        messaging.MsgTrack(config=config)
+
     def _clean_track(self):
         track = messaging.MsgTrack()
-        track.track = {}
+        track.data = {}
         track.total_messages_tracked = 0
         return track
 

tests/test_plugin.py

@@ -26,6 +26,7 @@ class TestPlugin(unittest.TestCase):
         stats.APRSDStats(self.config)
         packets.WatchList(config=self.config)
         packets.SeenList(config=self.config)
+        messaging.MsgTrack(config=self.config)
 
     @mock.patch.object(fake.FakeBaseNoThreadsPlugin, "process")
     def test_base_plugin_no_threads(self, mock_process):
@@ -161,7 +162,7 @@ class TestQueryPlugin(TestPlugin):
     @mock.patch("aprsd.messaging.MsgTrack.restart_delayed")
     def test_query_restart_delayed(self, mock_restart):
         track = messaging.MsgTrack()
-        track.track = {}
+        track.data = {}
         packet = fake.fake_packet(message="!4")
         query = query_plugin.QueryPlugin(self.config)