INT-4021: SimpleMessageStore: Disable lazy-load

JIRA: https://jira.spring.io/browse/INT-4021

Previously the `SimpleMessageStore` unconditionally followed with the super class options
and provided the `lazy-load` functionality by default.

* Since `persistent` and `lazy-load` logic does not make sense for the `in-memory` store, disable it in the `SimpleMessageStore`
* Fir `AggregatorTests` for better coverage.
* NOTE: The same message can't be persisted in the Persistent `MessageStore`. The store key is fully based on the `messageId`.
And also we provide the header which indicates that the messages has been stored before. See `JdbcMessageStore.addMessage()` for example:
```
if (message.getHeaders().containsKey(SAVED_KEY)) {
      Message<T> saved = (Message<T>) getMessage(message.getHeaders().getId());
      if (saved != null) {
		if (saved.equals(message)) {
			return message;
		} // We need to save it under its own id
	}
}
```
* Fix (S)FTP Streaming tests to use `AbstractPersistentAcceptOnceFileListFilter` instead of raw `AcceptOnceFileListFilter`,
which relies on the object identity, but neither `FtpFile`, nor  `ChannelSftp.LsEntry` provides good `equals()` and `hashCode()` implementations.
* Make `StompIntegrationTests` as `DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD` because the sporadic failure on the Travis around wrong value from the queue isn't clear (yet).
Looks like some interim event is generated by the Websocket Container on the Tomcat.

spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java

@@ -128,6 +128,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
 		this.groupCapacity = groupCapacity;
 		this.lockRegistry = lockRegistry;
 		this.upperBoundTimeout = upperBoundTimeout;
+		disableLazyLoadMessageGroups();
+	}
+
+	private void disableLazyLoadMessageGroups() {
+		super.setLazyLoadMessageGroups(false);
 	}
 
 	/**
@@ -161,6 +166,11 @@ public class SimpleMessageStore extends AbstractMessageGroupStore
 		this.lockRegistry = lockRegistry;
 	}
 
+	@Override
+	public void setLazyLoadMessageGroups(boolean lazyLoadMessageGroups) {
+		throw new UnsupportedOperationException("The lazy-load isn't supported for in-memory 'SimpleMessageStore'");
+	}
+
 	@Override
 	@ManagedAttribute
 	public long getMessageCount() {

spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java

@@ -27,6 +27,10 @@ import static org.mockito.Mockito.mock;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.commons.logging.Log;
@@ -42,6 +46,7 @@ import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.store.MessageGroup;
+import org.springframework.integration.store.SimpleMessageGroupFactory;
 import org.springframework.integration.store.SimpleMessageStore;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
@@ -58,6 +63,7 @@ import org.springframework.util.StopWatch;
  * @author Marius Bogoevici
  * @author Iwein Fuld
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class AggregatorTests {
 
@@ -67,7 +73,7 @@ public class AggregatorTests {
 
 	private final SimpleMessageStore store = new SimpleMessageStore(50);
 
-	List<MessageGroupExpiredEvent> expiryEvents = new ArrayList<MessageGroupExpiredEvent>();
+	private final List<MessageGroupExpiredEvent> expiryEvents = new ArrayList<MessageGroupExpiredEvent>();
 
 	@Before
 	public void configureAggregator() {
@@ -92,7 +98,7 @@ public class AggregatorTests {
 	}
 
 	@Test
-	public void testAggPerf() {
+	public void testAggPerf() throws InterruptedException, ExecutionException, TimeoutException {
 		AggregatingMessageHandler handler = new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor());
 		handler.setCorrelationStrategy(new CorrelationStrategy() {
 
@@ -100,20 +106,36 @@ public class AggregatorTests {
 			public Object getCorrelationKey(Message<?> message) {
 				return "foo";
 			}
+
 		});
 		handler.setReleaseStrategy(new MessageCountReleaseStrategy(60000));
 		handler.setExpireGroupsUponCompletion(true);
 		handler.setSendPartialResultOnExpiry(true);
 		DirectChannel outputChannel = new DirectChannel();
 		handler.setOutputChannel(outputChannel);
+
+		final CompletableFuture<Collection<?>> resultFuture = new CompletableFuture<>();
 		outputChannel.subscribe(new MessageHandler() {
 
 			@Override
 			public void handleMessage(Message<?> message) throws MessagingException {
-				logger.warn("Received " + ((Collection<?>) message.getPayload()).size());
+				Collection<?> payload = (Collection<?>) message.getPayload();
+				logger.warn("Received " + payload.size());
+				resultFuture.complete(payload);
 			}
 
 		});
+
+		SimpleMessageStore store = new SimpleMessageStore();
+
+		SimpleMessageGroupFactory messageGroupFactory =
+				new SimpleMessageGroupFactory(SimpleMessageGroupFactory.GroupType.BLOCKING_QUEUE);
+
+		store.setMessageGroupFactory(messageGroupFactory);
+
+		handler.setMessageStore(store);
+
+
 		Message<?> message = new GenericMessage<String>("foo");
 		StopWatch stopwatch = new StopWatch();
 		stopwatch.start();
@@ -129,10 +151,14 @@ public class AggregatorTests {
 		stopwatch.stop();
 		logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() +
 				" (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)");
+
+		Collection<?> result = resultFuture.get(10, TimeUnit.SECONDS);
+		assertNotNull(result);
+		assertEquals(60000, result.size());
 	}
 
 	@Test
-	public void testCustomAggPerf() {
+	public void testCustomAggPerf() throws InterruptedException, ExecutionException, TimeoutException {
 		class CustomHandler extends AbstractMessageHandler {
 
 			// custom aggregator, only handles a single correlation
@@ -172,11 +198,15 @@ public class AggregatorTests {
 
 		DirectChannel outputChannel = new DirectChannel();
 		CustomHandler handler = new CustomHandler(outputChannel);
+
+		final CompletableFuture<Collection<?>> resultFuture = new CompletableFuture<>();
 		outputChannel.subscribe(new MessageHandler() {
 
 			@Override
 			public void handleMessage(Message<?> message) throws MessagingException {
-				logger.warn("Received " + ((Collection<?>) message.getPayload()).size());
+				Collection<?> payload = (Collection<?>) message.getPayload();
+				logger.warn("Received " + payload.size());
+				resultFuture.complete(payload);
 			}
 
 		});
@@ -195,6 +225,10 @@ public class AggregatorTests {
 		stopwatch.stop();
 		logger.warn("Sent " + 120000 + " in " + stopwatch.getTotalTimeSeconds() +
 				" (10k in " + stopwatch.getLastTaskTimeMillis() + "ms)");
+
+		Collection<?> result = resultFuture.get(10, TimeUnit.SECONDS);
+		assertNotNull(result);
+		assertEquals(60000, result.size());
 	}
 
 	@Test

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java

@@ -35,10 +35,11 @@ import org.springframework.integration.annotation.Transformer;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageSource;
-import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.FtpTestSupport;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.transformer.StreamTransformer;
 import org.springframework.messaging.Message;
@@ -49,6 +50,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 4.3
  *
  */
@@ -68,7 +70,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 		received = (Message<byte[]>) this.data.receive(10000);
 		assertNotNull(received);
 		assertThat(new String(received.getPayload()), equalTo("source2"));
-		assertNull(this.data.receive(0));
+		assertNull(this.data.receive(10));
 	}
 
 	@Configuration
@@ -93,7 +95,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 		public MessageSource<InputStream> ftpMessageSource() {
 			FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);
 			messageSource.setRemoteDirectory("ftpSource/");
-			messageSource.setFilter(new AcceptOnceFileListFilter<FTPFile>());
+			messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming"));
 			return messageSource;
 		}
 

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java

@@ -34,10 +34,11 @@ import org.springframework.integration.annotation.Transformer;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageSource;
-import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.integration.scheduling.PollerMetadata;
 import org.springframework.integration.sftp.SftpTestSupport;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
 import org.springframework.integration.transformer.StreamTransformer;
 import org.springframework.messaging.Message;
@@ -50,6 +51,7 @@ import com.jcraft.jsch.ChannelSftp.LsEntry;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 4.3
  *
  */
@@ -69,7 +71,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 		received = (Message<byte[]>) this.data.receive(10000);
 		assertNotNull(received);
 		assertThat(new String(received.getPayload()), equalTo("source2"));
-		assertNull(this.data.receive(0));
+		assertNull(this.data.receive(10));
 	}
 
 	@Configuration
@@ -94,7 +96,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 		public MessageSource<InputStream> ftpMessageSource() {
 			SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null);
 			messageSource.setRemoteDirectory("sftpSource/");
-			messageSource.setFilter(new AcceptOnceFileListFilter<LsEntry>());
+			messageSource.setFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming"));
 			return messageSource;
 		}
 

spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java

@@ -31,7 +31,6 @@ import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
-import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -57,7 +56,6 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.event.inbound.ApplicationEventListeningMessageProducer;
 import org.springframework.integration.test.support.LogAdjustingTestSupport;
-import org.springframework.integration.test.util.TestUtils;
 import org.springframework.integration.transformer.ExpressionEvaluatingTransformer;
 import org.springframework.integration.websocket.ClientWebSocketContainer;
 import org.springframework.integration.websocket.IntegrationWebSocketContainer;
@@ -108,7 +106,7 @@ import org.springframework.web.socket.sockjs.client.WebSocketTransport;
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
-@DirtiesContext
+@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
 public class StompIntegrationTests extends LogAdjustingTestSupport {
 
 	@Value("#{server.serverContext}")
@@ -133,12 +131,6 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
 		super("org.springframework", "org.springframework.integration");
 	}
 
-	@Before
-	public void setup() {
-		this.webSocketInputChannel.clear();
-		this.webSocketEvents.clear();
-	}
-
 
 	@Test
 	public void sendMessageToController() throws Exception {
@@ -331,7 +323,6 @@ public class StompIntegrationTests extends LogAdjustingTestSupport {
 		StompHeaderAccessor stompHeaderAccessor = StompHeaderAccessor.create(StompCommand.MESSAGE);
 		stompHeaderAccessor.setDestination(destination);
 		Message<byte[]> message = MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.toMessageHeaders());
-		Object sessions = TestUtils.getPropertyValue(subscriptionRegistry, "subscriptionRegistry.sessions");
 		MultiValueMap<String, String> subscriptions = subscriptionRegistry.findSubscriptions(message);
 		return !subscriptions.isEmpty();
 	}