class Maildir < Source
SCAN_INTERVAL = 30 # seconds
+ MYHOSTNAME = Socket.gethostname
## remind me never to use inheritance again.
- yaml_properties :uri, :cur_offset, :usual, :archived, :id, :labels
- def initialize uri, last_date=nil, usual=true, archived=false, id=nil, labels=[]
+ yaml_properties :uri, :cur_offset, :usual, :archived, :id, :labels, :mtimes
+ def initialize uri, last_date=nil, usual=true, archived=false, id=nil, labels=[], mtimes={}
super uri, last_date, usual, archived, id
uri = URI(Source.expand_filesystem_uri(uri))
raise ArgumentError, "not a maildir URI" unless uri.scheme == "maildir"
raise ArgumentError, "maildir URI cannot have a host: #{uri.host}" if uri.host
+ raise ArgumentError, "maildir URI must have a path component" unless uri.path
@dir = uri.path
@labels = (labels || []).freeze
@ids_to_fns = {}
@last_scan = nil
@mutex = Mutex.new
+ #the mtime from the subdirs in the maildir with the unix epoch as default.
+ #these are used to determine whether scanning the directory for new mail
+ #is a worthwhile effort
+ @mtimes = { 'cur' => Time.at(0), 'new' => Time.at(0) }.merge(mtimes || {})
+ @dir_ids = { 'cur' => [], 'new' => [] }
end
def file_path; @dir end
start = @ids.index(cur_offset || start_offset) or raise OutOfSyncSourceError, "Unknown message id #{cur_offset || start_offset}." # couldn't find the most recent email
end
+ def store_message date, from_email, &block
+ stored = false
+ new_fn = new_maildir_basefn + ':2,S'
+ Dir.chdir(@dir) do |d|
+ tmp_path = File.join(@dir, 'tmp', new_fn)
+ new_path = File.join(@dir, 'new', new_fn)
+ begin
+ sleep 2 if File.stat(tmp_path)
+
+ File.stat(tmp_path)
+ rescue Errno::ENOENT #this is what we want.
+ begin
+ File.open(tmp_path, 'w') do |f|
+ yield f #provide a writable interface for the caller
+ f.fsync
+ end
+
+ File.link tmp_path, new_path
+ stored = true
+ ensure
+ File.unlink tmp_path if File.exists? tmp_path
+ end
+ end #rescue Errno...
+ end #Dir.chdir
+
+ stored
+ end
+
+ def each_raw_message_line id
+ scan_mailbox
+ with_file_for(id) do |f|
+ until f.eof?
+ yield f.gets
+ end
+ end
+ end
+
def load_header id
scan_mailbox
- with_file_for(id) { |f| MBox::read_header f }
+ with_file_for(id) { |f| parse_raw_email_header f }
end
def load_message id
def raw_message id
scan_mailbox
- with_file_for(id) { |f| f.readlines.join }
+ with_file_for(id) { |f| f.read }
end
- def scan_mailbox
+ def scan_mailbox opts={}
+ return unless @ids.empty? || opts[:rescan]
return if @last_scan && (Time.now - @last_scan) < SCAN_INTERVAL
- cdir = File.join(@dir, 'cur')
- ndir = File.join(@dir, 'new')
-
- raise FatalSourceError, "#{cdir} not a directory" unless File.directory? cdir
- raise FatalSourceError, "#{ndir} not a directory" unless File.directory? ndir
+ initial_poll = @ids.empty?
+ Redwood::log "scanning maildir #@dir..."
begin
- @ids, @ids_to_fns = @mutex.synchronize do
- ids, ids_to_fns = [], {}
- (Dir[File.join(cdir, "*")] + Dir[File.join(ndir, "*")]).map do |fn|
- id = make_id fn
- ids << id
- ids_to_fns[id] = fn
- end
- [ids.sort, ids_to_fns]
+ @mtimes.each_key do |d|
+ subdir = File.join(@dir, d)
+ raise FatalSourceError, "#{subdir} not a directory" unless File.directory? subdir
+
+ mtime = File.mtime subdir
+
+ #only scan the dir if the mtime is more recent (or we haven't polled
+ #since startup)
+ if @mtimes[d] < mtime || initial_poll
+ @mtimes[d] = mtime
+ @dir_ids[d] = []
+ Dir[File.join(subdir, '*')].map do |fn|
+ id = make_id fn
+ @dir_ids[d] << id
+ @ids_to_fns[id] = fn
+ end
+ else
+ Redwood::log "no poll on #{d}. mtime on indicates no new messages."
+ end
end
+ @ids = @dir_ids.values.flatten.uniq.sort!
rescue SystemCallError, IOError => e
raise FatalSourceError, "Problem scanning Maildir directories: #{e.message}."
end
+ Redwood::log "done scanning maildir"
@last_scan = Time.now
end
+ synchronized :scan_mailbox
def each
scan_mailbox
end
def end_offset
- scan_mailbox
- @ids.last
+ scan_mailbox :rescan => true
+ @ids.last + 1
end
def pct_done; 100.0 * (@ids.index(cur_offset) || 0).to_f / (@ids.length - 1).to_f; end
private
def make_id fn
+ #doing this means 1 syscall instead of 2 (File.mtime, File.size).
+ #makes a noticeable difference on nfs.
+ stat = File.stat(fn)
# use 7 digits for the size. why 7? seems nice.
- sprintf("%d%07d", File.mtime(fn), File.size(fn)).to_i
+ sprintf("%d%07d", stat.mtime, stat.size % 10000000).to_i
+ end
+
+ def new_maildir_basefn
+ Kernel::srand()
+ "#{Time.now.to_i.to_s}.#{$$}#{Kernel.rand(1000000)}.#{MYHOSTNAME}"
end
def with_file_for id
projects / sup / blobdiff