wip queue

Author: moggers87

salmon/queue.py

@@ -188,3 +188,11 @@ def oversize(self, key):
             return os.path.getsize(file_name) > self.pop_limit, file_name
         else:
             return False, None
+
+
+class QueueWithMetadata(Queue):
+    """Just like Queue, except it stores envolope data"""
+    def push(self, message, metadata):
+    def pop(self):
+        key, message = super().pop()
+        return key, message, metadata

salmon/server.py

@@ -292,40 +292,37 @@ def process_message(self, Peer, From, To, Data, **kwargs):
         return _deliver(self, Peer, From, To, Data, **kwargs)
 
 
-class SMTPOnlyOneRcpt(SMTP):
-    async def smtp_RCPT(self, arg):
-        if self.envelope.rcpt_tos:
-            await self.push(SMTP_MULTIPLE_RCPTS_ERROR)
-        else:
-            await super().smtp_RCPT(arg)
-
-
 class SMTPHandler:
     def __init__(self, executor=None):
         self.executor = executor
 
     async def handle_DATA(self, server, session, envelope):
-        status = await server.loop.run_in_executor(self.executor, partial(
-            _deliver,
-            self,
-            session.peer,
-            envelope.mail_from,
-            envelope.rcpt_tos[0],
-            envelope.content,
-        ))
-        return status or "250 Ok"
+        try:
+            await server.loop.run_in_executor(self.executor, partial())
+            server.push_to_queue(session, envelope)
+            status = "250 Ok"
+        except Exception:
+            # TODO log exception
+            status = "550 Server error"
+        return status
 
 
 class AsyncSMTPReceiver(Controller):
     """Receives emails and hands it to the Router for further processing."""
-    def __init__(self, handler=None, **kwargs):
+    def __init__(self, handler=None, queue=None, **kwargs):
         if handler is None:
             handler = SMTPHandler()
+        if queue is None:
+            queue = QueueReceiver(IN_QUEUE)
+        self.queue = queue
         super().__init__(handler, **kwargs)
 
     def factory(self):
-        # TODO implement a queue
-        return SMTPOnlyOneRcpt(self.handler, enable_SMTPUTF8=self.enable_SMTPUTF8, ident=ROUTER_VERSION_STRING)
+        return SMTP(self.handler, enable_SMTPUTF8=self.enable_SMTPUTF8, ident=ROUTER_VERSION_STRING)
+
+    def stop(self):
+        super().stop()
+        self.queue.stop()
 
 
 class LMTPHandler:
@@ -430,7 +427,9 @@ def start(self, one_shot=False):
 
             logging.debug("Pulled message with key: %r off", key)
             self.workers.apply_async(self.process_message, args=(msg,))
+        self.stop()
 
+    def stop(self):
         self.workers.close()
         self.workers.join()