lib/inbox-sync/sync.rb in inbox-sync-0.3.0 vs lib/inbox-sync/sync.rb in inbox-sync-0.4.0

@@ -1,10 +1,11 @@
 require 'net/imap'
 require 'net/smtp'
 
 require 'inbox-sync/config'
 require 'inbox-sync/filter_actions'
+require 'inbox-sync/sync/mail_item_group'
 require 'inbox-sync/notice/sync_mail_item_error'
 
 module InboxSync
 
   class Sync
@@ -51,32 +52,43 @@
       logout if logged_in?
       @source_imap = @notify_smtp = nil
       logger.info "=== #{config_log_detail(@config.source)} sync finished. ==="
     end
 
-    def run(runner=nil)
-      return if runner && runner.shutdown?
-      each_source_mail_item(runner) do |mail_item|
-        begin
-          logger.debug "** #{mail_item.inspect}"
-          response = send_to_dest(mail_item)
-          dest_uid = parse_append_response_uid(response)
-          apply_dest_filters(dest_uid)
-        rescue Exception => err
-          log_error(err)
-          notify(Notice::SyncMailItemError.new(@notify_smtp, @config.notify, {
-            :error => err,
-            :mail_item => mail_item,
-            :sync => self
-          }))
-        ensure
-          archive_on_source(mail_item)
-          mail_item = nil
-        end
+    # this splits the mail_items list into `@config.max_threads` lists
+    # this spreads mails evenly across the groups with earlier items
+    # appearing earliest in each list
+
+    def mail_item_groups
+      num_groups = @config.max_threads
+      groups = []
+      num_groups.times { groups << MailItemGroup.new(self) }
+      get_mail_items.each_with_index do |item, i|
+        groups[i % num_groups].add(item)
       end
+      groups
     end
 
+    def run(mail_item)
+      begin
+        logger.debug "** #{mail_item.inspect}"
+        response = send_to_dest(mail_item)
+        dest_uid = parse_append_response_uid(response)
+        apply_dest_filters(dest_uid)
+      rescue Exception => err
+        log_error(err)
+        notify(Notice::SyncMailItemError.new(@notify_smtp, @config.notify, {
+          :error => err,
+          :mail_item => mail_item,
+          :sync => self
+        }))
+      ensure
+        archive_on_source(mail_item)
+        mail_item = nil
+      end
+    end
+
     def notify(notice)
       logger.info "** sending '#{notice.subject}' to #{notice.to.inspect}"
       begin
         notice.send
       rescue Exception => err
@@ -96,25 +108,10 @@
       logout_imap(@source_imap, @config.source)
       @logged_in = false
       true
     end
 
-    def each_source_mail_item(runner=nil)
-      logger.info "* find: #{config_log_detail(@config.source)}, #{@config.source.inbox.inspect}..."
-      items = MailItem.find(@source_imap)
-      logger.info "* ...found #{items.size} mail items"
-
-      items.each do |mail_item|
-        if runner && runner.shutdown?
-          logger.info "* the runner has been shutdown - aborting the sync"
-          break
-        end
-        yield mail_item
-      end
-      items = nil
-    end
-
     # Send a mail item to the destination:
     # The idea here is that destinations may not accept corrupted or invalid
     # mail items.  If appending the original mail item in any way fails,
     # create a stripped down version of the mail item and try to append that.
     # If appending the stripped down version still fails, error on up
@@ -219,9 +216,16 @@
     def expunge_imap(imap, config)
       if config.expunge
         logger.debug "* EXPUNGE #{config.inbox.inspect}: #{config_log_detail(config)}"
         imap.expunge
       end
+    end
+
+    def get_mail_items
+      logger.info "* find: #{config_log_detail(@config.source)}, #{@config.source.inbox.inspect}..."
+      items = MailItem.find(@source_imap)
+      logger.info "* ...found #{items.size} mail items"
+      items
     end
 
     def logout_imap(imap, config)
       logger.debug "* LOGOUT: #{config_log_detail(config)}"
       imap.logout