Update of /cvsroot/mailman/mailman3/src/mailman/queues
In directory sc8-pr-cvs1.sourceforge.net:/tmp/cvs-serv31133/src/mailman/queues
Added Files:
Tag: 3.0
__init__.py bounce.py incoming.py maildir.py outgoing.py
retry.py runner.py switchboard.py
Log Message:
Initial checkin of Mailman3 source
--- NEW FILE: __init__.py ---
--- NEW FILE: bounce.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program 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 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Bounce queue runner."""
import re
import logging
from email.Utils import parseaddr
from mailman.config import configuration
from mailman.interfaces.qrunner import IQRunner
from mailman.queues.runner import Runner
from mailman.smtp.bounceproc import BounceProc
log = logging.getLogger('delivery.bounces')
class BounceRunner(Runner):
__implements__ = IQRunner
QRNAME = 'bounce'
def __init__(self, slice=None):
super(BounceRunner, self).__init__(slice)
# This object will collect bounce events we've received. Every once
# in a while, a thread will wake up, flip this file, and then process
# the bounce events out of it.
self._bounces = BounceProc()
def _dispose(self, mlist, msg, msgdata):
# Decode the recipient headers for VERP'd bounce addresses
bounces = []
headers = []
for h in ('to', 'delivered-to', 'envelope-to', 'apparently-to'):
headers.extend(msg.get_all(h, []))
for val in headers:
addr = parseaddr(val)[1]
if not addr:
log.info('No address found: %s', val)
continue
mo = configuration.smtp.verpre.search(addr)
if not mo:
log.info('No VERP found: %s', addr)
recip = '[EMAIL PROTECTED]' % mo.group('mailbox', 'host')
bounces.append(recip)
self._bounces.registerBounces(bounces)
def _cleanup(self):
self._bounces.close()
super(BounceRunner, self)._cleanup()
--- NEW FILE: incoming.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program 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 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Incoming queue runner."""
import sys
import logging
from mailman import errors
from mailman.interfaces.qrunner import IQRunner
from mailman.queues.runner import Runner
log = logging.getLogger('mailman.queues')
class IncomingRunner(Runner):
__implements__ = IQRunner
QRNAME = 'incoming'
def _dispose(self, mlist, msg, msgdata):
# Process the message through the handler pipeline. The handler
# pipeline can actually come from one of three places: the message
# metadata, the mlist, or the global pipeline.
#
# If a message was requeued due to an uncaught exception, its metadata
# will contain the retry pipeline. Use this above all else.
# Otherwise, if the mlist has a `pipeline' attribute, it should be
# used. Final fallback is the global pipeline.
pipeline = self._get_pipeline(mlist, msg, msgdata)
msgdata['pipeline'] = pipeline
more = self._dopipeline(mlist, msg, msgdata, pipeline)
if not more:
del msgdata['pipeline']
return more
# Overridable
def _get_pipeline(self, mlist, msg, msgdata):
# We must return a copy of the list, otherwise, the first message that
# flows through the pipeline will empty it out!
return msgdata.get('pipeline',
getattr(mlist, 'pipeline'))
#config.GLOBAL_PIPELINE))[:]
def _dopipeline(self, mlist, msg, msgdata, pipeline):
while pipeline:
handler = pipeline.pop(0)
modname = 'mailman.handlers.' + handler
__import__(modname)
try:
sys.modules[modname].process(mlist, msg, msgdata)
except errors.DiscardMessage:
log.info('message discarded: %s', msg['message-id'])
except errors.HoldMessage:
# Let the approval process take it from here. The message no
# longer needs to be queued.
pass
except errors.RejectMessage, e:
self.registerBounce(mlist, msg, msgdata, e)
except:
# Push this pipeline module back on the stack, then re-raise
# the exception. The message will be shunted, and when
# unshunted, processing will start at the last handler.
pipeline.insert(0, handler)
raise
# We've successfully completed handling of this message
return False
def registerBounce(self, mlist, msg, msgdata, e):
# XXX Register bounces
pass
--- NEW FILE: maildir.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program 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 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Maildir queue runner.
Most MTAs can be configured to deliver messages to a `maildir'[1].
In maildir delivery, the MTA drops messages for a mailing list into the
qfiles/maildir/<listname>/new directory. This qrunner will read messages from
the maildir's new/ directory and inject them into Mailman's qfiles/incoming
directory for processing in the normal pipeline. This delivery mechanism
contrasts with mail program delivery, where incoming messages end up in
qfiles/incoming via the MTA executing the scripts/post script (and likewise
for the other -aliases for each mailing list).
The advantage to maildir delivery is that it is more efficient; there's no
need to fork an intervening program just to take the message from the MTA's
standard output.
[1] http://cr.yp.to/proto/maildir.html
We're going to use the :info flag == 1, experimental status flag for our own
purposes. The :1 can be followed by one of these letters:
- P means that our maildir qrunner is in the process of parsing and enqueuing
the message. If successful, it will unlink the file.
- X means something failed during the parse/enqueue phase. An error message
will be logged and the file will be renamed <filename>:1,X. The maildir
qrunner will never automatically return to this file, but once the problem
is fixed, you can manually move the file back to the new/ directory and the
maildir qrunner will attempt to re-process it. At some point we may do this
automatically.
"""
import os
import re
import errno
import logging
from email.Parser import Parser
from email.Utils import parseaddr
from mailman import errors
from mailman.config import configuration
from mailman.database import listdb
from mailman.interfaces.qrunner import IQRunner
from mailman.message import Message
from mailman.queues.runner import Runner
from mailman.queues.switchboard import Switchboard
log = logging.getLogger('mailman.queues')
# Anatomy of a Mailman address:
#
# <listname>[-<command>[EMAIL PROTECTED]
#
# Split an address into the above constituent parts, returning an object
# containing the following attributes. When an optional part is missing, the
# attribute's value will be None.
class AddressParts:
command = None
recip = None
domain = None
def splitaddr(listname, address):
addrparts = AddressParts()
# Split on the rightmost at sign
parts = address.split('@', 1)
if len(parts) <> 2:
return None
localpart, addrparts.domain = parts
if localpart == listname:
# This is a list posting
return addrparts
# Strip off the listname bit from the localpart. What's left must be the
# command and any VERP encoded information.
command = localpart[len(listname)+1:]
i = command.find('+')
if i < 0:
addrparts.command = command
return addrparts
addrparts.command = command[:i]
verprecip = command[i+1:]
# Find the rightmost = sign, which must separate out the recipient's
# domain and their localpart, then concatenate.
i = verprecip.rfind('=')
if i <= 0:
return None
addrparts.recip = '[EMAIL PROTECTED]' % (verprecip[:i], verprecip[i+1:])
return addrparts
class MaildirRunner(Runner):
__implements__ = IQRunner
QRNAME = 'maildir'
# This class is much different than most runners because it pulls files of
# a different format than the other runners. The files this runner reads
# are just single message files as dropped into the directory by the MTA.
# This runner will read the file, and enqueue it in the expected qfiles
# directory for normal processing.
#
# We're only inheriting from Runner to get the run() implementation.
def __init__(self, slice=None):
# Ignore the arguments because maildir format allows us to handle the
# files with multiple processes without the need for locks.
self._slice = '[ignore]'
self._qrcount = '[ignore]'
self._kids = {}
self._stop = False
self._parser = Parser(Message)
self._maildir = os.path.join(configuration.dirs.qdir, 'maildir')
# We create a switchboard for this runner, but then throw it away. We
# just want the side effect of making sure the directory exists.
Switchboard(self._maildir)
def _oneloop(self):
# Cruise through each list's new/ directory looking for new messages.
# If any are found, we'll parse the message file, create the message
# object and metadata, and enqueue the message to the incoming
# directory. BAW: we may eventually want to use multiple thread or
# processes so that maildir processing isn't totally sequential on the
# list's new/ directory.
for listname in os.listdir(self._maildir):
# We won't worry about whether the list actually exists at this
# point. If it doesn't, the incoming qrunner will barf later.
new = os.path.join(self._maildir, listname, 'new')
cur = os.path.join(self._maildir, listname, 'cur')
try:
files = os.listdir(new)
except OSError, e:
if e.errno <> errno.ENOENT: raise
# Nothing's been delivered to this mailing list yet.
continue
for file in files:
srcname = os.path.join(new, file)
dstname = os.path.join(cur, file + ':1,P')
xdstname = os.path.join(cur, file + ':1,X')
try:
os.rename(srcname, dstname)
except OSError, e:
if e.errno == errno.ENOENT:
# Some other MaildirRunner beat us to it
continue
log.error('Could not rename maildir file: %s', srcname)
raise
# Now open, read, parse, and enqueue this message
try:
fp = open(dstname)
try:
msg = self._parser.parse(fp)
finally:
fp.close()
self._process(listname, msg)
os.unlink(dstname)
except Exception, e:
log.error('maildir exception: %s', e)
os.rename(dstname, xdstname)
def _process(self, listname, msg):
# Parse the various mailbox to fields to see command this message is
# destined for.
vals = []
for header in configuration.smtp.envto:
vals.extend(msg.get_all(header, []))
# Find the first one that parses out to the given listname
dest = None
for val in vals:
to = parseaddr(val)[1]
dest = splitaddr(listname, to)
if dest:
break
if not dest:
raise errors.NoValidDestination
# Now craft the metadata dictionary for this message, and open the
# switchboard for this message's queue.
msgdata = {'listname': listname}
if dest.command is None:
dir = 'incoming'
elif dest.command == 'bounces':
dir = 'bounce'
elif dest.command == 'confirm':
msgdata['toconfirm'] = True
dir = 'command'
elif dest.command in ('join', 'subscribe'):
msgdata['tojoin'] = True
dir = 'command'
elif dest.command in ('leave', 'unsubscribe'):
msgdata['toleave'] = True
dir = 'command'
elif dest.command == 'owner':
msgdata['toowner'] = True
dir = 'incoming'
elif dest.command == 'request':
msgdata['torequest'] = True
dir = 'command'
else:
raise errors.NoValidDestination
# Enqueue the message; we're done
sb = Switchboard(os.path.join(configuration.dirs.qdir, dir))
sb.enqueue(msg, metadata)
--- NEW FILE: outgoing.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program 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 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Outgoing queue runner."""
from mailman.interfaces.qrunner import IQRunner
from mailman.smtp import send
from mailman.queues.runner import Runner
class OutgoingRunner(Runner):
__implements__ = IQRunner
QRNAME = 'outgoing'
def _dispose(self, mlist, msg, msgdata):
send(mlist, msg, msgdata)
# Because send() does all its work in a thread, it can't fail in the
# main thread. Sub-threads will requeue the message if necessary.
return False
--- NEW FILE: retry.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program 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 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Retry queue."""
import os
import logging
import datetime
from mailman.config import configuration
from mailman.interfaces.qrunner import IQRunner
from mailman.queues.runner import Runner
from mailman.queues.switchboard import Switchboard
from mailman.queues.outgoing import OutgoingRunner
log = logging.getLogger('mailman.smtp')
class RetryRunner(Runner):
__implements__ = IQRunner
QRNAME = 'retry'
def __init__(self, slice=None):
super(RetryRunner, self).__init__(slice)
outdir = os.path.join(configuration.dirs.qdir, OutgoingRunner.QRNAME)
self._outq = Switchboard(outdir)
def _dispose(self, mlist, msg, msgdata):
__pychecker__ = 'no-argsused'
now = datetime.datetime.now()
# Did we wait long enough to retry this message? If not, then just
# requeue it to the retry queue.
if now < msgdata['last_attempt'] + configuration.smtp.retry_interval:
return True
# Did we make progress the last time we tried? If so, try again.
if len(msgdata['recips']) < msgdata['last_recipcnt']:
self._outq.enqueue(msg, msgdata)
return False
# How many times have we tried to make progress on this message? If
# we've exhausted our retries, just log and discard.
retries = msgdata.get('retrycnt', configuration.smtp.retry_count) - 1
if retries < 0:
log.info('retries exhausted, discarding: %s', msg['message-id'])
return False
# Okay, retry it
self._outq.enqueue(msg, msgdata)
return False
--- NEW FILE: runner.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program 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 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Generic queue runner class.
"""
import os
import time
import logging
import traceback
from cStringIO import StringIO
from mailman import errors
from mailman import utils
from mailman.config import configuration
from mailman.mlist import MailList
from mailman.queues.switchboard import Switchboard
log = logging.getLogger('mailman.queues')
class Runner(object):
QRNAME = None
SLEEPTIME = utils.seconds(1)
def __init__(self, slice=None):
self._slice = slice
self._qrclass, self._qrcount = configuration.smtp.qrunners[self.QRNAME]
self._kids = {}
# Create our queue directory switchboard
dir = os.path.join(configuration.dirs.qdir, self.QRNAME)
self._switchboard = Switchboard(dir, slice, self._qrcount)
# Create the shunt switchboard
dir = os.path.join(configuration.dirs.qdir, 'shunt')
self._shunt = Switchboard(dir)
self._stop = False
def stop(self):
self._stop = True
def run(self):
log.info('qrunner started: %s [%s:%s]',
self.QRNAME, self._slice, self._qrcount)
# Start the main loop for this queue runner.
try:
try:
while True:
# Once through the loop that processes all the files in
# the queue directory.
filecnt = self._oneloop()
# Do the periodic work for the subclass. BAW: this
# shouldn't be called here. There should be one more
# _doperiodic() call at the end of the _oneloop() loop.
self._doperiodic()
# If the stop flag is set, we're done.
if self._stop:
break
# Give the runner an opportunity to snooze for a while,
# but pass it the file count so it can decide whether to
# do more work now or not.
self._snooze(filecnt)
except KeyboardInterrupt:
pass
finally:
# We've broken out of our main loop, so we want to reap all the
# subprocesses we've created and do any other necessary cleanups.
self._cleanup()
log.info('qrunner done: %s [%s:%s]',
self.QRNAME, self._slice, self._qrcount)
def _oneloop(self):
# First, list all the files in our queue directory.
# Switchboard.files() is guaranteed to hand us the files in FIFO
# order. Return an integer count of the number of files that were
# available for this qrunner to process.
files = self._switchboard.files()
for filebase in files:
# Ask the switchboard for the message and metadata objects
# associated with this filebase.
msg, msgdata = self._switchboard.dequeue(filebase)
# Now that we've dequeued the message, we want to be incredibly
# anal about making sure that no uncaught exception could cause us
# to lose the message. All runners that implement _dispose() must
# guarantee that exceptions are caught and dealt with properly.
# Still, there may be a bug in the infrastructure, and we do not
# want those to cause messages to be lost. Any uncaught
# exceptions will cause the message to be stored in the shunt
# queue for human intervention.
try:
self._onefile(msg, msgdata)
except Exception, e:
self._traceback(e)
# Put a marker in the metadata for unshunting
msgdata['whichq'] = self._switchboard.whichq()
filebase = self._shunt.enqueue(msg, msgdata)
log.error('SHUNTING: %s', filebase)
# Other work we want to do each time through the loop
utils.reap(self._kids, once=True)
self._doperiodic()
if self._shortcircuit():
break
return len(files)
def _onefile(self, msg, msgdata):
# Do some common sanity checking on the message metadata. It's got to
# be destined for a particular mailing list. This switchboard is used
# to shunt off badly formatted messages. We don't want to just trash
# them because they may be fixable with human intervention. Just get
# them out of our site though.
#
# Find out which mailing list this message is destined for.
listname = msgdata.get('listname')
try:
mlist = MailList(listname)
except errors.NoListError:
log.error('Dequeuing message for missing list: %s', listname)
self._shunt.enqueue(msg, msgdata)
return
# Start an explicit transaction for the processing of this message.
txn = mlist.begin()
try:
# Process this message
keepqueued = self._dispose(mlist, msg, msgdata)
except:
txn.abort()
raise
else:
txn.commit()
# Keep tabs on any child processes that got spawned.
kids = msgdata.get('_kids')
if kids:
self._kids.update(kids)
if keepqueued:
self._switchboard.enqueue(msg, msgdata)
def _traceback(self, exc):
log.error('Uncaught runner exception: %s', exc)
s = StringIO()
traceback.print_exc(file=s)
log.error(s.getvalue())
#
# Subclasses can override these methods.
#
def _cleanup(self):
"""Clean up upon exit from the main processing loop.
Called when the Runner's main loop is stopped, this should perform
any necessary resource deallocation. Its return value is irrelevant.
"""
utils.reap(self._kids)
def _dispose(self, mlist, msg, msgdata):
"""Dispose of a single message destined for a mailing list.
Called for each message that the Runner is responsible for, this is
the primary overridable method for processing each message.
Subclasses, must provide implementation for this method.
mlist is the MailList instance this message is destined for.
msg is the Message object representing the message.
msgdata is a dictionary of message metadata.
"""
raise NotImplementedError
def _doperiodic(self):
"""Do some processing `every once in a while'.
Called every once in a while both from the Runner's main loop, and
from the Runner's hash slice processing loop. You can do whatever
special periodic processing you want here, and the return value is
irrelevant.
"""
pass
def _snooze(self, filecnt):
"""Sleep for a little while.
filecnt is the number of messages in the queue the last time through.
Sub-runners can decide to continue to do work, or sleep for a while
based on this value. By default, we only snooze if there was nothing
to do last time around.
"""
if filecnt or self.SLEEPTIME <= 0:
return
time.sleep(self.SLEEPTIME)
def _shortcircuit(self):
"""Return a true value if the individual file processing loop should
exit before it's finished processing each message in the current slice
of hash space. A false value tells _oneloop() to continue processing
until the current snapshot of hash space is exhausted.
You could, for example, implement a throttling algorithm here.
"""
return self._stop
--- NEW FILE: switchboard.py ---
# Copyright (C) 2004 Barry A. Warsaw
#
# This program 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 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 59 Temple
# Place - Suite 330, Boston, MA 02111-1307, USA.
"""Reading and writing message objects and message metadata.
"""
import os
import sha
import email
import errno
import cPickle
import logging
import datetime
from mailman.message import Message
from mailman.config.versions import qfile_schema_version
log = logging.getLogger('mailman.errors')
# 20 bytes of all bits set, maximum sha.digest() value
shamax = 0xffffffffffffffffffffffffffffffffffffffffL
class Switchboard:
def __init__(self, whichq, slice=None, numslices=1):
self._whichq = whichq
# Create the directory if it doesn't yet exist.
omask = os.umask(0) # rwxrws---
try:
try:
os.mkdir(self._whichq, 0770)
except OSError, e:
if e.errno <> errno.EEXIST: raise
finally:
os.umask(omask)
# Fast track for no slices
self._lower = None
self._upper = None
# BAW: test performance and end-cases of this algorithm
if numslices <> 1:
self._lower = ((shamax+1) * slice) / numslices
self._upper = (((shamax+1) * (slice+1)) / numslices) - 1
def whichq(self):
return self._whichq
def enqueue(self, _msg, _msgdata={}, **_kws):
# Calculate the SHA hexdigest of the message to get a unique base
# filename. We're also going to use the digest as a hash into the set
# of parallel qrunner processes.
data = _msgdata.copy()
data.update(_kws)
# Always add the metadata schema version number
data['version'] = qfile_schema_version
# Filter out volatile entries.
for k in data.keys():
if k.startswith('_'):
del data[k]
# Encode the current time into the file name for FIFO sorting in
# files(). The file name consists of two parts separated by a `+':
# the received time for this message (i.e. when it first showed up on
# this system) and the sha hex digest.
now = datetime.datetime.now()
rcvtime = data.setdefault('received_time', now)
# Get some data for the input to the sha hash
msgsave = cPickle.dumps(_msg, 1)
hashfood = msgsave + repr(now)
filebase = rcvtime.isoformat() + '+' + sha.new(hashfood).hexdigest()
# We only write one file to represent the message, but the file
# contains two pickles. The first pickle is the metadata and the
# second pickle is the message object. We play name switcheroo to
# close any race conditions.
filename = os.path.join(self._whichq, filebase + '.pck')
tmpfile = filename + '.tmp'
omask = os.umask(007)
try:
fp = open(tmpfile, 'w')
try:
cPickle.dump(data, fp, 1)
fp.write(msgsave)
fp.flush()
os.fsync(fp.fileno())
finally:
fp.close()
finally:
os.umask(omask)
os.rename(tmpfile, filename)
return filebase
def dequeue(self, filebase):
filename = os.path.join(self._whichq, filebase + '.pck')
# Now read the metadata and message object pickles from this file
fp = open(filename)
try:
data = cPickle.load(fp)
msg = cPickle.load(fp)
finally:
fp.close()
os.unlink(filename)
return msg, data
def files(self):
files = []
lower = self._lower
upper = self._upper
for f in os.listdir(self._whichq):
# By ignoring anything that doesn't end in .pck, we ignore
# tempfiles and avoid a race condition.
if not f.endswith('.pck'):
continue
# Parse out the hash and only return files which lie within our
# hash space.
filebase = os.path.splitext(f)[0]
when, digest = filebase.split('+')
# Throw out any files which don't match our bitrange. BAW: test
# performance and end-cases of this algorithm.
if not lower or (lower <= long(digest, 16) < upper):
files.append(filebase)
# FIFO sort, by nature of the leading ISO timestamp. If more than one
# file matches in the fractional seconds, the hash will break the
# time, but this should be so rare and to not worry about.
files.sort()
return files
_______________________________________________
Mailman-checkins mailing list
[EMAIL PROTECTED]
Unsubscribe: http://mail.python.org/mailman/options/mailman-checkins/archive%40jab.org
