diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml index 6cfd28f958..b46652504e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests-context.xml @@ -20,7 +20,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java index 70d81c2197..707c8a3069 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -39,6 +39,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @@ -64,7 +65,7 @@ public class StreamTransformerParserTests { @Test public void directChannelWithStringMessage() { this.directInput.send(new GenericMessage(new ByteArrayInputStream("foo".getBytes()))); - Message result = output.receive(0); + Message result = output.receive(10000); assertNotNull(result); assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload()); } @@ -72,7 +73,7 @@ public class StreamTransformerParserTests { @Test public void queueChannelWithStringMessage() { this.queueInput.send(new GenericMessage(new ByteArrayInputStream("foo".getBytes()))); - Message result = output.receive(3000); + Message result = output.receive(10000); assertNotNull(result); assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload()); } @@ -80,7 +81,7 @@ public class StreamTransformerParserTests { @Test public void charset() { this.charsetChannel.send(new GenericMessage(new ByteArrayInputStream("foo".getBytes()))); - Message result = output.receive(0); + Message result = output.receive(10000); assertNotNull(result); assertEquals("foo", result.getPayload()); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml index 9de7188e78..c7f995517b 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml @@ -65,6 +65,7 @@ + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java index 012569c3be..fa4f318ef8 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,6 +31,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.hamcrest.Matchers; +import org.junit.After; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -39,6 +40,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.core.serializer.support.SerializationFailedException; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.store.MessageGroup; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; @@ -83,17 +85,28 @@ public class JdbcMessageStoreChannelIntegrationTests { @Autowired private MessageChannel routingSlip; + @Autowired + @Qualifier("service-activator") + private AbstractEndpoint serviceActivator; + @Before public void clear() { Service.reset(1); for (MessageGroup group : messageStore) { messageStore.removeMessageGroup(group.getGroupId()); } + + this.serviceActivator.start(); + } + + @After + public void tearDown() { + this.serviceActivator.stop(); } @Test public void testSendAndActivate() throws Exception { - input.send(new GenericMessage("foo")); + input.send(new GenericMessage<>("foo")); Service.await(10000); assertEquals(1, Service.messages.size()); } @@ -101,7 +114,7 @@ public class JdbcMessageStoreChannelIntegrationTests { @Test public void testSendAndActivateWithRollback() throws Exception { Service.fail = true; - input.send(new GenericMessage("foo")); + input.send(new GenericMessage<>("foo")); Service.await(10000); assertThat(Service.messages.size(), Matchers.greaterThanOrEqualTo(1)); // After a rollback in the poller the message is still waiting to be delivered @@ -128,28 +141,24 @@ public class JdbcMessageStoreChannelIntegrationTests { @Repeat(2) public void testTransactionalSendAndReceive() throws Exception { - boolean result = new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + boolean result = new TransactionTemplate(transactionManager).execute(status -> { - @Override - public Boolean doInTransaction(TransactionStatus status) { - - synchronized (storeLock) { - - boolean result = input.send(new GenericMessage("foo"), 500L); - // This will time out because the transaction has not committed yet - try { - Service.await(3000); - fail("Expected timeout"); - } - catch (Exception e) { - // expected - } - - return result; + synchronized (storeLock) { + boolean result1 = input.send(new GenericMessage<>("foo"), 100L); + // This will time out because the transaction has not committed yet + try { + Service.await(100); + fail("Expected timeout"); + } + catch (Exception e) { + // expected } + return result1; + } + }); assertTrue("Could not send message", result); @@ -184,7 +193,7 @@ public class JdbcMessageStoreChannelIntegrationTests { } @Test - public void testSameTransactionSendAndReceive() throws Exception { + public void testSameTransactionSendAndReceive() { final StopWatch stopWatch = new StopWatch(); DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition(); @@ -239,7 +248,7 @@ public class JdbcMessageStoreChannelIntegrationTests { @Test public void testWithRoutingSlip() { try { - this.routingSlip.send(new GenericMessage("foo")); + this.routingSlip.send(new GenericMessage<>("foo")); fail("MessageDeliveryException expected"); } catch (Exception e) { @@ -255,7 +264,7 @@ public class JdbcMessageStoreChannelIntegrationTests { private static boolean fail = false; - private static List messages = new CopyOnWriteArrayList(); + private static List messages = new CopyOnWriteArrayList<>(); private static CountDownLatch latch; diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml index 583f07ce63..bc8619cefb 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml @@ -59,6 +59,7 @@ + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java index d70c57d62a..227529a3f3 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,6 +26,7 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.junit.After; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -33,6 +34,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.store.MessageGroup; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; @@ -44,6 +46,11 @@ import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; import org.springframework.util.StopWatch; +/** + * @author Dave Syer + * @author Gary Russell + * @author Artem Bilan + */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @DirtiesContext // close at the end after class @@ -65,54 +72,58 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests { @Autowired private PlatformTransactionManager transactionManager; + @Autowired + @Qualifier("service-relay") + private AbstractEndpoint serviceRelay; + @Before public void clear() { - for (MessageGroup group : messageStore) { - messageStore.removeMessageGroup(group.getGroupId()); + for (MessageGroup group : this.messageStore) { + this.messageStore.removeMessageGroup(group.getGroupId()); } } - @Test - // @Repeat(50) - public void testSameTransactionDifferentChannelSendAndReceive() throws Exception { + @After + public void tearDown() { + this.serviceRelay.stop(); + } + @Test + public void testSameTransactionDifferentChannelSendAndReceive() throws Exception { Service.reset(1); - assertNull(durable.receive(100L)); - assertNull(relay.receive(100L)); + assertNull(this.durable.receive(100L)); + assertNull(this.relay.receive(100L)); final StopWatch stopWatch = new StopWatch(); - boolean result = new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + boolean result = + new TransactionTemplate(this.transactionManager) + .execute(status -> { - @Override - public Boolean doInTransaction(TransactionStatus status) { + synchronized (this.storeLock) { - synchronized (storeLock) { + boolean result1 = this.relay.send(new GenericMessage<>("foo"), 500L); + // This will time out because the transaction has not committed yet + try { + Service.await(100); + fail("Expected timeout"); + } + catch (Exception e) { + // expected + } - boolean result = relay.send(new GenericMessage("foo"), 500L); - // This will time out because the transaction has not committed yet - try { - Service.await(1000); - fail("Expected timeout"); - } - catch (Exception e) { - // expected - } + try { + stopWatch.start(); + // It hasn't arrive yet because we are still in the sending transaction + assertNull(this.durable.receive(100L)); + } + finally { + stopWatch.stop(); + } - try { - stopWatch.start(); - // It hasn't arrive yet because we are still in the sending transaction - assertNull(durable.receive(100L)); - } - finally { - stopWatch.stop(); - } + return result1; - return result; - - } - - } - }); + } + }); assertTrue("Could not send message", result); // If the poll blocks in the RDBMS there is no way for the queue to respect the timeout @@ -130,25 +141,21 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests { * * With the storeLock: It doesn't deadlock as long as the lock is injected into the poller as well. */ - new TransactionTemplate(transactionManager).execute(new TransactionCallback() { + new TransactionTemplate(this.transactionManager) + .execute(status -> { + synchronized (this.storeLock) { - @Override - public Void doInTransaction(TransactionStatus status) { - synchronized (storeLock) { + try { + stopWatch.start(); + this.durable.receive(100L); + return null; + } + finally { + stopWatch.stop(); + } - try { - stopWatch.start(); - durable.receive(100L); - return null; } - finally { - stopWatch.stop(); - } - - } - } - - }); + }); // If the poll blocks in the RDBMS there is no way for the queue to respect the timeout assertTrue("Timed out waiting for receive", stopWatch.getTotalTimeMillis() < 10000); @@ -156,6 +163,7 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests { } public static class Service { + private static boolean fail = false; private static List messages = new CopyOnWriteArrayList(); @@ -182,6 +190,7 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests { } return input; } + } }