lib/sup/poll.rb (10166B) - raw
1 require 'thread'
2
3 module Redwood
4
5 class PollManager
6 include Redwood::Singleton
7
8 HookManager.register "before-add-message", <<EOS
9 Executes immediately before a message is added to the index.
10 Variables:
11 message: the new message
12 EOS
13
14 HookManager.register "before-poll", <<EOS
15 Executes immediately before a poll for new messages commences.
16 No variables.
17 EOS
18
19 HookManager.register "after-poll", <<EOS
20 Executes immediately after a poll for new messages completes.
21 Variables:
22 num: the total number of new messages added in this poll
23 num_inbox: the number of new messages added in this poll which
24 appear in the inbox (i.e. were not auto-archived).
25 num_total: the total number of messages
26 num_inbox_total: the total number of new messages in the inbox.
27 num_inbox_total_unread: the total number of unread messages in the inbox
28 num_updated: the total number of updated messages
29 num_deleted: the total number of deleted messages
30 labels: the labels that were applied
31 from_and_subj: an array of (from email address, subject) pairs
32 from_and_subj_inbox: an array of (from email address, subject) pairs for
33 only those messages appearing in the inbox
34 EOS
35
36 def initialize
37 @delay = $config[:poll_interval] || 300
38 @mutex = Mutex.new
39 @thread = nil
40 @last_poll = nil
41 @polling = Mutex.new
42 @poll_sources = nil
43 @mode = nil
44 @should_clear_running_totals = false
45 clear_running_totals # defines @running_totals
46 UpdateManager.register self
47 end
48
49 def poll_with_sources
50 @mode ||= PollMode.new
51
52 if HookManager.enabled? "before-poll"
53 HookManager.run("before-poll")
54 else
55 BufferManager.flash "Polling for new messages..."
56 end
57
58 num, numi, numu, numd, from_and_subj, from_and_subj_inbox, loaded_labels = @mode.poll
59 clear_running_totals if @should_clear_running_totals
60 @running_totals[:num] += num
61 @running_totals[:numi] += numi
62 @running_totals[:numu] += numu
63 @running_totals[:numd] += numd
64 @running_totals[:loaded_labels] += loaded_labels || []
65
66
67 if HookManager.enabled? "after-poll"
68 hook_args = { :num => num, :num_inbox => numi,
69 :num_total => @running_totals[:num], :num_inbox_total => @running_totals[:numi],
70 :num_updated => @running_totals[:numu],
71 :num_deleted => @running_totals[:numd],
72 :labels => @running_totals[:loaded_labels],
73 :from_and_subj => from_and_subj, :from_and_subj_inbox => from_and_subj_inbox,
74 :num_inbox_total_unread => lambda { Index.num_results_for :labels => [:inbox, :unread] } }
75
76 HookManager.run("after-poll", hook_args)
77 else
78 if @running_totals[:num] > 0
79 flash_msg = "Loaded #{@running_totals[:num].pluralize 'new message'}, #{@running_totals[:numi]} to inbox. " if @running_totals[:num] > 0
80 flash_msg += "Updated #{@running_totals[:numu].pluralize 'message'}. " if @running_totals[:numu] > 0
81 flash_msg += "Deleted #{@running_totals[:numd].pluralize 'message'}. " if @running_totals[:numd] > 0
82 flash_msg += "Labels: #{@running_totals[:loaded_labels].map{|l| l.to_s}.join(', ')}." if @running_totals[:loaded_labels].size > 0
83 BufferManager.flash flash_msg
84 else
85 BufferManager.flash "No new messages."
86 end
87 end
88
89 end
90
91 def poll
92 if @polling.try_lock
93 @poll_sources = SourceManager.usual_sources
94 num, numi = poll_with_sources
95 @polling.unlock
96 [num, numi]
97 else
98 debug "poll already in progress."
99 return
100 end
101 end
102
103 def poll_unusual
104 if @polling.try_lock
105 @poll_sources = SourceManager.unusual_sources
106 num, numi = poll_with_sources
107 @polling.unlock
108 [num, numi]
109 else
110 debug "poll_unusual already in progress."
111 return
112 end
113 end
114
115 def start
116 @thread = Redwood::reporting_thread("periodic poll") do
117 while true
118 sleep @delay / 2
119 poll if @last_poll.nil? || (Time.now - @last_poll) >= @delay
120 end
121 end
122 end
123
124 def stop
125 @thread.kill if @thread
126 @thread = nil
127 end
128
129 def do_poll
130 total_num = total_numi = total_numu = total_numd = 0
131 from_and_subj = []
132 from_and_subj_inbox = []
133 loaded_labels = Set.new
134
135 @mutex.synchronize do
136 @poll_sources.each do |source|
137 begin
138 yield "Loading from #{source}... "
139 rescue SourceError => e
140 warn "problem getting messages from #{source}: #{e.message}"
141 next
142 end
143
144 msg = ""
145 num = numi = numu = numd = 0
146 poll_from source do |action,m,old_m,progress|
147 if action == :delete
148 yield "Deleting #{m.id}"
149 loaded_labels.merge m.labels
150 numd += 1
151 elsif action == :update
152 yield "Message at #{m.source_info} is an update of an old message. Updating labels from #{old_m.labels.to_a * ','} => #{m.labels.to_a * ','}"
153 loaded_labels.merge m.labels
154 numu += 1
155 elsif action == :add
156 if old_m
157 new_locations = (m.locations - old_m.locations)
158 if not new_locations.empty?
159 yield "Message at #{new_locations[0].info} has changed its source location. Updating labels from #{old_m.labels.to_a * ','} => #{m.labels.to_a * ','}"
160 numu += 1
161 else
162 yield "Skipping already-imported message at #{m.locations[-1].info}"
163 end
164 else
165 yield "Found new message at #{m.source_info} with labels #{m.labels.to_a * ','}"
166 loaded_labels.merge m.labels
167 num += 1
168 from_and_subj << [m.from && m.from.longname, m.subj]
169 if (m.labels & [:inbox, :spam, :deleted, :killed]) == Set.new([:inbox])
170 from_and_subj_inbox << [m.from && m.from.longname, m.subj]
171 numi += 1
172 end
173 end
174 else fail
175 end
176 end
177 msg += "Found #{num} messages, #{numi} to inbox. " unless num == 0
178 msg += "Updated #{numu} messages. " unless numu == 0
179 msg += "Deleted #{numd} messages." unless numd == 0
180 yield msg unless msg == ""
181 total_num += num
182 total_numi += numi
183 total_numu += numu
184 total_numd += numd
185 end
186
187 loaded_labels = loaded_labels - LabelManager::HIDDEN_RESERVED_LABELS - [:inbox, :killed]
188 yield "Done polling; loaded #{total_num} new messages total"
189 @last_poll = Time.now
190 end
191 [total_num, total_numi, total_numu, total_numd, from_and_subj, from_and_subj_inbox, loaded_labels]
192 end
193
194 ## like Source#poll, but yields successive Message objects, which have their
195 ## labels and locations set correctly. The Messages are saved to or removed
196 ## from the index after being yielded.
197 def poll_from source, opts={}
198 debug "trying to acquire poll lock for: #{source}..."
199 if source.try_lock
200 begin
201 source.poll do |sym, args|
202 case sym
203 when :add
204 m = Message.build_from_source source, args[:info]
205 old_m = Index.build_message m.id
206 m.labels += args[:labels]
207 m.labels.delete :inbox if source.archived?
208 m.labels.delete :unread if source.read?
209 m.labels.delete :unread if m.source_marked_read? # preserve read status if possible
210 m.labels.each { |l| LabelManager << l }
211 m.labels = old_m.labels + (m.labels - [:unread, :inbox]) if old_m
212 m.locations = old_m.locations + m.locations if old_m
213 HookManager.run "before-add-message", :message => m
214 yield :add, m, old_m, args[:progress] if block_given?
215 Index.sync_message m, true
216
217 if Index.message_joining_killed? m
218 m.labels += [:killed]
219 Index.sync_message m, true
220 end
221
222 ## We need to add or unhide the message when it either did not exist
223 ## before at all or when it was updated. We do *not* add/unhide when
224 ## the same message was found at a different location
225 if old_m
226 UpdateManager.relay self, :updated, m
227 elsif !old_m or not old_m.locations.member? m.location
228 UpdateManager.relay self, :added, m
229 end
230 when :delete
231 Index.each_message({:location => [source.id, args[:info]]}, false) do |m|
232 m.locations.delete Location.new(source, args[:info])
233 Index.sync_message m, false
234 if m.locations.size == 0
235 yield :delete, m, [source,args[:info]], args[:progress] if block_given?
236 Index.delete m.id
237 UpdateManager.relay self, :location_deleted, m
238 end
239 end
240 when :update
241 Index.each_message({:location => [source.id, args[:old_info]]}, false) do |m|
242 old_m = Index.build_message m.id
243 m.locations.delete Location.new(source, args[:old_info])
244 m.locations.push Location.new(source, args[:new_info])
245 ## Update labels that might have been modified remotely
246 m.labels -= source.supported_labels?
247 m.labels += args[:labels]
248 yield :update, m, old_m if block_given?
249 Index.sync_message m, true
250 UpdateManager.relay self, :updated, m
251 end
252 end
253 end
254
255 rescue SourceError => e
256 warn "problem getting messages from #{source}: #{e.message}"
257
258 ensure
259 source.go_idle
260 source.unlock
261 end
262 else
263 debug "source #{source} is already being polled."
264 end
265 end
266
267 def handle_idle_update sender, idle_since; @should_clear_running_totals = false; end
268 def handle_unidle_update sender, idle_since; @should_clear_running_totals = true; clear_running_totals; end
269 def clear_running_totals; @running_totals = {:num => 0, :numi => 0, :numu => 0, :numd => 0, :loaded_labels => Set.new}; end
270 end
271
272 end