Closes #167: Allow to use RedisQueue for email sending

install/requirements.txt

@@ -49,7 +49,6 @@ stevedore==1.1.0
 tg.devtools==2.3.7
 git+https://github.com/algoo/tgapp-resetpassword.git
 tgext.admin==0.6.4
-tgext.asyncjob==0.3.1
 tgext.crud==0.7.3
 tgext.pluggable==0.6.2
 transaction==1.4.4
@@ -65,3 +64,5 @@ zope.sqlalchemy==0.7.6
 PyYAML
 redis==2.10.5
 typing==3.5.3.0
+rq==0.7.1
+click==6.7

tracim/development.ini.base

@@ -192,8 +192,13 @@ email.notification.smtp.port = 25
 email.notification.smtp.user = your_smtp_user
 email.notification.smtp.password = your_smtp_password
 
-# Asyncjob tracker, can be memory or redis default is memory
-asyncjob.tracker = memory
+## Email sending configuration
+# processing_mode may be sync or async,
+# with async, please configure redis below
+email.processing_mode = sync
+# email.async.redis.host = localhost
+# email.async.redis.port = 6379
+# email.async.redis.db = 0
 
 ## Radical (CalDav server) configuration
 # radicale.server.host = 0.0.0.0

tracim/setup.py

@@ -43,6 +43,7 @@ install_requires=[
     "python-ldap-test==0.2.1",
     "unicode-slugify==0.1.3",
     "pytz==2014.7",
+    'rq==0.7.1',
     ]
 
 setup(
@@ -76,6 +77,7 @@ setup(
             'ldap_server = tracim.command.ldap_test_server:LDAPTestServerCommand',
             'user_create = tracim.command.user:CreateUserCommand',
             'user_update = tracim.command.user:UpdateUserCommand',
+            'mail sender = tracim.command.mail:MailSenderCommend',
         ]
     },
     dependency_links=[

tracim/tracim/command/mail.py

@@ -0,0 +1,16 @@
+# -*- coding: utf-8 -*-
+from rq import Connection, Worker
+
+from tracim.command import AppContextCommand
+
+
+class MailSenderCommend(AppContextCommand):
+    def get_description(self):
+        return '''Run rq worker for mail sending'''
+
+    def take_action(self, parsed_args):
+        super().take_action(parsed_args)
+
+        with Connection():
+            w = Worker(['mail_sender'])
+            w.work()

tracim/tracim/config/app_cfg.py

@@ -19,8 +19,6 @@ from urllib.parse import urlparse
 import tg
 from paste.deploy.converters import asbool
 from tg.configuration.milestones import environment_loaded
-from tgext.asyncjob.trackers.memory import MemoryProgressTracker
-from tgext.asyncjob.trackers.redisdb import RedisProgressTracker
 
 from tgext.pluggable import plug
 from tgext.pluggable import replace_template
@@ -30,10 +28,9 @@ from tracim.lib.utils import lazy_ugettext as l_
 import tracim
 from tracim import model
 from tracim.config import TracimAppConfig
-from tracim.lib import app_globals, helpers
-from tracim.lib.auth.wrapper import AuthConfigWrapper
 from tracim.lib.base import logger
 from tracim.lib.daemons import DaemonsManager
+from tracim.lib.daemons import MailSenderDaemon
 from tracim.lib.daemons import RadicaleDaemon
 from tracim.lib.daemons import WsgiDavDaemon
 from tracim.model.data import ActionDescription
@@ -106,6 +103,7 @@ def start_daemons(manager: DaemonsManager):
 
     manager.run('radicale', RadicaleDaemon)
     manager.run('webdav', WsgiDavDaemon)
+    manager.run('mail_sender', MailSenderDaemon)
 
 environment_loaded.register(lambda: start_daemons(daemons))
 
@@ -341,32 +339,36 @@ class CFG(object):
         if not self.WSGIDAV_CLIENT_BASE_URL.endswith('/'):
             self.WSGIDAV_CLIENT_BASE_URL += '/'
 
-        self.ASYNC_JOB_TRACKER = tg.config.get(
-            'asyncjob.tracker',
-            'memory',
-        )
+        self.EMAIL_PROCESSING_MODE = tg.config.get(
+            'email.processing_mode',
+            'sync',
+        ).upper()
 
-        if self.ASYNC_JOB_TRACKER not in ('memory', 'redis'):
+        if self.EMAIL_PROCESSING_MODE not in (
+                self.CST.ASYNC,
+                self.CST.SYNC,
+        ):
             raise Exception(
-                'asyncjob.tracker configuration '
-                'can ''be "memory" or "redis", not "{0}"'.format(
-                    self.ASYNC_JOB_TRACKER,
+                'email.processing_mode '
+                'can ''be "{}" or "{}", not "{}"'.format(
+                    self.CST.ASYNC,
+                    self.CST.SYNC,
+                    self.EMAIL_PROCESSING_MODE,
                 )
             )
 
-        if self.ASYNC_JOB_TRACKER == 'redis':
-            self.ASYNC_JOB_TRACKER_REDIS_HOST = tg.config.get(
-                'asyncjob.tracker.redis.host',
-                'localhost',
-            )
-            self.ASYNC_JOB_TRACKER_REDIS_PORT = int(tg.config.get(
-                'asyncjob.tracker.redis.port',
-                6379,
-            ))
-            self.ASYNC_JOB_TRACKER_REDIS_DB = int(tg.config.get(
-                'asyncjob.tracker.redis.db',
-                15,
-            ))
+        self.EMAIL_SENDER_REDIS_HOST = tg.config.get(
+            'email.async.redis.host',
+            'localhost',
+        )
+        self.EMAIL_SENDER_REDIS_PORT = int(tg.config.get(
+            'email.async.redis.port',
+            6379,
+        ))
+        self.EMAIL_SENDER_REDIS_DB = int(tg.config.get(
+            'email.async.redis.db',
+            0,
+        ))
 
     def get_tracker_js_content(self, js_tracker_file_path = None):
         js_tracker_file_path = tg.config.get('js_tracker_path', None)
@@ -408,30 +410,3 @@ class CFG(object):
 base_config.variable_provider = lambda: {
     'CFG': CFG.get_instance()
 }
-
-
-def plug_asyncjob():
-    cfg = CFG.get_instance()
-
-    # # Manual creation of async job tracker to be able to log it
-    async_job_tracker = cfg.ASYNC_JOB_TRACKER
-    if async_job_tracker == 'redis':
-        async_job_progress_tracker = RedisProgressTracker(
-            host=cfg.ASYNC_JOB_TRACKER_REDIS_HOST,
-            port=cfg.ASYNC_JOB_TRACKER_REDIS_PORT,
-            db=cfg.ASYNC_JOB_TRACKER_REDIS_DB,
-        )
-    else:
-        async_job_progress_tracker = MemoryProgressTracker()
-
-    logger.info(
-        cfg,
-        'Async job track using {0}'.format(str(async_job_progress_tracker)),
-    )
-    plug(
-        base_config,
-        'tgext.asyncjob',
-        progress_tracker=async_job_progress_tracker,
-    )
-
-environment_loaded.register(lambda: plug_asyncjob())

tracim/tracim/lib/daemons.py

@@ -1,20 +1,19 @@
+import logging
 import threading
 from configparser import DuplicateSectionError
-from datetime import datetime
 from wsgiref.simple_server import make_server
 import signal
 
 import collections
-import time
-
-import io
-import yaml
 
 from radicale import Application as RadicaleApplication
 from radicale import HTTPServer as BaseRadicaleHTTPServer
 from radicale import HTTPSServer as BaseRadicaleHTTPSServer
 from radicale import RequestHandler as RadicaleRequestHandler
 from radicale import config as radicale_config
+from rq import Connection as RQConnection
+from rq import Worker as BaseRQWorker
+from redis import Redis
 
 from tracim.lib.base import logger
 from tracim.lib.exceptions import AlreadyRunningDaemon
@@ -148,6 +147,43 @@ class Daemon(threading.Thread):
         raise NotImplementedError()
 
 
+class MailSenderDaemon(Daemon):
+    # NOTE: use *args and **kwargs because parent __init__ use strange
+    # * parameter
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.worker = None  # type: RQWorker
+
+    def append_thread_callback(self, callback: collections.Callable) -> None:
+        logger.warning('MailSenderDaemon not implement append_thread_callback')
+        pass
+
+    def stop(self) -> None:
+        self.worker.request_stop('TRACIM STOP', None)
+
+    def run(self) -> None:
+        from tracim.config.app_cfg import CFG
+        cfg = CFG.get_instance()
+
+        with RQConnection(Redis(
+            host=cfg.EMAIL_SENDER_REDIS_HOST,
+            port=cfg.EMAIL_SENDER_REDIS_PORT,
+            db=cfg.EMAIL_SENDER_REDIS_DB,
+        )):
+            self.worker = RQWorker(['mail_sender'])
+            self.worker.work()
+
+
+class RQWorker(BaseRQWorker):
+    def _install_signal_handlers(self):
+        # TODO BS 20170126: RQ WWorker is designed to work in main thread
+        # So we have to disable these signals (we implement server stop in
+        # MailSenderDaemon.stop method). When bug
+        # https://github.com/tracim/tracim/issues/166 will be fixed, ensure
+        # This worker terminate correctly.
+        pass
+
+
 class RadicaleHTTPSServer(TracimSocketServerMixin, BaseRadicaleHTTPSServer):
     pass
 

tracim/tracim/lib/email.py

@@ -1,17 +1,50 @@
 # -*- coding: utf-8 -*-
-
-from email.mime.multipart import MIMEMultipart
 import smtplib
+from email.message import Message
+from email.mime.multipart import MIMEMultipart
 from email.mime.text import MIMEText
 
+import typing
 from mako.template import Template
-from tgext.asyncjob import asyncjob_perform
+from redis import Redis
+from rq import Queue
 from tg.i18n import ugettext as _
 
 from tracim.lib.base import logger
 from tracim.model import User
 
 
+def send_email_through(
+        send_callable: typing.Callable[[Message], None],
+        message: Message,
+) -> None:
+    """
+    Send mail encapsulation to send it in async or sync mode.
+    TODO BS 20170126: A global mail/sender management should be a good
+                      thing. Actually, this method is an fast solution.
+    :param send_callable: A callable who get message on first parameter
+    :param message: The message who have to be sent
+    """
+    from tracim.config.app_cfg import CFG
+    cfg = CFG.get_instance()
+
+    if cfg.EMAIL_PROCESSING_MODE == CFG.CST.SYNC:
+        send_callable(message)
+    elif cfg.EMAIL_PROCESSING_MODE == CFG.CST.ASYNC:
+        queue = Queue('mail_sender', connection=Redis(
+            host=cfg.EMAIL_SENDER_REDIS_HOST,
+            port=cfg.EMAIL_SENDER_REDIS_PORT,
+            db=cfg.EMAIL_SENDER_REDIS_DB,
+        ))
+        queue.enqueue(send_callable, message)
+    else:
+        raise NotImplementedError(
+            'Mail sender processing mode {} is not implemented'.format(
+                cfg.EMAIL_PROCESSING_MODE,
+            )
+        )
+
+
 class SmtpConfiguration(object):
     """
     Container class for SMTP configuration used in Tracim
@@ -139,11 +172,7 @@ class EmailManager(object):
         message.attach(part1)
         message.attach(part2)
 
-        asyncjob_perform(async_email_sender.send_mail, message)
-
-        # Note: The following action allow to close the SMTP connection.
-        # This will work only if the async jobs are done in the right order
-        asyncjob_perform(async_email_sender.disconnect)
+        send_email_through(async_email_sender.send_mail, message)
 
     def _render(self, mako_template_filepath: str, context: dict):
         """

tracim/tracim/lib/notifications.py

@@ -10,6 +10,7 @@ from mako.template import Template
 
 from tracim.lib.base import logger
 from tracim.lib.email import SmtpConfiguration
+from tracim.lib.email import send_email_through
 from tracim.lib.email import EmailSender
 from tracim.lib.user import UserApi
 from tracim.lib.workspace import WorkspaceApi
@@ -23,8 +24,6 @@ from tracim.model.data import Content, UserRoleInWorkspace, ContentType, \
 from tracim.model.auth import User
 
 
-from tgext.asyncjob import asyncjob_perform
-
 class INotifier(object):
     """
     Interface for Notifier instances
@@ -142,7 +141,6 @@ class RealNotifier(object):
                 # TODO - D.A - 2014-11-06
                 # This feature must be implemented in order to be able to scale to large communities
                 raise NotImplementedError('Sending emails through ASYNC mode is not working yet')
-                asyncjob_perform(EmailNotifier(self._smtp_config, global_config).notify_content_update, self._user.user_id, content.content_id)
             else:
                 logger.info(self, 'Sending email in SYNC mode')
                 EmailNotifier(self._smtp_config, global_config).notify_content_update(self._user.user_id, content.content_id)
@@ -278,19 +276,7 @@ class EmailNotifier(object):
             message.attach(part1)
             message.attach(part2)
 
-            message_str = message.as_string()
-            # asyncjob_perform(async_email_sender.send_mail, message)
-            # FIXME: Temporary hack to enable email sending in
-            # uwsgi/prod environment
-            async_email_sender.send_mail(message)
-            # s.send_message(message)
-
-        # Note: The following action allow to close the SMTP connection.
-        # This will work only if the async jobs are done in the right order
-        # FIXME: Temporary hack to enable email sending in
-        # uwsgi/prod environment
-        # asyncjob_perform(async_email_sender.disconnect)
-
+            send_email_through(async_email_sender.send_mail, message)
 
     def _build_email_body(self, mako_template_filepath: str, role: UserRoleInWorkspace, content: Content, actor: User) -> str:
         """
@@ -360,7 +346,7 @@ class EmailNotifier(object):
                 title_diff = ''
                 if previous_revision.label != content.label:
                     title_diff = htmldiff(previous_revision.label, content.label)
-                content_text = l_('<p id="content-body-intro">Here is an overview of the changes:</p>')+ \
+                content_text = str(l_('<p id="content-body-intro">Here is an overview of the changes:</p>'))+ \
                     title_diff + \
                     htmldiff(previous_revision.description, content.description)
 
@@ -370,7 +356,7 @@ class EmailNotifier(object):
                 title_diff = ''
                 if previous_revision.label != content.label:
                     title_diff = htmldiff(previous_revision.label, content.label)
-                content_text = l_('<p id="content-body-intro">Here is an overview of the changes:</p>')+ \
+                content_text = str(l_('<p id="content-body-intro">Here is an overview of the changes:</p>'))+ \
                     title_diff + \
                     htmldiff(previous_revision.description, content.description)