By using this site, you agree to our updated Privacy Policy and our Terms of Use. Manage your Cookies Settings.
424,851 Members | 1,128 Online
Bytes IT Community
+ Ask a Question
Need help? Post your question and get tips & solutions from a community of 424,851 IT Pros & Developers. It's quick & easy.

PLEASE HELP--2 instances of the application reading the messages using IMAP

P: 1
I have an application, which uses pop3 to read the messages from the mailbox, and it has been working fine for so many year. We recently have started changing this application to use java mail IMAP 4 instead of Pop3.

Primary reason for that :- Is we want to better meet out SLA, so we have decided to go with the 2 instances of this application, and hence decided to use IMAP4 to concurrently accessing the mailboxes via IMAP4.

My issues:- Now when, these 2 differet instances is trying to read same mailbox exactly at same time, It is reading the same message twice, as 1 instance is not aware of state changes in the 2nd instance.

Here is code that checks the mailboxes. The primary method, which checks a particular mailbox is checkMailbox

Please help with your suggestions, in terms of how to avoid reading the same message multiple times by these 2 different instances will be highly appreciated.
================================================== ========

Expand|Select|Wrap|Line Numbers
  1.  import java.io.ByteArrayOutputStream;
  2. import java.io.File;
  3. import java.io.FileOutputStream;
  4. import java.io.IOException;
  5. import java.io.OutputStream;
  6.  
  7. import javax.mail.Address;
  8. import javax.mail.Flags;
  9. import javax.mail.Folder;
  10. import javax.mail.Message;
  11. import javax.mail.MessagingException;
  12. import javax.mail.Session;
  13. import javax.mail.Store;
  14. import javax.mail.URLName;
  15. import javax.mail.internet.MailDateFormat;
  16. import javax.mail.internet.MimeMessage;
  17. import javax.mail.Flags;
  18. import javax.mail.event.*;
  19.  
  20. import mypackage.core.application.AppManager;
  21. import mypackage.core.base.ClydeException;
  22. import mypackage.core.utility.ClydeUtil;
  23. import mypackage.core.utility.MsgTrace;
  24.  
  25. import com.northwesternmutual.mail.util.LoggingWrapper;
  26. import com.northwesternmutual.mailreader.application.MailReaderAppInit;
  27. import com.northwesternmutual.mailreader.handler.access.INewMailHandler;
  28. import com.northwesternmutual.mailreader.handler.access.NewMailHandlerHome;
  29.  
  30. /**
  31. * Timer class that polls a mailbox and processes any messages
  32. */
  33. class MailboxWatcher implements Runnable 
  34. {
  35. private MailBox mailbox = null;
  36. MailDateFormat mailDateFormat = new MailDateFormat();
  37. File problemCheckpointFile = null;
  38. String problemMessageId = null;
  39.  
  40. private boolean checkForCorruptedFrom = ClydeUtil.getZimbuEnvVar("REJECT_DUAL_FROM", "N").equalsIgnoreCase("Y");
  41. private boolean quarantineAvailable = ClydeUtil.getZimbuEnvVar("QUARANTINE_CHECKPOINT_DIR", false) != null;
  42. private boolean deleteAfterQuarantine = quarantineAvailable && ClydeUtil.getZimbuEnvVar("DELETE_AFTER_QUARANTINE", "N").equalsIgnoreCase("Y");
  43.  
  44. /**
  45. * MailboxWatcher constructor comment.
  46. * @param mailBox
  47. * MailBox mailbox to poll
  48. */
  49. public MailboxWatcher(MailBox mailBox) {
  50. super();
  51. mailbox = mailBox;
  52. }
  53. /**
  54. * Check for new messages. 
  55. * @param aAlarm
  56. * the alarm ringing me
  57. */
  58. protected synchronized void checkMailbox() 
  59. {
  60. Store store = null;
  61. Folder f = null;
  62. try
  63. {
  64. Session s = Session.getDefaultInstance(System.getProperties());
  65. //s.setDebug(true);
  66. // open the mailbox
  67. store = s.getStore(mailbox.getIMAP4Url());
  68.  
  69. store.connect();
  70.  
  71.  
  72. // And get the folder we want.
  73. f = store.getFolder("INBOX");
  74.  
  75. if(f.isOpen()){
  76. //f.close(false);
  77. return;
  78. }
  79. f.open(Folder.READ_WRITE);
  80.  
  81. Message[] messages = f.getMessages();
  82.  
  83.  
  84.  
  85.  
  86. LoggingWrapper.trace(messages.length+ " messages @ " +mailbox, 
  87. MailReaderAppInit.GENERAL_CATEGORY, MsgTrace.MINIMAL);
  88.  
  89. if (messages.length > 0)
  90. {
  91. // Loop while there are messages to process, and the application
  92. // isn't on
  93. // its way down. Since this thread will stick around we check
  94. // isInitialized().
  95. for (int loop = 0; 
  96. loop < messages.length && AppManager.isInitialized();
  97. loop++)
  98. {
  99. boolean flag = false;
  100. //Before the message processing continue we will mark the message system flag as SEEN
  101.  
  102. LoggingWrapper.trace(messages[loop].getMessageNumber()+ "with messages messaage number got changed " , 
  103. MailReaderAppInit.GENERAL_CATEGORY, MsgTrace.MINIMAL);
  104. // If we don't process the next messag,e then don't continue
  105.  
  106. if (!processNextMessage(messages[loop]))
  107. {
  108.  
  109. loop = messages.length;
  110. f.expunge();
  111. }
  112.  
  113. }
  114. }
  115.  
  116. f.close(true);
  117. }
  118. catch (MessagingException me)
  119. {
  120. ClydeException.raise(ClydeException.INTERNAL_ERROR,
  121. "EMCR001",
  122. "Error polling mailbox " + mailbox,
  123. me.getMessage(),
  124. true,
  125. false,
  126. me);
  127. }catch(Exception ex){
  128.  
  129. }
  130. finally
  131. {
  132. //We close the folder and delete the messages that succeeded
  133. try
  134. {
  135. if (f !=null && f.isOpen())
  136. {
  137. f.close(true);
  138. }
  139. }
  140. catch (MessagingException me) {}
  141.  
  142. try
  143. {
  144. if (store !=null && store.isConnected())
  145. {
  146. store.close();
  147. }
  148. }
  149. catch (MessagingException me) {}
  150. }
  151. }
  152.  
  153. /**
  154. * Get a unique value from the mail messages
  155. * @return java.lang.String
  156. * @param msg
  157. * javax.mail.Message
  158. */
  159. private String getUniqueMessageId(Message msg) 
  160. throws MessagingException
  161. {
  162. // This works 99% of the time
  163. String msgIds[] = msg.getHeader("Message-Id");
  164. if (msgIds == null)
  165. {
  166. // this is needed when a message is returned to sender from the OpenMail
  167. // server
  168. // .... I think OM is broke in this regard.
  169. msgIds = msg.getHeader("In-Reply-To");
  170. if (msgIds == null)
  171. {
  172. ClydeException.raise(ClydeException.INTERNAL_ERROR, 
  173. "EMCR014",
  174. "Couldn't get UNIQUE message id",
  175. "No messge Id, or reply-to");
  176. }
  177. }
  178.  
  179. String id = msgIds[0];
  180. // Strip off the < and the >
  181. id = id.substring(1, id.length()-1);
  182. // And we'll also remove the @ since it's kind of ugly
  183. id = id.replace('@', '_'); 
  184. return id;
  185. }
  186.  
  187. /**
  188. * returns false to stop processing this mailbox or true to continue
  189. */ 
  190. private boolean processNextMessage(Message msg)
  191. throws MessagingException
  192. {
  193. INewMailHandler handler = NewMailHandlerHome.instance().getHandler();
  194.  
  195. String messageId = getUniqueMessageId(msg);
  196.  
  197. Message copyOfMessage = copyMessage(msg);
  198.  
  199. // Then we send this message out.
  200. try
  201. {
  202. isCorruptedFrom(copyOfMessage);
  203. if (handler.message(mailbox.getMessageQueue(),
  204. mailbox.getAttachmentQueue(),
  205. copyOfMessage))
  206. {
  207. msg.setFlag(Flags.Flag.DELETED, true);
  208.  
  209. }
  210. else 
  211. {
  212. // This stops sending all messages if the handler failed
  213. return false;
  214. }
  215. }
  216. catch (MessagingException me)
  217. {
  218. quarantineAndDelete(msg, messageId, copyOfMessage, me);
  219. return false;
  220. }
  221. catch (ClydeException ce)
  222. {
  223. if (ce.techMsg().indexOf("2030") > -1 ||
  224. ce.techMsg().indexOf("2031") > -1)
  225. {
  226. quarantineAndDelete(msg, messageId, copyOfMessage, ce);
  227. return false;
  228. }
  229. else
  230. {
  231. throw ce;
  232. }
  233. }
  234.  
  235. return true;
  236. }
  237.  
  238.  
  239. /**
  240. * Returns a read/write version of the message. POP3 messages are read only, and
  241. * we want to be able to update it a bit.
  242. * @param msg
  243. * @return Message
  244. * @throws MessagingException
  245. */
  246. private Message copyMessage(Message msg) throws MessagingException {
  247. // ..First mark which mailbox we read it from.
  248. Message copyOfMessage = new MimeMessage((MimeMessage)msg);
  249. fixExciteDotComBoundaryProblem(msg, copyOfMessage);
  250. copyOfMessage.addHeader("X-MailReader-MailBox", mailbox.getUserId());
  251. flushOutCopyOfMessage(copyOfMessage);
  252. return copyOfMessage;
  253. }
  254.  
  255.  
  256. /**
  257. * Some WebMail providers don't appear to follow the RFC spec for mail messages.
  258. * Specifically, they add an extra semicolon to the Content-Type. See RFC 2387
  259. * section 3.4
  260. * @param msg
  261. * @param copyOfMessage
  262. * @throws MessagingException
  263. */
  264. private void fixExciteDotComBoundaryProblem(Message msg, Message copyOfMessage)
  265. throws MessagingException 
  266. {
  267. // Check for a mishandled content-type header... Darn WebMail providers...
  268. String contentTypes[] = msg.getHeader("Content-Type");
  269. if (contentTypes != null && contentTypes.length > 0)
  270. {
  271. String contentType = contentTypes[0].trim();
  272. if (contentType.endsWith(";"))
  273. {
  274. contentType = contentType.substring(0,contentType.length()-1);
  275. }
  276. copyOfMessage.setHeader("Content-Type", contentType);
  277. }
  278. }
  279.  
  280. /**
  281. * Certain fields aren't flushed out unless the message writeTo() is called.
  282. * This prevents problems in the send.
  283. * @param copyOfMessage
  284. * @throws MessagingException
  285. */
  286. private void flushOutCopyOfMessage(Message copyOfMessage)
  287. throws MessagingException 
  288. {
  289. // Flush out extra fields by dumping it out...
  290. OutputStream os = new ByteArrayOutputStream(10240);
  291. try { copyOfMessage.writeTo(os); } catch (IOException io) {}
  292. try { os.close(); } catch (IOException e) {}
  293. }
  294.  
  295. /**
  296. * Log a message, then if quarantine is available, quarantine it. If
  297. * deleteAfterQuarantine is set, then the message is removed from the mailbox
  298. * @param f
  299. * @param msg
  300. * @param messageId
  301. * @param copyOfMessage
  302. * @param originalException
  303. * @throws MessagingException
  304. */
  305. private void quarantineAndDelete(Message msg,
  306. String messageId,
  307. Message copyOfMessage,
  308. Throwable originalThrowable)
  309. throws MessagingException 
  310. {
  311. try 
  312. {
  313. ClydeException.raise(ClydeException.INTERNAL_ERROR,
  314. "QUAR001", 
  315. "Attempting Quarantining message",
  316. "Quarantining message: "+getUniqueMessageId(msg) +" from INBOX",
  317. true, false, originalThrowable);
  318. }
  319. catch (ClydeException justForLogging) {}
  320.  
  321. if (quarantineAvailable)
  322. {
  323. quarantineMessage(copyOfMessage);
  324. if (deleteAfterQuarantine)
  325. {
  326. msg.setFlag(Flags.Flag.DELETED, true);
  327. }
  328. }
  329. }
  330.  
  331. /**
  332. * Method isAutoReplyMessage.
  333. * @param msg
  334. * @return boolean
  335. */
  336. private boolean isAutoReplyMessage(Message msg) throws MessagingException {
  337. // X-OpenMail-Autoreplied: TRUE
  338. String headers[] = msg.getHeader("X-OpenMail-Autoreplied");
  339. // If the headers aren't there, it definetly isn't an autoreply
  340. if (headers == null || headers.length == 0)
  341. return false;
  342. if (headers[0].equalsIgnoreCase("true"))
  343. return true;
  344. return false;
  345. }
  346.  
  347. /**
  348. * Method isCorruptedFrom
  349. * @param msg
  350. * @return boolean
  351. */
  352. private boolean isCorruptedFrom(Message msg) 
  353. throws MessagingException 
  354. {
  355. if (checkForCorruptedFrom == true)
  356. {
  357. Address froms[] = msg.getFrom();
  358. System.out.println("From address:-"+froms[0]);
  359.  
  360. // If the headers aren't there, it definetly isn't an autoreply
  361. if (froms != null && froms.length > 1)
  362. {
  363. throw new MessagingException("Message contains corrupted From: " + msg.getHeader("From")[0]);
  364. }
  365. }
  366. return false;
  367. }
  368.  
  369.  
  370. private void quarantineMessage(Message msg)
  371. throws MessagingException
  372. {
  373. try
  374. {
  375. File quarantine = new File(ClydeUtil.getZimbuEnvVar("QUARANTINE_CHECKPOINT_DIR"), "quarantine." getUniqueMessageId(msg)".msg");
  376. FileOutputStream fos = new FileOutputStream(quarantine);
  377. msg.writeTo(fos);
  378. fos.close();
  379. LoggingWrapper.trace("Quarantining message: "+getUniqueMessageId(msg) " into " quarantine + " from INBOX",
  380. MailReaderAppInit.GENERAL_CATEGORY, MsgTrace.VERBOSE);
  381.  
  382. try 
  383. {
  384. ClydeException.raise(ClydeException.INTERNAL_ERROR,
  385. "QUAR001", 
  386. "Quarantining message",
  387. "Quarantining message: "+getUniqueMessageId(msg) " into " quarantine + " from INBOX");
  388. }
  389. catch (ClydeException justForLogging) {}
  390. }
  391. catch (IOException io)
  392. {
  393. throw new MessagingException("Problem quarantining message", io);
  394. }
  395. }
  396.  
  397. /**
  398. * @see java.lang.Runnable
  399. */
  400. public void run() 
  401. {
  402. try {
  403.  
  404.  
  405. checkMailbox();
  406.  
  407.  
  408. } catch (Throwable th)
  409. {
  410. ClydeUtil.dump(th);
  411. LoggingWrapper.trace("checkMailbox threw an uncaught throwable", MailReaderAppInit.GENERAL_CATEGORY, 
  412. MsgTrace.MINIMAL);
  413. }
  414. finally
  415. {
  416. mailbox.setNextPollTime();
  417. }
  418. }
  419. }
Jan 10 '08 #1
Share this Question
Share on Google+
1 Reply


Plater
Expert 5K+
P: 7,872
It was my impression of both pop/imap standards that the server for these should not be allowing concurrent access to the same mailbox. (For just such reasons I would imagine)
A valid mailbox connection access is supposed to mark the mailbox as locked until such a time as the connection is terminated.
It would seem that your server is not doing that?

Is it possible for you to set up some sort of 'mutex' type system, i.e. implement your own methods for locking a mailbox while it's in use (however brief). If the idle disconnect is short for the connections, you could just have your application block until the mailbox becomes unlocked. As long as the individual access times to your server remain short, there wouldn't really be a slow down.
Feb 18 '08 #2

Post your reply

Sign in to post your reply or Sign up for a free account.