offlineimap rev 542
"Automatic Subversion Change Mailer" <[email protected]> Mon, 28 Jul 2003 16:07:31 -0500 (CDT)
| Newsgroups | gmane.mail.imap.offlineimap.subversion |
|---|---|
| Message-ID | <[email protected]> |
You are receiving this message because
all commits get sent to this address.
Author: jgoerzen
Date: 2003-07-28 16:06:12 -0500 (Mon, 28 Jul 2003)
New Revision: 542
Modified:
offlineimap/branches/twisted/offlineimap/folder/IMAP.py
offlineimap/branches/twisted/offlineimap/imapserver.py
offlineimap/branches/twisted/offlineimap/imaputil.py
offlineimap/branches/twisted/offlineimap/repository/IMAP.py
Log:
Committed expiremental new connection allocator (this probably has syntax
errors and thus will likely not work)
Diff:
Modified: offlineimap/branches/twisted/offlineimap/folder/IMAP.py
==============================================================================
--- offlineimap/branches/twisted/offlineimap/folder/IMAP.py 2003-07-28 20:10:20 UTC (rev 541)
+++ offlineimap/branches/twisted/offlineimap/folder/IMAP.py 2003-07-28 21:06:12 UTC (rev 542)
@@ -23,9 +23,12 @@
from StringIO import StringIO
from copy import copy
from twisted.internet import defer
+from imaputil import rd
-class IMAPFolder(BaseFolder):
+class IMAPFolder(BaseFolder, imaputil.AcquireMixin):
+
def __init__(self, imapserver, name, visiblename, accountname, repository):
+ self._initAcquires()
self.config = imapserver.config
self.expunge = repository.getexpunge()
self.name = name
@@ -54,24 +57,18 @@
def getvisiblename(self):
return self.visiblename
- def getuidvalidity(self):
- imapobj = self.imapserver.acquireconnection()
- try:
- d = imapobj.examine(self.getfullname())
- d.addCallback(lambda x: x['UIDVALIDITY'])
- return d
- #return long(imapobj.untagged_responses['UIDVALIDITY'][0])
- finally:
- self.imapserver.releaseconnection(imapobj)
+ def _getuidvalidity_acquired(self, imapobj):
+ d = imapobj.examine(self.getfullname())
+ d.addCallback(lambda x: x['UIDVALIDITY'])
+ return d
- def cachemessagelist(self):
- imapobj = self.imapserver.acquireconnection()
+ def _cachemessagelist_acquired(self, imapobj):
maindefer = defer.Deferred()
d = imapobj.examine(self.getfullname())
- d.addCallback(self.cachemessagelist_starter, maindefer, imapobj)
+ d.addCallback(self._cachemessagelist_starter, maindefer, imapobj)
return maindefer
- def cachemessagelist_starter(self, examineresult, maindefer, imapobj):
+ def _cachemessagelist_starter(self, examineresult, maindefer, imapobj):
self.messagelist = {}
try:
@@ -92,9 +89,10 @@
df = imapobj.fetchFlags('1:%d' % maxmsgid)
du = imapobj.fetchUID('1:%d' % maxmsgid)
dl = defer.DeferredList([df, du])
- dl.addCallback(self.cachemessagelist_worker, maindefer)
+ dl.addCallback(self._cachemessagelist_worker, maindefer)
+ return dl
- def cachemessagelist_worker(self, flags_uids, maindefer):
+ def _cachemessagelist_worker(self, flags_uids, maindefer):
#if not len(flags_uids):
# # FIXME: this sometimes happens. Is it right?
# maindefer.callback(None)
@@ -111,34 +109,18 @@
print self.getname(), "Cached at", self.messagelist
maindefer.callback(None)
- def cachemessagelist_old(self):
- response = imapobj.fetch('1:%d' % maxmsgid, '(FLAGS UID)')[1]
- for messagestr in response:
- # Discard the message number.
- messagestr = string.split(messagestr, maxsplit = 1)[1]
- options = imaputil.flags2hash(messagestr)
- if not options.has_key('UID'):
- UIBase.getglobalui().warn('No UID in message with options %s' %\
- str(options),
- minor = 1)
- else:
- uid = long(options['UID'])
- flags = imaputil.flagsimap2maildir(options['FLAGS'])
- self.messagelist[uid] = {'uid': uid, 'flags': flags}
-
def getmessagelist(self):
return self.messagelist
- def getmessage(self, uid):
- imapobj = self.imapserver.acquireconnection()
+ def _getmessage_acquired(self, imapobj, uid):
d = imapobj.examine(self.getfullname())
d.addCallback(lambda x, uid: \
imapobj.fetchSpecific(uid, 1, peek = 1),
str(uid))
- d.addCallback(self.getmessage_process)
+ d.addCallback(self._getmessage_process)
return d
- def getmessage_process(self, message):
+ def _getmessage_process(self, message):
key = message.keys()[0]
data = message[key][0]
i = data.index('BODY')
@@ -178,10 +160,11 @@
ui.debug('imap', 'savemessage_addheader: trailer = ' + repr(trailer))
return leader + newline + trailer
- def savemessage_searchforheader(self, imapobj, headername, headervalue):
- if imapobj.untagged_responses.has_key('APPENDUID'):
- return long(imapobj.untagged_responses['APPENDUID'][0].split(' ')[1])
-
+ def _savemessage_searchforheader(self, imapobj, headername, headervalue):
+ # FIXME: HANDLE APPENDUID
+ #if imapobj.untagged_responses.has_key('APPENDUID'):
+ # return long(imapobj.untagged_responses['APPENDUID'][0].split(' ')[1])
+ # THIS WHOLE FUNCTION IS HOSED.
ui = UIBase.getglobalui()
ui.debug('imap', 'savemessage_searchforheader called for %s: %s' % \
(headername, headervalue))
@@ -203,18 +186,13 @@
matchinguids.sort()
return long(matchinguids[0])
- def savemessage(self, uid, content, flags):
- d = self.imapserver.acquireconnection()
- d.addCallback(self.savemessage_step1, uid, content, flags)
- return d
-
- def savemessage_step1(self, imapobj, uid, content, flags):
- ui = UIBase.getglobalui()
+ def _savemessage_acquired(self, imapobj, uid, content, flags):
+ ui = UIBase.getobalui()
ui.debug('imap', 'savemessage_step1: called')
d = imapobj.select(self.getfullname())
- d.addCallback(self.savemessage_step2, imapobj, uid, content, flags)
+ d.addCallback(self._savemessage_step2, imapobj, uid, content, flags)
- def savemessage_step2(self, ignore, imapobj, uid, content, flags):
+ def _savemessage_step2(self, ignore, imapobj, uid, content, flags):
# This backend always assigns a new uid, so the uid arg is ignored.
# In order to get the new uid, we need to save off the message ID.
@@ -289,8 +267,7 @@
ui.debug('imap', 'savemessage: returning %d' % uid)
return uid
- def savemessageflags(self, uid, flags):
- imapobj = self.imapserver.acquireconnection()
+ def _savemessageflags_acquired(self, imapobj, uid, flags):
d = imapobj.select(self.getfullname())
d.addCallback(self.savemessageflags_worker, imapobj, uid, flags)
d.addCallback(self.savemessageflags_store, uid, flags)
@@ -328,22 +305,21 @@
def deletemessagesflags(self, uidlist, flags):
return self.processmessagesflags('-', uidlist, flags)
- def processmessagesflags(self, operation, uidlist, flags):
- imapobj = self.imapserver.acquireconnection()
- try:
- try:
- imapobj.select(self.getfullname())
- except imapobj.readonly:
- UIBase.getglobalui().flagstoreadonly(self, uidlist, flags)
- return
- r = imapobj.uid('store',
- imaputil.listjoin(uidlist),
- operation + 'FLAGS',
- imaputil.flagsmaildir2imap(flags))
- assert r[0] == 'OK', 'Error with store: ' + r[1]
- r = r[1]
- finally:
- self.imapserver.releaseconnection(imapobj)
+ def _processmessagesflags_acquired(self, imapobj,
+ operation, uidlist, flags):
+ imapobj.select(self.getfullname())
+ # HANDLE READONLY
+ #except imapobj.readonly:
+ # UIBase.getglobalui().flagstoreadonly(self, uidlist, flags)
+ # return
+ ####
+ raise NotImplementedError, "FIXME: this function is not yet done."
+ r = imapobj.uid('store',
+ imaputil.listjoin(uidlist),
+ operation + 'FLAGS',
+ imaputil.flagsmaildir2imap(flags))
+ assert r[0] == 'OK', 'Error with store: ' + r[1]
+ r = r[1]
# Some IMAP servers do not always return a result. Therefore,
# only update the ones that it talks about, and manually fix
# the others.
@@ -389,24 +365,14 @@
dflags = self.addmessagesflags_noconvert(uidlist, ['T'])
if self.expunge:
- # FIXME
- imapobj = self.imapserver.acquireconnection()
- try:
- dflags.addCallback(lambda ignore, io, folder: io.select(folder),
- imapobj, self.getfullname())
- # FIXME: handle readonly
- #except imapobj.readonly:
- # UIBase.getglobalui().deletereadonly(self, uidlist)
- # return
- dflags.addCallback(lambda ignore, io: io.expunge(),
- imapobj)
- finally:
- self.imapserver.releaseconnection(imapobj)
-
- dflags.addCallback(self.deletemessage_noconvert_worker,
- uidlist)
+ dflags.addCallback(lambda x: self.expunge())
return dflags
+ def _expunge_acquired(self, imapobj):
+ d = imapobj.select(self.getfullname())
+ d.addCallback(lambda ignore, io: io.expunge(), imapobj)
+ return d
+
def deletemessage_noconvert_worker(self, ignore, uidlist):
for uid in uidlist:
del self.messagelist[uid]
Modified: offlineimap/branches/twisted/offlineimap/imapserver.py
==============================================================================
--- offlineimap/branches/twisted/offlineimap/imapserver.py 2003-07-28 20:10:20 UTC (rev 541)
+++ offlineimap/branches/twisted/offlineimap/imapserver.py 2003-07-28 21:06:12 UTC (rev 542)
@@ -22,6 +22,7 @@
from twisted.internet import defer, protocol, reactor
from twisted.protocols.imap4 import IMAP4Client
from twisted.internet.ssl import ClientContextFactory
+from imaputil import rd
class MyIMAP(IMAP4Client):
def connectionMade(self):
@@ -72,6 +73,8 @@
self.maxconnections = maxconnections
self.availableconnections = []
self.assignedconnections = []
+ self.waitingforconnections = []
+ self.openingcount = 0
self.lastowner = {}
#self.semaphore = BoundedSemaphore(self.maxconnections)
self.connectionlock = Lock()
@@ -126,7 +129,8 @@
print "reference is", self.reference
d.addCallback(self.listcallback, self.reference, '')
d.addCallback(self.setdelim)
- d.addCallback(self.handOverConnection)
+ d.addCallback(lambda x, imap: self._handleAvailableConnection(imap),
+ imapobj)
def listcallback(self, discarded, reference, path):
@@ -151,73 +155,83 @@
UIBase.getglobalui().warn(str(error))
self.plainauth(imapobj)
- def acquireconnection(self):
+ def acquireconnection(self, d):
"""Fetches a connection from the pool, making sure to create a new one
if needed, to obey the maximum connection limits, etc.
Opens a connection to the server and returns an appropriate
- object."""
+ object.
- # FIXME: stub for now.
- return self.imapobj
+ Pass to it a Deferred. This is the object that you should be returning
+ from your calling function. When you hit callback() on it, the
+ connection will be eventually closed."""
+ d2 = self._grabConnection()
+ d2.addCallback(self._gotConnection, d)
+ d2.chainCallback(d)
- def initconnection(self):
- print "initconection called."
+ def _getConnectionCount(self):
+ return len(self.availableconnections) + \
+ len(self.assignedconnections) + \
+ self.openingcount
+
+ def _grabConnection(self):
+ if len(self.availableconnections):
+ # Something is available. Just grab it.
+ conn = self.availableconnections.pop(0)
+ self.assignedconnections.append(conn)
+ return rd(conn)
+
+ # Noting is available... We're going to have to add the caller
+ # to the wait queue regardless of what happens next. So...
d = defer.Deferred()
- if hasattr(self, 'imapobj'):
- print "Acquireconnection returning existing object."
- d.callback(self.imapobj)
+ self.waitingforconnections.append(d)
+ if self._getConnectionCount() >= self.maxconnections:
+ # Noting is available, and we can't create anything new,
+ # so we put the caller into the wait queue.
return d
+ # OK, at this point, nothing is available, but we DO have space
+ # to generate a new connection. Do it.
+ self._newconnection()
+ return d
+ def _gotConnection(self, connection, d):
+ d.addBoth(self._releaseConnection, connection)
+ return connection
+
+ def _releaseConnection(self, stuff, connection):
+ # Take it out of the active list
+ self.assignedconnections.remove(connection)
+ # And hand it off to the next consumer or queue
+ self._handleAvailableConnection(connection)
+
+ def _handleAvailableConnection(self, connection)
+ if len(self.waitingforconnections):
+ # Somebody is waiting for connections. Give it to him and
+ # wake him up.
+ self.waitingforconnections.pop(0).callback(connection)
+ else:
+ # Just add it to the available queue.
+ self.availableconnections.append(connection)
+
+ def _newconnection(self):
+ print "_newconection called."
+
if self.usessl:
print "Using SSL."
reactor.connectSSL(self.hostname, self.port,
IMAPClientFactory(self),
ClientContextFactory())
else:
+ print "Not using SSL"
reactor.connectTCP(self.hostname, self.port,
IMAPClientFactory(self))
- self.connectiondefer = d
- return d
-
def connectionMade(self, protocol):
print "This is IMAPServer.connectionMade; handing over to plainauth"
- self.imapobj = protocol
print "Connectionmade: imapobj is", self.imapobj
- self.plainauth(protocol)
+ return self.plainauth(protocol)
+
+ # FIXME:
- def handOverConnection(self, d = None):
- print "This is IMAPServer.handOverConnection"
- self.connectiondefer.callback(self.imapobj)
- del self.connectiondefer
- return
-
- self.semaphore.acquire()
- self.connectionlock.acquire()
- imapobj = None
-
- if len(self.availableconnections): # One is available.
- # Try to find one that previously belonged to this thread
- # as an optimization. Start from the back since that's where
- # they're popped on.
- threadid = thread.get_ident()
- imapobj = None
- for i in range(len(self.availableconnections) - 1, -1, -1):
- tryobj = self.availableconnections[i]
- if self.lastowner[tryobj] == threadid:
- imapobj = tryobj
- del(self.availableconnections[i])
- break
- if not imapobj:
- imapobj = self.availableconnections[0]
- del(self.availableconnections[0])
- self.assignedconnections.append(imapobj)
- self.lastowner[imapobj] = thread.get_ident()
- self.connectionlock.release()
- return imapobj
-
- self.connectionlock.release() # Release until need to modify data
-
success = 0
while not success:
# Generate a new connection.
Modified: offlineimap/branches/twisted/offlineimap/imaputil.py
==============================================================================
--- offlineimap/branches/twisted/offlineimap/imaputil.py 2003-07-28 20:10:20 UTC (rev 541)
+++ offlineimap/branches/twisted/offlineimap/imaputil.py 2003-07-28 21:06:12 UTC (rev 542)
@@ -195,7 +195,22 @@
return ",".join(retval)
+class helper(self, destfunc, args, kwargs):
+ d = defer.Deferred()
+ d.addCallbacks(destfunc, callbackArgs=args,
+ callbackKeywords=kwargs)
+ self.imapserver.acquireconnection(d)
+ return d
-
-
-
+class AcquireMixin:
+ def _initAcquires(self):
+ acquirelist = []
+ for item in dir(self):
+ match = re.search('^_(.+)_acquired$', item)
+ if match:
+ acquirelist.append(match.group(1))
+ for acquirename in acquirelist:
+ newname = '_' + name + '_acquired'
+ func = lambda self, *args, **kwargs: \
+ helper(self, getattr(self, newname), args, kwargs)
+ setattr(self, name, func)
Modified: offlineimap/branches/twisted/offlineimap/repository/IMAP.py
==============================================================================
--- offlineimap/branches/twisted/offlineimap/repository/IMAP.py 2003-07-28 20:10:20 UTC (rev 541)
+++ offlineimap/branches/twisted/offlineimap/repository/IMAP.py 2003-07-28 21:06:12 UTC (rev 542)
@@ -20,12 +20,13 @@
from offlineimap import folder, imaputil, imapserver
from offlineimap.folder.UIDMaps import MappedIMAPFolder
import re, types, os
-from threading import *
+from imaputil import rd
-class IMAPRepository(BaseRepository):
+class IMAPRepository(BaseRepository, imaputil.AcquireMixin):
def __init__(self, reposname, account):
"""Initialize an IMAPRepository object."""
BaseRepository.__init__(self, reposname, account)
+ self._initAcquires()
self.imapserver = imapserver.ConfigedIMAPServer(self)
self.folders = None
self.nametrans = lambda foldername: foldername
@@ -129,17 +130,15 @@
def getfolders(self):
if self.folders != None:
- return self.folders
- imapobj = self.imapserver.acquireconnection()
- #imapobj.addCallback(self.getfolders_real)
- #return imapobj
- # FIXME: use self.imapserver.reference
- print "Reference: '%s'" % self.imapserver.reference
- d = imapobj.list('', '*')
+ return rd(self.folders)
+ return self.getfolders_needdata()
+
+ def _getfolders_needdata_acquired(self, imapobj)
+ d = imapobj.list(self.imapserver.reference, '*')
d.addCallback(self.getfolders_real)
return d
- def getfolders_real(self, listresult):
+ def _getfolders_real(self, listresult):
retval = []
for flags, delim, name in listresult:
print type(flags)
@@ -174,6 +173,8 @@
return retval
def makefolder(self, foldername):
+ #FIXME#
+ raise NotImplementedError
#if self.getreference() != '""':
# newname = self.getreference() + self.getsep() + foldername
#else: