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