From 413d5354a1e770851aa0bd8dfd876288172d66b3 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 1 Nov 2012 17:14:31 -0400 Subject: [PATCH] INT-2803 Fix Lazy Fetch of Email Messages INT-2805 Update JavaMail to 1.4.5 There is a need to fetch the entire email message before the folder is closed. Once the folder is closed, you cannot perform any more operations on the message. Prior to RC1, the message was copied, which forced an eager fetch. Add code to copy the message. Also, transaction synchronization operations need access to a folder instance to perform operations, such as delete, on a message. Add a wrapper to lazily create a folder instance in message.getFolder() when needed. Add documentation to explain that messages must be re-fetched before performing transaction synchronization operations. JavaMail 1.4.5 is now Open Source, which makes debugging much easier. --- build.gradle | 2 +- .../mail/AbstractMailReceiver.java | 36 +++- .../mail/ImapMailReceiverTests.java | 164 ++++++++++++------ .../src/test/resources/test.mail | 50 ++++++ src/reference/docbook/mail.xml | 22 ++- 5 files changed, 215 insertions(+), 59 deletions(-) create mode 100644 spring-integration-mail/src/test/resources/test.mail diff --git a/build.gradle b/build.gradle index 0155bcdd67..56a4e0ea6b 100644 --- a/build.gradle +++ b/build.gradle @@ -569,7 +569,7 @@ project('spring-integration-mail') { dependencies { compile project(":spring-integration-core") compile "org.springframework:spring-context-support:$springVersion" - compile("javax.mail:mail:1.4.4", provided) + compile("javax.mail:mail:1.4.5", provided) compile("javax.activation:activation:$javaxActivationVersion", optional) testCompile project(":spring-integration-test") } diff --git a/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java b/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java index d9a89a1d4b..33f8c1c4be 100755 --- a/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java +++ b/spring-integration-mail/src/main/java/org/springframework/integration/mail/AbstractMailReceiver.java @@ -207,7 +207,7 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl protected void openFolder() throws MessagingException { this.openSession(); if (this.folder == null) { - this.folder = this.store.getFolder(this.url); + this.folder = obtainFolderInstance(); } if (this.folder == null || !this.folder.exists()) { throw new IllegalStateException("no such folder [" + this.url.getFile() + "]"); @@ -221,6 +221,10 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl this.folder.open(this.folderOpenMode); } + private Folder obtainFolderInstance() throws MessagingException { + return this.store.getFolder(this.url); + } + public Message[] receive() throws javax.mail.MessagingException { synchronized (this.folderMonitor) { try { @@ -263,6 +267,11 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl if (this.shouldDeleteMessages()) { this.deleteMessages(filteredMessages); } + // Copy messages to cause an eager fetch + for (int i = 0; i < filteredMessages.length; i++) { + MimeMessage mimeMessage = new IntegrationMimeMessage((MimeMessage) filteredMessages[i]); + filteredMessages[i] = mimeMessage; + } } private void setMessageFlags(Message[] filteredMessages) throws MessagingException { @@ -382,4 +391,29 @@ public abstract class AbstractMailReceiver extends IntegrationObjectSupport impl return this.store; } + /** + * Since we copy the message to eagerly fetch the message, it has no folder. + * However, we need to make a folder available in case the user wants to + * perform operations on the message in the folder later in the flow. + * @author Gary Russell + * @since 2.2 + * + */ + public class IntegrationMimeMessage extends MimeMessage { + + public IntegrationMimeMessage(MimeMessage source) throws MessagingException { + super(source); + } + + @Override + public Folder getFolder() { + try { + return AbstractMailReceiver.this.obtainFolderInstance(); + } + catch (MessagingException e) { + throw new org.springframework.integration.MessagingException("Unable to obtain the mail folder", e); + } + } + + } } diff --git a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java index 962518e4b0..91854aa0c9 100644 --- a/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java +++ b/spring-integration-mail/src/test/java/org/springframework/integration/mail/ImapMailReceiverTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.mail; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertSame; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; @@ -24,14 +25,19 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.IOException; +import java.io.OutputStream; import java.lang.reflect.Field; import java.util.Properties; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import javax.mail.BodyPart; import javax.mail.Flags; import javax.mail.Flags.Flag; import javax.mail.Folder; import javax.mail.Message; +import javax.mail.Multipart; import javax.mail.Store; import javax.mail.URLName; import javax.mail.internet.MimeMessage; @@ -42,10 +48,10 @@ import org.junit.Test; import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; - import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.core.io.ClassPathResource; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.PollableChannel; @@ -53,8 +59,10 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand import org.springframework.integration.history.MessageHistory; import org.springframework.integration.mail.config.ImapIdleChannelAdapterParserTests; import org.springframework.integration.test.util.TestUtils; +import org.springframework.util.FileCopyUtils; import com.sun.mail.imap.IMAPFolder; +import com.sun.mail.imap.IMAPMessage; /** * @author Oleg Zhurakousky @@ -62,9 +70,9 @@ import com.sun.mail.imap.IMAPFolder; * */ public class ImapMailReceiverTests { - + private AtomicInteger failed = new AtomicInteger(0); - + @Test public void receiveAndMarkAsReadDontDelete() throws Exception{ AbstractMailReceiver receiver = new ImapMailReceiver(); @@ -80,7 +88,7 @@ public class ImapMailReceiverTests { Message msg1 = mock(MimeMessage.class); Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { DirectFieldAccessor accessor = new DirectFieldAccessor(invocation.getMock()); @@ -88,17 +96,17 @@ public class ImapMailReceiverTests { if (folderOpenMode != Folder.READ_WRITE){ throw new IllegalArgumentException("Folder had to be open in READ_WRITE mode"); } - + return null; } }).when(receiver).openFolder(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; @@ -116,13 +124,13 @@ public class ImapMailReceiverTests { receiver.setShouldDeleteMessages(true); receiver = spy(receiver); receiver.afterPropertiesSet(); - + Field folderField = AbstractMailReceiver.class.getDeclaredField("folder"); folderField.setAccessible(true); Folder folder = mock(Folder.class); when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER)); folderField.set(receiver, folder); - + Message msg1 = mock(MimeMessage.class); Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; @@ -136,13 +144,13 @@ public class ImapMailReceiverTests { return null; } }).when(receiver).openFolder(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; @@ -159,14 +167,14 @@ public class ImapMailReceiverTests { ((ImapMailReceiver)receiver).setShouldMarkMessagesAsRead(false); receiver = spy(receiver); receiver.afterPropertiesSet(); - + Field folderField = AbstractMailReceiver.class.getDeclaredField("folder"); folderField.setAccessible(true); Folder folder = mock(Folder.class); when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER)); folderField.set(receiver, folder); - - + + Message msg1 = mock(MimeMessage.class); Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; @@ -175,13 +183,13 @@ public class ImapMailReceiverTests { return null; } }).when(receiver).openFolder(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; @@ -199,13 +207,13 @@ public class ImapMailReceiverTests { ((ImapMailReceiver)receiver).setShouldMarkMessagesAsRead(false); receiver = spy(receiver); receiver.afterPropertiesSet(); - + Field folderField = AbstractMailReceiver.class.getDeclaredField("folder"); folderField.setAccessible(true); Folder folder = mock(Folder.class); when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER)); folderField.set(receiver, folder); - + Message msg1 = mock(MimeMessage.class); Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; @@ -219,13 +227,13 @@ public class ImapMailReceiverTests { return null; } }).when(receiver).openFolder(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; @@ -243,13 +251,13 @@ public class ImapMailReceiverTests { AbstractMailReceiver receiver = new ImapMailReceiver(); receiver = spy(receiver); receiver.afterPropertiesSet(); - + Field folderField = AbstractMailReceiver.class.getDeclaredField("folder"); folderField.setAccessible(true); Folder folder = mock(Folder.class); when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER)); folderField.set(receiver, folder); - + Message msg1 = mock(MimeMessage.class); Message msg2 = mock(MimeMessage.class); final Message[] messages = new Message[]{msg1, msg2}; @@ -263,13 +271,13 @@ public class ImapMailReceiverTests { return null; } }).when(receiver).openFolder(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; @@ -283,22 +291,22 @@ public class ImapMailReceiverTests { @Test @Ignore public void testMessageHistory() throws Exception{ - ApplicationContext context = + ApplicationContext context = new ClassPathXmlApplicationContext("ImapIdleChannelAdapterParserTests-context.xml", ImapIdleChannelAdapterParserTests.class); ImapIdleChannelAdapter adapter = context.getBean("simpleAdapter", ImapIdleChannelAdapter.class); - + AbstractMailReceiver receiver = new ImapMailReceiver(); receiver = spy(receiver); receiver.afterPropertiesSet(); - + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); adapterAccessor.setPropertyValue("mailReceiver", receiver); - + MimeMessage mailMessage = mock(MimeMessage.class); Flags flags = mock(Flags.class); when(mailMessage.getFlags()).thenReturn(flags); final Message[] messages = new Message[]{mailMessage}; - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { DirectFieldAccessor accesor = new DirectFieldAccessor((invocation.getMock())); @@ -308,19 +316,19 @@ public class ImapMailReceiverTests { return null; } }).when(receiver).openFolder(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; } }).when(receiver).fetchMessages(messages); - + PollableChannel channel = context.getBean("channel", PollableChannel.class); adapter.start(); @@ -334,16 +342,17 @@ public class ImapMailReceiverTests { @Test public void testIdleChannelAdapterException() throws Exception{ - ApplicationContext context = + ApplicationContext context = new ClassPathXmlApplicationContext("ImapIdleChannelAdapterParserTests-context.xml", ImapIdleChannelAdapterParserTests.class); ImapIdleChannelAdapter adapter = context.getBean("simpleAdapter", ImapIdleChannelAdapter.class); //ImapMailReceiver receiver = (ImapMailReceiver) TestUtils.getPropertyValue(adapter, "mailReceiver"); - - - + + + DirectChannel channel = new DirectChannel(); channel.subscribe(new AbstractReplyProducingMessageHandler() { + @Override protected Object handleRequestMessage(org.springframework.integration.Message requestMessage) { throw new RuntimeException("Failed"); } @@ -351,58 +360,58 @@ public class ImapMailReceiverTests { adapter.setOutputChannel(channel); QueueChannel errorChannel = new QueueChannel(); adapter.setErrorChannel(errorChannel); - + AbstractMailReceiver receiver = new ImapMailReceiver(); receiver = spy(receiver); receiver.afterPropertiesSet(); - + Field folderField = AbstractMailReceiver.class.getDeclaredField("folder"); folderField.setAccessible(true); Folder folder = mock(IMAPFolder.class); when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER)); folderField.set(receiver, folder); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return true; } }).when(folder).isOpen(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; } }).when(receiver).openFolder(); - + DirectFieldAccessor adapterAccessor = new DirectFieldAccessor(adapter); adapterAccessor.setPropertyValue("mailReceiver", receiver); - + MimeMessage mailMessage = mock(MimeMessage.class); Flags flags = mock(Flags.class); when(mailMessage.getFlags()).thenReturn(flags); final Message[] messages = new Message[]{mailMessage}; - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return messages; } }).when(receiver).searchForNewMessages(); - + doAnswer(new Answer() { public Object answer(InvocationOnMock invocation) throws Throwable { return null; } }).when(receiver).fetchMessages(messages); - + adapter.start(); org.springframework.integration.Message replMessage = errorChannel.receive(10000); assertNotNull(replMessage); assertEquals("Failed", ((Exception) replMessage.getPayload()).getCause().getMessage()); } - + @Test // see INT-1801 public void testImapLifecycleForRaceCondition() throws Exception{ - + for (int i = 0; i < 1000; i++) { final ImapMailReceiver receiver = new ImapMailReceiver("imap://foo"); Store store = mock(Store.class); @@ -412,12 +421,12 @@ public class ImapMailReceiverTests { when(folder.search((SearchTerm) Mockito.any())).thenReturn(new Message[]{}); when(store.getFolder(Mockito.any(URLName.class))).thenReturn(folder); when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER)); - - + + DirectFieldAccessor df = new DirectFieldAccessor(receiver); df.setPropertyValue("store", store); receiver.afterPropertiesSet(); - + new Thread(new Runnable() { public void run(){ try { @@ -428,10 +437,10 @@ public class ImapMailReceiverTests { failed.getAndIncrement(); } } - + } }).start(); - + new Thread(new Runnable() { public void run(){ try { @@ -445,4 +454,55 @@ public class ImapMailReceiverTests { } assertEquals(0, failed.get()); } + + @Test + public void testAttachments() throws Exception { + final ImapMailReceiver receiver = new ImapMailReceiver("imap://foo"); + Store store = mock(Store.class); + Folder folder = mock(Folder.class); + when(folder.exists()).thenReturn(true); + when(folder.isOpen()).thenReturn(true); + final AtomicBoolean closed = new AtomicBoolean(); + doAnswer(new Answer (){ + public Object answer(InvocationOnMock invocation) throws Throwable { + closed.set(true); + return null; + } + }).when(folder).close(Mockito.anyBoolean()); + IMAPMessage message = mock(IMAPMessage.class); + when(folder.search((SearchTerm) Mockito.any())).thenReturn(new Message[]{message}); + when(store.getFolder(Mockito.any(URLName.class))).thenReturn(folder); + when(folder.getPermanentFlags()).thenReturn(new Flags(Flags.Flag.USER)); + DirectFieldAccessor df = new DirectFieldAccessor(receiver); + df.setPropertyValue("store", store); + receiver.afterPropertiesSet(); + + Multipart multiPart = mock(Multipart.class); + when(multiPart.getCount()).thenReturn(1); + when(message.getContent()).thenReturn(multiPart); + final BodyPart bodyPart = mock(BodyPart.class); + doAnswer(new Answer(){ + + public Object answer(InvocationOnMock invocation) throws Throwable { + if (closed.get()) { + throw new IOException("Folder is closed"); + } + return bodyPart; + } + }).when(multiPart).getBodyPart(Mockito.anyInt()); + when(bodyPart.getContent()).thenReturn("bar"); + doAnswer(new Answer () { + + public Object answer(InvocationOnMock invocation) throws Throwable { + OutputStream os = (OutputStream) invocation.getArguments()[0]; + FileCopyUtils.copy(new ClassPathResource("test.mail").getInputStream(), os); + return null; + } + }).when(message).writeTo(Mockito.any(OutputStream.class)); + Message[] messages = receiver.receive(); + Object content = messages[0].getContent(); + assertEquals("bar\n", ((Multipart) content).getBodyPart(0).getContent()); + + assertSame(folder, messages[0].getFolder()); + } } diff --git a/spring-integration-mail/src/test/resources/test.mail b/spring-integration-mail/src/test/resources/test.mail new file mode 100644 index 0000000000..9e8349f8a5 --- /dev/null +++ b/spring-integration-mail/src/test/resources/test.mail @@ -0,0 +1,50 @@ +Delivered-To: yyyyy@gmail.com +Received: by 10.64.27.133 with SMTP id t5csp748412ieg; + Thu, 1 Nov 2012 09:38:30 -0700 (PDT) +Received: by 10.220.40.16 with SMTP id i16mr23706195vce.31.1351787910470; + Thu, 01 Nov 2012 09:38:30 -0700 (PDT) +Return-Path: +Received: from mail-vb0-f51.google.com (mail-vb0-f51.google.com [209.85.212.51]) + by mx.google.com with ESMTPS id t8si2299833vcw.15.2012.11.01.09.38.30 + (version=TLSv1/SSLv3 cipher=OTHER); + Thu, 01 Nov 2012 09:38:30 -0700 (PDT) +Received-SPF: pass (google.com: domain of xxxxx@gmail.com designates 209.85.212.51 as permitted sender) client-ip=209.85.212.51; +Authentication-Results: mx.google.com; spf=pass (google.com: domain of xxxxxx@gmail.com designates 209.85.212.51 as permitted sender) smtp.mail=xxxxx@gmail.com; dkim=pass header.i=@gmail.com +Received: by mail-vb0-f51.google.com with SMTP id fn1so2898461vbb.24 + for ; Thu, 01 Nov 2012 09:38:30 -0700 (PDT) +Received: by 10.220.227.70 with SMTP id iz6mr23760024vcb.45.1351787910197; + Thu, 01 Nov 2012 09:38:30 -0700 (PDT) +Return-Path: +Received: from [192.168.1.7] (pool-72-78-102-80.phlapa.fios.verizon.net. [72.78.102.80]) + by mx.google.com with ESMTPS id g5sm3780201vez.6.2012.11.01.09.38.29 + (version=TLSv1/SSLv3 cipher=OTHER); + Thu, 01 Nov 2012 09:38:29 -0700 (PDT) +Sender: Gary Russell +Message-ID: <5092A55B.8050000@foo.bar> +Date: Thu, 01 Nov 2012 12:37:47 -0400 +From: Gary Russell +User-Agent: Mozilla/5.0 (X11; Linux x86_64; rv:16.0) Gecko/20121026 Thunderbird/16.0.2 +MIME-Version: 1.0 +To: yyyyy@gmail.com +Subject: Test +Content-Type: multipart/mixed; + boundary="------------040903000701040401040200" + +This is a multi-part message in MIME format. +--------------040903000701040401040200 +Content-Type: text/plain; charset=ISO-8859-1; format=flowed +Content-Transfer-Encoding: 7bit + +bar + +--------------040903000701040401040200 +Content-Type: text/plain; charset=UTF-8; + name="foo.txt" +Content-Transfer-Encoding: 7bit +Content-Disposition: attachment; + filename="foo.txt" + +foo + +--------------040903000701040401040200-- + diff --git a/src/reference/docbook/mail.xml b/src/reference/docbook/mail.xml index 4ffdda3659..fe2d17670c 100644 --- a/src/reference/docbook/mail.xml +++ b/src/reference/docbook/mail.xml @@ -320,15 +320,27 @@ public class Mover { public void process(MimeMessage message) throws Exception{ Folder folder = message.getFolder(); - Store store = null; - if (!folder.isOpen()){ - folder.open(Folder.READ_WRITE); - store = folder.getStore(); + folder.open(Folder.READ_WRITE); + String messageId = message.getMessageID(); + Message[] messages = folder.getMessages(); + FetchProfile contentsProfile = new FetchProfile(); + contentsProfile.add(FetchProfile.Item.ENVELOPE); + contentsProfile.add(FetchProfile.Item.CONTENT_INFO); + contentsProfile.add(FetchProfile.Item.FLAGS); + folder.fetch(messages, contentsProfile); + // find this message and mark for deletion + for (int i = 0; i < messages.length; i++) { + if (((MimeMessage) messages[i]).getMessageID().equals(messageId)) { + messages[i].setFlag(Flags.Flag.DELETED, true); + break; + } } - message.setFlag(Flags.Flag.DELETED, true); + Folder fooFolder = store.getFolder("FOO")); fooFolder.appendMessages(new MimeMessage[]{message}); folder.expunge(); + folder.close(true); + fooFolder.close(false); } }]]>