summaryrefslogtreecommitdiff
path: root/src/leap/mail/imap
diff options
context:
space:
mode:
authorKali Kaneko <kali@leap.se>2014-02-05 23:44:23 -0400
committerKali Kaneko <kali@leap.se>2014-02-17 11:37:03 -0400
commit06556ec6dc56a4859736fc2782779ee2eb9c1f55 (patch)
treefceb5909a88da9bc185d7474eba6e1ddc50c8fa9 /src/leap/mail/imap
parentc955c7015b5986af40b2253ac98846f4547e5e00 (diff)
defer parse to thread
Diffstat (limited to 'src/leap/mail/imap')
-rw-r--r--src/leap/mail/imap/memorystore.py4
-rw-r--r--src/leap/mail/imap/messages.py72
2 files changed, 29 insertions, 47 deletions
diff --git a/src/leap/mail/imap/memorystore.py b/src/leap/mail/imap/memorystore.py
index d0321ae..8deddda 100644
--- a/src/leap/mail/imap/memorystore.py
+++ b/src/leap/mail/imap/memorystore.py
@@ -230,6 +230,8 @@ class MemoryStore(object):
be fired.
:type notify_on_disk: bool
"""
+ from twisted.internet import reactor
+
log.msg("adding new doc to memstore %r (%r)" % (mbox, uid))
key = mbox, uid
@@ -251,7 +253,7 @@ class MemoryStore(object):
if not notify_on_disk:
# Caller does not care, just fired and forgot, so we pass
# a defer that will inmediately have its callback triggered.
- observer.callback(uid)
+ reactor.callLater(0, observer.callback, uid)
def put_message(self, mbox, uid, message, notify_on_disk=True):
"""
diff --git a/src/leap/mail/imap/messages.py b/src/leap/mail/imap/messages.py
index 25fc55f..89beaaa 100644
--- a/src/leap/mail/imap/messages.py
+++ b/src/leap/mail/imap/messages.py
@@ -78,7 +78,7 @@ def try_unique_query(curried):
# TODO we could take action, like trigger a background
# process to kill dupes.
name = getattr(curried, 'expected', 'doc')
- logger.warning(
+ logger.debug(
"More than one %s found for this mbox, "
"we got a duplicate!!" % (name,))
return query.pop()
@@ -720,9 +720,6 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser):
# ensure that we have a recent-flags and a hdocs-sec doc
self._get_or_create_rdoc()
- # Not for now...
- #self._get_or_create_hdocset()
-
def _get_empty_doc(self, _type=FLAGS_DOC):
"""
Returns an empty doc for storing different message parts.
@@ -758,21 +755,26 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser):
hdocset[fields.MBOX_KEY] = self.mbox
self._soledad.create_doc(hdocset)
+ @deferred_to_thread
def _do_parse(self, raw):
"""
Parse raw message and return it along with
relevant information about its outer level.
+ This is done in a separate thread, and the callback is passed
+ to `_do_add_msg` method.
+
:param raw: the raw message
:type raw: StringIO or basestring
- :return: msg, chash, size, multi
+ :return: msg, parts, chash, size, multi
:rtype: tuple
"""
msg = self._get_parsed_msg(raw)
chash = self._get_hash(msg)
size = len(msg.as_string())
multi = msg.is_multipart()
- return msg, chash, size, multi
+ parts = walk.get_parts(msg)
+ return msg, parts, chash, size, multi
def _populate_flags(self, flags, uid, chash, size, multi):
"""
@@ -879,19 +881,25 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser):
flags = tuple()
leap_assert_type(flags, tuple)
- d = defer.Deferred()
- self._do_add_msg(raw, flags, subject, date, notify_on_disk, d)
- return d
+ observer = defer.Deferred()
+
+ d = self._do_parse(raw)
+ d.addCallback(self._do_add_msg, flags, subject, date,
+ notify_on_disk, observer)
+ return observer
- # We SHOULD defer this (or the heavy load here) to the thread pool,
+ # We SHOULD defer the heavy load here) to the thread pool,
# but it gives troubles with the QSocketNotifier used by Qt...
- def _do_add_msg(self, raw, flags, subject, date, notify_on_disk, observer):
+ def _do_add_msg(self, parse_result, flags, subject,
+ date, notify_on_disk, observer):
"""
Helper that creates a new message document.
Here lives the magic of the leap mail. Well, in soledad, really.
See `add_msg` docstring for parameter info.
+ :param parse_result: a tuple with the results of `self._do_parse`
+ :type parse_result: tuple
:param observer: a deferred that will be fired with the message
uid when the adding succeed.
:type observer: deferred
@@ -902,26 +910,17 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser):
# TODO add the linked-from info !
# TODO add reference to the original message
- # parse
- msg, chash, size, multi = self._do_parse(raw)
+ from twisted.internet import reactor
+ msg, parts, chash, size, multi = parse_result
# check for uniqueness --------------------------------
- # XXX profiler says that this test is costly.
- # So we probably should just do an in-memory check and
- # move the complete check to the soledad writer?
# Watch out! We're reserving a UID right after this!
existing_uid = self._fdoc_already_exists(chash)
if existing_uid:
- logger.warning("We already have that message in this "
- "mailbox, unflagging as deleted")
uid = existing_uid
msg = self.get_msg_by_uid(uid)
- msg.setFlags((fields.DELETED_FLAG,), -1)
-
- # XXX if this is deferred to thread again we should not use
- # the callback in the deferred thread, but return and
- # call the callback from the caller fun...
- observer.callback(uid)
+ reactor.callLater(0, msg.setFlags, (fields.DELETED_FLAG,), -1)
+ reactor.callLater(0, observer.callback, uid)
return
uid = self.memstore.increment_last_soledad_uid(self.mbox)
@@ -930,7 +929,6 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser):
fd = self._populate_flags(flags, uid, chash, size, multi)
hd = self._populate_headr(msg, chash, subject, date)
- parts = walk.get_parts(msg)
body_phash_fun = [walk.get_body_phash_simple,
walk.get_body_phash_multi][int(multi)]
body_phash = body_phash_fun(walk.get_payloads(msg))
@@ -949,9 +947,9 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser):
self.set_recent_flag(uid)
msg_container = MessageWrapper(fd, hd, cdocs)
- self.memstore.create_message(self.mbox, uid, msg_container,
- observer=observer,
- notify_on_disk=notify_on_disk)
+ self.memstore.create_message(
+ self.mbox, uid, msg_container,
+ observer=observer, notify_on_disk=notify_on_disk)
#
# getters: specific queries
@@ -982,14 +980,6 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser):
{'doc_id': rdoc.doc_id, 'set': rflags})
return rflags
- #else:
- # fallback for cases without memory store
- #with self._rdoc_lock:
- #rdoc = self._get_recent_doc()
- #self.__rflags = set(rdoc.content.get(
- #fields.RECENTFLAGS_KEY, []))
- #return self.__rflags
-
def _set_recent_flags(self, value):
"""
Setter for the recent-flags set for this mailbox.
@@ -997,16 +987,6 @@ class MessageCollection(WithMsgFields, IndexedDB, MailParser, MBoxParser):
if self.memstore is not None:
self.memstore.set_recent_flags(self.mbox, value)
- #else:
- # fallback for cases without memory store
- #with self._rdoc_lock:
- #rdoc = self._get_recent_doc()
- #newv = set(value)
- #self.__rflags = newv
- #rdoc.content[fields.RECENTFLAGS_KEY] = list(newv)
- # XXX should deferLater 0 it?
- #self._soledad.put_doc(rdoc)
-
recent_flags = property(
_get_recent_flags, _set_recent_flags,
doc="Set of UIDs with the recent flag for this mailbox.")