Add message sending logic

README.md

@@ -8,9 +8,15 @@ Django >= 1.11, Celery >= 4.1.1
 
 ### setup
 
-- add `git+https://github.com/jar3b/django-queued-mailer.git` to requirements.txt
+- Add `git+https://github.com/jar3b/django-queued-mailer.git` to requirements.txt
-- install package `pip install --upgrade git+https://github.com/jar3b/django-queued-mailer.git`
+- Install package `pip install --upgrade git+https://github.com/jar3b/django-queued-mailer.git`
-- run celery worker
+- Modify Django settings: set proper email backend
+
+```
+EMAIL_BACKEND = 'queued_mailer.backend.EmailBackend'
+```
+ 
+- Run celery worker
 
 ```
 celery -A project worker

queued_mailer/backend.py

@@ -1,10 +1,10 @@
 # coding: utf-8
-
 from django.core.mail.backends.base import BaseEmailBackend
 from django.utils.encoding import force_text
 
 from .logger import logger
 from .tasks import send_message
+from .utils import get_email_connection
 
 
 def _get_message_recipients(email_message):
@@ -20,9 +20,11 @@ class EmailBackend(BaseEmailBackend):
 
     def send_messages(self, email_messages):
         num_sent = 0
+        connection = get_email_connection()
+
         for email in email_messages:
             try:
-                send_message.apply_async([email, ])
+                send_message.apply_async([email, connection])
                 num_sent += 1
             except Exception as e:
                 logger.error("cannot send message %s: %r" % (_get_message_recipients(email), e))

queued_mailer/exceptions.py

@@ -0,0 +1,2 @@
+class EmailTransportException(Exception):
+    pass

queued_mailer/tasks.py

@@ -1,21 +1,44 @@
 # coding: utf-8
-
+import smtplib
-from time import sleep
+import time
+from socket import error as socket_error
 
 import celery
 from celery.utils.log import get_task_logger
+from django.core.mail import EmailMessage
 
+from queued_mailer.utils import get_email_connection
+from .exceptions import EmailTransportException
 from .settings import TASK_QUEUE_NAME
 
 logger = get_task_logger(__name__)
 
 
-@celery.task(bind=True, queue=TASK_QUEUE_NAME, acks_late=True, default_retry_delay=30)
+@celery.task(bind=True, queue=TASK_QUEUE_NAME, acks_late=True,
-def send_message(self, email):
+             default_retry_delay=20, autoretry_for=(EmailTransportException,), retry_kwargs={'max_retries': 5})
-    logger.warning('task started')
+def send_message(self, email, connection=None):
+    logger.debug('task started')
+
+    start_time = time.time()
+
+    if not isinstance(email, EmailMessage):
+        raise Exception('Invalid message class, only django.core.mail.EmailMessage is supported')
+
     try:
-        sleep(5)
+        if connection is None:
-        logger.warning('task success')
+            connection = get_email_connection()
+
+        email.connection = connection
+        email.send()
+    except (socket_error, smtplib.SMTPSenderRefused,
+            smtplib.SMTPRecipientsRefused,
+            smtplib.SMTPDataError,
+            smtplib.SMTPAuthenticationError) as e:
+        logger.warning('message transport failure: %r' % e)
+        raise EmailTransportException(e)
     except Exception as e:
-        logger.warning('task error')
+        logger.error('message send error: %r' % e)
-        raise self.retry(exc=e)
+        raise
+
+    elapsed_time = time.time() - start_time
+    logger.warning("message sent, elasped time %s" % elapsed_time)

queued_mailer/utils.py

@@ -0,0 +1,7 @@
+from django.core.mail import get_connection
+
+from .settings import EMAIL_BACKEND
+
+
+def get_email_connection():
+    return get_connection(backend=EMAIL_BACKEND)