events: Move subscriber classes from svn_support to rhodecode.subscribers.

These classes are generic and can be used by other components as well.
Thereofore they should not live inside of the svn_support package.

rhodecode/subscribers.py

@@ -1,70 +1,143 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
+import logging
 import pylons
+import Queue
+import subprocess32
 
 from pyramid.i18n import get_localizer
 from pyramid.threadlocal import get_current_request
+from threading import Thread
 
 from rhodecode.translation import _ as tsf
 
 
+log = logging.getLogger(__name__)
+
+
 def add_renderer_globals(event):
     # Put pylons stuff into the context. This will be removed as soon as
     # migration to pyramid is finished.
     conf = pylons.config._current_obj()
     event['h'] = conf.get('pylons.h')
     event['c'] = pylons.tmpl_context
     event['url'] = pylons.url
 
     # TODO: When executed in pyramid view context the request is not available
     # in the event. Find a better solution to get the request.
     request = event['request'] or get_current_request()
 
     # Add Pyramid translation as '_' to context
     event['_'] = request.translate
     event['localizer'] = request.localizer
 
 
 def add_localizer(event):
     request = event.request
     localizer = get_localizer(request)
 
     def auto_translate(*args, **kwargs):
         return localizer.translate(tsf(*args, **kwargs))
 
     request.localizer = localizer
     request.translate = auto_translate
 
 
 def scan_repositories_if_enabled(event):
     """
     This is subscribed to the `pyramid.events.ApplicationCreated` event. It
     does a repository scan if enabled in the settings.
     """
     from rhodecode.model.scm import ScmModel
     from rhodecode.lib.utils import repo2db_mapper, get_rhodecode_base_path
     settings = event.app.registry.settings
     vcs_server_enabled = settings['vcs.server.enable']
     import_on_startup = settings['startup.import_repos']
     if vcs_server_enabled and import_on_startup:
         repositories = ScmModel().repo_scan(get_rhodecode_base_path())
         repo2db_mapper(repositories, remove_obsolete=False)
+
+
+class Subscriber(object):
+    def __call__(self, event):
+        self.run(event)
+
+    def run(self, event):
+        raise NotImplementedError('Subclass has to implement this.')
+
+
+class AsyncSubscriber(Subscriber):
+    def __init__(self):
+        self._stop = False
+        self._eventq = Queue.Queue()
+        self._worker = self.create_worker()
+        self._worker.start()
+
+    def __call__(self, event):
+        self._eventq.put(event)
+
+    def create_worker(self):
+        worker = Thread(target=self.do_work)
+        worker.daemon = True
+        return worker
+
+    def stop_worker(self):
+        self._stop = False
+        self._eventq.put(None)
+        self._worker.join()
+
+    def do_work(self):
+        while not self._stop:
+            event = self._eventq.get()
+            if event is not None:
+                self.run(event)
+
+
+class AsyncSubprocessSubscriber(AsyncSubscriber):
+
+    def __init__(self, cmd, timeout=None):
+        super(AsyncSubprocessSubscriber, self).__init__()
+        self._cmd = cmd
+        self._timeout = timeout
+
+    def run(self, event):
+        cmd = self._cmd
+        timeout = self._timeout
+        log.debug('Executing command %s.', cmd)
+
+        try:
+            output = subprocess32.check_output(
+                cmd, timeout=timeout, stderr=subprocess32.STDOUT)
+            log.debug('Command finished %s', cmd)
+            if output:
+                log.debug('Command output: %s', output)
+        except subprocess32.TimeoutExpired as e:
+            log.exception('Timeout while executing command.')
+            if e.output:
+                log.error('Command output: %s', e.output)
+        except subprocess32.CalledProcessError as e:
+            log.exception('Error while executing command.')
+            if e.output:
+                log.error('Command output: %s', e.output)
+        except:
+            log.exception(
+                'Exception while executing command %s.', cmd)

rhodecode/svn_support/__init__.py

@@ -1,87 +1,88 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 import os
 import shlex
 
 # Do not use `from rhodecode import events` here, it will be overridden by the
 # events module in this package due to pythons import mechanism.
 from rhodecode.events import RepoGroupEvent
+from rhodecode.subscribers import AsyncSubprocessSubscriber
 from rhodecode.config.middleware import (
     _bool_setting, _string_setting, _int_setting)
 
 from .events import ModDavSvnConfigChange
-from .subscribers import generate_config_subscriber, AsyncSubprocessSubscriber
+from .subscribers import generate_config_subscriber
 from . import config_keys
 
 
 log = logging.getLogger(__name__)
 
 
 def includeme(config):
     settings = config.registry.settings
     _sanitize_settings_and_apply_defaults(settings)
 
     if settings[config_keys.generate_config]:
         # Add subscriber to generate the Apache mod dav svn configuration on
         # repository group events.
         config.add_subscriber(generate_config_subscriber, RepoGroupEvent)
 
         # Prepare reload command to pass it to the subprocess module and add a
         # subscriber to execute it on configuration changes.
         reload_cmd = shlex.split(settings[config_keys.reload_command])
         reload_timeout = settings[config_keys.reload_timeout] or None
         config_change_subscriber = AsyncSubprocessSubscriber(
             cmd=reload_cmd, timeout=reload_timeout)
         config.add_subscriber(config_change_subscriber, ModDavSvnConfigChange)
 
 
 def _sanitize_settings_and_apply_defaults(settings):
     """
     Set defaults, convert to python types and validate settings.
     """
     _bool_setting(settings, config_keys.generate_config, 'false')
     _bool_setting(settings, config_keys.list_parent_path, 'true')
     _int_setting(settings, config_keys.reload_timeout, 10)
     _string_setting(settings, config_keys.config_file_path, '', lower=False)
     _string_setting(settings, config_keys.location_root, '/', lower=False)
     _string_setting(settings, config_keys.reload_command, '', lower=False)
 
     # Convert negative timeout values to zero.
     if settings[config_keys.reload_timeout] < 0:
         settings[config_keys.reload_timeout] = 0
 
     # Append path separator to location root.
     settings[config_keys.location_root] = _append_path_sep(
         settings[config_keys.location_root])
 
     # Validate settings.
     if settings[config_keys.generate_config]:
         assert len(settings[config_keys.config_file_path]) > 0
 
 
 def _append_path_sep(path):
     """
     Append the path separator if missing.
     """
     if isinstance(path, basestring) and not path.endswith(os.path.sep):
         path += os.path.sep
     return path

rhodecode/svn_support/subscribers.py

@@ -1,109 +1,40 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
-import Queue
-import subprocess32
-from threading import Thread
 
 
 from .utils import generate_mod_dav_svn_config
 
 
 log = logging.getLogger(__name__)
 
 
 def generate_config_subscriber(event):
     """
     Subscriber to the `rhodcode.events.RepoGroupEvent`. This triggers the
     automatic generation of mod_dav_svn config file on repository group
     changes.
     """
     try:
         generate_mod_dav_svn_config(event.request.registry)
     except Exception:
         log.exception(
             'Exception while generating subversion mod_dav_svn configuration.')
-
-
-class Subscriber(object):
-    def __call__(self, event):
-        self.run(event)
-
-    def run(self, event):
-        raise NotImplementedError('Subclass has to implement this.')
-
-
-class AsyncSubscriber(Subscriber):
-    def __init__(self):
-        self._stop = False
-        self._eventq = Queue.Queue()
-        self._worker = self.create_worker()
-        self._worker.start()
-
-    def __call__(self, event):
-        self._eventq.put(event)
-
-    def create_worker(self):
-        worker = Thread(target=self.do_work)
-        worker.daemon = True
-        return worker
-
-    def stop_worker(self):
-        self._stop = False
-        self._eventq.put(None)
-        self._worker.join()
-
-    def do_work(self):
-        while not self._stop:
-            event = self._eventq.get()
-            if event is not None:
-                self.run(event)
-
-
-class AsyncSubprocessSubscriber(AsyncSubscriber):
-
-    def __init__(self, cmd, timeout=None):
-        super(AsyncSubprocessSubscriber, self).__init__()
-        self._cmd = cmd
-        self._timeout = timeout
-
-    def run(self, event):
-        cmd = self._cmd
-        timeout = self._timeout
-        log.debug('Executing command %s.', cmd)
-
-        try:
-            output = subprocess32.check_output(
-                cmd, timeout=timeout, stderr=subprocess32.STDOUT)
-            log.debug('Command finished %s', cmd)
-            if output:
-                log.debug('Command output: %s', output)
-        except subprocess32.TimeoutExpired as e:
-            log.exception('Timeout while executing command.')
-            if e.output:
-                log.error('Command output: %s', e.output)
-        except subprocess32.CalledProcessError as e:
-            log.exception('Error while executing command.')
-            if e.output:
-                log.error('Command output: %s', e.output)
-        except:
-            log.exception(
-                'Exception while executing command %s.', cmd)