From 6c774638cc2827791cb8894602548205ffa862c1 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 2 May 2016 13:30:02 -0400 Subject: [PATCH] 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 saved = (Message) 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. --- .../integration/store/SimpleMessageStore.java | 10 +++++ .../aggregator/AggregatorTests.java | 44 ++++++++++++++++--- .../FtpStreamingMessageSourceTests.java | 8 ++-- .../SftpStreamingMessageSourceTests.java | 8 ++-- .../client/StompIntegrationTests.java | 11 +---- 5 files changed, 60 insertions(+), 21 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index c736440a68..ff2a6c2cf6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/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() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java index 5a14bcaaea..790cdb2699 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatorTests.java +++ b/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 expiryEvents = new ArrayList(); + private final List expiryEvents = new ArrayList(); @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> 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("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> 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 diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java index 48e52391fb..7954be4db7 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java +++ b/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) 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 ftpMessageSource() { FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null); messageSource.setRemoteDirectory("ftpSource/"); - messageSource.setFilter(new AcceptOnceFileListFilter()); + messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming")); return messageSource; } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java index 8fe9387021..49dc3d92b4 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java +++ b/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) 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 ftpMessageSource() { SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(), null); messageSource.setRemoteDirectory("sftpSource/"); - messageSource.setFilter(new AcceptOnceFileListFilter()); + messageSource.setFilter(new SftpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming")); return messageSource; } diff --git a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java b/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java index 9be482ff9c..f6bbd7bb3f 100644 --- a/spring-integration-websocket/src/test/java/org/springframework/integration/websocket/client/StompIntegrationTests.java +++ b/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 message = MessageBuilder.createMessage(new byte[0], stompHeaderAccessor.toMessageHeaders()); - Object sessions = TestUtils.getPropertyValue(subscriptionRegistry, "subscriptionRegistry.sessions"); MultiValueMap subscriptions = subscriptionRegistry.findSubscriptions(message); return !subscriptions.isEmpty(); }