KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > net > suberic > pooka > thread > LoadMessageThread


1 package net.suberic.pooka.thread;
2
3 import net.suberic.pooka.*;
4 import net.suberic.pooka.event.*;
5 import net.suberic.pooka.gui.LoadMessageTracker;
6 import net.suberic.pooka.gui.MessageProxy;
7 import net.suberic.pooka.gui.FolderInternalFrame;
8 import javax.mail.*;
9 import javax.mail.internet.*;
10 import java.util.*;
11
12 /**
13  * This class does the actual loading of the header information from
14  * the messages in the folder. It also is set up to communicate with
15  * a JProgessBar to show how far the loading has gotten.
16  *
17  * More specifically, this thread takes an array of Messages and a
18  * Vector of Strings which are the column values which are to be put
19  * into the table. It then loads the values into a Vector of Vectors,
20  * each of which contains the information for the Table for a group
21  * of Messages. It then throws a ChangeEvent to the listening
22  * FolderTableModel. The FolderTableModel can then get the information
23  * using the getNewMessages() function.
24  */
25
26 public class LoadMessageThread extends Thread JavaDoc {
27   private FolderInfo folderInfo;
28   private List columnValues;
29   private List loadQueue = new LinkedList();
30   private List priorityLoadQueue = new LinkedList();
31   private List messageLoadedListeners = new LinkedList();
32   private int updateCheckMilliseconds = 60000;
33   private int updateMessagesCount = 10;
34   private boolean sleeping = false;
35
36   private boolean stopped = false;
37
38   public static int NORMAL = 5;
39   public static int HIGH = 10;
40
41   /**
42    * This creates a new LoadMessageThread from a FolderInfo object.
43    */
44     
45   public LoadMessageThread(FolderInfo newFolderInfo) {
46     super("Load Message Thread - " + newFolderInfo.getFolderID());
47     folderInfo = newFolderInfo;
48     this.setPriority(1);
49   }
50   
51   public void run() {
52     int uptime = 0;
53     while (! stopped) {
54       try {
55     loadWaitingMessages();
56       } catch (Exception JavaDoc e) {
57     if (getFolderInfo().getLogger().isLoggable(java.util.logging.Level.WARNING)) {
58       e.printStackTrace();
59     }
60       }
61
62       try {
63     sleeping = true;
64     if (updateCheckMilliseconds < 1) {
65       while (updateCheckMilliseconds < 1)
66         sleep(60000);
67     } else {
68       sleep(updateCheckMilliseconds);
69     }
70     sleeping = false;
71       } catch (InterruptedException JavaDoc ie) {
72     sleeping = false;
73       }
74     }
75   }
76   
77   
78   /**
79    * Loads the messages in the queue.
80    */
81   public void loadWaitingMessages() {
82     
83     int updateCounter = 0;
84     int loadedMessageCount = 0;
85     MessageProxy mp;
86     
87     // start this load section.
88 int queueSize = getQueueSize();
89     int totalMessageCount = queueSize;
90     if (! stopped && queueSize > 0) {
91       folderInfo.getLogger().log(java.util.logging.Level.FINE, folderInfo.getFolderID() + " loading " + queueSize + " messages.");
92       
93       MessageLoadedListener display = getFolderInfo().getFolderDisplayUI();
94       if (display != null)
95     this.addMessageLoadedListener(display);
96       
97       fireMessageLoadedEvent(MessageLoadedEvent.LOADING_STARTING, 0, totalMessageCount);
98       
99       // get the batch sizes.
100 int fetchBatchSize = 50;
101       int loadBatchSize = 25;
102       try {
103     fetchBatchSize = Integer.parseInt(Pooka.getProperty("Pooka.fetchBatchSize", "50"));
104       } catch (NumberFormatException JavaDoc nfe) {
105       }
106       
107       try {
108     loadBatchSize = Integer.parseInt(Pooka.getProperty("Pooka.loadBatchSize", "25"));
109       } catch (NumberFormatException JavaDoc nfe) {
110       }
111       
112       FetchProfile fetchProfile = getFolderInfo().getFetchProfile();
113
114       // we'll stay in this while loop until the queue is empty
115 for (List messages = retrieveNextBatch(fetchBatchSize); ! stopped && messages != null; messages=retrieveNextBatch(fetchBatchSize)) {
116     totalMessageCount = messages.size() + getQueueSize() + loadedMessageCount;
117     if (Pooka.getProperty("Pooka.openFoldersInBackGround", "false").equalsIgnoreCase("true")) {
118       synchronized(folderInfo.getFolderThread().getRunLock()) {
119         try {
120           folderInfo.getFolderThread().setCurrentActionName("Loading messages.");
121           // break when either we've been stopped or we're out of messages,
122 for (int batchCount = 0; ! stopped && batchCount < messages.size(); batchCount++) {
123         mp=(MessageProxy)messages.get(batchCount);
124         
125         // if the message hasn't been fetched, then fetch fetchBatchSize
126 // worth of messages.
127 if (! mp.getMessageInfo().hasBeenFetched()) {
128           try {
129             List fetchList = new ArrayList();
130             for (int j = batchCount; fetchList.size() < fetchBatchSize && j < messages.size(); j++) {
131               MessageInfo fetchInfo = ((MessageProxy) messages.get(j)).getMessageInfo();
132               if (! fetchInfo.hasBeenFetched()) {
133             fetchList.add(fetchInfo);
134             //fetchInfo.setFetched(true);
135 }
136             }
137           
138             MessageInfo[] toFetch = new MessageInfo[fetchList.size()];
139             toFetch = (MessageInfo[]) fetchList.toArray(toFetch);
140             getFolderInfo().fetch(toFetch, fetchProfile);
141           } catch(MessagingException me) {
142             if (folderInfo.getLogger().isLoggable(java.util.logging.Level.WARNING)) {
143               
144               System.out.println("caught error while fetching for folder " + getFolderInfo().getFolderID() + ": " + me);
145               me.printStackTrace();
146             }
147           }
148           
149         }
150         
151         // now load each individual messageproxy.
152 // and refresh each message.
153 try {
154           if (! mp.isLoaded())
155           mp.loadTableInfo();
156           if (mp.needsRefresh())
157             mp.refreshMessage();
158           else if (! mp.matchedFilters()) {
159             mp.matchFilters();
160         }
161         } catch (Exception JavaDoc e) {
162           if (folderInfo.getLogger().isLoggable(java.util.logging.Level.WARNING)) {
163             e.printStackTrace();
164           }
165         }
166         
167         loadedMessageCount++;
168         if (++updateCounter >= getUpdateMessagesCount()) {
169           fireMessageLoadedEvent(MessageLoadedEvent.MESSAGES_LOADED, loadedMessageCount, totalMessageCount);
170           updateCounter = 0;
171         }
172           }
173         } finally {
174           folderInfo.getFolderThread().setCurrentActionName("");
175         }
176       } // end synchronized
177 } else {
178       // break when either we've been stopped or we're out of messages,
179 for (int batchCount = 0; ! stopped && batchCount < messages.size(); batchCount++) {
180         mp=(MessageProxy)messages.get(batchCount);
181         
182         // if the message hasn't been fetched, then fetch fetchBatchSize
183 // worth of messages.
184 if (! mp.getMessageInfo().hasBeenFetched()) {
185           try {
186         List fetchList = new ArrayList();
187         for (int j = batchCount; fetchList.size() < fetchBatchSize && j < messages.size(); j++) {
188           MessageInfo fetchInfo = ((MessageProxy) messages.get(j)).getMessageInfo();
189           if (! fetchInfo.hasBeenFetched()) {
190             fetchList.add(fetchInfo);
191             //fetchInfo.setFetched(true);
192 }
193         }
194         
195         MessageInfo[] toFetch = new MessageInfo[fetchList.size()];
196         toFetch = (MessageInfo[]) fetchList.toArray(toFetch);
197         synchronized(folderInfo.getFolderThread().getRunLock()) {
198           folderInfo.getFolderThread().setCurrentActionName("Loading messages.");
199           getFolderInfo().fetch(toFetch, fetchProfile);
200           folderInfo.getFolderThread().setCurrentActionName("");
201         }
202           } catch(MessagingException me) {
203         if (getFolderInfo().getLogger().isLoggable(java.util.logging.Level.WARNING)) {
204           System.out.println("caught error while fetching for folder " + getFolderInfo().getFolderID() + ": " + me);
205           me.printStackTrace();
206         }
207           }
208           
209         }
210         
211         // now load each individual messageproxy.
212 // and refresh each message.
213 try {
214           synchronized(folderInfo.getFolderThread().getRunLock()) {
215         try {
216           folderInfo.getFolderThread().setCurrentActionName("Loading messages.");
217           if (! mp.isLoaded())
218             mp.loadTableInfo();
219           if (mp.needsRefresh())
220             mp.refreshMessage();
221           else if (! mp.matchedFilters()) {
222             mp.matchFilters();
223           }
224         } finally {
225           folderInfo.getFolderThread().setCurrentActionName("");
226         }
227           } // synchronized
228 } catch (Exception JavaDoc e) {
229           if (folderInfo.getLogger().isLoggable(java.util.logging.Level.WARNING)) {
230         e.printStackTrace();
231           }
232         }
233         
234         loadedMessageCount++;
235         if (++updateCounter >= getUpdateMessagesCount()) {
236           fireMessageLoadedEvent(MessageLoadedEvent.MESSAGES_LOADED, loadedMessageCount, totalMessageCount);
237           updateCounter = 0;
238         }
239       }
240     }
241       }
242
243       if (updateCounter > 0)
244     fireMessageLoadedEvent(MessageLoadedEvent.MESSAGES_LOADED, loadedMessageCount, totalMessageCount);
245       
246       fireMessageLoadedEvent(MessageLoadedEvent.LOADING_COMPLETE, loadedMessageCount, totalMessageCount);
247       
248       if (display != null)
249     removeMessageLoadedListener(display);
250     }
251   }
252   
253   /**
254    * Fires a new MessageLoadedEvent to each registered MessageLoadedListener.
255    */
256   public void fireMessageLoadedEvent(int type, int numMessages, int max) {
257     for (int i = 0; i < messageLoadedListeners.size(); i ++) {
258       ((MessageLoadedListener)messageLoadedListeners.get(i)).handleMessageLoaded(new MessageLoadedEvent(this, type, numMessages, max));
259     }
260   }
261   
262   /**
263    * Adds a MessageLoadedListener to the messageLoadedListener list.
264    */
265   public void addMessageLoadedListener(MessageLoadedListener newListener) {
266     if (messageLoadedListeners.indexOf(newListener) == -1)
267       messageLoadedListeners.add(newListener);
268   }
269   
270   /**
271    * Removes a MessageLoadedListener from the messageLoadedListener list,
272    * if it's in the list.
273    */
274   
275   public void removeMessageLoadedListener(MessageLoadedListener remListener) {
276     if (messageLoadedListeners.indexOf(remListener) > -1)
277       messageLoadedListeners.remove(remListener);
278   }
279   
280   /**
281    * Adds the MessageProxy(s) to the loadQueue.
282    */
283   public synchronized void loadMessages(MessageProxy mp) {
284     loadMessages(mp, NORMAL);
285   }
286
287   /**
288    * Adds the MessageProxy(s) to the loadQueue.
289    */
290   public synchronized void loadMessages(MessageProxy mp, int pPriority) {
291     if (pPriority > NORMAL) {
292       if (! priorityLoadQueue.contains(mp))
293     priorityLoadQueue.add(mp);
294       loadQueue.remove(mp);
295     } else {
296       if (! priorityLoadQueue.contains(mp) && ! loadQueue.contains(mp))
297     loadQueue.add(mp);
298     }
299     
300     if (this.isSleeping())
301       this.interrupt();
302   }
303   
304   /**
305    * Adds the MessageProxy(s) to the loadQueue.
306    */
307   public synchronized void loadMessages(MessageProxy[] mp) {
308     loadMessages(mp, NORMAL);
309   }
310
311   /**
312    * Adds the MessageProxy(s) to the loadQueue.
313    */
314   public synchronized void loadMessages(MessageProxy[] mp, int pPriority) {
315     loadMessages(Arrays.asList(mp), pPriority);
316   }
317   
318   /**
319    * Adds the MessageProxy(s) to the loadQueue.
320    */
321   public synchronized void loadMessages(List mp) {
322     loadMessages(mp, NORMAL);
323   }
324
325   /**
326    * Adds the MessageProxy(s) to the loadQueue.
327    */
328   public synchronized void loadMessages(List mp, int pPriority) {
329     if (mp != null && mp.size() > 0) {
330       if (pPriority > NORMAL) {
331     loadQueue.removeAll(mp);
332     addUniqueReversed(priorityLoadQueue, mp);
333       } else {
334     List copy = new ArrayList(mp);
335     copy.removeAll(priorityLoadQueue);
336     addUniqueReversed(loadQueue, copy);
337       }
338     }
339     
340     if (this.isSleeping())
341       this.interrupt();
342   }
343   
344   /**
345    * retrieves all the messages from the loadQueue, and resets that
346    * List to 0 (an empty List).
347    *
348    * generally, use retrieveNextBatch() instead.
349    */
350   public synchronized List retrieveLoadQueue() {
351     List returnValue = new LinkedList();
352     returnValue.addAll(priorityLoadQueue);
353     returnValue.addAll(loadQueue);
354     loadQueue = new LinkedList();
355     priorityLoadQueue = new LinkedList();
356     return returnValue;
357   }
358   
359   /**
360    * Adds all of the entries in toAdd to targetList, in reversed order.
361    */
362   private void addUniqueReversed(List targetList, List toAdd) {
363     for (int i = toAdd.size() - 1; i >= 0; i--) {
364       Object JavaDoc current = toAdd.get(i);
365       if (current != null && ! targetList.contains(current))
366     targetList.add(current);
367     }
368   }
369
370   /**
371    * Retrieves the next pCount messages from the queue, or returns null
372    * if there are no entries in the queue.
373    */
374   public synchronized List retrieveNextBatch(int pCount) {
375     int plqLength = priorityLoadQueue.size();
376     int lqLength = loadQueue.size();
377
378     // check to see if we actually have anything in the queue.
379 if (plqLength + lqLength > 0) {
380       List returnValue = new LinkedList();
381
382       // adding the priority queue first
383 if (plqLength > 0) {
384     // if the priority queue is larger than (or the same size as) the
385 // requested count, then just return it.
386 if (plqLength >= pCount) {
387       List subList = priorityLoadQueue.subList(0, pCount - 1);
388       returnValue.addAll(subList);
389       subList.clear();
390       return returnValue;
391     } else {
392       // just add of the priority queue, and go on.
393 returnValue.addAll(priorityLoadQueue);
394       priorityLoadQueue.clear();
395     }
396       }
397
398       // add in the normal queue now.
399 if (lqLength > 0) {
400     int newCount = pCount - plqLength;
401     if (lqLength >= newCount) {
402       List subList = loadQueue.subList(0, newCount -1);
403       returnValue.addAll(subList);
404       subList.clear();
405     } else {
406       returnValue.addAll(loadQueue);
407       loadQueue.clear();
408     }
409       }
410       
411       return returnValue;
412     } else {
413       return null;
414     }
415   }
416
417   public int getUpdateMessagesCount() {
418     return updateMessagesCount;
419   }
420   
421   public void setUpdateMessagesCount(int newValue) {
422     updateMessagesCount = newValue;
423   }
424
425   /**
426    * Returns the total amount left in the queue.
427    */
428   public synchronized int getQueueSize() {
429     return loadQueue.size() + priorityLoadQueue.size();
430   }
431
432   public List getColumnValues() {
433     return columnValues;
434   }
435   
436   public void setColumnValues(List newValue) {
437     columnValues=newValue;
438   }
439   
440   public FolderInfo getFolderInfo() {
441     return folderInfo;
442   }
443   
444   public boolean isSleeping() {
445     return sleeping;
446   }
447
448   /**
449    * Stops the thread.
450    */
451   public void stopThread() {
452     stopped = true;
453   }
454 }
455
456
457
458
459
460
461
Popular Tags