KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > apache > james > test > mock > james > InMemorySpoolRepository


1 /****************************************************************
2  * Licensed to the Apache Software Foundation (ASF) under one *
3  * or more contributor license agreements. See the NOTICE file *
4  * distributed with this work for additional information *
5  * regarding copyright ownership. The ASF licenses this file *
6  * to you under the Apache License, Version 2.0 (the *
7  * "License"); you may not use this file except in compliance *
8  * with the License. You may obtain a copy of the License at *
9  * *
10  * http://www.apache.org/licenses/LICENSE-2.0 *
11  * *
12  * Unless required by applicable law or agreed to in writing, *
13  * software distributed under the License is distributed on an *
14  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY *
15  * KIND, either express or implied. See the License for the *
16  * specific language governing permissions and limitations *
17  * under the License. *
18  ****************************************************************/
19
20
21 package org.apache.james.test.mock.james;
22
23 import org.apache.avalon.framework.activity.Disposable;
24 import org.apache.avalon.framework.container.ContainerUtil;
25 import org.apache.james.core.MailImpl;
26 import org.apache.james.services.SpoolRepository;
27 import org.apache.james.test.mock.avalon.MockLogger;
28 import org.apache.james.util.Lock;
29 import org.apache.mailet.Mail;
30
31 import javax.mail.MessagingException JavaDoc;
32
33 import java.util.ArrayList JavaDoc;
34 import java.util.Collection JavaDoc;
35 import java.util.ConcurrentModificationException JavaDoc;
36 import java.util.Hashtable JavaDoc;
37 import java.util.Iterator JavaDoc;
38
39 /**
40  * Implementation of a MailRepository on a FileSystem.
41  *
42  * Requires a configuration element in the .conf.xml file of the form:
43  * <repository destinationURL="file://path-to-root-dir-for-repository"
44  * type="MAIL"
45  * model="SYNCHRONOUS"/>
46  * Requires a logger called MailRepository.
47  *
48  * @version 1.0.0, 24/04/1999
49  */
50 public class InMemorySpoolRepository
51     implements SpoolRepository, Disposable {
52
53     /**
54      * Whether 'deep debugging' is turned on.
55      */
56     protected final static boolean DEEP_DEBUG = true;
57     private Lock lock;
58     private MockLogger logger;
59     private Hashtable JavaDoc spool;
60
61     private MockLogger getLogger() {
62         if (logger == null) {
63             logger = new MockLogger();
64         }
65         return logger;
66     }
67
68     /**
69      * Releases a lock on a message identified by a key
70      *
71      * @param key the key of the message to be unlocked
72      *
73      * @return true if successfully released the lock, false otherwise
74      */
75     public boolean unlock(String JavaDoc key) {
76         if (lock.unlock(key)) {
77             if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
78                 StringBuffer JavaDoc debugBuffer =
79                     new StringBuffer JavaDoc(256)
80                             .append("Unlocked ")
81                             .append(key)
82                             .append(" for ")
83                             .append(Thread.currentThread().getName())
84                             .append(" @ ")
85                             .append(new java.util.Date JavaDoc(System.currentTimeMillis()));
86                 getLogger().debug(debugBuffer.toString());
87             }
88             return true;
89         } else {
90             return false;
91         }
92     }
93
94     /**
95      * Obtains a lock on a message identified by a key
96      *
97      * @param key the key of the message to be locked
98      *
99      * @return true if successfully obtained the lock, false otherwise
100      */
101     public boolean lock(String JavaDoc key) {
102         if (lock.lock(key)) {
103             if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
104                 StringBuffer JavaDoc debugBuffer =
105                     new StringBuffer JavaDoc(256)
106                             .append("Locked ")
107                             .append(key)
108                             .append(" for ")
109                             .append(Thread.currentThread().getName())
110                             .append(" @ ")
111                             .append(new java.util.Date JavaDoc(System.currentTimeMillis()));
112                 getLogger().debug(debugBuffer.toString());
113             }
114 // synchronized (this) {
115 // notifyAll();
116 // }
117 return true;
118         } else {
119             return false;
120         }
121     }
122
123     /**
124      * Stores a message in this repository. Shouldn't this return the key
125      * under which it is stored?
126      *
127      * @param mc the mail message to store
128      */
129     public void store(Mail mc) throws MessagingException JavaDoc {
130         try {
131             String JavaDoc key = mc.getName();
132             //Remember whether this key was locked
133 boolean wasLocked = true;
134             synchronized (this) {
135                 wasLocked = lock.isLocked(key);
136     
137                 if (!wasLocked) {
138                     //If it wasn't locked, we want a lock during the store
139 lock(key);
140                 }
141             }
142             try {
143                 MailImpl m = new MailImpl(mc,mc.getName());
144                 m.setState(mc.getState());
145                 spool.put(mc.getName(),m);
146             } finally {
147                 if (!wasLocked) {
148                     // If it wasn't locked, we need to unlock now
149 unlock(key);
150                     synchronized (this) {
151                         notify();
152                     }
153                 }
154             }
155
156             if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
157                 StringBuffer JavaDoc logBuffer =
158                     new StringBuffer JavaDoc(64)
159                             .append("Mail ")
160                             .append(key)
161                             .append(" stored.");
162                 getLogger().debug(logBuffer.toString());
163             }
164
165         } catch (Exception JavaDoc e) {
166             getLogger().error("Exception storing mail: " + e,e);
167             throw new MessagingException JavaDoc("Exception caught while storing Message Container: ",e);
168         }
169     }
170
171     /**
172      * Retrieves a message given a key. At the moment, keys can be obtained
173      * from list() in superinterface Store.Repository
174      *
175      * @param key the key of the message to retrieve
176      * @return the mail corresponding to this key, null if none exists
177      */
178     public Mail retrieve(String JavaDoc key) throws MessagingException JavaDoc {
179         if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
180             getLogger().debug("Retrieving mail: " + key);
181         }
182         try {
183             Mail mc = null;
184             try {
185                 mc = new MailImpl((Mail) spool.get(key),key);
186                 mc.setState(((Mail) spool.get(key)).getState());
187             }
188             catch (RuntimeException JavaDoc re){
189                 StringBuffer JavaDoc exceptionBuffer = new StringBuffer JavaDoc(128);
190                 if(re.getCause() instanceof Error JavaDoc){
191                     exceptionBuffer.append("Error when retrieving mail, not deleting: ")
192                             .append(re.toString());
193                 }else{
194                     exceptionBuffer.append("Exception retrieving mail: ")
195                             .append(re.toString())
196                             .append(", so we're deleting it.");
197                     remove(key);
198                 }
199                 getLogger().warn(exceptionBuffer.toString());
200                 return null;
201             }
202             return mc;
203         } catch (Exception JavaDoc me) {
204             getLogger().error("Exception retrieving mail: " + me);
205             throw new MessagingException JavaDoc("Exception while retrieving mail: " + me.getMessage());
206         }
207     }
208
209     /**
210      * Removes a specified message
211      *
212      * @param mail the message to be removed from the repository
213      */
214     public void remove(Mail mail) throws MessagingException JavaDoc {
215         remove(mail.getName());
216     }
217
218
219     /**
220      * Removes a Collection of mails from the repository
221      * @param mails The Collection of <code>MailImpl</code>'s to delete
222      * @throws MessagingException
223      * @since 2.2.0
224      */
225     public void remove(Collection JavaDoc mails) throws MessagingException JavaDoc {
226         Iterator JavaDoc delList = mails.iterator();
227         while (delList.hasNext()) {
228             remove((Mail)delList.next());
229         }
230     }
231
232     /**
233      * Removes a message identified by key.
234      *
235      * @param key the key of the message to be removed from the repository
236      */
237     public void remove(String JavaDoc key) throws MessagingException JavaDoc {
238         if (lock(key)) {
239             try {
240                 if (spool != null) {
241                     Object JavaDoc o = spool.remove(key);
242                     ContainerUtil.dispose(o);
243                 }
244             } finally {
245                 unlock(key);
246             }
247         } else {
248             StringBuffer JavaDoc exceptionBuffer =
249                 new StringBuffer JavaDoc(64)
250                         .append("Cannot lock ")
251                         .append(key)
252                         .append(" to remove it");
253             throw new MessagingException JavaDoc(exceptionBuffer.toString());
254         }
255     }
256
257     /**
258      * List string keys of messages in repository.
259      *
260      * @return an <code>Iterator</code> over the list of keys in the repository
261      *
262      */
263     public Iterator JavaDoc list() {
264         // Fix ConcurrentModificationException by cloning
265 // the keyset before getting an iterator
266 final ArrayList JavaDoc clone;
267         synchronized(spool) {
268             clone = new ArrayList JavaDoc(spool.keySet());
269         }
270         return clone.iterator();
271     }
272
273     
274     /**
275      * <p>Returns an arbitrarily selected mail deposited in this Repository.
276      * Usage: SpoolManager calls accept() to see if there are any unprocessed
277      * mails in the spool repository.</p>
278      *
279      * <p>Synchronized to ensure thread safe access to the underlying spool.</p>
280      *
281      * @return the mail
282      */
283     public synchronized Mail accept() throws InterruptedException JavaDoc {
284         if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
285             getLogger().debug("Method accept() called");
286         }
287         return accept(new SpoolRepository.AcceptFilter () {
288             public boolean accept (String JavaDoc _, String JavaDoc __, long ___, String JavaDoc ____) {
289                 return true;
290             }
291
292             public long getWaitTime () {
293                 return 0;
294             }
295         });
296     }
297
298     /**
299      * <p>Returns an arbitrarily selected mail deposited in this Repository that
300      * is either ready immediately for delivery, or is younger than it's last_updated plus
301      * the number of failed attempts times the delay time.
302      * Usage: RemoteDeliverySpool calls accept() with some delay and should block until an
303      * unprocessed mail is available.</p>
304      *
305      * <p>Synchronized to ensure thread safe access to the underlying spool.</p>
306      *
307      * @return the mail
308      */
309     public synchronized Mail accept(final long delay) throws InterruptedException JavaDoc
310     {
311         if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
312             getLogger().debug("Method accept(delay) called");
313         }
314         return accept(new SpoolRepository.AcceptFilter () {
315             long youngest = 0;
316                 
317                 public boolean accept (String JavaDoc key, String JavaDoc state, long lastUpdated, String JavaDoc errorMessage) {
318                     if (state.equals(Mail.ERROR)) {
319                         //Test the time...
320 long timeToProcess = delay + lastUpdated;
321                 
322                         if (System.currentTimeMillis() > timeToProcess) {
323                             //We're ready to process this again
324 return true;
325                         } else {
326                             //We're not ready to process this.
327 if (youngest == 0 || youngest > timeToProcess) {
328                                 //Mark this as the next most likely possible mail to process
329 youngest = timeToProcess;
330                             }
331                             return false;
332                         }
333                     } else {
334                         //This mail is good to go... return the key
335 return true;
336                     }
337                 }
338         
339                 public long getWaitTime () {
340                     if (youngest == 0) {
341                         return 0;
342                     } else {
343                         long duration = youngest - System.currentTimeMillis();
344                         youngest = 0; //get ready for next round
345 return duration <= 0 ? 1 : duration;
346                     }
347                 }
348             });
349     }
350
351
352     /**
353      * Returns an arbitrarily select mail deposited in this Repository for
354      * which the supplied filter's accept method returns true.
355      * Usage: RemoteDeliverySpool calls accept(filter) with some a filter which determines
356      * based on number of retries if the mail is ready for processing.
357      * If no message is ready the method will block until one is, the amount of time to block is
358      * determined by calling the filters getWaitTime method.
359      *
360      * <p>Synchronized to ensure thread safe access to the underlying spool.</p>
361      *
362      * @return the mail
363      */
364     public synchronized Mail accept(SpoolRepository.AcceptFilter filter) throws InterruptedException JavaDoc {
365         if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
366             getLogger().debug("Method accept(Filter) called");
367         }
368         while (!Thread.currentThread().isInterrupted()) try {
369             for (Iterator JavaDoc it = list(); it.hasNext(); ) {
370                 String JavaDoc s = it.next().toString();
371                 if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
372                     StringBuffer JavaDoc logBuffer =
373                         new StringBuffer JavaDoc(64)
374                                 .append("Found item ")
375                                 .append(s)
376                                 .append(" in spool.");
377                     getLogger().debug(logBuffer.toString());
378                 }
379                 if (lock(s)) {
380                     if ((DEEP_DEBUG) && (getLogger().isDebugEnabled())) {
381                         getLogger().debug("accept(Filter) has locked: " + s);
382                     }
383                     try {
384                         Mail mail = retrieve(s);
385                         // Retrieve can return null if the mail is no longer on the spool
386 // (i.e. another thread has gotten to it first).
387 // In this case we simply continue to the next key
388 if (mail == null || !filter.accept (mail.getName(),
389                                                             mail.getState(),
390                                                             mail.getLastUpdated().getTime(),
391                                                             mail.getErrorMessage())) {
392                             unlock(s);
393                             continue;
394                         }
395                         return mail;
396                     } catch (javax.mail.MessagingException JavaDoc e) {
397                         unlock(s);
398                         getLogger().error("Exception during retrieve -- skipping item " + s, e);
399                     }
400                 }
401             }
402
403             //We did not find any... let's wait for a certain amount of time
404 wait (filter.getWaitTime());
405         } catch (InterruptedException JavaDoc ex) {
406             throw ex;
407         } catch (ConcurrentModificationException JavaDoc cme) {
408             // Should never get here now that list methods clones keyset for iterator
409 getLogger().error("CME in spooler - please report to http://james.apache.org", cme);
410         }
411         throw new InterruptedException JavaDoc();
412     }
413
414     /**
415      *
416      */
417     public InMemorySpoolRepository() {
418         spool = new Hashtable JavaDoc();
419         lock = new Lock();
420     }
421
422     public int size() {
423         return spool.size();
424     }
425
426     public void clear() {
427         if (spool != null) {
428             Iterator JavaDoc i = list();
429             while (i.hasNext()) {
430                 String JavaDoc key = (String JavaDoc) i.next();
431                 try {
432                     remove(key);
433                 } catch (MessagingException JavaDoc e) {
434                 }
435             }
436         }
437     }
438
439     public void dispose() {
440         clear();
441     }
442
443     public String JavaDoc toString() {
444         StringBuffer JavaDoc result = new StringBuffer JavaDoc();
445         result.append(super.toString());
446         Iterator JavaDoc i = list();
447         while (i.hasNext()) {
448             result.append("\n\t"+i.next());
449         }
450         return result.toString();
451     }
452     
453 }
454
Popular Tags