Removed unused threading code

This changes the RX thread to send the packet into a queue instead of
starting a new thread for every packet.

aprsd/cmds/listen.py

@@ -135,7 +135,7 @@ def listen(
     keepalive.start()
 
     LOG.debug("Create APRSDListenThread")
-    listen_thread = APRSDListenThread(threads.msg_queues, config=config)
+    listen_thread = APRSDListenThread(threads.packet_queue, config=config)
     LOG.debug("Start APRSDListenThread")
     listen_thread.start()
     LOG.debug("keepalive Join")

aprsd/cmds/server.py

@@ -96,10 +96,15 @@ def server(ctx, flush):
     plugin_manager.setup_plugins()
 
     rx_thread = rx.APRSDPluginRXThread(
-        msg_queues=threads.msg_queues,
+        packet_queue=threads.packet_queue,
         config=config,
     )
+    process_thread = rx.APRSDPluginProcessPacketThread(
+        config=config,
+        packet_queue=threads.packet_queue,
+    )
     rx_thread.start()
+    process_thread.start()
 
     packets.PacketTrack().restart()
 

aprsd/cmds/webchat.py

@@ -63,6 +63,9 @@ class SentMessages(objectstore.ObjectStoreMixin):
             cls._instance = super().__new__(cls)
         return cls._instance
 
+    def is_initialized(self):
+        return True
+
     @wrapt.synchronized(lock)
     def add(self, msg):
         self.data[msg.msgNo] = self.create(msg.msgNo)

aprsd/plugins/email.py

@@ -80,7 +80,6 @@ class EmailPlugin(plugin.APRSDRegexCommandPluginBase):
     def create_threads(self):
         if self.enabled:
             return APRSDEmailThread(
-                msg_queues=threads.msg_queues,
                 config=self.config,
             )
 
@@ -502,9 +501,8 @@ def resend_email(config, count, fromcall):
 
 
 class APRSDEmailThread(threads.APRSDThread):
-    def __init__(self, msg_queues, config):
+    def __init__(self, config):
         super().__init__("EmailThread")
-        self.msg_queues = msg_queues
         self.config = config
         self.past = datetime.datetime.now()
 

aprsd/threads/__init__.py

@@ -7,7 +7,4 @@ from .keep_alive import KeepAliveThread  # noqa: F401
 from .rx import APRSDRXThread  # noqa: F401
 
 
-rx_msg_queue = queue.Queue(maxsize=20)
-msg_queues = {
-    "rx": rx_msg_queue,
-}
+packet_queue = queue.Queue(maxsize=20)

aprsd/threads/aprsd.py

@@ -1,6 +1,5 @@
 import abc
 import logging
-from queue import Queue
 import threading
 
 import wrapt
@@ -16,7 +15,6 @@ class APRSDThreadList:
 
     threads_list = []
     lock = threading.Lock()
-    global_queue = Queue()
 
     def __new__(cls, *args, **kwargs):
         if cls._instance is None:
@@ -26,7 +24,6 @@ class APRSDThreadList:
 
     @wrapt.synchronized(lock)
     def add(self, thread_obj):
-        thread_obj.set_global_queue(self.global_queue)
         self.threads_list.append(thread_obj)
 
     @wrapt.synchronized(lock)
@@ -35,7 +32,6 @@ class APRSDThreadList:
 
     @wrapt.synchronized(lock)
     def stop_all(self):
-        self.global_queue.put_nowait({"quit": True})
         """Iterate over all threads and call stop on them."""
         for th in self.threads_list:
             LOG.info(f"Stopping Thread {th.name}")
@@ -50,30 +46,15 @@ class APRSDThreadList:
 
 class APRSDThread(threading.Thread, metaclass=abc.ABCMeta):
 
-    global_queue = None
-
     def __init__(self, name):
         super().__init__(name=name)
         self.thread_stop = False
         APRSDThreadList().add(self)
 
-    def set_global_queue(self, global_queue):
-        self.global_queue = global_queue
-
     def _should_quit(self):
         """ see if we have a quit message from the global queue."""
         if self.thread_stop:
             return True
-        if self.global_queue.empty():
-            return False
-        msg = self.global_queue.get(timeout=1)
-        if not msg:
-            return False
-        if "quit" in msg and msg["quit"] is True:
-            # put the message back on the queue for others
-            self.global_queue.put_nowait(msg)
-            self.thread_stop = True
-            return True
 
     def stop(self):
         self.thread_stop = True

aprsd/threads/rx.py

@@ -1,5 +1,6 @@
 import abc
 import logging
+import queue
 import time
 
 import aprslib
@@ -12,9 +13,9 @@ LOG = logging.getLogger("APRSD")
 
 
 class APRSDRXThread(APRSDThread):
-    def __init__(self, msg_queues, config):
+    def __init__(self, packet_queue, config):
         super().__init__("RX_MSG")
-        self.msg_queues = msg_queues
+        self.packet_queue = packet_queue
         self.config = config
         self._client = client.factory.create()
 
@@ -68,11 +69,7 @@ class APRSDPluginRXThread(APRSDRXThread):
         # LOG.debug(raw)
         packet.log(header="RX")
         packets.PacketList().rx(packet)
-        thread = APRSDPluginProcessPacketThread(
-            config=self.config,
-            packet=packet,
-        )
-        thread.start()
+        self.packet_queue.put(packet)
 
 
 class APRSDProcessPacketThread(APRSDThread):
@@ -83,11 +80,10 @@ class APRSDProcessPacketThread(APRSDThread):
     will ack a message before sending the packet to the subclass
     for processing."""
 
-    def __init__(self, config, packet):
+    def __init__(self, config, packet_queue):
         self.config = config
-        self.packet = packet
-        name = self.packet.raw[:10]
-        super().__init__(f"RXPKT-{name}")
+        self.packet_queue = packet_queue
+        super().__init__("ProcessPKT")
         self._loop_cnt = 1
 
     def process_ack_packet(self, packet):
@@ -99,9 +95,18 @@ class APRSDProcessPacketThread(APRSDThread):
         return
 
     def loop(self):
+        try:
+            packet = self.packet_queue.get(block=True, timeout=1)
+            if packet:
+                self.process_packet(packet)
+        except queue.Empty:
+            pass
+        self._loop_cnt += 1
+        return True
+
+    def process_packet(self, packet):
         """Process a packet received from aprs-is server."""
         LOG.debug(f"RXPKT-LOOP {self._loop_cnt}")
-        packet = self.packet
         our_call = self.config["aprsd"]["callsign"].lower()
 
         from_call = packet.from_call
@@ -147,7 +152,7 @@ class APRSDProcessPacketThread(APRSDThread):
         return False
 
     @abc.abstractmethod
-    def process_our_message_packet(self, *args, **kwargs):
+    def process_our_message_packet(self, packet):
         """Process a MessagePacket destined for us!"""
 
     def process_other_packet(self, packet, for_us=False):
@@ -210,9 +215,7 @@ class APRSDPluginProcessPacketThread(APRSDProcessPacketThread):
                             to_call=from_call,
                             message_text=reply,
                         )
-                        LOG.warning("Calling msg_pkg.send()")
                         msg_pkt.send()
-                        LOG.warning("Calling msg_pkg.send() --- DONE")
 
             # If the message was for us and we didn't have a
             # response, then we send a usage statement.

examples/plugins/example_plugin.py

@@ -6,7 +6,7 @@ from aprsd import plugin
 LOG = logging.getLogger("APRSD")
 
 
-class HelloPlugin(plugin.APRSDPluginBase):
+class HelloPlugin(plugin.APRSDRegexCommandPluginBase):
     """Hello World."""
 
     version = "1.0"
@@ -14,7 +14,7 @@ class HelloPlugin(plugin.APRSDPluginBase):
     command_regex = "^[hH]"
     command_name = "hello"
 
-    def command(self, fromcall, message, ack):
+    def command(self, packet):
         LOG.info("HelloPlugin")
-        reply = f"Hello '{fromcall}'"
+        reply = f"Hello '{packet.from_call}'"
         return reply

tests/cmds/test_webchat.py

@@ -90,7 +90,7 @@ class TestSendMessageCommand(unittest.TestCase):
         mock_emit.called_once()
 
     @mock.patch("aprsd.config.parse_config")
-    @mock.patch("aprsd.packets.PacketList.add")
+    @mock.patch("aprsd.packets.PacketList.rx")
     @mock.patch("aprsd.cmds.webchat.socketio.emit")
     def test_process_our_message_packet(
         self, mock_parse_config,