Use my lazr.config megamerge branch for now, even though it's still under

development.

Completely rework the way switchboards and queue runners are initialized,
i.e. driven from the configuration file instead of hard coded.

The various queue runner directories are no longer available thorugh the
config object directly.  Get them from config.switchboards.

Provide minimal mailman.cfg and testing.cfg configuration files.

Neuter styles for now until they can be consolidated with lazr.config.

14 files changed, 84 insertions, 66 deletions

diff --git a/mailman/queue/__init__.py b/mailman/queue/__init__.py
index 27ea83320..5aa72bb14 100644
--- a/mailman/queue/__init__.py
+++ b/mailman/queue/__init__.py
@@ -32,6 +32,7 @@ __all__ = [
 
 
 import os
+import sys
 import time
 import email
 import errno
@@ -43,13 +44,16 @@ import marshal
 import traceback
 
 from cStringIO import StringIO
+from lazr.config import as_boolean, as_timedelta
+from string import Template
 from zope.interface import implements
 
-from mailman import i18n
 from mailman import Message
 from mailman import Utils
+from mailman import i18n
 from mailman.config import config
-from mailman.interfaces import IRunner, ISwitchboard
+from mailman.interfaces.runner import IRunner
+from mailman.interfaces.switchboard import ISwitchboard
 
 # 20 bytes of all bits set, maximum hashlib.sha.digest() value
 shamax = 0xffffffffffffffffffffffffffffffffffffffffL
@@ -57,6 +61,7 @@ shamax = 0xffffffffffffffffffffffffffffffffffffffffL
 # Small increment to add to time in case two entries have the same time.  This
 # prevents skipping one of two entries with the same time until the next pass.
 DELTA = .0001
+DOT = '.'
 
 elog = logging.getLogger('mailman.error')
 dlog = logging.getLogger('mailman.debug')
@@ -66,11 +71,24 @@ dlog = logging.getLogger('mailman.debug')
 class Switchboard:
     implements(ISwitchboard)
 
-    def __init__(self, whichq, slice=None, numslices=1, recover=False):
+    @staticmethod
+    def initialize():
+        """Initialize the global switchboards for input/output."""
+        for conf in config.qrunner_configs:
+            name = conf.name.split('.')[-1]
+            assert name not in config.switchboards, (
+                'Duplicate qrunner name: %s' % name)
+            substitutions = config.paths
+            substitutions['name'] = name
+            path = Template(conf.path).safe_substitute(substitutions)
+            config.switchboards[name] = Switchboard(path)
+
+    def __init__(self, queue_directory,
+                 slice=None, numslices=1, recover=False):
         """Create a switchboard object.
 
-        :param whichq: The queue directory.
-        :type whichq: str
+        :param queue_directory: The queue directory.
+        :type queue_directory: str
         :param slice: The slice number for this switchboard, or None.  If not
             None, it must be [0..`numslices`).
         :type slice: int or None
@@ -80,9 +98,11 @@ class Switchboard:
         :param recover: True if backup files should be recovered.
         :type recover: bool
         """
-        self._whichq = whichq
+        assert (numslices & (numslices - 1)) == 0, (
+            'Not a power of 2: %s' % numslices)
+        self.queue_directory = queue_directory
         # Create the directory if it doesn't yet exist.
-        Utils.makedirs(self._whichq, 0770)
+        Utils.makedirs(self.queue_directory, 0770)
         # Fast track for no slices
         self._lower = None
         self._upper = None
@@ -93,11 +113,6 @@ class Switchboard:
         if recover:
             self.recover_backup_files()
 
-    @property
-    def queue_directory(self):
-        """See `ISwitchboard`."""
-        return self._whichq
-
     def enqueue(self, _msg, _metadata=None, **_kws):
         """See `ISwitchboard`."""
         if _metadata is None:
@@ -125,7 +140,7 @@ class Switchboard:
         # and the sha hex digest.
         rcvtime = data.setdefault('received_time', now)
         filebase = repr(rcvtime) + '+' + hashlib.sha1(hashfood).hexdigest()
-        filename = os.path.join(self._whichq, filebase + '.pck')
+        filename = os.path.join(self.queue_directory, filebase + '.pck')
         tmpfile = filename + '.tmp'
         # Always add the metadata schema version number
         data['version'] = config.QFILE_SCHEMA_VERSION
@@ -148,8 +163,8 @@ class Switchboard:
     def dequeue(self, filebase):
         """See `ISwitchboard`."""
         # Calculate the filename from the given filebase.
-        filename = os.path.join(self._whichq, filebase + '.pck')
-        backfile = os.path.join(self._whichq, filebase + '.bak')
+        filename = os.path.join(self.queue_directory, filebase + '.pck')
+        backfile = os.path.join(self.queue_directory, filebase + '.bak')
         # Read the message object and metadata.
         with open(filename) as fp:
             # Move the file to the backup file name for processing.  If this
@@ -172,13 +187,13 @@ class Switchboard:
         return msg, data
 
     def finish(self, filebase, preserve=False):
-        bakfile = os.path.join(self._whichq, filebase + '.bak')
+        bakfile = os.path.join(self.queue_directory, filebase + '.bak')
         try:
             if preserve:
-                psvfile = os.path.join(config.SHUNTQUEUE_DIR,
-                                       filebase + '.psv')
+                shunt_dir = config.switchboards['shunt'].queue_directory
+                psvfile = os.path.join(shunt_dir, filebase + '.psv')
                 # Create the directory if it doesn't yet exist.
-                Utils.makedirs(config.SHUNTQUEUE_DIR, 0770)
+                Utils.makedirs(shunt_dir, 0770)
                 os.rename(bakfile, psvfile)
             else:
                 os.unlink(bakfile)
@@ -196,7 +211,7 @@ class Switchboard:
         times = {}
         lower = self._lower
         upper = self._upper
-        for f in os.listdir(self._whichq):
+        for f in os.listdir(self.queue_directory):
             # By ignoring anything that doesn't end in .pck, we ignore
             # tempfiles and avoid a race condition.
             filebase, ext = os.path.splitext(f)
@@ -220,8 +235,8 @@ class Switchboard:
         # to exist at the same time, so the move is enough to ensure that our
         # normal dequeuing process will handle them.
         for filebase in self.get_files('.bak'):
-            src = os.path.join(self._whichq, filebase + '.bak')
-            dst = os.path.join(self._whichq, filebase + '.pck')
+            src = os.path.join(self.queue_directory, filebase + '.bak')
+            dst = os.path.join(self.queue_directory, filebase + '.pck')
             os.rename(src, dst)
 
 
@@ -229,27 +244,31 @@ class Switchboard:
 class Runner:
     implements(IRunner)
 
-    QDIR = None
-    SLEEPTIME = None
-
-    def __init__(self, slice=None, numslices=1):
+    def __init__(self, name, slice=None):
         """Create a queue runner.
 
         :param slice: The slice number for this queue runner.  This is passed
             directly to the underlying `ISwitchboard` object.
         :type slice: int or None
-        :param numslices: The number of slices for this queue.  Must be a
-            power of 2.
-        :type numslices: int
         """
-        # Create our own switchboard.  Don't use the switchboard cache because
-        # we want to provide slice and numslice arguments.
-        self._switchboard = Switchboard(self.QDIR, slice, numslices, True)
-        # Create the shunt switchboard
-        self._shunt = Switchboard(config.SHUNTQUEUE_DIR)
+        # Grab the configuration section.
+        self.name = name
+        section = getattr(config, 'qrunner.' + name)
+        substitutions = config.paths
+        substitutions['name'] = name
+        self.queue_directory = Template(section.path).safe_substitute(
+            substitutions)
+        numslices = int(section.instances)
+        self.switchboard = Switchboard(
+            self.queue_directory, slice, numslices, True)
+        self.sleep_time = as_timedelta(section.sleep_time)
+        # sleep_time is a timedelta; turn it into a float for time.sleep().
+        self.sleep_float = (86400 * self.sleep_time.days +
+                            self.sleep_time.seconds +
+                            self.sleep_time.microseconds / 1000000.0)
+        self.max_restarts = int(section.max_restarts)
+        self.start = as_boolean(section.start)
         self._stop = False
-        if self.SLEEPTIME is None:
-            self.SLEEPTIME = config.QRUNNER_SLEEP_TIME
 
     def __repr__(self):
         return '<%s at %s>' % (self.__class__.__name__, id(self))
@@ -286,13 +305,13 @@ class Runner:
         dlog.debug('[%s] starting oneloop', me)
         # List all the files in our queue directory.  The switchboard is
         # guaranteed to hand us the files in FIFO order.
-        files = self._switchboard.files
+        files = self.switchboard.files
         for filebase in files:
             dlog.debug('[%s] processing filebase: %s', me, filebase)
             try:
                 # Ask the switchboard for the message and metadata objects
                 # associated with this queue file.
-                msg, msgdata = self._switchboard.dequeue(filebase)
+                msg, msgdata = self.switchboard.dequeue(filebase)
             except Exception, e:
                 # This used to just catch email.Errors.MessageParseError, but
                 # other problems can occur in message parsing, e.g.
@@ -302,14 +321,14 @@ class Runner:
                 self._log(e)
                 elog.error('Skipping and preserving unparseable message: %s',
                            filebase)
-                self._switchboard.finish(filebase, preserve=True)
+                self.switchboard.finish(filebase, preserve=True)
                 config.db.abort()
                 continue
             try:
                 dlog.debug('[%s] processing onefile', me)
                 self._process_one_file(msg, msgdata)
                 dlog.debug('[%s] finishing filebase: %s', me, filebase)
-                self._switchboard.finish(filebase)
+                self.switchboard.finish(filebase)
             except Exception, e:
                 # All runners that implement _dispose() must guarantee that
                 # exceptions are caught and dealt with properly.  Still, there
@@ -319,14 +338,14 @@ class Runner:
                 # intervention.
                 self._log(e)
                 # Put a marker in the metadata for unshunting.
-                msgdata['whichq'] = self._switchboard.queue_directory
+                msgdata['whichq'] = self.switchboard.queue_directory
                 # It is possible that shunting can throw an exception, e.g. a
                 # permissions problem or a MemoryError due to a really large
                 # message.  Try to be graceful.
                 try:
                     new_filebase = self._shunt.enqueue(msg, msgdata)
                     elog.error('SHUNTING: %s', new_filebase)
-                    self._switchboard.finish(filebase)
+                    self.switchboard.finish(filebase)
                 except Exception, e:
                     # The message wasn't successfully shunted.  Log the
                     # exception and try to preserve the original queue entry
@@ -335,13 +354,13 @@ class Runner:
                     elog.error(
                         'SHUNTING FAILED, preserving original entry: %s',
                         filebase)
-                    self._switchboard.finish(filebase, preserve=True)
+                    self.switchboard.finish(filebase, preserve=True)
                 config.db.abort()
             # Other work we want to do each time through the loop.
             dlog.debug('[%s] doing periodic', me)
             self._do_periodic()
             dlog.debug('[%s] checking short circuit', me)
-            if self._short_curcuit():
+            if self._short_circuit():
                 dlog.debug('[%s] short circuiting', me)
                 break
             dlog.debug('[%s] commiting', me)
@@ -380,7 +399,7 @@ class Runner:
             msgdata['lang'] = language
             keepqueued = self._dispose(mlist, msg, msgdata)
         if keepqueued:
-            self._switchboard.enqueue(msg, msgdata)
+            self.switchboard.enqueue(msg, msgdata)
 
     def _log(self, exc):
         elog.error('Uncaught runner exception: %s', exc)
@@ -401,10 +420,10 @@ class Runner:
 
     def _snooze(self, filecnt):
         """See `IRunner`."""
-        if filecnt or float(self.SLEEPTIME) <= 0:
+        if filecnt or self.sleep_float <= 0:
             return
-        time.sleep(float(self.SLEEPTIME))
+        time.sleep(self.sleep_float)
 
-    def _short_curcuit(self):
+    def _short_circuit(self):
         """See `IRunner`."""
         return self._stop
diff --git a/mailman/queue/archive.py b/mailman/queue/archive.py
index 6dda70387..e9fd5f7ad 100644
--- a/mailman/queue/archive.py
+++ b/mailman/queue/archive.py
@@ -30,14 +30,14 @@ from datetime import datetime
 from email.Utils import parsedate_tz, mktime_tz, formatdate
 from locknix.lockfile import Lock
 
-from mailman.configuration import config
+from mailman import Defaults
 from mailman.core.plugins import get_plugins
 from mailman.queue import Runner
 
 
 
 class ArchiveRunner(Runner):
-    QDIR = config.ARCHQUEUE_DIR
+    """The archive runner."""
 
     def _dispose(self, mlist, msg, msgdata):
         # Support clobber_date, i.e. setting the date in the archive to the
@@ -48,9 +48,9 @@ class ArchiveRunner(Runner):
         received_time = formatdate(msgdata['received_time'])
         if not original_date:
             clobber = True
-        elif config.ARCHIVER_CLOBBER_DATE_POLICY == 1:
+        elif Defaults.ARCHIVER_CLOBBER_DATE_POLICY == 1:
             clobber = True
-        elif config.ARCHIVER_CLOBBER_DATE_POLICY == 2:
+        elif Defaults.ARCHIVER_CLOBBER_DATE_POLICY == 2:
             # What's the timestamp on the original message?
             timetup = parsedate_tz(original_date)
             now = datetime.now()
@@ -60,7 +60,7 @@ class ArchiveRunner(Runner):
                 else:
                     utc_timestamp = datetime.fromtimestamp(mktime_tz(timetup))
                     clobber = (abs(now - utc_timestamp) > 
-                               config.ARCHIVER_ALLOWABLE_SANE_DATE_SKEW)
+                               Defaults.ARCHIVER_ALLOWABLE_SANE_DATE_SKEW)
             except (ValueError, OverflowError):
                 # The likely cause of this is that the year in the Date: field
                 # is horribly incorrect, e.g. (from SF bug # 571634):
diff --git a/mailman/queue/bounce.py b/mailman/queue/bounce.py
index 0c5788174..18fa08437 100644
--- a/mailman/queue/bounce.py
+++ b/mailman/queue/bounce.py
@@ -30,7 +30,7 @@ from email.Utils import parseaddr
 from mailman import Utils
 from mailman.Bouncers import BouncerAPI
 from mailman.Message import UserNotification
-from mailman.configuration import config
+from mailman.config import config
 from mailman.i18n import _
 from mailman.queue import Runner, Switchboard
 
diff --git a/mailman/queue/command.py b/mailman/queue/command.py
index a02827de0..986758ce1 100644
--- a/mailman/queue/command.py
+++ b/mailman/queue/command.py
@@ -43,7 +43,7 @@ from zope.interface import implements
 from mailman import Message
 from mailman import Utils
 from mailman.app.replybot import autorespond_to_sender
-from mailman.configuration import config
+from mailman.config import config
 from mailman.i18n import _
 from mailman.interfaces import ContinueProcessing, IEmailResults
 from mailman.queue import Runner
diff --git a/mailman/queue/docs/archiver.txt b/mailman/queue/docs/archiver.txt
index 98ead0aa7..ed7c26d45 100644
--- a/mailman/queue/docs/archiver.txt
+++ b/mailman/queue/docs/archiver.txt
@@ -18,8 +18,7 @@ interface.  By default, there's a Pipermail archiver.
     ... First post!
     ... """)
 
-    >>> from mailman.queue import Switchboard
-    >>> archiver_queue = Switchboard(config.ARCHQUEUE_DIR)
+    >>> archiver_queue = config.switchboards['archive']
     >>> ignore = archiver_queue.enqueue(msg, {}, listname=mlist.fqdn_listname)
 
     >>> from mailman.queue.archive import ArchiveRunner
diff --git a/mailman/queue/http.py b/mailman/queue/http.py
index cb6940b29..c626e766e 100644
--- a/mailman/queue/http.py
+++ b/mailman/queue/http.py
@@ -25,7 +25,7 @@ from cStringIO import StringIO
 from wsgiref.simple_server import make_server, WSGIRequestHandler
 
 from mailman.Cgi.wsgi_app import mailman_app
-from mailman.configuration import config
+from mailman.config import config
 from mailman.queue import Runner
 
 hlog = logging.getLogger('mailman.http')
diff --git a/mailman/queue/incoming.py b/mailman/queue/incoming.py
index d4decd435..aaab56b21 100644
--- a/mailman/queue/incoming.py
+++ b/mailman/queue/incoming.py
@@ -26,7 +26,7 @@ prepared for delivery.  Rejections, discards, and holds are processed
 immediately.
 """
 
-from mailman.configuration import config
+from mailman.config import config
 from mailman.core.chains import process
 from mailman.queue import Runner
 
diff --git a/mailman/queue/lmtp.py b/mailman/queue/lmtp.py
index b4688653b..18f430e55 100644
--- a/mailman/queue/lmtp.py
+++ b/mailman/queue/lmtp.py
@@ -43,7 +43,7 @@ import asyncore
 from email.utils import parseaddr
 
 from mailman.Message import Message
-from mailman.configuration import config
+from mailman.config import config
 from mailman.database.transaction import txn
 from mailman.queue import Runner, Switchboard
 
diff --git a/mailman/queue/maildir.py b/mailman/queue/maildir.py
index 71bac67dc..3f7dd5556 100644
--- a/mailman/queue/maildir.py
+++ b/mailman/queue/maildir.py
@@ -57,7 +57,7 @@ from email.Parser import Parser
 from email.Utils import parseaddr
 
 from mailman.Message import Message
-from mailman.configuration import config
+from mailman.config import config
 from mailman.queue import Runner
 
 log = logging.getLogger('mailman.error')
diff --git a/mailman/queue/news.py b/mailman/queue/news.py
index 8415f7d95..70ffae71b 100644
--- a/mailman/queue/news.py
+++ b/mailman/queue/news.py
@@ -29,7 +29,7 @@ from email.utils import getaddresses, make_msgid
 COMMASPACE = ', '
 
 from mailman import Utils
-from mailman.configuration import config
+from mailman.config import config
 from mailman.interfaces import NewsModeration
 from mailman.queue import Runner
 
diff --git a/mailman/queue/outgoing.py b/mailman/queue/outgoing.py
index 3ab67eaad..9eb287e6b 100644
--- a/mailman/queue/outgoing.py
+++ b/mailman/queue/outgoing.py
@@ -27,7 +27,7 @@ import logging
 from datetime import datetime
 
 from mailman import Message
-from mailman.configuration import config
+from mailman.config import config
 from mailman.core import errors
 from mailman.queue import Runner, Switchboard
 from mailman.queue.bounce import BounceMixin
diff --git a/mailman/queue/pipeline.py b/mailman/queue/pipeline.py
index 665bfce77..4d4be7790 100644
--- a/mailman/queue/pipeline.py
+++ b/mailman/queue/pipeline.py
@@ -23,7 +23,7 @@ headers, calculates message recipients, and more.
 """
 
 from mailman.app.pipelines import process
-from mailman.configuration import config
+from mailman.config import config
 from mailman.queue import Runner
 
 
diff --git a/mailman/queue/retry.py b/mailman/queue/retry.py
index f8e1b4665..03cf86848 100644
--- a/mailman/queue/retry.py
+++ b/mailman/queue/retry.py
@@ -17,7 +17,7 @@
 
 import time
 
-from mailman.configuration import config
+from mailman.config import config
 from mailman.queue import Runner, Switchboard
 
 
diff --git a/mailman/queue/virgin.py b/mailman/queue/virgin.py
index 503337ea1..917d702ca 100644
--- a/mailman/queue/virgin.py
+++ b/mailman/queue/virgin.py
@@ -24,7 +24,7 @@ recipient.
 """
 
 from mailman.app.pipelines import process
-from mailman.configuration import config
+from mailman.config import config
 from mailman.queue import Runner