git.cworth.org Git - sup/blob - lib/sup/mbox/loader.rb

   1 require 'rmail'
   2 require 'uri'
   3
   4 module Redwood
   5 module MBox
   6
   7 class Loader < Source
   8   yaml_properties :uri, :cur_offset, :usual, :archived, :id, :labels
   9   def initialize uri_or_fp, start_offset=nil, usual=true, archived=false, id=nil, labels=[]
  10     super uri_or_fp, start_offset, usual, archived, id
  11
  12     @mutex = Mutex.new
  13     @labels = (labels || []).freeze
  14
  15     case uri_or_fp
  16     when String
  17       uri = URI(uri_or_fp)
  18       raise ArgumentError, "not an mbox uri" unless uri.scheme == "mbox"
  19       raise ArgumentError, "mbox uri ('#{uri}') cannot have a host: #{uri.host}" if uri.host
  20       @f = File.open uri.path
  21     else
  22       @f = uri_or_fp
  23     end
  24   end
  25
  26   def file_path; URI(uri).path end
  27
  28   def self.suggest_labels_for path
  29     ## heuristic: use the filename as a label, unless the file
  30     ## has a path that probably represents an inbox.
  31     if File.dirname(path) =~ /\b(var|usr|spool)\b/
  32       []
  33     else
  34       [File.basename(path).intern]
  35     end
  36   end
  37
  38   def check
  39     if (cur_offset ||= start_offset) > end_offset
  40       raise OutOfSyncSourceError, "mbox file is smaller than last recorded message offset. Messages have probably been deleted by another client."
  41     end
  42   end
  43
  44   def start_offset; 0; end
  45   def end_offset; File.size @f; end
  46
  47   def load_header offset
  48     header = nil
  49     @mutex.synchronize do
  50       @f.seek offset
  51       l = @f.gets
  52       unless l =~ BREAK_RE
  53         raise OutOfSyncSourceError, "mismatch in mbox file offset #{offset.inspect}: #{l.inspect}."
  54       end
  55       header = MBox::read_header @f
  56     end
  57     header
  58   end
  59
  60   def load_message offset
  61     @mutex.synchronize do
  62       @f.seek offset
  63       begin
  64         RMail::Mailbox::MBoxReader.new(@f).each_message do |input|
  65           return RMail::Parser.read(input)
  66         end
  67       rescue RMail::Parser::Error => e
  68         raise FatalSourceError, "error parsing mbox file: #{e.message}"
  69       end
  70     end
  71   end
  72
  73   def raw_header offset
  74     ret = ""
  75     @mutex.synchronize do
  76       @f.seek offset
  77       until @f.eof? || (l = @f.gets) =~ /^$/
  78         ret += l
  79       end
  80     end
  81     ret
  82   end
  83
  84   def raw_full_message offset
  85     ret = ""
  86     each_raw_full_message_line(offset) { |l| ret += l }
  87     ret
  88   end
  89
  90   ## apparently it's a million times faster to call this directly if
  91   ## we're just moving messages around on disk, than reading things
  92   ## into memory with raw_full_message.
  93   ##
  94   ## i hoped never to have to move shit around on disk but
  95   ## sup-sync-back has to do it.
  96   def each_raw_full_message_line offset
  97     @mutex.synchronize do
  98       @f.seek offset
  99       yield @f.gets
 100       until @f.eof? || (l = @f.gets) =~ BREAK_RE
 101         yield l
 102       end
 103     end
 104   end
 105
 106   def next
 107     returned_offset = nil
 108     next_offset = cur_offset
 109
 110     begin
 111       @mutex.synchronize do
 112         @f.seek cur_offset
 113
 114         ## cur_offset could be at one of two places here:
 115
 116         ## 1. before a \n and a mbox separator, if it was previously at
 117         ##    EOF and a new message was added; or,
 118         ## 2. at the beginning of an mbox separator (in all other
 119         ##    cases).
 120
 121         l = @f.gets or raise "next while at EOF"
 122         if l =~ /^\s*$/ # case 1
 123           returned_offset = @f.tell
 124           @f.gets # now we're at a BREAK_RE, so skip past it
 125         else # case 2
 126           returned_offset = cur_offset
 127           ## we've already skipped past the BREAK_RE, so just go
 128         end
 129
 130         while(line = @f.gets)
 131           break if line =~ BREAK_RE
 132           next_offset = @f.tell
 133         end
 134       end
 135     rescue SystemCallError, IOError => e
 136       raise FatalSourceError, "Error reading #{@f.path}: #{e.message}"
 137     end
 138
 139     self.cur_offset = next_offset
 140     [returned_offset, @labels]
 141   end
 142 end
 143
 144 end
 145 end