Massive digester rework.

This moves actual digest creation and sending into a new queue runner.  That
way, digest creation does not block the main incoming processor.  Lots of code
clean up and reorganization too, and many fixes.

digest_last_sent_at: Make this a DateTime
digest_size_threshold: Make this a Float
digest_volume_frequency: Make this an Enum

DigestFrequency: new enum

Fix up IMailingList interface to reflect reality.

In Runner base class, allow for msg.get_sender() to return a False value.  If
this happens (as it can with the empty marker messages used to signal the
digester that there's work to do), just use the list's preferred language.

Give the retry.py and incoming.py queue runner modules some proper epilogues.

The EmptyingRunner (for testing) logs itself as the runner its wrapping.

The digest file name now ends in .mmdf since that's the format we're using.

Add 'create_list' to doctest globals.

Add a Mailbox class to add support for the context manager protocol.

16 files changed, 1511 insertions, 891 deletions

diff --git a/src/mailman/app/digests.py b/src/mailman/app/digests.py
new file mode 100644
index 000000000..812ae1649
--- /dev/null
+++ b/src/mailman/app/digests.py
@@ -0,0 +1,426 @@
+# Copyright (C) 1998-2009 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman 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 General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Add the message to the list's current digest and possibly send it."""
+
+# Messages are accumulated to a Unix mailbox compatible file containing all
+# the messages destined for the digest.  This file must be parsable by the
+# mailbox.UnixMailbox class (i.e. it must be ^From_ quoted).
+#
+# When the file reaches the size threshold, it is moved to the qfiles/digest
+# directory and the DigestRunner will craft the MIME, rfc1153, and
+# (eventually) URL-subject linked digests from the mbox.
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'ToDigest',
+    ]
+
+
+import os
+import re
+import copy
+import time
+import logging
+
+from StringIO import StringIO          # cStringIO can't handle unicode.
+from email.charset import Charset
+from email.generator import Generator
+from email.header import decode_header, make_header, Header
+from email.mime.base import MIMEBase
+from email.mime.message import MIMEMessage
+from email.mime.text import MIMEText
+from email.parser import Parser
+from email.utils import formatdate, getaddresses, make_msgid
+from zope.interface import implements
+
+from mailman import Message
+from mailman import Utils
+from mailman import i18n
+from mailman.Mailbox import Mailbox
+from mailman.Mailbox import Mailbox
+from mailman.config import config
+from mailman.core import errors
+from mailman.interfaces.handler import IHandler
+from mailman.interfaces.member import DeliveryMode, DeliveryStatus
+from mailman.pipeline.decorate import decorate
+from mailman.pipeline.scrubber import process as scrubber
+
+
+_ = i18n._
+
+UEMPTYSTRING = ''
+EMPTYSTRING = ''
+
+log = logging.getLogger('mailman.error')
+
+
+
+def process(mlist, msg, msgdata):
+    # Short circuit non-digestable lists.
+    if not mlist.digestable or msgdata.get('isdigest'):
+        return
+    mboxfile = os.path.join(mlist.data_path, 'digest.mbox')
+    mboxfp = open(mboxfile, 'a+')
+    mbox = Mailbox(mboxfp)
+    mbox.AppendMessage(msg)
+    # Calculate the current size of the accumulation file.  This will not tell
+    # us exactly how big the MIME, rfc1153, or any other generated digest
+    # message will be, but it's the most easily available metric to decide
+    # whether the size threshold has been reached.
+    mboxfp.flush()
+    size = os.path.getsize(mboxfile)
+    if size / 1024.0 >= mlist.digest_size_threshold:
+        # This is a bit of a kludge to get the mbox file moved to the digest
+        # queue directory.
+        try:
+            # Enclose in try/except here because a error in send_digest() can
+            # silently stop regular delivery.  Unsuccessful digest delivery
+            # should be tried again by cron and the site administrator will be
+            # notified of any error explicitly by the cron error message.
+            mboxfp.seek(0)
+            send_digests(mlist, mboxfp)
+            os.unlink(mboxfile)
+        except Exception, errmsg:
+            # Bare except is generally prohibited in Mailman, but we can't
+            # forecast what exceptions can occur here.
+            log.exception('send_digests() failed: %s', errmsg)
+    mboxfp.close()
+
+
+
+def send_digests(mlist, mboxfp):
+    # Set the digest volume and time
+    if mlist.digest_last_sent_at:
+        bump = False
+        # See if we should bump the digest volume number
+        timetup = time.localtime(mlist.digest_last_sent_at)
+        now = time.localtime(time.time())
+        freq = mlist.digest_volume_frequency
+        if freq == 0 and timetup[0] < now[0]:
+            # Yearly
+            bump = True
+        elif freq == 1 and timetup[1] <> now[1]:
+            # Monthly, but we take a cheap way to calculate this.  We assume
+            # that the clock isn't going to be reset backwards.
+            bump = True
+        elif freq == 2 and (timetup[1] % 4 <> now[1] % 4):
+            # Quarterly, same caveat
+            bump = True
+        elif freq == 3:
+            # Once again, take a cheap way of calculating this
+            weeknum_last = int(time.strftime('%W', timetup))
+            weeknum_now = int(time.strftime('%W', now))
+            if weeknum_now > weeknum_last or timetup[0] > now[0]:
+                bump = True
+        elif freq == 4 and timetup[7] <> now[7]:
+            # Daily
+            bump = True
+        if bump:
+            mlist.bump_digest_volume()
+    mlist.digest_last_sent_at = time.time()
+    # Wrapper around actually digest crafter to set up the language context
+    # properly.  All digests are translated to the list's preferred language.
+    with i18n.using_language(mlist.preferred_language):
+        send_i18n_digests(mlist, mboxfp)
+
+
+
+def send_i18n_digests(mlist, mboxfp):
+    mbox = Mailbox(mboxfp)
+    # Prepare common information (first lang/charset)
+    lang = mlist.preferred_language
+    lcset = Utils.GetCharSet(lang)
+    lcset_out = Charset(lcset).output_charset or lcset
+    # Common Information (contd)
+    realname = mlist.real_name
+    volume = mlist.volume
+    issue = mlist.next_digest_number
+    digestid = _('$realname Digest, Vol $volume, Issue $issue')
+    digestsubj = Header(digestid, lcset, header_name='Subject')
+    # Set things up for the MIME digest.  Only headers not added by
+    # CookHeaders need be added here.
+    # Date/Message-ID should be added here also.
+    mimemsg = Message.Message()
+    mimemsg['Content-Type'] = 'multipart/mixed'
+    mimemsg['MIME-Version'] = '1.0'
+    mimemsg['From'] = mlist.request_address
+    mimemsg['Subject'] = digestsubj
+    mimemsg['To'] = mlist.posting_address
+    mimemsg['Reply-To'] = mlist.posting_address
+    mimemsg['Date'] = formatdate(localtime=1)
+    mimemsg['Message-ID'] = make_msgid()
+    # Set things up for the rfc1153 digest
+    plainmsg = StringIO()
+    rfc1153msg = Message.Message()
+    rfc1153msg['From'] = mlist.request_address
+    rfc1153msg['Subject'] = digestsubj
+    rfc1153msg['To'] = mlist.posting_address
+    rfc1153msg['Reply-To'] = mlist.posting_address
+    rfc1153msg['Date'] = formatdate(localtime=1)
+    rfc1153msg['Message-ID'] = make_msgid()
+    separator70 = '-' * 70
+    separator30 = '-' * 30
+    # In the rfc1153 digest, the masthead contains the digest boilerplate plus
+    # any digest header.  In the MIME digests, the masthead and digest header
+    # are separate MIME subobjects.  In either case, it's the first thing in
+    # the digest, and we can calculate it now, so go ahead and add it now.
+    mastheadtxt = Utils.maketext(
+        'masthead.txt',
+        {'real_name' :        mlist.real_name,
+         'got_list_email':    mlist.posting_address,
+         'got_listinfo_url':  mlist.script_url('listinfo'),
+         'got_request_email': mlist.request_address,
+         'got_owner_email':   mlist.owner_address,
+         }, mlist=mlist)
+    # MIME
+    masthead = MIMEText(mastheadtxt.encode(lcset), _charset=lcset)
+    masthead['Content-Description'] = digestid
+    mimemsg.attach(masthead)
+    # RFC 1153
+    print >> plainmsg, mastheadtxt
+    print >> plainmsg
+    # Now add the optional digest header
+    if mlist.digest_header:
+        headertxt = decorate(mlist, mlist.digest_header, _('digest header'))
+        # MIME
+        header = MIMEText(headertxt.encode(lcset), _charset=lcset)
+        header['Content-Description'] = _('Digest Header')
+        mimemsg.attach(header)
+        # RFC 1153
+        print >> plainmsg, headertxt
+        print >> plainmsg
+    # Now we have to cruise through all the messages accumulated in the
+    # mailbox file.  We can't add these messages to the plainmsg and mimemsg
+    # yet, because we first have to calculate the table of contents
+    # (i.e. grok out all the Subjects).  Store the messages in a list until
+    # we're ready for them.
+    #
+    # Meanwhile prepare things for the table of contents
+    toc = StringIO()
+    print >> toc, _("Today's Topics:\n")
+    # Now cruise through all the messages in the mailbox of digest messages,
+    # building the MIME payload and core of the RFC 1153 digest.  We'll also
+    # accumulate Subject: headers and authors for the table-of-contents.
+    messages = []
+    msgcount = 0
+    msg = mbox.next()
+    while msg is not None:
+        if msg == '':
+            # It was an unparseable message
+            msg = mbox.next()
+            continue
+        msgcount += 1
+        messages.append(msg)
+        # Get the Subject header
+        msgsubj = msg.get('subject', _('(no subject)'))
+        subject = Utils.oneline(msgsubj, in_unicode=True)
+        # Don't include the redundant subject prefix in the toc
+        mo = re.match('(re:? *)?(%s)' % re.escape(mlist.subject_prefix),
+                      subject, re.IGNORECASE)
+        if mo:
+            subject = subject[:mo.start(2)] + subject[mo.end(2):]
+        username = ''
+        addresses = getaddresses([Utils.oneline(msg.get('from', ''),
+                                                in_unicode=True)])
+        # Take only the first author we find
+        if isinstance(addresses, list) and addresses:
+            username = addresses[0][0]
+            if not username:
+                username = addresses[0][1]
+        if username:
+            username = ' ({0})'.format(username)
+        # Put count and Wrap the toc subject line
+        wrapped = Utils.wrap('{0:2}. {1}'.format(msgcount, subject), 65)
+        slines = wrapped.split('\n')
+        # See if the user's name can fit on the last line
+        if len(slines[-1]) + len(username) > 70:
+            slines.append(username)
+        else:
+            slines[-1] += username
+        # Add this subject to the accumulating topics
+        first = True
+        for line in slines:
+            if first:
+                print >> toc, ' ', line
+                first = False
+            else:
+                print >> toc, '     ', line.lstrip()
+        # We do not want all the headers of the original message to leak
+        # through in the digest messages.  For this phase, we'll leave the
+        # same set of headers in both digests, i.e. those required in RFC 1153
+        # plus a couple of other useful ones.  We also need to reorder the
+        # headers according to RFC 1153.  Later, we'll strip out headers for
+        # for the specific MIME or plain digests.
+        keeper = {}
+        all_keepers = set(
+            header for header in
+            config.digests.mime_digest_keep_headers.split() +
+            config.digests.plain_digest_keep_headers.split())
+        for keep in all_keepers:
+            keeper[keep] = msg.get_all(keep, [])
+        # Now remove all unkempt headers :)
+        for header in msg.keys():
+            del msg[header]
+        # And add back the kept header in the RFC 1153 designated order
+        for keep in all_keepers:
+            for field in keeper[keep]:
+                msg[keep] = field
+        # And a bit of extra stuff
+        msg['Message'] = repr(msgcount)
+        # Get the next message in the digest mailbox
+        msg = mbox.next()
+    # Now we're finished with all the messages in the digest.  First do some
+    # sanity checking and then on to adding the toc.
+    if msgcount == 0:
+        # Why did we even get here?
+        return
+    toctext = toc.getvalue()
+    # MIME
+    try:
+        tocpart = MIMEText(toctext.encode(lcset), _charset=lcset)
+    except UnicodeError:
+        tocpart = MIMEText(toctext.encode('utf-8'), _charset='utf-8')
+    tocpart['Content-Description']= _("Today's Topics ($msgcount messages)")
+    mimemsg.attach(tocpart)
+    # RFC 1153
+    print >> plainmsg, toctext
+    print >> plainmsg
+    # For RFC 1153 digests, we now need the standard separator
+    print >> plainmsg, separator70
+    print >> plainmsg
+    # Now go through and add each message
+    mimedigest = MIMEBase('multipart', 'digest')
+    mimemsg.attach(mimedigest)
+    first = True
+    for msg in messages:
+        # MIME.  Make a copy of the message object since the rfc1153
+        # processing scrubs out attachments.
+        mimedigest.attach(MIMEMessage(copy.deepcopy(msg)))
+        # rfc1153
+        if first:
+            first = False
+        else:
+            print >> plainmsg, separator30
+            print >> plainmsg
+        # Use Mailman.pipeline.scrubber.process() to get plain text
+        try:
+            msg = scrubber(mlist, msg)
+        except errors.DiscardMessage:
+            print >> plainmsg, _('[Message discarded by content filter]')
+            continue
+        # Honor the default setting
+        for h in config.digests.plain_digest_keep_headers.split():
+            if msg[h]:
+                uh = Utils.wrap('{0}: {1}'.format(
+                    h, Utils.oneline(msg[h], in_unicode=True)))
+                uh = '\n\t'.join(uh.split('\n'))
+                print >> plainmsg, uh
+        print >> plainmsg
+        # If decoded payload is empty, this may be multipart message.
+        # -- just stringfy it.
+        payload = msg.get_payload(decode=True) \
+                  or msg.as_string().split('\n\n',1)[1]
+        mcset = msg.get_content_charset('us-ascii')
+        try:
+            payload = unicode(payload, mcset, 'replace')
+        except (LookupError, TypeError):
+            # unknown or empty charset
+            payload = unicode(payload, 'us-ascii', 'replace')
+        print >> plainmsg, payload
+        if not payload.endswith('\n'):
+            print >> plainmsg
+    # Now add the footer
+    if mlist.digest_footer:
+        footertxt = decorate(mlist, mlist.digest_footer)
+        # MIME
+        footer = MIMEText(footertxt.encode(lcset), _charset=lcset)
+        footer['Content-Description'] = _('Digest Footer')
+        mimemsg.attach(footer)
+        # RFC 1153
+        # BAW: This is not strictly conformant RFC 1153.  The trailer is only
+        # supposed to contain two lines, i.e. the "End of ... Digest" line and
+        # the row of asterisks.  If this screws up MUAs, the solution is to
+        # add the footer as the last message in the RFC 1153 digest.  I just
+        # hate the way that VM does that and I think it's confusing to users,
+        # so don't do it unless there's a clamor.
+        print >> plainmsg, separator30
+        print >> plainmsg
+        print >> plainmsg, footertxt
+        print >> plainmsg
+    # Do the last bit of stuff for each digest type
+    signoff = _('End of ') + digestid
+    # MIME
+    # BAW: This stuff is outside the normal MIME goo, and it's what the old
+    # MIME digester did.  No one seemed to complain, probably because you
+    # won't see it in an MUA that can't display the raw message.  We've never
+    # got complaints before, but if we do, just wax this.  It's primarily
+    # included for (marginally useful) backwards compatibility.
+    mimemsg.postamble = signoff
+    # rfc1153
+    print >> plainmsg, signoff
+    print >> plainmsg, '*' * len(signoff)
+    # Do our final bit of housekeeping, and then send each message to the
+    # outgoing queue for delivery.
+    mlist.next_digest_number += 1
+    virginq = config.switchboards['virgin']
+    # Calculate the recipients lists
+    plainrecips = set()
+    mimerecips = set()
+    # When someone turns off digest delivery, they will get one last digest to
+    # ensure that there will be no gaps in the messages they receive.
+    # Currently, this dictionary contains the email addresses of those folks
+    # who should get one last digest.  We need to find the corresponding
+    # IMember records.
+    digest_members = set(mlist.digest_members.members)
+    for address in mlist.one_last_digest:
+        member = mlist.digest_members.get_member(address)
+        if member:
+            digest_members.add(member)
+    for member in digest_members:
+        if member.delivery_status <> DeliveryStatus.enabled:
+            continue
+        # Send the digest to the case-preserved address of the digest members.
+        email_address = member.address.original_address
+        if member.delivery_mode == DeliveryMode.plaintext_digests:
+            plainrecips.add(email_address)
+        elif member.delivery_mode == DeliveryMode.mime_digests:
+            mimerecips.add(email_address)
+        else:
+            raise AssertionError(
+                'Digest member "{0}" unexpected delivery mode: {1}'.format(
+                    email_address, member.delivery_mode))
+    # Zap this since we're now delivering the last digest to these folks.
+    mlist.one_last_digest.clear()
+    # MIME
+    virginq.enqueue(mimemsg,
+                    recips=mimerecips,
+                    listname=mlist.fqdn_listname,
+                    isdigest=True)
+    # RFC 1153
+    # If the entire digest message can't be encoded by list charset, fall
+    # back to 'utf-8'.
+    try:
+        rfc1153msg.set_payload(plainmsg.getvalue().encode(lcset), lcset)
+    except UnicodeError:
+        rfc1153msg.set_payload(plainmsg.getvalue().encode('utf-8'), 'utf-8')
+    virginq.enqueue(rfc1153msg,
+                    recips=plainrecips,
+                    listname=mlist.fqdn_listname,
+                    isdigest=True)
diff --git a/src/mailman/config/mailman.cfg b/src/mailman/config/mailman.cfg
index 2bf528bea..d68099ca3 100644
--- a/src/mailman/config/mailman.cfg
+++ b/src/mailman/config/mailman.cfg
@@ -25,7 +25,7 @@ class: mailman.queue.archive.ArchiveRunner
 
 [qrunner.bad]
 class: mailman.queue.fake.BadRunner
-# The shunt runner is just a placeholder for its switchboard.
+# The bad runner is just a placeholder for its switchboard.
 start: no
 
 [qrunner.bounces]
@@ -66,4 +66,7 @@ start: no
 [qrunner.virgin]
 class: mailman.queue.virgin.VirginRunner
 
+[qrunner.digest]
+class: mailman.queue.digest.DigestRunner
+
 [style.default]
diff --git a/src/mailman/database/mailinglist.py b/src/mailman/database/mailinglist.py
index 8803a5fa4..e380a71b1 100644
--- a/src/mailman/database/mailinglist.py
+++ b/src/mailman/database/mailinglist.py
@@ -66,7 +66,7 @@ class MailingList(Model):
     admin_responses = Pickle()
     postings_responses = Pickle()
     request_responses = Pickle()
-    digest_last_sent_at = Float()
+    digest_last_sent_at = DateTime()
     one_last_digest = Pickle()
     volume = Int()
     last_post_time = DateTime()
@@ -108,8 +108,8 @@ class MailingList(Model):
     digest_header = Unicode()
     digest_is_default = Bool()
     digest_send_periodic = Bool()
-    digest_size_threshold = Int()
-    digest_volume_frequency = Int()
+    digest_size_threshold = Float()
+    digest_volume_frequency = Enum()
     digestable = Bool()
     discard_these_nonmembers = Pickle()
     emergency = Bool()
diff --git a/src/mailman/database/mailman.sql b/src/mailman/database/mailman.sql
index b098ed13b..61d48285a 100644
--- a/src/mailman/database/mailman.sql
+++ b/src/mailman/database/mailman.sql
@@ -37,7 +37,7 @@ CREATE TABLE mailinglist (
         admin_responses BLOB,
         postings_responses BLOB,
         request_responses BLOB,
-        digest_last_sent_at NUMERIC(10, 2),
+        digest_last_sent_at TIMESTAMP,
         one_last_digest BLOB,
         volume INTEGER,
         last_post_time TIMESTAMP,
@@ -77,7 +77,7 @@ CREATE TABLE mailinglist (
         digest_is_default BOOLEAN,
         digest_send_periodic BOOLEAN,
         digest_size_threshold INTEGER,
-        digest_volume_frequency INTEGER,
+        digest_volume_frequency TEXT,
         digestable BOOLEAN,
         discard_these_nonmembers BLOB,
         emergency BOOLEAN,
diff --git a/src/mailman/interfaces/mailinglist.py b/src/mailman/interfaces/mailinglist.py
index cd7b11d64..2885f60ab 100644
--- a/src/mailman/interfaces/mailinglist.py
+++ b/src/mailman/interfaces/mailinglist.py
@@ -21,6 +21,7 @@ from __future__ import absolute_import, unicode_literals
 
 __metaclass__ = type
 __all__ = [
+    'DigestFrequency',
     'IMailingList',
     'Personalization',
     'ReplyToMunging',
@@ -42,7 +43,6 @@ class Personalization(Enum):
     full = 2
 
 
-
 class ReplyToMunging(Enum):
     # The Reply-To header is passed through untouched
     no_munging = 0
@@ -52,6 +52,14 @@ class ReplyToMunging(Enum):
     explicit_header = 2
 
 
+class DigestFrequency(Enum):
+    yearly = 0
+    monthly = 1
+    quarterly = 2
+    weekly = 3
+    daily = 4
+
+
 
 class IMailingList(Interface):
     """A mailing list."""
@@ -155,7 +163,7 @@ class IMailingList(Interface):
         """A monotonically increasing integer sequentially assigned to each
         list posting.""")
 
-    last_digest_date = Attribute(
+    digest_last_sent_at = Attribute(
         """The date and time a digest of this mailing list was last sent.""")
 
     owners = Attribute(
@@ -197,27 +205,23 @@ class IMailingList(Interface):
         role.
         """)
 
-    volume_number = Attribute(
+    volume = Attribute(
         """A monotonically increasing integer sequentially assigned to each
         new digest volume.  The volume number may be bumped either
         automatically (i.e. on a defined schedule) or manually.  When the
         volume number is bumped, the digest number is always reset to 1.""")
 
-    digest_number = Attribute(
+    next_digest_number = Attribute(
         """A sequence number for a specific digest in a given volume.  When
         the digest volume number is bumped, the digest number is reset to
         1.""")
 
-    def bump():
-        """Bump the digest's volume number to the next integer in the
-        sequence, and reset the digest number to 1.
-        """
     message_count = Attribute(
         """The number of messages in the digest currently being collected.""")
 
-    digest_size = Attribute(
-        """The approximate size in kilobytes of the digest currently being
-        collected.""")
+    digest_size_threshold = Attribute(
+        """The maximum (approximate) size in kilobytes of the digest currently
+        being collected.""")
 
     messages = Attribute(
         """An iterator over all the messages in the digest currently being
diff --git a/src/mailman/pipeline/docs/digests.txt b/src/mailman/pipeline/docs/digests.txt
index cb939f7ca..6aafef62c 100644
--- a/src/mailman/pipeline/docs/digests.txt
+++ b/src/mailman/pipeline/docs/digests.txt
@@ -6,13 +6,7 @@ as individual messages when immediately posted.  There are several forms of
 digests, although only two are currently supported: MIME digests and RFC 1153
 (a.k.a. plain text) digests.
 
-    >>> from mailman.pipeline.to_digest import process
-    >>> mlist = config.db.list_manager.create(u'_xtest@example.com')
-    >>> mlist.preferred_language = u'en'
-    >>> mlist.real_name = u'XTest'
-    >>> mlist.subject_prefix = u'[_XTest] '
-    >>> mlist.one_last_digest = set()
-    >>> switchboard = config.switchboards['virgin']
+    >>> mlist = create_list('xtest@example.com')
 
 This is a helper function used to iterate through all the accumulated digest
 messages, in the order in which they were posted.  This makes it easier to
@@ -21,515 +15,97 @@ update the tests when we switch to a different mailbox format.
     >>> from mailman.testing.helpers import digest_mbox
     >>> from itertools import count
     >>> from string import Template
-    >>> def makemsg():
+
+    >>> def message_factory():
     ...     for i in count(1):
     ...         text = Template("""\
     ... From: aperson@example.com
-    ... To: _xtest@example.com
+    ... To: xtest@example.com
     ... Subject: Test message $i
     ...
     ... Here is message $i
     ... """).substitute(i=i)
     ...         yield message_from_string(text)
+    >>> message_factory = message_factory()
 
 
 Short circuiting
 ----------------
 
 When a message is posted to the mailing list, it is generally added to a
-running collection of messages.  For now, this is a Unix mailbox file,
-although in the future this may end up being converted to a maildir style
-mailbox.  In any event, there are several factors that would bypass the
-storing of posted messages to the mailbox.  For example, the mailing list may
-not allow digests...
+mailbox, unless the mailing list does not allow digests.
 
     >>> mlist.digestable = False
-    >>> msg = makemsg().next()
+    >>> msg = next(message_factory)
+    >>> process = config.handlers['to-digest'].process
     >>> process(mlist, msg, {})
-    >>> sum(1 for mboxmsg in digest_mbox(mlist))
+    >>> sum(1 for msg in digest_mbox(mlist))
     0
-    >>> switchboard.files
+    >>> digest_queue = config.switchboards['digest']
+    >>> digest_queue.files
     []
 
 ...or they may allow digests but the message is already a digest.
 
     >>> mlist.digestable = True
     >>> process(mlist, msg, dict(isdigest=True))
-    >>> sum(1 for mboxmsg in digest_mbox(mlist))
+    >>> sum(1 for msg in digest_mbox(mlist))
     0
-    >>> switchboard.files
+    >>> digest_queue.files
     []
 
 
 Sending a digest
 ----------------
 
-For messages which are not digests, but which are posted to a digestable
+For messages which are not digests, but which are posted to a digesting
 mailing list, the messages will be stored until they reach a criteria
 triggering the sending of the digest.  If none of those criteria are met, then
 the message will just sit in the mailbox for a while.
 
     >>> mlist.digest_size_threshold = 10000
     >>> process(mlist, msg, {})
-    >>> switchboard.files
+    >>> digest_queue.files
     []
     >>> digest = digest_mbox(mlist)
-    >>> sum(1 for mboxmsg in digest)
+    >>> sum(1 for msg in digest)
     1
     >>> import os
     >>> os.remove(digest._path)
 
-When the size of the digest mbox reaches the maximum size threshold, a digest
-is crafted and sent out.  This puts two messages in the virgin queue, an HTML
-digest and an RFC 1153 plain text digest.  The size threshold is in KB.
+When the size of the digest mailbox reaches the maximum size threshold, a
+marker message is placed into the digest runner's queue.  The digest is not
+actually crafted by the handler.
 
     >>> mlist.digest_size_threshold = 1
     >>> mlist.volume = 2
     >>> mlist.next_digest_number = 10
     >>> size = 0
-    >>> for msg in makemsg():
+    >>> for msg in message_factory:
     ...     process(mlist, msg, {})
     ...     size += len(str(msg))
-    ...     if size > mlist.digest_size_threshold * 1024:
+    ...     if size >= mlist.digest_size_threshold * 1024:
     ...         break
-    >>> sum(1 for mboxmsg in digest_mbox(mlist))
-    0
-    >>> len(switchboard.files)
-    2
-    >>> for filebase in switchboard.files:
-    ...     qmsg, qdata = switchboard.dequeue(filebase)
-    ...     switchboard.finish(filebase)
-    ...     if qmsg.is_multipart():
-    ...         mimemsg = qmsg
-    ...         mimedata = qdata
-    ...     else:
-    ...         rfc1153msg = qmsg
-    ...         rfc1153data = qdata
-    >>> print mimemsg.as_string()
-    Content-Type: multipart/mixed; boundary="..."
-    MIME-Version: 1.0
-    From: _xtest-request@example.com
-    Subject: XTest Digest, Vol 2, Issue 10
-    To: _xtest@example.com
-    Reply-To: _xtest@example.com
-    Date: ...
-    Message-ID: ...
-    <BLANKLINE>
-    --...
-    Content-Type: text/plain; charset="us-ascii"
-    MIME-Version: 1.0
-    Content-Transfer-Encoding: 7bit
-    Content-Description: XTest Digest, Vol 2, Issue 10
-    <BLANKLINE>
-    Send XTest mailing list submissions to
-        _xtest@example.com
-    <BLANKLINE>
-    To subscribe or unsubscribe via the World Wide Web, visit
-        http://lists.example.com/listinfo/_xtest@example.com
-    or, via email, send a message with subject or body 'help' to
-        _xtest-request@example.com
-    <BLANKLINE>
-    You can reach the person managing the list at
-        _xtest-owner@example.com
-    <BLANKLINE>
-    When replying, please edit your Subject line so it is more specific
-    than "Re: Contents of XTest digest..."
-    <BLANKLINE>
-    --...
-    Content-Type: text/plain; charset="us-ascii"
-    MIME-Version: 1.0
-    Content-Transfer-Encoding: 7bit
-    Content-Description: Today's Topics (8 messages)
-    <BLANKLINE>
-    Today's Topics:
-    <BLANKLINE>
-       1. Test message 1 (aperson@example.com)
-       2. Test message 2 (aperson@example.com)
-       3. Test message 3 (aperson@example.com)
-       4. Test message 4 (aperson@example.com)
-       5. Test message 5 (aperson@example.com)
-       6. Test message 6 (aperson@example.com)
-       7. Test message 7 (aperson@example.com)
-       8. Test message 8 (aperson@example.com)
-    <BLANKLINE>
-    --...
-    Content-Type: multipart/digest; boundary="..."
-    MIME-Version: 1.0
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    From: aperson@example.com
-    To: _xtest@example.com
-    Subject: Test message 1
-    Message: 1
-    <BLANKLINE>
-    Here is message 1
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    From: aperson@example.com
-    To: _xtest@example.com
-    Subject: Test message 2
-    Message: 2
-    <BLANKLINE>
-    Here is message 2
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    From: aperson@example.com
-    To: _xtest@example.com
-    Subject: Test message 3
-    Message: 3
-    <BLANKLINE>
-    Here is message 3
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    From: aperson@example.com
-    To: _xtest@example.com
-    Subject: Test message 4
-    Message: 4
-    <BLANKLINE>
-    Here is message 4
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    From: aperson@example.com
-    To: _xtest@example.com
-    Subject: Test message 5
-    Message: 5
-    <BLANKLINE>
-    Here is message 5
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    From: aperson@example.com
-    To: _xtest@example.com
-    Subject: Test message 6
-    Message: 6
-    <BLANKLINE>
-    Here is message 6
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    From: aperson@example.com
-    To: _xtest@example.com
-    Subject: Test message 7
-    Message: 7
-    <BLANKLINE>
-    Here is message 7
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    From: aperson@example.com
-    To: _xtest@example.com
-    Subject: Test message 8
-    Message: 8
-    <BLANKLINE>
-    Here is message 8
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    --...
-    >>> dump_msgdata(mimedata)
-    _parsemsg: False
-    isdigest : True
-    listname : _xtest@example.com
-    recips   : set([])
-    version  : 3
-
-
-    >>> print rfc1153msg.as_string()
-    From: _xtest-request@example.com
-    Subject: XTest Digest, Vol 2, Issue 10
-    To: _xtest@example.com
-    Reply-To: _xtest@example.com
-    Date: ...
-    Message-ID: ...
-    MIME-Version: 1.0
-    Content-Type: text/plain; charset="us-ascii"
-    Content-Transfer-Encoding: 7bit
-    <BLANKLINE>
-    Send XTest mailing list submissions to
-        _xtest@example.com
-    <BLANKLINE>
-    To subscribe or unsubscribe via the World Wide Web, visit
-        http://lists.example.com/listinfo/_xtest@example.com
-    or, via email, send a message with subject or body 'help' to
-        _xtest-request@example.com
-    <BLANKLINE>
-    You can reach the person managing the list at
-        _xtest-owner@example.com
-    <BLANKLINE>
-    When replying, please edit your Subject line so it is more specific
-    than "Re: Contents of XTest digest..."
-    <BLANKLINE>
-    <BLANKLINE>
-    Today's Topics:
-    <BLANKLINE>
-       1. Test message 1 (aperson@example.com)
-       2. Test message 2 (aperson@example.com)
-       3. Test message 3 (aperson@example.com)
-       4. Test message 4 (aperson@example.com)
-       5. Test message 5 (aperson@example.com)
-       6. Test message 6 (aperson@example.com)
-       7. Test message 7 (aperson@example.com)
-       8. Test message 8 (aperson@example.com)
-    <BLANKLINE>
-    <BLANKLINE>
-    ----------------------------------------------------------------------
-    <BLANKLINE>
-    Message: 1
-    From: aperson@example.com
-    Subject: Test message 1
-    To: _xtest@example.com
-    Message-ID: ...
-    <BLANKLINE>
-    Here is message 1
-    <BLANKLINE>
-    <BLANKLINE>
-    ------------------------------
-    <BLANKLINE>
-    Message: 2
-    From: aperson@example.com
-    Subject: Test message 2
-    To: _xtest@example.com
-    Message-ID: ...
-    <BLANKLINE>
-    Here is message 2
-    <BLANKLINE>
-    <BLANKLINE>
-    ------------------------------
-    <BLANKLINE>
-    Message: 3
-    From: aperson@example.com
-    Subject: Test message 3
-    To: _xtest@example.com
-    Message-ID: ...
-    <BLANKLINE>
-    Here is message 3
-    <BLANKLINE>
-    <BLANKLINE>
-    ------------------------------
-    <BLANKLINE>
-    Message: 4
-    From: aperson@example.com
-    Subject: Test message 4
-    To: _xtest@example.com
-    Message-ID: ...
-    <BLANKLINE>
-    Here is message 4
-    <BLANKLINE>
-    <BLANKLINE>
-    ------------------------------
-    <BLANKLINE>
-    Message: 5
-    From: aperson@example.com
-    Subject: Test message 5
-    To: _xtest@example.com
-    Message-ID: ...
-    <BLANKLINE>
-    Here is message 5
-    <BLANKLINE>
-    <BLANKLINE>
-    ------------------------------
-    <BLANKLINE>
-    Message: 6
-    From: aperson@example.com
-    Subject: Test message 6
-    To: _xtest@example.com
-    Message-ID: ...
-    <BLANKLINE>
-    Here is message 6
-    <BLANKLINE>
-    <BLANKLINE>
-    ------------------------------
-    <BLANKLINE>
-    Message: 7
-    From: aperson@example.com
-    Subject: Test message 7
-    To: _xtest@example.com
-    Message-ID: ...
-    <BLANKLINE>
-    Here is message 7
-    <BLANKLINE>
-    <BLANKLINE>
-    ------------------------------
-    <BLANKLINE>
-    Message: 8
-    From: aperson@example.com
-    Subject: Test message 8
-    To: _xtest@example.com
-    Message-ID: ...
-    <BLANKLINE>
-    Here is message 8
-    <BLANKLINE>
-    <BLANKLINE>
-    End of XTest Digest, Vol 2, Issue 10
-    ************************************
-    <BLANKLINE>
-    >>> dump_msgdata(rfc1153data)
-    _parsemsg: False
-    isdigest : True
-    listname : _xtest@example.com
-    recips   : set([])
-    version  : 3
-
-
-Internationalized digests
--------------------------
-
-When messages come in with a content-type character set different than that of
-the list's preferred language, recipients will get an internationalized
-digest.  French is not enabled by default site-wide, so enable that now.
-
-    >>> config.languages.enable_language('fr')
 
-    # Simulate the site administrator setting the default server language to
-    # French in the configuration file.  Without this, the English template
-    # will be found and the masthead won't be translated.
-    >>> config.push('french', """
-    ... [mailman]
-    ... default_language: fr
-    ... """)
+    >>> sum(1 for msg in digest_mbox(mlist))
+    0
+    >>> len(digest_queue.files)
+    1
 
-    >>> mlist.preferred_language = u'fr'
-    >>> msg = message_from_string("""\
-    ... From: aperson@example.org
-    ... To: _xtest@example.com
-    ... Subject: =?iso-2022-jp?b?GyRCMGxIVhsoQg==?=
-    ... MIME-Version: 1.0
-    ... Content-Type: text/plain; charset=iso-2022-jp
-    ... Content-Transfer-Encoding: 7bit
-    ...
-    ... \x1b$B0lHV\x1b(B
-    ... """)
+The digest has been moved to a unique file.
 
-Set the digest threshold to zero so that the digests will be sent immediately.
+    >>> from mailman.utilities.mailbox import Mailbox
+    >>> from mailman.testing.helpers import get_queue_messages
+    >>> item = get_queue_messages('digest')[0]
+    >>> for msg in Mailbox(item.msgdata['digest_path']):
+    ...     print msg['subject']
+    Test message 2
+    Test message 3
+    Test message 4
+    Test message 5
+    Test message 6
+    Test message 7
+    Test message 8
+    Test message 9
 
-    >>> mlist.digest_size_threshold = 0
-    >>> process(mlist, msg, {})
-    >>> sum(1 for mboxmsg in digest_mbox(mlist))
-    0
-    >>> len(switchboard.files)
-    2
-    >>> for filebase in switchboard.files:
-    ...     qmsg, qdata = switchboard.dequeue(filebase)
-    ...     switchboard.finish(filebase)
-    ...     if qmsg.is_multipart():
-    ...         mimemsg = qmsg
-    ...         mimedata = qdata
-    ...     else:
-    ...         rfc1153msg = qmsg
-    ...         rfc1153data = qdata
-    >>> print mimemsg.as_string()
-    Content-Type: multipart/mixed; boundary="..."
-    MIME-Version: 1.0
-    From: _xtest-request@example.com
-    Subject: Groupe XTest, Vol. 2, Parution 11
-    To: _xtest@example.com
-    Reply-To: _xtest@example.com
-    Date: ...
-    Message-ID: ...
-    <BLANKLINE>
-    --...
-    Content-Type: text/plain; charset="iso-8859-1"
-    MIME-Version: 1.0
-    Content-Transfer-Encoding: quoted-printable
-    Content-Description: Groupe XTest, Vol. 2, Parution 11
-    <BLANKLINE>
-    Envoyez vos messages pour la liste XTest =E0
-        _xtest@example.com
-    <BLANKLINE>
-    Pour vous (d=E9s)abonner par le web, consultez
-        http://lists.example.com/listinfo/_xtest@example.com
-    <BLANKLINE>
-    ou, par courriel, envoyez un message avec =AB=A0help=A0=BB dans le corps ou
-    dans le sujet =E0
-        _xtest-request@example.com
-    <BLANKLINE>
-    Vous pouvez contacter l'administrateur de la liste =E0 l'adresse
-        _xtest-owner@example.com
-    <BLANKLINE>
-    Si vous r=E9pondez, n'oubliez pas de changer l'objet du message afin
-    qu'il soit plus sp=E9cifique que =AB=A0Re: Contenu du groupe de XTest...=A0=
-    =BB
-    <BLANKLINE>
-    --...
-    Content-Type: text/plain; charset="utf-8"
-    MIME-Version: 1.0
-    Content-Transfer-Encoding: base64
-    Content-Description: Today's Topics (1 messages)
-    <BLANKLINE>
-    VGjDqG1lcyBkdSBqb3VyIDoKCiAgIDEuIOS4gOeVqiAoYXBlcnNvbkBleGFtcGxlLm9yZykK
-    <BLANKLINE>
-    --...
-    Content-Type: multipart/digest; boundary="..."
-    MIME-Version: 1.0
-    <BLANKLINE>
-    --...
-    Content-Type: message/rfc822
-    MIME-Version: 1.0
-    <BLANKLINE>
-    Content-Transfer-Encoding: 7bit
-    From: aperson@example.org
-    MIME-Version: 1.0
-    To: _xtest@example.com
-    Content-Type: text/plain; charset=iso-2022-jp
-    Subject: =?iso-2022-jp?b?GyRCMGxIVhsoQg==?=
-    Message: 1
-    <BLANKLINE>
-    一番
-    <BLANKLINE>
-    <BLANKLINE>
-    --...
-    --...
-    >>> dump_msgdata(mimedata)
-    _parsemsg: False
-    isdigest : True
-    listname : _xtest@example.com
-    recips   : set([])
-    version  : 3
-
-    >>> print rfc1153msg.as_string()
-    From: _xtest-request@example.com
-    Subject: Groupe XTest, Vol. 2, Parution 11
-    To: _xtest@example.com
-    Reply-To: _xtest@example.com
-    Date: ...
-    Message-ID: ...
-    MIME-Version: 1.0
-    Content-Type: text/plain; charset="utf-8"
-    Content-Transfer-Encoding: base64
-    <BLANKLINE>
-    ...
-    <BLANKLINE>
-    >>> dump_msgdata(rfc1153data)
-    _parsemsg: False
-    isdigest : True
-    listname : _xtest@example.com
-    recips   : set([])
-    version  : 3
+Digests are actually crafted and sent by a separate digest queue runner.
diff --git a/src/mailman/pipeline/to_digest.py b/src/mailman/pipeline/to_digest.py
index b85764ac9..ebb40a77c 100644
--- a/src/mailman/pipeline/to_digest.py
+++ b/src/mailman/pipeline/to_digest.py
@@ -15,15 +15,7 @@
 # You should have received a copy of the GNU General Public License along with
 # GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
 
-"""Add the message to the list's current digest and possibly send it."""
-
-# Messages are accumulated to a Unix mailbox compatible file containing all
-# the messages destined for the digest.  This file must be parsable by the
-# mailbox.UnixMailbox class (i.e. it must be ^From_ quoted).
-#
-# When the file reaches the size threshold, it is moved to the qfiles/digest
-# directory and the DigestRunner will craft the MIME, rfc1153, and
-# (eventually) URL-subject linked digests from the mbox.
+"""Add the message to the list's current digest."""
 
 from __future__ import absolute_import, unicode_literals
 
@@ -34,396 +26,16 @@ __all__ = [
 
 
 import os
-import re
-import copy
-import time
-import logging
+import datetime
 
-from StringIO import StringIO          # cStringIO can't handle unicode.
-from email.charset import Charset
-from email.generator import Generator
-from email.header import decode_header, make_header, Header
-from email.mime.base import MIMEBase
-from email.mime.message import MIMEMessage
-from email.mime.text import MIMEText
-from email.parser import Parser
-from email.utils import formatdate, getaddresses, make_msgid
 from zope.interface import implements
 
-from mailman import Message
-from mailman import Utils
-from mailman import i18n
-from mailman.Mailbox import Mailbox
-from mailman.Mailbox import Mailbox
+from mailman.Message import Message
 from mailman.config import config
-from mailman.core import errors
+from mailman.i18n import _
 from mailman.interfaces.handler import IHandler
-from mailman.interfaces.member import DeliveryMode, DeliveryStatus
-from mailman.pipeline.decorate import decorate
-from mailman.pipeline.scrubber import process as scrubber
-
-
-_ = i18n._
-
-UEMPTYSTRING = ''
-EMPTYSTRING = ''
-
-log = logging.getLogger('mailman.error')
-
-
-
-def process(mlist, msg, msgdata):
-    # Short circuit non-digestable lists.
-    if not mlist.digestable or msgdata.get('isdigest'):
-        return
-    mboxfile = os.path.join(mlist.data_path, 'digest.mbox')
-    mboxfp = open(mboxfile, 'a+')
-    mbox = Mailbox(mboxfp)
-    mbox.AppendMessage(msg)
-    # Calculate the current size of the accumulation file.  This will not tell
-    # us exactly how big the MIME, rfc1153, or any other generated digest
-    # message will be, but it's the most easily available metric to decide
-    # whether the size threshold has been reached.
-    mboxfp.flush()
-    size = os.path.getsize(mboxfile)
-    if size / 1024.0 >= mlist.digest_size_threshold:
-        # This is a bit of a kludge to get the mbox file moved to the digest
-        # queue directory.
-        try:
-            # Enclose in try/except here because a error in send_digest() can
-            # silently stop regular delivery.  Unsuccessful digest delivery
-            # should be tried again by cron and the site administrator will be
-            # notified of any error explicitly by the cron error message.
-            mboxfp.seek(0)
-            send_digests(mlist, mboxfp)
-            os.unlink(mboxfile)
-        except Exception, errmsg:
-            # Bare except is generally prohibited in Mailman, but we can't
-            # forecast what exceptions can occur here.
-            log.exception('send_digests() failed: %s', errmsg)
-    mboxfp.close()
-
-
-
-def send_digests(mlist, mboxfp):
-    # Set the digest volume and time
-    if mlist.digest_last_sent_at:
-        bump = False
-        # See if we should bump the digest volume number
-        timetup = time.localtime(mlist.digest_last_sent_at)
-        now = time.localtime(time.time())
-        freq = mlist.digest_volume_frequency
-        if freq == 0 and timetup[0] < now[0]:
-            # Yearly
-            bump = True
-        elif freq == 1 and timetup[1] <> now[1]:
-            # Monthly, but we take a cheap way to calculate this.  We assume
-            # that the clock isn't going to be reset backwards.
-            bump = True
-        elif freq == 2 and (timetup[1] % 4 <> now[1] % 4):
-            # Quarterly, same caveat
-            bump = True
-        elif freq == 3:
-            # Once again, take a cheap way of calculating this
-            weeknum_last = int(time.strftime('%W', timetup))
-            weeknum_now = int(time.strftime('%W', now))
-            if weeknum_now > weeknum_last or timetup[0] > now[0]:
-                bump = True
-        elif freq == 4 and timetup[7] <> now[7]:
-            # Daily
-            bump = True
-        if bump:
-            mlist.bump_digest_volume()
-    mlist.digest_last_sent_at = time.time()
-    # Wrapper around actually digest crafter to set up the language context
-    # properly.  All digests are translated to the list's preferred language.
-    with i18n.using_language(mlist.preferred_language):
-        send_i18n_digests(mlist, mboxfp)
-
-
-
-def send_i18n_digests(mlist, mboxfp):
-    mbox = Mailbox(mboxfp)
-    # Prepare common information (first lang/charset)
-    lang = mlist.preferred_language
-    lcset = Utils.GetCharSet(lang)
-    lcset_out = Charset(lcset).output_charset or lcset
-    # Common Information (contd)
-    realname = mlist.real_name
-    volume = mlist.volume
-    issue = mlist.next_digest_number
-    digestid = _('$realname Digest, Vol $volume, Issue $issue')
-    digestsubj = Header(digestid, lcset, header_name='Subject')
-    # Set things up for the MIME digest.  Only headers not added by
-    # CookHeaders need be added here.
-    # Date/Message-ID should be added here also.
-    mimemsg = Message.Message()
-    mimemsg['Content-Type'] = 'multipart/mixed'
-    mimemsg['MIME-Version'] = '1.0'
-    mimemsg['From'] = mlist.request_address
-    mimemsg['Subject'] = digestsubj
-    mimemsg['To'] = mlist.posting_address
-    mimemsg['Reply-To'] = mlist.posting_address
-    mimemsg['Date'] = formatdate(localtime=1)
-    mimemsg['Message-ID'] = make_msgid()
-    # Set things up for the rfc1153 digest
-    plainmsg = StringIO()
-    rfc1153msg = Message.Message()
-    rfc1153msg['From'] = mlist.request_address
-    rfc1153msg['Subject'] = digestsubj
-    rfc1153msg['To'] = mlist.posting_address
-    rfc1153msg['Reply-To'] = mlist.posting_address
-    rfc1153msg['Date'] = formatdate(localtime=1)
-    rfc1153msg['Message-ID'] = make_msgid()
-    separator70 = '-' * 70
-    separator30 = '-' * 30
-    # In the rfc1153 digest, the masthead contains the digest boilerplate plus
-    # any digest header.  In the MIME digests, the masthead and digest header
-    # are separate MIME subobjects.  In either case, it's the first thing in
-    # the digest, and we can calculate it now, so go ahead and add it now.
-    mastheadtxt = Utils.maketext(
-        'masthead.txt',
-        {'real_name' :        mlist.real_name,
-         'got_list_email':    mlist.posting_address,
-         'got_listinfo_url':  mlist.script_url('listinfo'),
-         'got_request_email': mlist.request_address,
-         'got_owner_email':   mlist.owner_address,
-         }, mlist=mlist)
-    # MIME
-    masthead = MIMEText(mastheadtxt.encode(lcset), _charset=lcset)
-    masthead['Content-Description'] = digestid
-    mimemsg.attach(masthead)
-    # RFC 1153
-    print >> plainmsg, mastheadtxt
-    print >> plainmsg
-    # Now add the optional digest header
-    if mlist.digest_header:
-        headertxt = decorate(mlist, mlist.digest_header, _('digest header'))
-        # MIME
-        header = MIMEText(headertxt.encode(lcset), _charset=lcset)
-        header['Content-Description'] = _('Digest Header')
-        mimemsg.attach(header)
-        # RFC 1153
-        print >> plainmsg, headertxt
-        print >> plainmsg
-    # Now we have to cruise through all the messages accumulated in the
-    # mailbox file.  We can't add these messages to the plainmsg and mimemsg
-    # yet, because we first have to calculate the table of contents
-    # (i.e. grok out all the Subjects).  Store the messages in a list until
-    # we're ready for them.
-    #
-    # Meanwhile prepare things for the table of contents
-    toc = StringIO()
-    print >> toc, _("Today's Topics:\n")
-    # Now cruise through all the messages in the mailbox of digest messages,
-    # building the MIME payload and core of the RFC 1153 digest.  We'll also
-    # accumulate Subject: headers and authors for the table-of-contents.
-    messages = []
-    msgcount = 0
-    msg = mbox.next()
-    while msg is not None:
-        if msg == '':
-            # It was an unparseable message
-            msg = mbox.next()
-            continue
-        msgcount += 1
-        messages.append(msg)
-        # Get the Subject header
-        msgsubj = msg.get('subject', _('(no subject)'))
-        subject = Utils.oneline(msgsubj, in_unicode=True)
-        # Don't include the redundant subject prefix in the toc
-        mo = re.match('(re:? *)?(%s)' % re.escape(mlist.subject_prefix),
-                      subject, re.IGNORECASE)
-        if mo:
-            subject = subject[:mo.start(2)] + subject[mo.end(2):]
-        username = ''
-        addresses = getaddresses([Utils.oneline(msg.get('from', ''),
-                                                in_unicode=True)])
-        # Take only the first author we find
-        if isinstance(addresses, list) and addresses:
-            username = addresses[0][0]
-            if not username:
-                username = addresses[0][1]
-        if username:
-            username = ' ({0})'.format(username)
-        # Put count and Wrap the toc subject line
-        wrapped = Utils.wrap('{0:2}. {1}'.format(msgcount, subject), 65)
-        slines = wrapped.split('\n')
-        # See if the user's name can fit on the last line
-        if len(slines[-1]) + len(username) > 70:
-            slines.append(username)
-        else:
-            slines[-1] += username
-        # Add this subject to the accumulating topics
-        first = True
-        for line in slines:
-            if first:
-                print >> toc, ' ', line
-                first = False
-            else:
-                print >> toc, '     ', line.lstrip()
-        # We do not want all the headers of the original message to leak
-        # through in the digest messages.  For this phase, we'll leave the
-        # same set of headers in both digests, i.e. those required in RFC 1153
-        # plus a couple of other useful ones.  We also need to reorder the
-        # headers according to RFC 1153.  Later, we'll strip out headers for
-        # for the specific MIME or plain digests.
-        keeper = {}
-        all_keepers = set(
-            header for header in
-            config.digests.mime_digest_keep_headers.split() +
-            config.digests.plain_digest_keep_headers.split())
-        for keep in all_keepers:
-            keeper[keep] = msg.get_all(keep, [])
-        # Now remove all unkempt headers :)
-        for header in msg.keys():
-            del msg[header]
-        # And add back the kept header in the RFC 1153 designated order
-        for keep in all_keepers:
-            for field in keeper[keep]:
-                msg[keep] = field
-        # And a bit of extra stuff
-        msg['Message'] = repr(msgcount)
-        # Get the next message in the digest mailbox
-        msg = mbox.next()
-    # Now we're finished with all the messages in the digest.  First do some
-    # sanity checking and then on to adding the toc.
-    if msgcount == 0:
-        # Why did we even get here?
-        return
-    toctext = toc.getvalue()
-    # MIME
-    try:
-        tocpart = MIMEText(toctext.encode(lcset), _charset=lcset)
-    except UnicodeError:
-        tocpart = MIMEText(toctext.encode('utf-8'), _charset='utf-8')
-    tocpart['Content-Description']= _("Today's Topics ($msgcount messages)")
-    mimemsg.attach(tocpart)
-    # RFC 1153
-    print >> plainmsg, toctext
-    print >> plainmsg
-    # For RFC 1153 digests, we now need the standard separator
-    print >> plainmsg, separator70
-    print >> plainmsg
-    # Now go through and add each message
-    mimedigest = MIMEBase('multipart', 'digest')
-    mimemsg.attach(mimedigest)
-    first = True
-    for msg in messages:
-        # MIME.  Make a copy of the message object since the rfc1153
-        # processing scrubs out attachments.
-        mimedigest.attach(MIMEMessage(copy.deepcopy(msg)))
-        # rfc1153
-        if first:
-            first = False
-        else:
-            print >> plainmsg, separator30
-            print >> plainmsg
-        # Use Mailman.pipeline.scrubber.process() to get plain text
-        try:
-            msg = scrubber(mlist, msg)
-        except errors.DiscardMessage:
-            print >> plainmsg, _('[Message discarded by content filter]')
-            continue
-        # Honor the default setting
-        for h in config.digests.plain_digest_keep_headers.split():
-            if msg[h]:
-                uh = Utils.wrap('{0}: {1}'.format(
-                    h, Utils.oneline(msg[h], in_unicode=True)))
-                uh = '\n\t'.join(uh.split('\n'))
-                print >> plainmsg, uh
-        print >> plainmsg
-        # If decoded payload is empty, this may be multipart message.
-        # -- just stringfy it.
-        payload = msg.get_payload(decode=True) \
-                  or msg.as_string().split('\n\n',1)[1]
-        mcset = msg.get_content_charset('us-ascii')
-        try:
-            payload = unicode(payload, mcset, 'replace')
-        except (LookupError, TypeError):
-            # unknown or empty charset
-            payload = unicode(payload, 'us-ascii', 'replace')
-        print >> plainmsg, payload
-        if not payload.endswith('\n'):
-            print >> plainmsg
-    # Now add the footer
-    if mlist.digest_footer:
-        footertxt = decorate(mlist, mlist.digest_footer)
-        # MIME
-        footer = MIMEText(footertxt.encode(lcset), _charset=lcset)
-        footer['Content-Description'] = _('Digest Footer')
-        mimemsg.attach(footer)
-        # RFC 1153
-        # BAW: This is not strictly conformant RFC 1153.  The trailer is only
-        # supposed to contain two lines, i.e. the "End of ... Digest" line and
-        # the row of asterisks.  If this screws up MUAs, the solution is to
-        # add the footer as the last message in the RFC 1153 digest.  I just
-        # hate the way that VM does that and I think it's confusing to users,
-        # so don't do it unless there's a clamor.
-        print >> plainmsg, separator30
-        print >> plainmsg
-        print >> plainmsg, footertxt
-        print >> plainmsg
-    # Do the last bit of stuff for each digest type
-    signoff = _('End of ') + digestid
-    # MIME
-    # BAW: This stuff is outside the normal MIME goo, and it's what the old
-    # MIME digester did.  No one seemed to complain, probably because you
-    # won't see it in an MUA that can't display the raw message.  We've never
-    # got complaints before, but if we do, just wax this.  It's primarily
-    # included for (marginally useful) backwards compatibility.
-    mimemsg.postamble = signoff
-    # rfc1153
-    print >> plainmsg, signoff
-    print >> plainmsg, '*' * len(signoff)
-    # Do our final bit of housekeeping, and then send each message to the
-    # outgoing queue for delivery.
-    mlist.next_digest_number += 1
-    virginq = config.switchboards['virgin']
-    # Calculate the recipients lists
-    plainrecips = set()
-    mimerecips = set()
-    # When someone turns off digest delivery, they will get one last digest to
-    # ensure that there will be no gaps in the messages they receive.
-    # Currently, this dictionary contains the email addresses of those folks
-    # who should get one last digest.  We need to find the corresponding
-    # IMember records.
-    digest_members = set(mlist.digest_members.members)
-    for address in mlist.one_last_digest:
-        member = mlist.digest_members.get_member(address)
-        if member:
-            digest_members.add(member)
-    for member in digest_members:
-        if member.delivery_status <> DeliveryStatus.enabled:
-            continue
-        # Send the digest to the case-preserved address of the digest members.
-        email_address = member.address.original_address
-        if member.delivery_mode == DeliveryMode.plaintext_digests:
-            plainrecips.add(email_address)
-        elif member.delivery_mode == DeliveryMode.mime_digests:
-            mimerecips.add(email_address)
-        else:
-            raise AssertionError(
-                'Digest member "{0}" unexpected delivery mode: {1}'.format(
-                    email_address, member.delivery_mode))
-    # Zap this since we're now delivering the last digest to these folks.
-    mlist.one_last_digest.clear()
-    # MIME
-    virginq.enqueue(mimemsg,
-                    recips=mimerecips,
-                    listname=mlist.fqdn_listname,
-                    isdigest=True)
-    # RFC 1153
-    # If the entire digest message can't be encoded by list charset, fall
-    # back to 'utf-8'.
-    try:
-        rfc1153msg.set_payload(plainmsg.getvalue().encode(lcset), lcset)
-    except UnicodeError:
-        rfc1153msg.set_payload(plainmsg.getvalue().encode('utf-8'), 'utf-8')
-    virginq.enqueue(rfc1153msg,
-                    recips=plainrecips,
-                    listname=mlist.fqdn_listname,
-                    isdigest=True)
+from mailman.interfaces.mailinglist import DigestFrequency
+from mailman.utilities.mailbox import Mailbox
 
 
 
@@ -437,4 +49,75 @@ class ToDigest:
 
     def process(self, mlist, msg, msgdata):
         """See `IHandler`."""
-        process(mlist, msg, msgdata)
+        # Short circuit for non-digestable messages.
+        if not mlist.digestable or msgdata.get('isdigest'):
+            return
+        # Open the mailbox that will be used to collect the current digest.
+        mailbox_path = os.path.join(mlist.data_path, 'digest.mmdf')
+        # Lock the mailbox and append the message.
+        with Mailbox(mailbox_path, create=True) as mbox:
+            mbox.add(msg)
+        # Calculate the current size of the mailbox file.  This will not tell
+        # us exactly how big the resulting MIME and rfc1153 digest will
+        # actually be, but it's the most easily available metric to decide
+        # whether the size threshold has been reached.
+        size = os.path.getsize(mailbox_path)
+        if size >= mlist.digest_size_threshold * 1024.0:
+            # The digest is ready to send.  Because we don't want to hold up
+            # this process with crafting the digest, we're going to move the
+            # digest file to a safe place, then craft a fake message for the
+            # DigestRunner as a trigger for it to build and send the digest.
+            mailbox_dest = os.path.join(
+                mlist.data_path,
+                'digest.{0.volume}.{0.next_digest_number}.mmdf'.format(mlist))
+            volume = mlist.volume
+            digest_number = mlist.next_digest_number
+            bump(mlist)
+            os.rename(mailbox_path, mailbox_dest)
+            config.switchboards['digest'].enqueue(
+                Message(),
+                listname=mlist.fqdn_listname,
+                digest_path=mailbox_dest,
+                volume=volume,
+                digest_number=digest_number)
+
+
+
+def bump(mlist):
+    """Bump the digest number and volume."""
+    now = datetime.datetime.now()
+    if mlist.digest_last_sent_at is None:
+        # There has been no previous digest.
+        bump = False
+    elif mlist.digest_volume_frequency == DigestFrequency.yearly:
+        bump = (now.year > mlist.digest_last_sent_at.year)
+    elif mlist.digest_volume_frequency == DigestFrequency.monthly:
+        # Monthly.
+        this_month = now.year * 100 + now.month
+        digest_month = (mlist.digest_last_sent_at.year * 100 +
+                        mlist.digest_last_sent_at.month)
+        bump = (this_month > digest_month)
+    elif mlist.digest_volume_frequency == DigestFrequency.quarterly:
+        # Quarterly.
+        this_quarter = now.year * 100 + (now.month - 1) // 4
+        digest_quarter = (mlist.digest_last_sent_at.year * 100 +
+                          (mlist.digest_last_sent_at.month - 1) // 4)
+        bump = (this_quarter > digest_quarter)
+    elif mlist.digest_volume_frequency == DigestFrequency.weekly:
+        this_week = now.year * 100 + now.isocalendar()[1]
+        digest_week = (mlist.digest_last_sent_at.year * 100 +
+                       mlist.digest_last_sent_at.isocalendar()[1])
+        bump = (this_week > digest_week)
+    elif mlist.digest_volume_frequency == DigestFrequency.daily:
+        bump = (now.toordinal() > mlist.digest_last_sent_at.toordinal())
+    else:
+        raise AssertionError(
+            'Bad DigestFrequency: {0}'.format(
+                mlist.digest_volume_frequency))
+    if bump:
+        mlist.volume += 1
+        mlist.next_digest_number = 1
+    else:
+        # Just bump the digest number.
+        mlist.next_digest_number += 1
+    mlist.digest_last_sent_at = now
diff --git a/src/mailman/queue/__init__.py b/src/mailman/queue/__init__.py
index 6094bda9e..ead077e90 100644
--- a/src/mailman/queue/__init__.py
+++ b/src/mailman/queue/__init__.py
@@ -428,10 +428,13 @@ class Runner:
         # special care to reset the defaults, otherwise subsequent messages
         # may be translated incorrectly.
         sender = msg.get_sender()
-        member = mlist.members.get_member(sender)
-        language = (member.preferred_language
-                    if member is not None
-                    else mlist.preferred_language)
+        if sender:
+            member = mlist.members.get_member(sender)
+            language = (member.preferred_language
+                        if member is not None
+                        else mlist.preferred_language)
+        else:
+            language = mlist.preferred_language
         with i18n.using_language(language):
             msgdata['lang'] = language
             keepqueued = self._dispose(mlist, msg, msgdata)
diff --git a/src/mailman/queue/digest.py b/src/mailman/queue/digest.py
new file mode 100644
index 000000000..e066be993
--- /dev/null
+++ b/src/mailman/queue/digest.py
@@ -0,0 +1,365 @@
+# Copyright (C) 2009 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman 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 General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Digest queue runner."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'DigestRunner',
+    ]
+
+
+import re
+
+# cStringIO doesn't support unicode.
+from StringIO import StringIO
+from contextlib import nested
+from copy import deepcopy
+from email.header import Header
+from email.message import Message
+from email.mime.message import MIMEMessage
+from email.mime.multipart import MIMEMultipart
+from email.mime.text import MIMEText
+from email.utils import formatdate, getaddresses, make_msgid
+
+from mailman import i18n
+from mailman.Utils import GetCharSet, maketext, oneline, wrap
+from mailman.config import config
+from mailman.core.errors import DiscardMessage
+from mailman.i18n import _
+from mailman.pipeline.decorate import decorate
+from mailman.pipeline.scrubber import process as scrubber
+from mailman.queue import Runner
+from mailman.utilities.mailbox import Mailbox
+
+
+
+class Digester:
+    """Base digester class."""
+
+    def __init__(self, mlist, volume, digest_number):
+        self._mlist = mlist
+        self._charset = GetCharSet(mlist.preferred_language)
+        # This will be used in the Subject, so use $-strings.
+        realname = mlist.real_name
+        issue = digest_number
+        self._digest_id = _('$realname Digest, Vol $volume, Issue $issue')
+        self._subject = Header(self._digest_id,
+                               self._charset,
+                               header_name='Subject')
+        self._message = self._make_message()
+        self._message['From'] = mlist.request_address
+        self._message['Subject'] = self._subject
+        self._message['To'] = mlist.posting_address
+        self._message['Reply-To'] = mlist.posting_address
+        self._message['Date'] = formatdate(localtime=True)
+        self._message['Message-ID'] = make_msgid()
+        # In the rfc1153 digest, the masthead contains the digest boilerplate
+        # plus any digest header.  In the MIME digests, the masthead and
+        # digest header are separate MIME subobjects.  In either case, it's
+        # the first thing in the digest, and we can calculate it now, so go
+        # ahead and add it now.
+        self._masthead = maketext(
+            'masthead.txt', dict(
+                real_name=mlist.real_name,
+                got_list_email=mlist.posting_address,
+                got_listinfo_url=mlist.script_url('listinfo'),
+                got_request_email=mlist.request_address,
+                got_owner_email=mlist.owner_address,
+                ),
+            mlist=mlist)
+        # Set things up for the table of contents.
+        self._header = decorate(mlist, mlist.digest_header)
+        self._toc = StringIO()
+        print >> self._toc, _("Today's Topics:\n")
+
+    def add_to_toc(self, msg, count):
+        """Add a message to the table of contents."""
+        subject = msg.get('subject', _('(no subject)'))
+        subject = oneline(subject, in_unicode=True)
+        # Don't include the redundant subject prefix in the toc
+        mo = re.match('(re:? *)?({0})'.format(
+            re.escape(self._mlist.subject_prefix)),
+                      subject, re.IGNORECASE)
+        if mo:
+            subject = subject[:mo.start(2)] + subject[mo.end(2):]
+        # Take only the first author we find.
+        username = ''
+        addresses = getaddresses(
+            [oneline(msg.get('from', ''), in_unicode=True)])
+        if addresses:
+            username = addresses[0][0]
+            if not username:
+                username = addresses[0][1]
+        if username:
+            username = ' ({0})'.format(username)
+        lines = wrap('{0:2}. {1}'. format(count, subject), 65).split('\n')
+        # See if the user's name can fit on the last line
+        if len(lines[-1]) + len(username) > 70:
+            lines.append(username)
+        else:
+            lines[-1] += username
+        # Add this subject to the accumulating topics
+        first = True
+        for line in lines:
+            if first:
+                print >> self._toc, ' ', line
+                first = False
+            else:
+                print >> self._toc, '     ', line.lstrip()
+
+    def add_message(self, msg, count):
+        """Add the message to the digest."""
+        # We do not want all the headers of the original message to leak
+        # through in the digest messages.
+        keepers = {}
+        for header in self._keepers:
+            keepers[header] = msg.get_all(keeper, [])
+        # Remove all the unkempt <wink> headers.  Use .keys() to allow for
+        # destructive iteration...
+        for header in msg.keys():
+            del msg[header]
+        # ... and add them in the designated order.
+        for header in self._keepers:
+            for value in keepers[header]:
+                msg[header] = value
+        # Add some useful extra stuff.
+        msg['Message'] = unicode(count)
+
+
+
+class MIMEDigester(Digester):
+    """A MIME digester."""
+
+    def __init__(self, mlist, volume, digest_number):
+        super(MIMEDigester, self).__init__(mlist, volume, digest_number)
+        masthead = MIMEText(self._masthead.encode(self._charset),
+                            _charset=self._charset)
+        masthead['Content-Description'] = self._subject
+        self._message.attach(masthead)
+        # Add the optional digest header.
+        if mlist.digest_header:
+            header = MIMEText(self._header.encode(self._charset),
+                              _charset=self._charset)
+            header['Content-Description'] = _('Digest Header')
+            self._message.attach(header)
+        # Calculate the set of headers we're to keep in the MIME digest.
+        self._keepers = set(config.digests.mime_digest_keep_headers.split())
+
+    def _make_message(self):
+        return MIMEMultipart('mixed')
+
+    def add_toc(self, count):
+        """Add the table of contents."""
+        toc_text = self._toc.getvalue()
+        try:
+            toc_part = MIMEText(toc_text.encode(self._charset),
+                                _charset=self._charset)
+        except UnicodeError:
+            toc_part = MIMEText(toc_text.encode('utf-8'), _charset='utf-8')
+        toc_part['Content-Description']= _("Today's Topics ($count messages)")
+        self._message.attach(toc_part)
+
+    def add_message(self, msg, count):
+        """Add the message to the digest."""
+        # Make a copy of the message object, since the RFC 1153 processing
+        # scrubs out attachments.
+        self._message.attach(MIMEMessage(deepcopy(msg)))
+
+    def finish(self):
+        """Finish up the digest, producing the email-ready copy."""
+        if self._mlist.digest_footer:
+            footer_text = decorate(self._mlist, self._mlist.digest_footer)
+            footer = MIMEText(footer_text.encode(self._charset),
+                              _charset=self._charset)
+            footer['Content-Description'] = _('Digest Footer')
+            self._message.attach(footer)
+        # This stuff is outside the normal MIME goo, and it's what the old
+        # MIME digester did.  No one seemed to complain, probably because you
+        # won't see it in an MUA that can't display the raw message.  We've
+        # never got complaints before, but if we do, just wax this.  It's
+        # primarily included for (marginally useful) backwards compatibility.
+        self._message.postamble = _('End of ') + self._digest_id
+        return self._message
+
+
+
+class RFC1153Digester(Digester):
+    """A digester of the format specified by RFC 1153."""
+
+    def __init__(self, mlist, volume, digest_number):
+        super(RFC1153Digester, self).__init__(mlist, volume, digest_number)
+        self._separator70 = '-' * 70
+        self._separator30 = '-' * 30
+        self._text = StringIO()
+        print >> self._text, self._masthead
+        print >> self._text
+        # Add the optional digest header.
+        if mlist.digest_header:
+            print >> self._text, self._header
+            print >> self._text
+        # Calculate the set of headers we're to keep in the RFC1153 digest.
+        self._keepers = set(config.digests.plain_digest_keep_headers.split())
+
+    def _make_message(self):
+        return Message()
+
+    def add_toc(self, count):
+        """Add the table of contents."""
+        print >> self._text, self._toc.getvalue()
+        print >> self._text
+        print >> self._text, self._separator70
+        print >> self._text
+
+    def add_message(self, msg, count):
+        """Add the message to the digest."""
+        if count > 1:
+            print >> self._text, self._separator30
+            print >> self._text
+        # Scrub attachements.
+        try:
+            msg = scrubber(self._mlist, msg)
+        except DiscardMessage:
+            print >> self._text, _('[Message discarded by content filter]')
+            return
+        # Each message section contains a few headers.
+        for header in config.digests.plain_digest_keep_headers.split():
+            if header in msg:
+                value = oneline(msg[header], in_unicode=True)
+                value = wrap('{0}: {1}'.format(header, value))
+                value = '\n\t'.join(value.split('\n'))
+                print >> self._text, value
+        print >> self._text
+        # Add the payload.  If the decoded payload is empty, this may be a
+        # multipart message.  In that case, just stringify it.
+        payload = msg.get_payload(decode=True)
+        payload = (payload if payload else msg.as_string().split('\n\n', 1)[1])
+        try:
+            charset = msg.get_content_charset('us-ascii')
+            payload = unicode(payload, charset, 'replace')
+        except (LookupError, TypeError):
+            # Unknown or empty charset.
+            payload = unicode(payload, 'us-ascii', 'replace')
+        print >> self._text, payload
+        if not payload.endswith('\n'):
+            print >> self._text
+
+    def finish(self):
+        """Finish up the digest, producing the email-ready copy."""
+        if self._mlist.digest_footer:
+            footer_text = decorate(self._mlist, self._mlist.digest_footer)
+            # This is not strictly conformant RFC 1153.  The trailer is only
+            # supposed to contain two lines, i.e. the "End of ... Digest" line
+            # and the row of asterisks.  If this screws up MUAs, the solution
+            # is to add the footer as the last message in the RFC 1153 digest.
+            # I just hate the way that VM does that and I think it's confusing
+            # to users, so don't do it unless there's a clamor.
+            print >> self._text, self._separator30
+            print >> self._text
+            print >> self._text, footer_text
+            print >> self._text
+        # Add the sign-off.
+        sign_off = _('End of ') + self._digest_id
+        print >> self._text, sign_off
+        print >> self._text, '*' * len(sign_off)
+        # If the digest message can't be encoded by the list character set,
+        # fall back to utf-8.
+        text = self._text.getvalue()
+        try:
+            self._message.set_payload(text.encode(self._charset),
+                                      charset=self._charset)
+        except UnicodeError:
+            self._message.set_payload(text.encode('utf-8'), charset='utf-8')
+        return self._message
+
+
+
+class DigestRunner(Runner):
+    """The digest queue runner."""
+
+    def _dispose(self, mlist, msg, msgdata):
+        """See `IRunner`."""
+        volume = msgdata['volume']
+        digest_number = msgdata['digest_number']
+        with nested(Mailbox(msgdata['digest_path']),
+                    i18n.using_language(mlist.preferred_language)) as (
+            mailbox, language):
+            # Create the digesters.
+            mime_digest = MIMEDigester(mlist, volume, digest_number)
+            rfc1153_digest = RFC1153Digester(mlist, volume, digest_number)
+            # Cruise through all the messages in the mailbox, first building
+            # the table of contents and accumulating Subject: headers and
+            # authors.  The question really is whether it's better from a
+            # performance and memory footprint to go through the mailbox once
+            # and cache the messages in a list, or to cruise through the
+            # mailbox twice.  We'll do the latter, but it's a complete guess.
+            count = None
+            for count, (key, message) in enumerate(mailbox.iteritems(), 1):
+                mime_digest.add_to_toc(message, count)
+                rfc1153_digest.add_to_toc(message, count)
+            assert count is not None, 'No digest messages?'
+            # Add the table of contents.
+            mime_digest.add_toc(count)
+            rfc1153_digest.add_toc(count)
+            # Cruise through the set of messages a second time, adding them to
+            # the actual digest.
+            for count, (key, message) in enumerate(mailbox.iteritems(), 1):
+                mime_digest.add_message(message, count)
+                rfc1153_digest.add_message(message, count)
+            # Finish up the digests.
+            mime = mime_digest.finish()
+            rfc1153 = rfc1153_digest.finish()
+        # Calculate the recipients lists
+        mime_recipients = set()
+        rfc1153_recipients = set()
+        # When someone turns off digest delivery, they will get one last
+        # digest to ensure that there will be no gaps in the messages they
+        # receive.
+        digest_members = set(mlist.digest_members.members)
+        for address in mlist.one_last_digest:
+            member = mlist.digest_members.get_member(address)
+            if member:
+                digest_members.add(member)
+        for member in digest_members:
+            if member.delivery_status <> DeliveryStatus.enabled:
+                continue
+            # Send the digest to the case-preserved address of the digest
+            # members.
+            email_address = member.address.original_address
+            if member.delivery_mode == DeliveryMode.plaintext_digests:
+                rfc1153_recipients.add(email_address)
+            elif member.delivery_mode == DeliveryMode.mime_digests:
+                mime_recipients.add(email_address)
+            else:
+                raise AssertionError(
+                    'Digest member "{0}" unexpected delivery mode: {1}'.format(
+                        email_address, member.delivery_mode))
+        # Send the digests to the virgin queue for final delivery.
+        queue = config.switchboards['virgin']
+        queue.enqueue(mime,
+                      recips=mime_recipients,
+                      listname=mlist.fqdn_listname,
+                      isdigest=True)
+        queue.enqueue(rfc1153,
+                      recips=rfc1153_recipients,
+                      listname=mlist.fqdn_listname,
+                      isdigest=True)
+        # Now that we've delivered the last digest to folks who were waiting
+        # for it, clear that recipient set.
+        mlist.one_last_digest.clear()
diff --git a/src/mailman/queue/docs/digester.txt b/src/mailman/queue/docs/digester.txt
new file mode 100644
index 000000000..487549e90
--- /dev/null
+++ b/src/mailman/queue/docs/digester.txt
@@ -0,0 +1,484 @@
+Digesting
+=========
+
+Mailman crafts and sends digests by a separate digest queue runner process.
+This starts by a number of messages being posted to the mailing list.
+
+    >>> mlist = create_list('test@example.com')
+    >>> mlist.digest_size_threshold = 0.5
+    >>> mlist.volume = 1
+    >>> mlist.next_digest_number = 1
+    >>> size = 0
+
+    >>> from string import Template
+    >>> process = config.handlers['to-digest'].process
+    >>> for i in range(1, 5):
+    ...     text = Template("""\
+    ... From: aperson@example.com
+    ... To: xtest@example.com
+    ... Subject: Test message $i
+    ...
+    ... Here is message $i
+    ... """).substitute(i=i)
+    ...     msg = message_from_string(text)
+    ...     process(mlist, msg, {})
+    ...     size += len(text)
+    ...     if size >= mlist.digest_size_threshold * 1024:
+    ...         break
+
+The queue runner gets kicked off when a marker message gets dropped into the
+digest queue.  The message metadata points to the mailbox file containing the
+messages to put in the digest.
+
+    >>> digestq = config.switchboards['digest']
+    >>> len(digestq.files)
+    1
+
+    >>> from mailman.testing.helpers import get_queue_messages
+    >>> entry = get_queue_messages('digest')[0]
+
+The marker message is empty.
+
+    >>> print entry.msg.as_string()
+
+But the message metadata has a reference to the digest file.
+
+    >>> dump_msgdata(entry.msgdata)
+    _parsemsg    : False
+    digest_number: 1
+    digest_path  : .../lists/test@example.com/digest.1.1.mmdf
+    listname     : test@example.com
+    version      : 3
+    volume       : 1
+
+    # Put the messages back in the queue for the runner to handle.
+    >>> filebase = digestq.enqueue(entry.msg, entry.msgdata)
+
+There are 4 messages in the digest.
+
+    >>> from mailman.utilities.mailbox import Mailbox
+    >>> sum(1 for item in Mailbox(entry.msgdata['digest_path']))
+    4
+
+When the queue runner runs, it processes the digest mailbox, crafting both the
+plain text (RFC 1153) digest and the MIME digest.
+
+    >>> from mailman.queue.digest import DigestRunner
+    >>> from mailman.testing.helpers import make_testable_runner
+    >>> runner = make_testable_runner(DigestRunner)
+    >>> runner.run()
+
+The digest runner places both digests into the virgin queue for final
+delivery.
+
+    >>> messages = get_queue_messages('virgin')
+    >>> len(messages)
+    2
+
+The MIME digest is a multipart, and the RFC 1153 digest is the other one.
+
+    >>> if messages[0].msg.is_multipart():
+    ...     mime = messages[0].msg
+    ...     rfc1153 = messages[1].msg
+    ... else:
+    ...     mime = messages[1].msg
+    ...     rfc1153 = messages[0].msg
+
+The MIME digest has lots of good stuff, all contained in the multipart.
+
+    >>> print mime.as_string()
+    Content-Type: multipart/mixed; boundary="===============...=="
+    MIME-Version: 1.0
+    From: test-request@example.com
+    Subject: Test Digest, Vol 1, Issue 1
+    To: test@example.com
+    Reply-To: test@example.com
+    Date: ...
+    Message-ID: ...
+    <BLANKLINE>
+    --===============...==
+    Content-Type: text/plain; charset="us-ascii"
+    MIME-Version: 1.0
+    Content-Transfer-Encoding: 7bit
+    Content-Description: Test Digest, Vol 1, Issue 1
+    <BLANKLINE>
+    Send Test mailing list submissions to
+    	test@example.com
+    <BLANKLINE>
+    To subscribe or unsubscribe via the World Wide Web, visit
+    	http://lists.example.com/listinfo/test@example.com
+    or, via email, send a message with subject or body 'help' to
+    	test-request@example.com
+    <BLANKLINE>
+    You can reach the person managing the list at
+    	test-owner@example.com
+    <BLANKLINE>
+    When replying, please edit your Subject line so it is more specific
+    than "Re: Contents of Test digest..."
+    <BLANKLINE>
+    --===============...==
+    Content-Type: text/plain; charset="us-ascii"
+    MIME-Version: 1.0
+    Content-Transfer-Encoding: 7bit
+    Content-Description: Today's Topics (4 messages)
+    <BLANKLINE>
+    Today's Topics:
+    <BLANKLINE>
+       1. Test message 1 (aperson@example.com)
+       2. Test message 2 (aperson@example.com)
+       3. Test message 3 (aperson@example.com)
+       4. Test message 4 (aperson@example.com)
+    <BLANKLINE>
+    --===============...==
+    Content-Type: message/rfc822
+    MIME-Version: 1.0
+    <BLANKLINE>
+    From: aperson@example.com
+    To: xtest@example.com
+    Subject: Test message 1
+    <BLANKLINE>
+    Here is message 1
+    <BLANKLINE>
+    --===============...==
+    Content-Type: message/rfc822
+    MIME-Version: 1.0
+    <BLANKLINE>
+    From: aperson@example.com
+    To: xtest@example.com
+    Subject: Test message 2
+    <BLANKLINE>
+    Here is message 2
+    <BLANKLINE>
+    --===============...==
+    Content-Type: message/rfc822
+    MIME-Version: 1.0
+    <BLANKLINE>
+    From: aperson@example.com
+    To: xtest@example.com
+    Subject: Test message 3
+    <BLANKLINE>
+    Here is message 3
+    <BLANKLINE>
+    --===============...==
+    Content-Type: message/rfc822
+    MIME-Version: 1.0
+    <BLANKLINE>
+    From: aperson@example.com
+    To: xtest@example.com
+    Subject: Test message 4
+    <BLANKLINE>
+    Here is message 4
+    <BLANKLINE>
+    --===============...==
+    Content-Type: text/plain; charset="us-ascii"
+    MIME-Version: 1.0
+    Content-Transfer-Encoding: 7bit
+    Content-Description: Digest Footer
+    <BLANKLINE>
+    _______________________________________________
+    Test mailing list
+    test@example.com
+    http://lists.example.com/listinfo/test@example.com
+    <BLANKLINE>
+    --===============...==--
+
+The RFC 1153 contains the digest in a single plain text message.
+
+    >>> print rfc1153.as_string()
+    From: test-request@example.com
+    Subject: Test Digest, Vol 1, Issue 1
+    To: test@example.com
+    Reply-To: test@example.com
+    Date: ...
+    Message-ID: ...
+    MIME-Version: 1.0
+    Content-Type: text/plain; charset="us-ascii"
+    Content-Transfer-Encoding: 7bit
+    <BLANKLINE>
+    Send Test mailing list submissions to
+    	test@example.com
+    <BLANKLINE>
+    To subscribe or unsubscribe via the World Wide Web, visit
+    	http://lists.example.com/listinfo/test@example.com
+    or, via email, send a message with subject or body 'help' to
+    	test-request@example.com
+    <BLANKLINE>
+    You can reach the person managing the list at
+    	test-owner@example.com
+    <BLANKLINE>
+    When replying, please edit your Subject line so it is more specific
+    than "Re: Contents of Test digest..."
+    <BLANKLINE>
+    <BLANKLINE>
+    Today's Topics:
+    <BLANKLINE>
+       1. Test message 1 (aperson@example.com)
+       2. Test message 2 (aperson@example.com)
+       3. Test message 3 (aperson@example.com)
+       4. Test message 4 (aperson@example.com)
+    <BLANKLINE>
+    <BLANKLINE>
+    ----------------------------------------------------------------------
+    <BLANKLINE>
+    From: aperson@example.com
+    Subject: Test message 1
+    To: xtest@example.com
+    Message-ID: ...
+    <BLANKLINE>
+    Here is message 1
+    <BLANKLINE>
+    ------------------------------
+    <BLANKLINE>
+    From: aperson@example.com
+    Subject: Test message 2
+    To: xtest@example.com
+    Message-ID: ...
+    <BLANKLINE>
+    Here is message 2
+    <BLANKLINE>
+    ------------------------------
+    <BLANKLINE>
+    From: aperson@example.com
+    Subject: Test message 3
+    To: xtest@example.com
+    Message-ID: ...
+    <BLANKLINE>
+    Here is message 3
+    <BLANKLINE>
+    ------------------------------
+    <BLANKLINE>
+    From: aperson@example.com
+    Subject: Test message 4
+    To: xtest@example.com
+    Message-ID: ...
+    <BLANKLINE>
+    Here is message 4
+    <BLANKLINE>
+    ------------------------------
+    <BLANKLINE>
+    _______________________________________________
+    Test mailing list
+    test@example.com
+    http://lists.example.com/listinfo/test@example.com
+    <BLANKLINE>
+    <BLANKLINE>
+    End of Test Digest, Vol 1, Issue 1
+    **********************************
+    <BLANKLINE>
+
+
+Internationalized digests
+-------------------------
+
+When messages come in with a content-type character set different than that of
+the list's preferred language, recipients will get an internationalized
+digest.  French is not enabled by default site-wide, so enable that now.
+
+    >>> config.languages.enable_language('fr')
+
+    # Simulate the site administrator setting the default server language to
+    # French in the configuration file.  Without this, the English template
+    # will be found and the masthead won't be translated.
+    >>> config.push('french', """
+    ... [mailman]
+    ... default_language: fr
+    ... """)
+
+    >>> mlist.preferred_language = u'fr'
+    >>> msg = message_from_string("""\
+    ... From: aperson@example.org
+    ... To: test@example.com
+    ... Subject: =?iso-2022-jp?b?GyRCMGxIVhsoQg==?=
+    ... MIME-Version: 1.0
+    ... Content-Type: text/plain; charset=iso-2022-jp
+    ... Content-Transfer-Encoding: 7bit
+    ...
+    ... \x1b$B0lHV\x1b(B
+    ... """)
+
+Set the digest threshold to zero so that the digests will be sent immediately.
+
+    >>> mlist.digest_size_threshold = 0
+    >>> process(mlist, msg, {})
+
+The marker message is sitting in the digest queue.
+
+    >>> len(digestq.files)
+    1
+    >>> entry = get_queue_messages('digest')[0]
+    >>> dump_msgdata(entry.msgdata)
+    _parsemsg    : False
+    digest_number: 2
+    digest_path  : .../lists/test@example.com/digest.1.2.mmdf
+    listname     : test@example.com
+    version      : 3
+    volume       : 1
+
+The digest queue runner runs a loop, placing the two digests into the virgin
+queue.
+
+    # Put the messages back in the queue for the runner to handle.
+    >>> filebase = digestq.enqueue(entry.msg, entry.msgdata)
+    >>> runner.run()
+    >>> messages = get_queue_messages('virgin')
+    >>> len(messages)
+    2
+
+One of which is the MIME digest and the other of which is the RFC 1153 digest.
+
+    >>> if messages[0].msg.is_multipart():
+    ...     mime = messages[0].msg
+    ...     rfc1153 = messages[1].msg
+    ... else:
+    ...     mime = messages[1].msg
+    ...     rfc1153 = messages[0].msg
+
+You can see that the digests contain a mix of French and Japanese.
+
+    >>> print mime.as_string()
+    Content-Type: multipart/mixed; boundary="===============...=="
+    MIME-Version: 1.0
+    From: test-request@example.com
+    Subject: Groupe Test, Vol. 1, Parution 2
+    To: test@example.com
+    Reply-To: test@example.com
+    Date: ...
+    Message-ID: ...
+    <BLANKLINE>
+    --===============...==
+    Content-Type: text/plain; charset="iso-8859-1"
+    MIME-Version: 1.0
+    Content-Transfer-Encoding: quoted-printable
+    Content-Description: Groupe Test, Vol. 1, Parution 2
+    <BLANKLINE>
+    Envoyez vos messages pour la liste Test =E0
+    	test@example.com
+    <BLANKLINE>
+    Pour vous (d=E9s)abonner par le web, consultez
+    	http://lists.example.com/listinfo/test@example.com
+    <BLANKLINE>
+    ou, par courriel, envoyez un message avec =AB=A0help=A0=BB dans le corps ou
+    dans le sujet =E0
+    	test-request@example.com
+    <BLANKLINE>
+    Vous pouvez contacter l'administrateur de la liste =E0 l'adresse
+    	test-owner@example.com
+    <BLANKLINE>
+    Si vous r=E9pondez, n'oubliez pas de changer l'objet du message afin
+    qu'il soit plus sp=E9cifique que =AB=A0Re: Contenu du groupe de Test...=A0=
+    =BB
+    --===============...==
+    Content-Type: text/plain; charset="utf-8"
+    MIME-Version: 1.0
+    Content-Transfer-Encoding: base64
+    Content-Description: Today's Topics (1 messages)
+    <BLANKLINE>
+    VGjDqG1lcyBkdSBqb3VyIDoKCiAgIDEuIOS4gOeVqiAoYXBlcnNvbkBleGFtcGxlLm9yZykK
+    <BLANKLINE>
+    --===============...==
+    Content-Type: message/rfc822
+    MIME-Version: 1.0
+    <BLANKLINE>
+    From: aperson@example.org
+    To: test@example.com
+    Subject: =?iso-2022-jp?b?GyRCMGxIVhsoQg==?=
+    MIME-Version: 1.0
+    Content-Type: text/plain; charset=iso-2022-jp
+    Content-Transfer-Encoding: 7bit
+    <BLANKLINE>
+    一番
+    <BLANKLINE>
+    --===============...==
+    Content-Type: text/plain; charset="iso-8859-1"
+    MIME-Version: 1.0
+    Content-Transfer-Encoding: quoted-printable
+    Content-Description: =?utf-8?q?Pied_de_page_des_remises_group=C3=A9es?=
+    <BLANKLINE>
+    _______________________________________________
+    Test mailing list
+    test@example.com
+    http://lists.example.com/listinfo/test@example.com
+    <BLANKLINE>
+    --===============...==--
+
+The RFC 1153 digest will be encoded in UTF-8 since it contains a mixture of
+French and Japanese characters.
+
+    >>> print rfc1153.as_string()
+    From: test-request@example.com
+    Subject: Groupe Test, Vol. 1, Parution 2
+    To: test@example.com
+    Reply-To: test@example.com
+    Date: ...
+    Message-ID: ...
+    MIME-Version: 1.0
+    Content-Type: text/plain; charset="utf-8"
+    Content-Transfer-Encoding: base64
+    <BLANKLINE>
+    RW52b...
+    Y29tC...
+    Ly9sa...
+    dXJya...
+    cyBvd...
+    dmV6I...
+    dGVzd...
+    ZGUgY...
+    Zmlxd...
+    ZXMgZ...
+    LS0tL...
+    LS0tL...
+    c3RAZ...
+    d2RvZ...
+    LWpwC...
+    X19fX...
+    dEBle...
+    cGxlL...
+    KioqK...
+    <BLANKLINE>
+
+The content can be decoded to see the actual digest text.
+
+    # We must display the repr of the decoded value because doctests cannot
+    # handle the non-ascii characters.
+    >>> [repr(line) for line in rfc1153.get_payload(decode=True).splitlines()]
+    ["'Envoyez vos messages pour la liste Test \\xc3\\xa0'",
+     "'\\ttest@example.com'",
+     "''",
+     "'Pour vous (d\\xc3\\xa9s)abonner par le web, consultez'",
+     "'\\thttp://lists.example.com/listinfo/test@example.com'",
+     "''",
+     "'ou, par courriel, envoyez un message avec \\xc2\\xab\\xc2\\xa0...
+     "'dans le sujet \\xc3\\xa0'",
+     "'\\ttest-request@example.com'",
+     "''",
+     '"Vous pouvez contacter l\'administrateur de la liste \\xc3\\xa0 ...
+     "'\\ttest-owner@example.com'",
+     "''",
+     '"Si vous r\\xc3\\xa9pondez, n\'oubliez pas de changer l\'objet du ...
+     '"qu\'il soit plus sp\\xc3\\xa9cifique que \\xc2\\xab\\xc2\\xa0Re: ...
+     "''",
+     "'Th\\xc3\\xa8mes du jour :'",
+     "''",
+     "'   1. \\xe4\\xb8\\x80\\xe7\\x95\\xaa (aperson@example.org)'",
+     "''",
+     "''",
+     "'---------------------------------------------------------------------...
+     "''",
+     "'From: aperson@example.org'",
+     "'Subject: \\xe4\\xb8\\x80\\xe7\\x95\\xaa'",
+     "'To: test@example.com'",
+     "'Message-ID: ...
+     "'Content-Type: text/plain; charset=iso-2022-jp'",
+     "''",
+     "'\\xe4\\xb8\\x80\\xe7\\x95\\xaa'",
+     "''",
+     "'------------------------------'",
+     "''",
+     "'_______________________________________________'",
+     "'Test mailing list'",
+     "'test@example.com'",
+     "'http://lists.example.com/listinfo/test@example.com'",
+     "''",
+     "''",
+     "'Fin de Groupe Test, Vol. 1, Parution 2'",
+     "'**************************************'"]
diff --git a/src/mailman/queue/incoming.py b/src/mailman/queue/incoming.py
index 1adda6629..877255662 100644
--- a/src/mailman/queue/incoming.py
+++ b/src/mailman/queue/incoming.py
@@ -26,6 +26,14 @@ prepared for delivery.  Rejections, discards, and holds are processed
 immediately.
 """
 
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'IncomingRunner',
+    ]
+
+
 from mailman.core.chains import process
 from mailman.queue import Runner
 
@@ -35,6 +43,7 @@ class IncomingRunner(Runner):
     """The incoming queue runner."""
 
     def _dispose(self, mlist, msg, msgdata):
+        """See `IRunner`."""
         if msgdata.get('envsender') is None:
             msgdata['envsender'] = mlist.no_reply_address
         # Process the message through the mailing list's start chain.
diff --git a/src/mailman/queue/retry.py b/src/mailman/queue/retry.py
index 2b5a6afad..d2ca78add 100644
--- a/src/mailman/queue/retry.py
+++ b/src/mailman/queue/retry.py
@@ -15,6 +15,16 @@
 # You should have received a copy of the GNU General Public License along with
 # GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
 
+"""Retry delivery."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'RetryRunner',
+    ]
+
+
 import time
 
 from mailman.config import config
@@ -23,15 +33,13 @@ from mailman.queue import Runner
 
 
 class RetryRunner(Runner):
-    def __init__(self, slice=None, numslices=1):
-        Runner.__init__(self, slice, numslices)
-        self._outq = config.switchboards['out']
+    """Retry delivery."""
 
     def _dispose(self, mlist, msg, msgdata):
-        # Move it to the out queue for another retry
-        self._outq.enqueue(msg, msgdata)
+        # Move the message to the out queue for another try.
+        config.switchboards['outgoing'].enqueue(msg, msgdata)
         return False
 
     def _snooze(self, filecnt):
-        # We always want to snooze
+        # We always want to snooze.
         time.sleep(self.sleep_float)
diff --git a/src/mailman/styles/default.py b/src/mailman/styles/default.py
index 296ebbff1..0f7869a75 100644
--- a/src/mailman/styles/default.py
+++ b/src/mailman/styles/default.py
@@ -33,7 +33,8 @@ from zope.interface import implements
 from mailman import Utils
 from mailman.i18n import _
 from mailman.interfaces import Action, NewsModeration
-from mailman.interfaces.mailinglist import Personalization, ReplyToMunging
+from mailman.interfaces.mailinglist import (
+    DigestFrequency, Personalization, ReplyToMunging)
 from mailman.interfaces.styles import IStyle
 
 
@@ -120,7 +121,7 @@ $real_name mailing list
 $fqdn_listname
 ${listinfo_page}
 """
-        mlist.digest_volume_frequency = 1
+        mlist.digest_volume_frequency = DigestFrequency.monthly
         mlist.one_last_digest = {}
         mlist.next_digest_number = 1
         mlist.nondigestable = True
diff --git a/src/mailman/testing/helpers.py b/src/mailman/testing/helpers.py
index f92c5f012..10f5229d7 100644
--- a/src/mailman/testing/helpers.py
+++ b/src/mailman/testing/helpers.py
@@ -35,7 +35,6 @@ import errno
 import signal
 import socket
 import logging
-import mailbox
 import smtplib
 import threading
 
@@ -44,6 +43,7 @@ from Queue import Empty, Queue
 from mailman.bin.master import Loop as Master
 from mailman.config import config
 from mailman.testing.smtplistener import Server
+from mailman.utilities.mailbox import Mailbox
 
 
 log = logging.getLogger('mailman.debug')
@@ -69,6 +69,12 @@ def make_testable_runner(runner_class, name=None):
     class EmptyingRunner(runner_class):
         """Stop processing when the queue is empty."""
 
+        def __init__(self, *args, **kws):
+            super(EmptyingRunner, self).__init__(*args, **kws)
+            # We know it's an EmptyingRunner, so really we want to see the
+            # super class in the log files.
+            self.__class__.__name__ = runner_class.__name__
+
         def _do_periodic(self):
             """Stop when the queue is empty."""
             self._stop = (len(self.switchboard.files) == 0)
@@ -106,8 +112,8 @@ def digest_mbox(mlist):
     :param mlist: The mailing list.
     :return: The mailing list's pending digest as a mailbox.
     """
-    path = os.path.join(mlist.data_path, 'digest.mbox')
-    return mailbox.mbox(path)
+    path = os.path.join(mlist.data_path, 'digest.mmdf')
+    return Mailbox(path)
 
 
 
diff --git a/src/mailman/tests/test_documentation.py b/src/mailman/tests/test_documentation.py
index dd205c307..8408265a6 100644
--- a/src/mailman/tests/test_documentation.py
+++ b/src/mailman/tests/test_documentation.py
@@ -39,6 +39,7 @@ from email import message_from_string
 import mailman
 
 from mailman.Message import Message
+from mailman.app.lifecycle import create_list
 from mailman.config import config
 from mailman.testing.layers import SMTPLayer
 
@@ -102,6 +103,7 @@ def dump_msgdata(msgdata, *additional_skips):
         print '{0:{2}}: {1}'.format(key, msgdata[key], longest)
 
 
+
 def setup(testobj):
     """Test setup."""
     # In general, I don't like adding convenience functions, since I think
@@ -110,6 +112,7 @@ def setup(testobj):
     # hide some icky test implementation details.
     testobj.globs['commit'] = config.db.commit
     testobj.globs['config'] = config
+    testobj.globs['create_list'] = create_list
     testobj.globs['dump_msgdata'] = dump_msgdata
     testobj.globs['message_from_string'] = specialized_message_from_string
     testobj.globs['smtpd'] = SMTPLayer.smtpd
diff --git a/src/mailman/utilities/mailbox.py b/src/mailman/utilities/mailbox.py
new file mode 100644
index 000000000..811f4c348
--- /dev/null
+++ b/src/mailman/utilities/mailbox.py
@@ -0,0 +1,49 @@
+# Copyright (C) 2009 by the Free Software Foundation, Inc.
+#
+# This file is part of GNU Mailman.
+#
+# GNU Mailman is free software: you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option)
+# any later version.
+#
+# GNU Mailman 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 General Public License along with
+# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Module stuff."""
+
+from __future__ import absolute_import, unicode_literals
+
+__metaclass__ = type
+__all__ = [
+    'Mailbox',
+    ]
+
+
+# Use a single file format for the digest mailbox because this makes it easier
+# to calculate the current size of the mailbox.  This way, we don't have to
+# carry around or store the size of the mailbox, we can just stat the file to
+# get its size.  MMDF is slightly more sane than mbox; it's primary advantage
+# for us is that it does no 'From' mangling.
+# mangling.
+from mailbox import MMDF
+
+
+
+class Mailbox(MMDF):
+    """A mailbox that interoperates with the 'with' statement."""
+
+    def __enter__(self):
+        self.lock()
+        return self
+
+    def __exit__(self, *exc):
+        self.flush()
+        self.unlock()
+        # Don't suppress the exception.
+        return False