diff --git a/build.gradle b/build.gradle index 524d04e22c..65c2061d59 100644 --- a/build.gradle +++ b/build.gradle @@ -244,6 +244,7 @@ subprojects { subproject -> } } + check.dependsOn javadoc build.dependsOn jacocoTestReport } diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java index 011b29c20d..c580b99b6f 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java @@ -19,7 +19,6 @@ package org.springframework.integration.amqp.inbound; import java.util.Map; import org.springframework.amqp.core.AcknowledgeMode; -import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.support.AmqpHeaders; @@ -31,8 +30,6 @@ import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.util.Assert; -import com.rabbitmq.client.Channel; - /** * Adapter that receives Messages from an AMQP Queue, converts them into * Spring Integration Messages, and sends the results to a Message Channel. @@ -80,21 +77,16 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements @Override protected void onInit() { - this.messageListenerContainer.setMessageListener(new ChannelAwareMessageListener() { - - @Override - public void onMessage(Message message, Channel channel) throws Exception { - Object payload = AmqpInboundChannelAdapter.this.messageConverter.fromMessage(message); - Map headers = - AmqpInboundChannelAdapter.this.headerMapper.toHeadersFromRequest(message.getMessageProperties()); - if (AmqpInboundChannelAdapter.this.messageListenerContainer.getAcknowledgeMode() - == AcknowledgeMode.MANUAL) { - headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag()); - headers.put(AmqpHeaders.CHANNEL, channel); - } - sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build()); + this.messageListenerContainer.setMessageListener((ChannelAwareMessageListener) (message, channel) -> { + Object payload = this.messageConverter.fromMessage(message); + Map headers = + this.headerMapper.toHeadersFromRequest(message.getMessageProperties()); + if (this.messageListenerContainer.getAcknowledgeMode() + == AcknowledgeMode.MANUAL) { + headers.put(AmqpHeaders.DELIVERY_TAG, message.getMessageProperties().getDeliveryTag()); + headers.put(AmqpHeaders.CHANNEL, channel); } - + sendMessage(getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build()); }); this.messageListenerContainer.afterPropertiesSet(); super.onInit(); diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index c0f246adee..05dc6e9cc1 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -16,9 +16,7 @@ package org.springframework.integration.amqp.outbound; -import org.springframework.amqp.AmqpException; import org.springframework.amqp.core.AmqpTemplate; -import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; import org.springframework.amqp.rabbit.support.CorrelationData; @@ -107,14 +105,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint } else { this.amqpTemplate.convertAndSend(exchangeName, routingKey, requestMessage.getPayload(), - new MessagePostProcessor() { - @Override - public org.springframework.amqp.core.Message postProcessMessage( - org.springframework.amqp.core.Message message) throws AmqpException { - getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(), - message.getMessageProperties()); - return message; - } + message -> { + getHeaderMapper().fromHeadersToRequest(requestMessage.getHeaders(), + message.getMessageProperties()); + return message; }); } } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java index 3ce0455b5a..ec525151ca 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java @@ -50,8 +50,6 @@ import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.support.LogAdjustingTestSupport; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; @@ -112,15 +110,11 @@ public class ChannelTests extends LogAdjustingTestSupport { @DirtiesContext public void pubSubLostConnectionTest() throws Exception { final CyclicBarrier latch = new CyclicBarrier(2); - channel.subscribe(new MessageHandler() { - - @Override - public void handleMessage(Message message) throws MessagingException { - try { - latch.await(10, TimeUnit.SECONDS); - } - catch (Exception e) { - } + channel.subscribe(message -> { + try { + latch.await(10, TimeUnit.SECONDS); + } + catch (Exception e) { } }); this.channel.send(new GenericMessage("foo")); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java index 2af2040743..2b058eff3d 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java @@ -34,8 +34,6 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.junit.Test; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.amqp.core.AmqpAdmin; import org.springframework.amqp.core.AmqpTemplate; @@ -70,12 +68,7 @@ public class DispatcherHasNoSubscribersTests { when(channel.queueDeclare(anyString(), anyBoolean(), anyBoolean(), anyBoolean(), any(Map.class))) .thenReturn(declareOk); Connection connection = mock(Connection.class); - doAnswer(new Answer() { - @Override - public Channel answer(InvocationOnMock invocation) throws Throwable { - return channel; - } - }).when(connection).createChannel(anyBoolean()); + doAnswer(invocation -> channel).when(connection).createChannel(anyBoolean()); ConnectionFactory connectionFactory = mock(ConnectionFactory.class); when(connectionFactory.createConnection()).thenReturn(connection); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); @@ -103,12 +96,7 @@ public class DispatcherHasNoSubscribersTests { public void testPubSub() { final Channel channel = mock(Channel.class); Connection connection = mock(Connection.class); - doAnswer(new Answer() { - @Override - public Channel answer(InvocationOnMock invocation) throws Throwable { - return channel; - } - }).when(connection).createChannel(anyBoolean()); + doAnswer(invocation -> channel).when(connection).createChannel(anyBoolean()); ConnectionFactory connectionFactory = mock(ConnectionFactory.class); when(connectionFactory.createConnection()).thenReturn(connection); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); @@ -139,16 +127,12 @@ public class DispatcherHasNoSubscribersTests { channel, "container", SimpleMessageListenerContainer.class); Log logger = mock(Log.class); final ArrayList logList = new ArrayList(); - doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) - throws Throwable { - String message = (String) invocation.getArguments()[0]; - if (message.startsWith("Dispatcher has no subscribers")) { - logList.add(message); - } - return null; + doAnswer(invocation -> { + String message = (String) invocation.getArguments()[0]; + if (message.startsWith("Dispatcher has no subscribers")) { + logList.add(message); } + return null; }).when(logger).warn(anyString(), any(Exception.class)); when(logger.isWarnEnabled()).thenReturn(true); Object listener = container.getMessageListener(); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java index e2e5d138f1..505d093a39 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java @@ -26,8 +26,6 @@ import java.lang.reflect.Field; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.amqp.core.Address; import org.springframework.amqp.core.Message; @@ -47,8 +45,6 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessagingException; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -104,14 +100,11 @@ public class AmqpInboundGatewayParserTests { @Test public void verifyUsageWithHeaderMapper() throws Exception { DirectChannel requestChannel = context.getBean("requestChannel", DirectChannel.class); - requestChannel.subscribe(new MessageHandler() { - @Override - public void handleMessage(org.springframework.messaging.Message siMessage) - throws MessagingException { - org.springframework.messaging.Message replyMessage = MessageBuilder.fromMessage(siMessage).setHeader("bar", "bar").build(); - MessageChannel replyChannel = (MessageChannel) siMessage.getHeaders().getReplyChannel(); - replyChannel.send(replyMessage); - } + requestChannel.subscribe(siMessage -> { + org.springframework.messaging.Message replyMessage = MessageBuilder.fromMessage(siMessage) + .setHeader("bar", "bar").build(); + MessageChannel replyChannel = (MessageChannel) siMessage.getHeaders().getReplyChannel(); + replyChannel.send(replyMessage); }); final AmqpInboundGateway gateway = context.getBean("withHeaderMapper", AmqpInboundGateway.class); @@ -121,15 +114,12 @@ public class AmqpInboundGatewayParserTests { RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(gateway, "amqpTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); - Mockito.doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - Object[] args = invocation.getArguments(); - Message amqpReplyMessage = (Message) args[2]; - MessageProperties properties = amqpReplyMessage.getMessageProperties(); - assertEquals("bar", properties.getHeaders().get("bar")); - return null; - } + Mockito.doAnswer(invocation -> { + Object[] args = invocation.getArguments(); + Message amqpReplyMessage = (Message) args[2]; + MessageProperties properties = amqpReplyMessage.getMessageProperties(); + assertEquals("bar", properties.getHeaders().get("bar")); + return null; }).when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(Message.class), Mockito.any(CorrelationData.class)); ReflectionUtils.setField(amqpTemplateField, gateway, amqpTemplate); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java index 90575a39ae..b0b63aa0cd 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java @@ -43,8 +43,6 @@ import org.junit.runner.RunWith; import org.mockito.Matchers; import org.mockito.Mockito; import org.mockito.internal.stubbing.answers.DoesNothing; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; @@ -133,19 +131,16 @@ public class AmqpOutboundChannelAdapterParserTests { amqpTemplate = Mockito.spy(amqpTemplate); final AtomicBoolean shouldBePersistent = new AtomicBoolean(); - Mockito.doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - Object[] args = invocation.getArguments(); - org.springframework.amqp.core.Message amqpMessage = (org.springframework.amqp.core.Message) args[2]; - MessageProperties properties = amqpMessage.getMessageProperties(); - assertEquals("foo", properties.getHeaders().get("foo")); - assertEquals("foobar", properties.getHeaders().get("foobar")); - assertNull(properties.getHeaders().get("bar")); - assertEquals(shouldBePersistent.get() ? MessageDeliveryMode.PERSISTENT - : MessageDeliveryMode.NON_PERSISTENT, properties.getDeliveryMode()); - return null; - } + Mockito.doAnswer(invocation -> { + Object[] args = invocation.getArguments(); + org.springframework.amqp.core.Message amqpMessage = (org.springframework.amqp.core.Message) args[2]; + MessageProperties properties = amqpMessage.getMessageProperties(); + assertEquals("foo", properties.getHeaders().get("foo")); + assertEquals("foobar", properties.getHeaders().get("foobar")); + assertNull(properties.getHeaders().get("bar")); + assertEquals(shouldBePersistent.get() ? MessageDeliveryMode.PERSISTENT + : MessageDeliveryMode.NON_PERSISTENT, properties.getDeliveryMode()); + return null; }) .when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class)); @@ -197,16 +192,13 @@ public class AmqpOutboundChannelAdapterParserTests { RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); - Mockito.doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - Object[] args = invocation.getArguments(); - org.springframework.amqp.core.Message amqpMessage = (org.springframework.amqp.core.Message) args[2]; - MessageProperties properties = amqpMessage.getMessageProperties(); - assertEquals("hello", new String(amqpMessage.getBody())); - assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode()); - return null; - } + Mockito.doAnswer(invocation -> { + Object[] args = invocation.getArguments(); + org.springframework.amqp.core.Message amqpMessage = (org.springframework.amqp.core.Message) args[2]; + MessageProperties properties = amqpMessage.getMessageProperties(); + assertEquals("hello", new String(amqpMessage.getBody())); + assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode()); + return null; }) .when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java index 3b1c907295..21be0c7e17 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java @@ -30,8 +30,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.MessageProperties; @@ -116,22 +114,19 @@ public class AmqpOutboundGatewayParserTests { amqpTemplate = Mockito.spy(amqpTemplate); final AtomicBoolean shouldBePersistent = new AtomicBoolean(); - Mockito.doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - Object[] args = invocation.getArguments(); - org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2]; - MessageProperties properties = amqpRequestMessage.getMessageProperties(); - assertEquals("foo", properties.getHeaders().get("foo")); - assertEquals(shouldBePersistent.get() ? MessageDeliveryMode.PERSISTENT - : MessageDeliveryMode.NON_PERSISTENT, properties.getDeliveryMode()); - // mock reply AMQP message - MessageProperties amqpProperties = new MessageProperties(); - amqpProperties.setAppId("test.appId"); - amqpProperties.setHeader("foobar", "foobar"); - amqpProperties.setHeader("bar", "bar"); - return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties); - } + Mockito.doAnswer(invocation -> { + Object[] args = invocation.getArguments(); + org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2]; + MessageProperties properties = amqpRequestMessage.getMessageProperties(); + assertEquals("foo", properties.getHeaders().get("foo")); + assertEquals(shouldBePersistent.get() ? MessageDeliveryMode.PERSISTENT + : MessageDeliveryMode.NON_PERSISTENT, properties.getDeliveryMode()); + // mock reply AMQP message + MessageProperties amqpProperties = new MessageProperties(); + amqpProperties.setAppId("test.appId"); + amqpProperties.setHeader("foobar", "foobar"); + amqpProperties.setHeader("bar", "bar"); + return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties); }) .when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), any(CorrelationData.class)); @@ -184,22 +179,19 @@ public class AmqpOutboundGatewayParserTests { RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); - Mockito.doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - Object[] args = invocation.getArguments(); - org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2]; - MessageProperties properties = amqpRequestMessage.getMessageProperties(); - assertEquals("foo", properties.getHeaders().get("foo")); - // mock reply AMQP message - MessageProperties amqpProperties = new MessageProperties(); - amqpProperties.setAppId("test.appId"); - amqpProperties.setHeader("foobar", "foobar"); - amqpProperties.setHeader("bar", "bar"); - assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode()); - amqpProperties.setReceivedDeliveryMode(properties.getDeliveryMode()); - return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties); - } + Mockito.doAnswer(invocation -> { + Object[] args = invocation.getArguments(); + org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2]; + MessageProperties properties = amqpRequestMessage.getMessageProperties(); + assertEquals("foo", properties.getHeaders().get("foo")); + // mock reply AMQP message + MessageProperties amqpProperties = new MessageProperties(); + amqpProperties.setAppId("test.appId"); + amqpProperties.setHeader("foobar", "foobar"); + amqpProperties.setHeader("bar", "bar"); + assertEquals(MessageDeliveryMode.PERSISTENT, properties.getDeliveryMode()); + amqpProperties.setReceivedDeliveryMode(properties.getDeliveryMode()); + return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties); }) .when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), any(CorrelationData.class)); @@ -240,20 +232,17 @@ public class AmqpOutboundGatewayParserTests { RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); - Mockito.doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) { - Object[] args = invocation.getArguments(); - org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2]; - MessageProperties properties = amqpRequestMessage.getMessageProperties(); - assertNull(properties.getHeaders().get("foo")); - // mock reply AMQP message - MessageProperties amqpProperties = new MessageProperties(); - amqpProperties.setAppId("test.appId"); - amqpProperties.setHeader("foobar", "foobar"); - amqpProperties.setHeader("bar", "bar"); - return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties); - } + Mockito.doAnswer(invocation -> { + Object[] args = invocation.getArguments(); + org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2]; + MessageProperties properties = amqpRequestMessage.getMessageProperties(); + assertNull(properties.getHeaders().get("foo")); + // mock reply AMQP message + MessageProperties amqpProperties = new MessageProperties(); + amqpProperties.setAppId("test.appId"); + amqpProperties.setHeader("foobar", "foobar"); + amqpProperties.setHeader("bar", "bar"); + return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties); }) .when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), any(CorrelationData.class)); @@ -295,20 +284,17 @@ public class AmqpOutboundGatewayParserTests { RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class); amqpTemplate = Mockito.spy(amqpTemplate); - Mockito.doAnswer(new Answer() { - @Override - public org.springframework.amqp.core.Message answer(InvocationOnMock invocation) { - Object[] args = invocation.getArguments(); - org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2]; - MessageProperties properties = amqpRequestMessage.getMessageProperties(); - assertNull(properties.getHeaders().get("foo")); - // mock reply AMQP message - MessageProperties amqpProperties = new MessageProperties(); - amqpProperties.setAppId("test.appId"); - amqpProperties.setHeader("foobar", "foobar"); - amqpProperties.setHeader("bar", "bar"); - return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties); - } + Mockito.doAnswer(invocation -> { + Object[] args = invocation.getArguments(); + org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2]; + MessageProperties properties = amqpRequestMessage.getMessageProperties(); + assertNull(properties.getHeaders().get("foo")); + // mock reply AMQP message + MessageProperties amqpProperties = new MessageProperties(); + amqpProperties.setAppId("test.appId"); + amqpProperties.setHeader("foobar", "foobar"); + amqpProperties.setHeader("bar", "bar"); + return new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties); }) .when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class), any(CorrelationData.class)); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java index 434e3484a3..33a978b3da 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java @@ -26,8 +26,6 @@ import static org.mockito.Mockito.when; import org.junit.After; import org.junit.Test; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.NoSuchBeanDefinitionException; @@ -84,12 +82,7 @@ public class OutboundGatewayTests { @SuppressWarnings("unchecked") public void testExpressionsBeanResolver() throws Exception { ApplicationContext context = mock(ApplicationContext.class); - doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - return invocation.getArguments()[0] + "bar"; - } - }).when(context).getBean(anyString()); + doAnswer(invocation -> invocation.getArguments()[0] + "bar").when(context).getBean(anyString()); when(context.containsBean(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME)).thenReturn(true); when(context.getBean(ClassUtils.forName("org.springframework.integration.config.SpelPropertyAccessorRegistrar", context.getClassLoader()))) diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index a7557f9c88..8b4a5fe96a 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -30,8 +30,6 @@ import java.util.Map; import org.junit.Test; import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.MessageProperties; @@ -70,12 +68,7 @@ public class InboundEndpointTests { @Test public void testInt2809JavaTypePropertiesToAmqp() throws Exception { Connection connection = mock(Connection.class); - doAnswer(new Answer() { - @Override - public Channel answer(InvocationOnMock invocation) throws Throwable { - return mock(Channel.class); - } - }).when(connection).createChannel(anyBoolean()); + doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean()); ConnectionFactory connectionFactory = mock(ConnectionFactory.class); when(connectionFactory.createConnection()).thenReturn(connection); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); @@ -116,12 +109,7 @@ public class InboundEndpointTests { @Test public void testInt2809JavaTypePropertiesFromAmqp() throws Exception { Connection connection = mock(Connection.class); - doAnswer(new Answer() { - @Override - public Channel answer(InvocationOnMock invocation) throws Throwable { - return mock(Channel.class); - } - }).when(connection).createChannel(anyBoolean()); + doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean()); ConnectionFactory connectionFactory = mock(ConnectionFactory.class); when(connectionFactory.createConnection()).thenReturn(connection); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); @@ -154,12 +142,7 @@ public class InboundEndpointTests { @Test public void testMessageConverterJsonHeadersHavePrecedenceOverMessageHeaders() throws Exception { Connection connection = mock(Connection.class); - doAnswer(new Answer() { - @Override - public Channel answer(InvocationOnMock invocation) throws Throwable { - return mock(Channel.class); - } - }).when(connection).createChannel(anyBoolean()); + doAnswer(invocation -> mock(Channel.class)).when(connection).createChannel(anyBoolean()); ConnectionFactory connectionFactory = mock(ConnectionFactory.class); when(connectionFactory.createConnection()).thenReturn(connection); SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); @@ -170,38 +153,30 @@ public class InboundEndpointTests { final Channel rabbitChannel = mock(Channel.class); - channel.subscribe(new MessageTransformingHandler(new Transformer() { - - @Override - public Message transform(Message message) { - assertSame(rabbitChannel, message.getHeaders().get(AmqpHeaders.CHANNEL)); - assertEquals(123L, message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)); - return MessageBuilder.fromMessage(message) - .setHeader(JsonHeaders.TYPE_ID, "foo") - .setHeader(JsonHeaders.CONTENT_TYPE_ID, "bar") - .setHeader(JsonHeaders.KEY_TYPE_ID, "baz") - .build(); - } + channel.subscribe(new MessageTransformingHandler(message -> { + assertSame(rabbitChannel, message.getHeaders().get(AmqpHeaders.CHANNEL)); + assertEquals(123L, message.getHeaders().get(AmqpHeaders.DELIVERY_TAG)); + return MessageBuilder.fromMessage(message) + .setHeader(JsonHeaders.TYPE_ID, "foo") + .setHeader(JsonHeaders.CONTENT_TYPE_ID, "bar") + .setHeader(JsonHeaders.KEY_TYPE_ID, "baz") + .build(); })); RabbitTemplate rabbitTemplate = Mockito.mock(RabbitTemplate.class); - Mockito.doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - org.springframework.amqp.core.Message message = - (org.springframework.amqp.core.Message) invocation.getArguments()[2]; - Map headers = message.getMessageProperties().getHeaders(); - assertTrue(headers.containsKey(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); - assertNotEquals("foo", headers.get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); - assertFalse(headers.containsKey(JsonHeaders.CONTENT_TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); - assertFalse(headers.containsKey(JsonHeaders.KEY_TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); - assertFalse(headers.containsKey(JsonHeaders.TYPE_ID)); - assertFalse(headers.containsKey(JsonHeaders.KEY_TYPE_ID)); - assertFalse(headers.containsKey(JsonHeaders.CONTENT_TYPE_ID)); - return null; - } + Mockito.doAnswer(invocation -> { + org.springframework.amqp.core.Message message = + (org.springframework.amqp.core.Message) invocation.getArguments()[2]; + Map headers = message.getMessageProperties().getHeaders(); + assertTrue(headers.containsKey(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); + assertNotEquals("foo", headers.get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); + assertFalse(headers.containsKey(JsonHeaders.CONTENT_TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); + assertFalse(headers.containsKey(JsonHeaders.KEY_TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); + assertFalse(headers.containsKey(JsonHeaders.TYPE_ID)); + assertFalse(headers.containsKey(JsonHeaders.KEY_TYPE_ID)); + assertFalse(headers.containsKey(JsonHeaders.CONTENT_TYPE_ID)); + return null; }).when(rabbitTemplate).send(Mockito.anyString(), Mockito.anyString(), Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class)); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java index fa7f2e41a2..0850bce04f 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java @@ -40,17 +40,15 @@ import org.junit.ClassRule; import org.junit.Rule; import org.junit.Test; import org.mockito.Matchers; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.amqp.core.AmqpReplyTimeoutException; -import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.rabbit.AsyncRabbitTemplate; import org.springframework.amqp.rabbit.AsyncRabbitTemplate.RabbitMessageFuture; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter; +import org.springframework.amqp.rabbit.listener.adapter.ReplyingMessageListener; import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; @@ -61,7 +59,6 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.rule.Log4jLevelAdjuster; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.concurrent.SettableListenableFuture; @@ -107,20 +104,16 @@ public class AsyncAmqpGatewayTests { receiver.setBeanName("receiver"); receiver.setQueueNames("asyncQ1"); final CountDownLatch waitForAckBeforeReplying = new CountDownLatch(1); - MessageListenerAdapter messageListener = new MessageListenerAdapter(new Object() { - - @SuppressWarnings("unused") - public String handleMessage(String foo) { - try { - waitForAckBeforeReplying.await(10, TimeUnit.SECONDS); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - return foo.toUpperCase(); - } - - }); + MessageListenerAdapter messageListener = new MessageListenerAdapter( + (ReplyingMessageListener) foo -> { + try { + waitForAckBeforeReplying.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return foo.toUpperCase(); + }); receiver.setMessageListener(messageListener); receiver.afterPropertiesSet(); receiver.start(); @@ -129,15 +122,10 @@ public class AsyncAmqpGatewayTests { Log logger = spy(TestUtils.getPropertyValue(gateway, "logger", Log.class)); given(logger.isDebugEnabled()).willReturn(true); final CountDownLatch replyTimeoutLatch = new CountDownLatch(1); - willAnswer(new Answer() { - - @Override - public Void answer(InvocationOnMock invocation) throws Throwable { - invocation.callRealMethod(); - replyTimeoutLatch.countDown(); - return null; - } - + willAnswer(invocation -> { + invocation.callRealMethod(); + replyTimeoutLatch.countDown(); + return null; }).given(logger).debug(Matchers.startsWith("Reply not required and async timeout for")); new DirectFieldAccessor(gateway).setPropertyValue("logger", logger); QueueChannel outputChannel = new QueueChannel(); @@ -175,13 +163,7 @@ public class AsyncAmqpGatewayTests { // timeout tests asyncTemplate.setReceiveTimeout(10); - receiver.setMessageListener(new MessageListener() { - - @Override - public void onMessage(org.springframework.amqp.core.Message message) { - } - - }); + receiver.setMessageListener(message1 -> { }); // reply timeout with no requiresReply message = MessageBuilder.withPayload("bar").setErrorChannel(errorChannel).build(); gateway.handleMessage(message); @@ -203,13 +185,8 @@ public class AsyncAmqpGatewayTests { // error on sending result DirectChannel errorForce = new DirectChannel(); errorForce.setBeanName("errorForce"); - errorForce.subscribe(new MessageHandler() { - - @Override - public void handleMessage(Message message) throws MessagingException { - throw new RuntimeException("intentional"); - } - + errorForce.subscribe(message1 -> { + throw new RuntimeException("intentional"); }); gateway.setOutputChannel(errorForce); message = MessageBuilder.withPayload("qux").setErrorChannel(errorChannel).build(); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java index caf5a4483f..157c6818e2 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java @@ -28,8 +28,6 @@ import static org.mockito.Mockito.spy; import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.connection.ConnectionFactory; @@ -54,13 +52,9 @@ public class OutboundEndpointTests { AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate); final AtomicReference amqpMessage = new AtomicReference(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - amqpMessage.set((Message) invocation.getArguments()[2]); - return null; - } + doAnswer(invocation -> { + amqpMessage.set((Message) invocation.getArguments()[2]); + return null; }).when(amqpTemplate).send(anyString(), anyString(), any(Message.class), any(CorrelationData.class)); org.springframework.messaging.Message message = MessageBuilder.withPayload("foo") @@ -82,13 +76,9 @@ public class OutboundEndpointTests { endpoint.setHeaderMapper(mapper); final AtomicReference amqpMessage = new AtomicReference(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - amqpMessage.set((Message) invocation.getArguments()[2]); - return null; - } + doAnswer(invocation -> { + amqpMessage.set((Message) invocation.getArguments()[2]); + return null; }).when(amqpTemplate) .doSendAndReceiveWithTemporary(anyString(), anyString(), any(Message.class), any(CorrelationData.class)); org.springframework.messaging.Message message = MessageBuilder.withPayload("foo") diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java index d302be6fea..dbfa4b89e9 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java @@ -26,7 +26,6 @@ import org.junit.Test; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; -import org.springframework.amqp.rabbit.core.ChannelCallback; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter; import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter; @@ -36,7 +35,6 @@ import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import com.rabbitmq.client.AMQP.BasicProperties; -import com.rabbitmq.client.Channel; /** * @author Gary Russell @@ -73,14 +71,9 @@ public class JsonConverterCompatibilityTests { DefaultAmqpHeaderMapper.outboundMapper().fromHeadersToRequest(out.getHeaders(), messageProperties); final BasicProperties props = new DefaultMessagePropertiesConverter().fromMessageProperties(messageProperties, "UTF-8"); - this.rabbitTemplate.execute(new ChannelCallback() { - - @Override - public Void doInRabbit(Channel channel) throws Exception { - channel.basicPublish("", JSON_TESTQ, props, out.getPayload().getBytes()); - return null; - } - + this.rabbitTemplate.execute(channel -> { + channel.basicPublish("", JSON_TESTQ, props, out.getPayload().getBytes()); + return null; }); Object received = this.rabbitTemplate.receiveAndConvert(JSON_TESTQ); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java index ed723211ce..00f9e9a918 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java @@ -279,6 +279,7 @@ public final class MessageBuilder extends AbstractIntegrationMessageBuilder listener = new ApplicationListener() { - - @Override - public void onApplicationEvent(ApplicationEvent event) { - if (event instanceof PayloadApplicationEvent) { - String payload = (String) ((PayloadApplicationEvent) event).getPayload(); - if (payload.equals("hello")) { - receivedEvent = true; - } + ApplicationListener listener = event -> { + if (event instanceof PayloadApplicationEvent) { + String payload = (String) ((PayloadApplicationEvent) event).getPayload(); + if (payload.equals("hello")) { + receivedEvent = true; } } - }; this.context.addApplicationListener(listener); DirectChannel channel = context.getBean("input", DirectChannel.class); @@ -95,19 +89,14 @@ public class EventOutboundChannelAdapterParserTests { @Test public void withAdvice() { this.receivedEvent = false; - ApplicationListener listener = new ApplicationListener() { - - @Override - public void onApplicationEvent(ApplicationEvent event) { - Object source = event.getSource(); - if (source instanceof Message) { - String payload = (String) ((Message) source).getPayload(); - if (payload.equals("hello")) { - receivedEvent = true; - } + ApplicationListener listener = event -> { + Object source = event.getSource(); + if (source instanceof Message) { + String payload = (String) ((Message) source).getPayload(); + if (payload.equals("hello")) { + receivedEvent = true; } } - }; context.addApplicationListener(listener); DirectChannel channel = context.getBean("inputAdvice", DirectChannel.class); @@ -119,19 +108,14 @@ public class EventOutboundChannelAdapterParserTests { @Test //INT-2275 public void testInsideChain() { this.receivedEvent = false; - ApplicationListener listener = new ApplicationListener() { - - @Override - public void onApplicationEvent(ApplicationEvent event) { - Object source = event.getSource(); - if (source instanceof Message) { - String payload = (String) ((Message) source).getPayload(); - if (payload.equals("foobar")) { - receivedEvent = true; - } + ApplicationListener listener = event -> { + Object source = event.getSource(); + if (source instanceof Message) { + String payload = (String) ((Message) source).getPayload(); + if (payload.equals("foobar")) { + receivedEvent = true; } } - }; this.context.addApplicationListener(listener); DirectChannel channel = context.getBean("inputChain", DirectChannel.class); @@ -146,28 +130,23 @@ public class EventOutboundChannelAdapterParserTests { new ClassPathXmlApplicationContext("EventOutboundChannelAdapterParserTestsWithPollable-context.xml", EventOutboundChannelAdapterParserTests.class); final CyclicBarrier barrier = new CyclicBarrier(2); - ApplicationListener listener = new ApplicationListener() { - - @Override - public void onApplicationEvent(ApplicationEvent event) { - Object source = event.getSource(); - if (source instanceof Message) { - String payload = (String) ((Message) source).getPayload(); - if (payload.equals("hello")) { - receivedEvent = true; - try { - barrier.await(); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - catch (BrokenBarrierException e) { - throw new IllegalStateException("broken barrier", e); - } + ApplicationListener listener = event -> { + Object source = event.getSource(); + if (source instanceof Message) { + String payload = (String) ((Message) source).getPayload(); + if (payload.equals("hello")) { + receivedEvent = true; + try { + barrier.await(); + } + catch (InterruptedException e1) { + Thread.currentThread().interrupt(); + } + catch (BrokenBarrierException e2) { + throw new IllegalStateException("broken barrier", e2); } } } - }; context.addApplicationListener(listener); QueueChannel channel = context.getBean("input", QueueChannel.class); diff --git a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java index 3cd3c3dda6..6bc9268ecf 100644 --- a/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java +++ b/spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java @@ -46,7 +46,6 @@ import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessagingException; import com.rometools.rome.feed.synd.SyndEntry; @@ -128,12 +127,7 @@ public class FeedInboundChannelAdapterParserTests { @Ignore // goes against the real feed public void validateSuccessfulNewsRetrievalWithHttpUrl() throws Exception { final CountDownLatch latch = new CountDownLatch(3); - MessageHandler handler = spy(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - latch.countDown(); - } - }); + MessageHandler handler = spy((MessageHandler) message -> latch.countDown()); ConfigurableApplicationContext context = new ClassPathXmlApplicationContext( "FeedInboundChannelAdapterParserTests-http-context.xml", this.getClass()); DirectChannel feedChannel = context.getBean("feedChannel", DirectChannel.class); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index b552d8d0b2..8e74e50614 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -42,7 +42,6 @@ import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.remote.AbstractFileInfo; import org.springframework.integration.file.remote.MessageSessionCallback; import org.springframework.integration.file.remote.RemoteFileTemplate; -import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.support.FileExistsMode; @@ -520,15 +519,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply return doMput(requestMessage); } } - return this.remoteFileTemplate.execute(new SessionCallback() { - - @Override - public Object doInSession(Session session) throws IOException { - return AbstractRemoteFileOutboundGateway.this.messageSessionCallback.doInSession(session, - requestMessage); - } - - }); + return this.remoteFileTemplate.execute(session -> + AbstractRemoteFileOutboundGateway.this.messageSessionCallback.doInSession(session, requestMessage)); } private Object doLs(Message requestMessage) { @@ -537,13 +529,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply dir += this.remoteFileTemplate.getRemoteFileSeparator(); } final String fullDir = dir; - List payload = this.remoteFileTemplate.execute(new SessionCallback>() { - - @Override - public List doInSession(Session session) throws IOException { - return AbstractRemoteFileOutboundGateway.this.ls(session, fullDir); - } - }); + List payload = this.remoteFileTemplate.execute(session -> + AbstractRemoteFileOutboundGateway.this.ls(session, fullDir)); return this.getMessageBuilderFactory().withPayload(payload) .setHeader(FileHeaders.REMOTE_DIRECTORY, dir) .build(); @@ -567,14 +554,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply } } else { - payload = this.remoteFileTemplate.execute(new SessionCallback() { - - @Override - public File doInSession(Session session) throws IOException { - return get(requestMessage, session, remoteDir, remoteFilePath, remoteFilename, true); - - } - }); + payload = this.remoteFileTemplate.execute(session1 -> + get(requestMessage, session1, remoteDir, remoteFilePath, remoteFilename, true)); } return getMessageBuilderFactory().withPayload(payload) .setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir) @@ -588,13 +569,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage); final String remoteFilename = getRemoteFilename(remoteFilePath); final String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename); - List payload = this.remoteFileTemplate.execute(new SessionCallback>() { - - @Override - public List doInSession(Session session) throws IOException { - return mGet(requestMessage, session, remoteDir, remoteFilename); - } - }); + List payload = this.remoteFileTemplate.execute(session -> + mGet(requestMessage, session, remoteDir, remoteFilename)); return this.getMessageBuilderFactory().withPayload(payload) .setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir) .setHeader(FileHeaders.REMOTE_FILE, remoteFilename) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index df300bd44f..69b994b751 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java @@ -40,7 +40,6 @@ import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.filters.ReversibleFileListFilter; import org.springframework.integration.file.remote.RemoteFileTemplate; -import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.messaging.MessagingException; @@ -241,52 +240,40 @@ public abstract class AbstractInboundFileSynchronizer } final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class); try { - int transferred = this.remoteFileTemplate.execute(new SessionCallback() { - - @Override - public Integer doInSession(Session session) throws IOException { - F[] files = session.list(remoteDirectory); - if (!ObjectUtils.isEmpty(files)) { - List filteredFiles = filterFiles(files); - if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) { - rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); - List newList = new ArrayList<>(maxFetchSize); - for (int i = 0; i < maxFetchSize; i++) { - newList.add(filteredFiles.get(i)); - } - filteredFiles = newList; + int transferred = this.remoteFileTemplate.execute(session -> { + F[] files = session.list(remoteDirectory); + if (!ObjectUtils.isEmpty(files)) { + List filteredFiles = filterFiles(files); + if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) { + rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize)); + List newList = new ArrayList<>(maxFetchSize); + for (int i = 0; i < maxFetchSize; i++) { + newList.add(filteredFiles.get(i)); } - for (F file : filteredFiles) { - try { - if (file != null) { - copyFileToLocalDirectory( - remoteDirectory, file, localDirectory, - session); - } - } - catch (RuntimeException e) { - rollbackFromFileToListEnd(filteredFiles, file); - throw e; - } - catch (IOException e) { - rollbackFromFileToListEnd(filteredFiles, file); - throw e; + filteredFiles = newList; + } + for (F file : filteredFiles) { + try { + if (file != null) { + copyFileToLocalDirectory( + remoteDirectory, file, localDirectory, + session); } } - return filteredFiles.size(); - } - else { - return 0; + catch (RuntimeException e1) { + rollbackFromFileToListEnd(filteredFiles, file); + throw e1; + } + catch (IOException e2) { + rollbackFromFileToListEnd(filteredFiles, file); + throw e2; + } } + return filteredFiles.size(); } - - public void rollbackFromFileToListEnd(List filteredFiles, F file) { - if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) { - ((ReversibleFileListFilter) AbstractInboundFileSynchronizer.this.filter) - .rollback(file, filteredFiles); - } + else { + return 0; } - }); if (this.logger.isDebugEnabled()) { this.logger.debug(transferred + " files transferred"); @@ -297,6 +284,13 @@ public abstract class AbstractInboundFileSynchronizer } } + protected void rollbackFromFileToListEnd(List filteredFiles, F file) { + if (this.filter instanceof ReversibleFileListFilter) { + ((ReversibleFileListFilter) this.filter) + .rollback(file, filteredFiles); + } + } + protected void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory, Session session) throws IOException { String remoteFileName = this.getFilename(remoteFile); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java index 7390bdec51..988837a797 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java @@ -84,13 +84,7 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr super.doStart(); destroyProcess(); this.command = "tail " + this.options + " " + this.getFile().getAbsolutePath(); - this.getTaskExecutor().execute(new Runnable() { - - @Override - public void run() { - runExec(); - } - }); + this.getTaskExecutor().execute(() -> runExec()); } @Override @@ -145,48 +139,39 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr * Runs a thread that waits for the Process result. */ private void startProcessMonitor() { - this.getTaskExecutor().execute(new Runnable() { + this.getTaskExecutor().execute(() -> { + Process process = OSDelegatingFileTailingMessageProducer.this.process; + if (process == null) { + if (logger.isDebugEnabled()) { + logger.debug("Process destroyed before starting process monitor"); + } + return; + } - @Override - public void run() { - Process process = OSDelegatingFileTailingMessageProducer.this.process; - if (process == null) { - if (logger.isDebugEnabled()) { - logger.debug("Process destroyed before starting process monitor"); - } - return; + int result = Integer.MIN_VALUE; + try { + if (logger.isDebugEnabled()) { + logger.debug("Monitoring process " + process); } - - int result = Integer.MIN_VALUE; - try { - if (logger.isDebugEnabled()) { - logger.debug("Monitoring process " + process); - } - result = process.waitFor(); - if (logger.isInfoEnabled()) { - logger.info("tail process terminated with value " + result); - } + result = process.waitFor(); + if (logger.isInfoEnabled()) { + logger.info("tail process terminated with value " + result); } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - logger.error("Interrupted - stopping adapter", e); - stop(); - } - finally { - destroyProcess(); - } - if (isRunning()) { - if (logger.isInfoEnabled()) { - logger.info("Restarting tail process in " + getMissingFileDelay() + " milliseconds"); - } - getRequiredTaskScheduler().schedule(new Runnable() { - - @Override - public void run() { - runExec(); - } - }, new Date(System.currentTimeMillis() + getMissingFileDelay())); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + logger.error("Interrupted - stopping adapter", e); + stop(); + } + finally { + destroyProcess(); + } + if (isRunning()) { + if (logger.isInfoEnabled()) { + logger.info("Restarting tail process in " + getMissingFileDelay() + " milliseconds"); } + getRequiredTaskScheduler().schedule((Runnable) () -> runExec(), + new Date(System.currentTimeMillis() + getMissingFileDelay())); } }); } @@ -204,35 +189,31 @@ public class OSDelegatingFileTailingMessageProducer extends FileTailingMessagePr return; } final BufferedReader errorReader = new BufferedReader(new InputStreamReader(process.getErrorStream())); - this.getTaskExecutor().execute(new Runnable() { - - @Override - public void run() { - String statusMessage; + this.getTaskExecutor().execute(() -> { + String statusMessage; + if (logger.isDebugEnabled()) { + logger.debug("Reading stderr"); + } + try { + while ((statusMessage = errorReader.readLine()) != null) { + publish(statusMessage); + if (logger.isTraceEnabled()) { + logger.trace(statusMessage); + } + } + } + catch (IOException e1) { if (logger.isDebugEnabled()) { - logger.debug("Reading stderr"); + logger.debug("Exception on tail error reader", e1); } + } + finally { try { - while ((statusMessage = errorReader.readLine()) != null) { - publish(statusMessage); - if (logger.isTraceEnabled()) { - logger.trace(statusMessage); - } - } + errorReader.close(); } - catch (IOException e) { + catch (IOException e2) { if (logger.isDebugEnabled()) { - logger.debug("Exception on tail error reader", e); - } - } - finally { - try { - errorReader.close(); - } - catch (IOException e) { - if (logger.isDebugEnabled()) { - logger.debug("Exception while closing stderr", e); - } + logger.debug("Exception while closing stderr", e2); } } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java index c5c2311f16..16a3ac64fc 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java @@ -36,7 +36,6 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandler; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; @@ -87,15 +86,11 @@ public class FileInboundTransactionTests { public void testNoTx() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean crash = new AtomicBoolean(); - input.subscribe(new MessageHandler() { - - @Override - public void handleMessage(Message message) throws MessagingException { - if (crash.get()) { - throw new MessagingException("eek"); - } - latch.countDown(); + input.subscribe(message -> { + if (crash.get()) { + throw new MessagingException("eek"); } + latch.countDown(); }); pseudoTx.start(); File file = new File(tmpDir.getRoot(), "si-test1/foo"); @@ -123,15 +118,11 @@ public class FileInboundTransactionTests { public void testTx() throws Exception { final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean crash = new AtomicBoolean(); - txInput.subscribe(new MessageHandler() { - - @Override - public void handleMessage(Message message) throws MessagingException { - if (crash.get()) { - throw new MessagingException("eek"); - } - latch.countDown(); + txInput.subscribe(message -> { + if (crash.get()) { + throw new MessagingException("eek"); } + latch.countDown(); }); realTx.start(); File file = new File(tmpDir.getRoot(), "si-test2/baz"); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java index b0bdf28bf0..485db76dfc 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java @@ -142,28 +142,18 @@ public class FileReadingMessageSourceIntegrationTests { @Repeat(5) public void concurrentProcessing() throws Exception { CountDownLatch go = new CountDownLatch(1); - Runnable successfulConsumer = new Runnable() { - - @Override - public void run() { - Message received = pollableFileSource.receive(); - while (received == null) { - Thread.yield(); - received = pollableFileSource.receive(); - } + Runnable successfulConsumer = () -> { + Message received = pollableFileSource.receive(); + while (received == null) { + Thread.yield(); + received = pollableFileSource.receive(); } - }; - Runnable failingConsumer = new Runnable() { - - @Override - public void run() { - Message received = pollableFileSource.receive(); - if (received != null) { - pollableFileSource.onFailure(received); - } + Runnable failingConsumer = () -> { + Message received = pollableFileSource.receive(); + if (received != null) { + pollableFileSource.onFailure(received); } - }; CountDownLatch successfulDone = doConcurrently(3, successfulConsumer, go); CountDownLatch failingDone = doConcurrently(10, failingConsumer, go); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java index 04d87a3c01..2dc97258db 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java @@ -50,8 +50,6 @@ import org.junit.rules.TemporaryFolder; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.file.FileWritingMessageHandler.FlushPredicate; -import org.springframework.integration.file.FileWritingMessageHandler.MessageFlushPredicate; import org.springframework.integration.file.support.FileExistsMode; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; @@ -381,13 +379,7 @@ public class FileWritingMessageHandlerTests { final String anyFilename = "fooBar.test"; QueueChannel output = new QueueChannel(); handler.setOutputChannel(output); - handler.setFileNameGenerator(new FileNameGenerator() { - - @Override - public String generateFileName(Message message) { - return anyFilename; - } - }); + handler.setFileNameGenerator(message -> anyFilename); Message message = MessageBuilder.withPayload("test").build(); handler.handleMessage(message); File result = (File) output.receive(0).getPayload(); @@ -449,13 +441,7 @@ public class FileWritingMessageHandlerTests { File tempFolder = this.temp.newFolder(); FileWritingMessageHandler handler = new FileWritingMessageHandler(tempFolder); handler.setFileExistsMode(FileExistsMode.APPEND_NO_FLUSH); - handler.setFileNameGenerator(new FileNameGenerator() { - - @Override - public String generateFileName(Message message) { - return "foo.txt"; - } - }); + handler.setFileNameGenerator(message -> "foo.txt"); ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); taskScheduler.afterPropertiesSet(); handler.setTaskScheduler(taskScheduler); @@ -487,14 +473,9 @@ public class FileWritingMessageHandlerTests { handler.setFlushInterval(30000); final AtomicBoolean called = new AtomicBoolean(); - handler.setFlushPredicate(new MessageFlushPredicate() { - - @Override - public boolean shouldFlush(String fileAbsolutePath, long lastWrite, Message triggerMessage) { - called.set(true); - return true; - } - + handler.setFlushPredicate((fileAbsolutePath, lastWrite, triggerMessage) -> { + called.set(true); + return true; }); handler.handleMessage(new GenericMessage(new ByteArrayInputStream("box".getBytes()))); handler.trigger(new GenericMessage("foo")); @@ -503,14 +484,9 @@ public class FileWritingMessageHandlerTests { handler.handleMessage(new GenericMessage(new ByteArrayInputStream("bux".getBytes()))); called.set(false); - handler.flushIfNeeded(new FlushPredicate() { - - @Override - public boolean shouldFlush(String fileAbsolutePath, long lastWrite) { - called.set(true); - return true; - } - + handler.flushIfNeeded((fileAbsolutePath, lastWrite) -> { + called.set(true); + return true; }); assertThat(file.length(), equalTo(24L)); assertTrue(called.get()); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java index 1f7a618fb2..daab0a8104 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java @@ -20,7 +20,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; import java.io.File; -import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -29,8 +28,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.Test; import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.serializer.StringRedisSerializer; @@ -81,21 +78,17 @@ public class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisA store = Mockito.spy(store); - Mockito.doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - if (suspend.get()) { - latch2.countDown(); - try { - latch1.await(10, TimeUnit.SECONDS); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } + Mockito.doAnswer(invocation -> { + if (suspend.get()) { + latch2.countDown(); + try { + latch1.await(10, TimeUnit.SECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); } - return invocation.callRealMethod(); } + return invocation.callRealMethod(); }).when(store).replace(Mockito.anyString(), Mockito.anyString(), Mockito.anyString()); final FileSystemPersistentAcceptOnceFileListFilter filter = @@ -114,13 +107,8 @@ public class PersistentAcceptOnceFileListFilterExternalStoreTests extends RedisA suspend.set(true); file.setLastModified(file.lastModified() + 5000L); - Future result = Executors.newSingleThreadExecutor().submit(new Callable() { - - @Override - public Integer call() throws Exception { - return filter.filterFiles(new File[] {file}).size(); - } - }); + Future result = Executors.newSingleThreadExecutor() + .submit(() -> filter.filterFiles(new File[] { file }).size()); assertTrue(latch2.await(10, TimeUnit.SECONDS)); store.put("foo:" + file.getAbsolutePath(), "43"); latch1.countDown(); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java index 199cd6665e..9e35585985 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java @@ -54,8 +54,6 @@ import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; import org.mockito.ArgumentCaptor; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.beans.factory.BeanFactory; import org.springframework.expression.common.LiteralExpression; @@ -260,14 +258,10 @@ public class RemoteFileOutboundGatewayTests { gw.afterPropertiesSet(); Session session = mock(Session.class); final AtomicReference args = new AtomicReference(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - Object[] arguments = invocation.getArguments(); - args.set((String) arguments[0] + (String) arguments[1]); - return null; - } + doAnswer(invocation -> { + Object[] arguments = invocation.getArguments(); + args.set((String) arguments[0] + (String) arguments[1]); + return null; }).when(session).rename(anyString(), anyString()); when(sessionFactory.getSession()).thenReturn(session); Message requestMessage = MessageBuilder.withPayload("foo") @@ -287,14 +281,10 @@ public class RemoteFileOutboundGatewayTests { gw.afterPropertiesSet(); Session session = mock(Session.class); final AtomicReference args = new AtomicReference(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - Object[] arguments = invocation.getArguments(); - args.set((String) arguments[0] + (String) arguments[1]); - return null; - } + doAnswer(invocation -> { + Object[] arguments = invocation.getArguments(); + args.set((String) arguments[0] + (String) arguments[1]); + return null; }).when(session).rename(anyString(), anyString()); when(sessionFactory.getSession()).thenReturn(session); Message out = (Message) gw.handleRequestMessage(new GenericMessage("foo")); @@ -312,23 +302,15 @@ public class RemoteFileOutboundGatewayTests { gw.afterPropertiesSet(); Session session = mock(Session.class); final AtomicReference args = new AtomicReference(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - Object[] arguments = invocation.getArguments(); - args.set((String) arguments[0] + (String) arguments[1]); - return null; - } + doAnswer(invocation -> { + Object[] arguments = invocation.getArguments(); + args.set((String) arguments[0] + (String) arguments[1]); + return null; }).when(session).rename(anyString(), anyString()); final List madeDirs = new ArrayList(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - madeDirs.add((String) invocation.getArguments()[0]); - return null; - } + doAnswer(invocation -> { + madeDirs.add((String) invocation.getArguments()[0]); + return null; }).when(session).mkdir(anyString()); when(sessionFactory.getSession()).thenReturn(session); Message requestMessage = MessageBuilder.withPayload("foo") @@ -926,13 +908,9 @@ public class RemoteFileOutboundGatewayTests { gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); final AtomicReference written = new AtomicReference(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - written.set((String) invocation.getArguments()[1]); - return null; - } + doAnswer(invocation -> { + written.set((String) invocation.getArguments()[1]); + return null; }).when(session).write(any(InputStream.class), anyString()); tempFolder.newFile("baz.txt"); tempFolder.newFile("qux.txt"); @@ -964,13 +942,9 @@ public class RemoteFileOutboundGatewayTests { gw.afterPropertiesSet(); when(sessionFactory.getSession()).thenReturn(session); final AtomicReference written = new AtomicReference(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - written.set((String) invocation.getArguments()[1]); - return null; - } + doAnswer(invocation -> { + written.set((String) invocation.getArguments()[1]); + return null; }).when(session).write(any(InputStream.class), anyString()); tempFolder.newFile("baz.txt"); tempFolder.newFile("qux.txt"); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java index ccc10aadb6..60eb4767b0 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java @@ -35,8 +35,6 @@ import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.beans.factory.BeanFactory; import org.springframework.expression.ExpressionParser; @@ -45,11 +43,11 @@ import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.file.remote.session.CachingSessionFactory; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.util.SimplePool; import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; /** * @author Oleg Zhurakousky @@ -65,12 +63,10 @@ public class FileTransferringMessageHandlerTests { Session session = mock(Session.class); when(sf.getSession()).thenReturn(session); - doAnswer(new Answer() { - public Object answer(InvocationOnMock invocation) throws Throwable { - String path = (String) invocation.getArguments()[1]; - assertFalse(path.startsWith("/")); - return null; - } + doAnswer(invocation -> { + String path = (String) invocation.getArguments()[1]; + assertFalse(path.startsWith("/")); + return null; }).when(session).rename(Mockito.anyString(), Mockito.anyString()); ExpressionParser parser = new SpelExpressionParser(); FileTransferringMessageHandler handler = new FileTransferringMessageHandler(sf); @@ -90,12 +86,10 @@ public class FileTransferringMessageHandlerTests { final AtomicReference temporaryPath = new AtomicReference(); final AtomicReference finalPath = new AtomicReference(); when(sf.getSession()).thenReturn(session); - doAnswer(new Answer() { - public Object answer(InvocationOnMock invocation) throws Throwable { - temporaryPath.set((String) invocation.getArguments()[0]); - finalPath.set((String) invocation.getArguments()[1]); - return null; - } + doAnswer(invocation -> { + temporaryPath.set((String) invocation.getArguments()[0]); + finalPath.set((String) invocation.getArguments()[1]); + return null; }).when(session).rename(Mockito.anyString(), Mockito.anyString()); FileTransferringMessageHandler handler = new FileTransferringMessageHandler(sf); handler.setRemoteDirectoryExpression(new LiteralExpression("foo")); @@ -115,12 +109,10 @@ public class FileTransferringMessageHandlerTests { Session session = mock(Session.class); when(sf.getSession()).thenReturn(session); - doAnswer(new Answer() { - public Object answer(InvocationOnMock invocation) throws Throwable { - String path = (String) invocation.getArguments()[1]; - assertFalse(path.startsWith("/")); - return null; - } + doAnswer(invocation -> { + String path = (String) invocation.getArguments()[1]; + assertFalse(path.startsWith("/")); + return null; }).when(session).rename(Mockito.anyString(), Mockito.anyString()); ExpressionParser parser = new SpelExpressionParser(); FileTransferringMessageHandler handler = new FileTransferringMessageHandler(sf); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java index 9662f4b5c0..2168828dd6 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java @@ -92,12 +92,8 @@ public class CachingSessionFactoryTests { template.setBeanFactory(mock(BeanFactory.class)); template.afterPropertiesSet(); try { - template.get(new GenericMessage("foo"), new InputStreamCallback() { - - @Override - public void doWithInputStream(InputStream stream) throws IOException { - throw new RuntimeException("bar"); - } + template.get(new GenericMessage("foo"), (InputStreamCallback) stream -> { + throw new RuntimeException("bar"); }); fail("Expected exception"); } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java index 07a88a5adb..b0aa31468f 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java @@ -36,8 +36,6 @@ import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; -import org.springframework.context.ApplicationEvent; -import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.file.tail.FileTailingMessageProducerSupport.FileTailingEvent; import org.springframework.messaging.Message; @@ -128,20 +126,10 @@ public class FileTailingMessageProducerTests { throws Exception { this.adapter = adapter; final List events = new ArrayList(); - adapter.setApplicationEventPublisher(new ApplicationEventPublisher() { - - @Override - public void publishEvent(ApplicationEvent event) { - FileTailingEvent tailEvent = (FileTailingEvent) event; - logger.warn(event); - events.add(tailEvent); - } - - @Override - public void publishEvent(Object event) { - - } - + adapter.setApplicationEventPublisher(event -> { + FileTailingEvent tailEvent = (FileTailingEvent) event; + logger.warn(event); + events.add(tailEvent); }); adapter.setFile(new File(testDir, "foo")); QueueChannel outputChannel = new QueueChannel(); diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java index 86ee05c0ff..e5bed1607d 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java @@ -22,7 +22,6 @@ import java.io.File; import java.io.FileOutputStream; import java.io.IOException; import java.io.OutputStream; -import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.Executors; @@ -86,26 +85,18 @@ public class TailRule extends TestWatcher { fos.close(); final AtomicReference c = new AtomicReference(); final CountDownLatch latch = new CountDownLatch(1); - Future future = Executors.newSingleThreadExecutor().submit(new Callable() { - - @Override - public Process call() throws Exception { - final Process process = Runtime.getRuntime().exec(commandToTest + " " + file.getAbsolutePath()); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - try { - c.set(process.getInputStream().read()); - latch.countDown(); - } - catch (IOException e) { - logger.error("Error reading test stream", e); - } - } - }); - return process; - } + Future future = Executors.newSingleThreadExecutor().submit(() -> { + final Process process = Runtime.getRuntime().exec(commandToTest + " " + file.getAbsolutePath()); + Executors.newSingleThreadExecutor().execute(() -> { + try { + c.set(process.getInputStream().read()); + latch.countDown(); + } + catch (IOException e) { + logger.error("Error reading test stream", e); + } + }); + return process; }); try { Process process = future.get(10, TimeUnit.SECONDS); diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplate.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplate.java index 16f7af235b..44e7e0f278 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplate.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplate.java @@ -80,33 +80,28 @@ public class FtpRemoteFileTemplate extends RemoteFileTemplate { */ @Override public boolean exists(final String path) { - return doExecuteWithClient(new ClientCallback() { + return doExecuteWithClient(client -> { + try { + switch (FtpRemoteFileTemplate.this.existsMode) { - @Override - public Boolean doWithClient(FTPClient client) { - try { - switch (FtpRemoteFileTemplate.this.existsMode) { + case STAT: + return client.getStatus(path) != null; - case STAT: - return client.getStatus(path) != null; + case NLST: + String[] names = client.listNames(path); + return !ObjectUtils.isEmpty(names); - case NLST: - String[] names = client.listNames(path); - return !ObjectUtils.isEmpty(names); + case NLST_AND_DIRS: + return getSession().exists(path); - case NLST_AND_DIRS: - return getSession().exists(path); - - default: - throw new IllegalStateException("Unsupported 'existsMode': " + - FtpRemoteFileTemplate.this.existsMode); - } - } - catch (IOException e) { - throw new MessagingException("Failed to check the remote path for " + path, e); + default: + throw new IllegalStateException("Unsupported 'existsMode': " + + FtpRemoteFileTemplate.this.existsMode); } } - + catch (IOException e) { + throw new MessagingException("Failed to check the remote path for " + path, e); + } }); } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java index 834212f564..8db146114f 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java @@ -53,7 +53,6 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.ReflectionUtils; -import org.springframework.util.ReflectionUtils.MethodCallback; /** * @author Oleg Zhurakousky @@ -111,16 +110,10 @@ public class FtpInboundChannelAdapterParserTests { FileListFilter acceptAllFilter = context.getBean("acceptAllFilter", FileListFilter.class); assertTrue(TestUtils.getPropertyValue(inbound, "fileSource.scanner.filter.fileFilters", Collection.class).contains(acceptAllFilter)); final AtomicReference genMethod = new AtomicReference(); - ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class, new MethodCallback() { - - @Override - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - if ("generateLocalFileName".equals(method.getName())) { - method.setAccessible(true); - genMethod.set(method); - } - } - }); + ReflectionUtils.doWithMethods(AbstractInboundFileSynchronizer.class, method -> { + method.setAccessible(true); + genMethod.set(method); + }, method -> "generateLocalFileName".equals(method.getName())); assertEquals("FOO.afoo", genMethod.get().invoke(fisync, "foo")); assertEquals(42, inbound.getMaxFetchSize()); } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java index 17f9baf9ab..3087cc5a4e 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java @@ -136,16 +136,10 @@ public class FtpOutboundGatewayParserTests { //INT-3129 assertNotNull(TestUtils.getPropertyValue(gateway, "localFilenameGeneratorExpression")); final AtomicReference genMethod = new AtomicReference(); - ReflectionUtils.doWithMethods(FtpOutboundGateway.class, new ReflectionUtils.MethodCallback() { - - @Override - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - if ("generateLocalFileName".equals(method.getName())) { - method.setAccessible(true); - genMethod.set(method); - } - } - }); + ReflectionUtils.doWithMethods(FtpOutboundGateway.class, method -> { + method.setAccessible(true); + genMethod.set(method); + }, method -> "generateLocalFileName".equals(method.getName())); assertEquals("FOO.afoo", genMethod.get().invoke(gateway, new GenericMessage(""), "foo")); assertThat(TestUtils.getPropertyValue(gateway, "mputFilter"), Matchers.instanceOf(SimplePatternFileListFilter.class)); } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java index 0381fbe33f..cf8468fca8 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java @@ -52,7 +52,6 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.expression.common.LiteralExpression; -import org.springframework.integration.file.FileNameGenerator; import org.springframework.integration.file.remote.FileInfo; import org.springframework.integration.file.remote.RemoteFileTemplate; import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler; @@ -95,12 +94,7 @@ public class FtpOutboundTests { assertFalse(file.exists()); FileTransferringMessageHandler handler = new FileTransferringMessageHandler(sessionFactory); handler.setRemoteDirectoryExpression(new LiteralExpression("remote-target-dir")); - handler.setFileNameGenerator(new FileNameGenerator() { - @Override - public String generateFileName(Message message) { - return "handlerContent.test"; - } - }); + handler.setFileNameGenerator(message -> "handlerContent.test"); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); handler.handleMessage(new GenericMessage("String data")); @@ -119,12 +113,7 @@ public class FtpOutboundTests { assertFalse(file.exists()); FileTransferringMessageHandler handler = new FileTransferringMessageHandler(sessionFactory); handler.setRemoteDirectoryExpression(new LiteralExpression("remote-target-dir")); - handler.setFileNameGenerator(new FileNameGenerator() { - @Override - public String generateFileName(Message message) { - return "handlerContent.test"; - } - }); + handler.setFileNameGenerator(message -> "handlerContent.test"); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); handler.handleMessage(new GenericMessage("byte[] data".getBytes())); @@ -141,12 +130,7 @@ public class FtpOutboundTests { FileTransferringMessageHandler handler = new FileTransferringMessageHandler(sessionFactory); handler.setRemoteDirectoryExpression(new LiteralExpression(targetDir.getName())); - handler.setFileNameGenerator(new FileNameGenerator() { - @Override - public String generateFileName(Message message) { - return ((File) message.getPayload()).getName() + ".test"; - } - }); + handler.setFileNameGenerator(message -> ((File) message.getPayload()).getName() + ".test"); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); @@ -167,12 +151,7 @@ public class FtpOutboundTests { FileTransferringMessageHandler handler = new FileTransferringMessageHandler(sessionFactory); handler.setRemoteDirectoryExpression(new LiteralExpression(targetDir.getName())); - handler.setFileNameGenerator(new FileNameGenerator() { - @Override - public String generateFileName(Message message) { - return ((File) message.getPayload()).getName() + ".test"; - } - }); + handler.setFileNameGenerator(message -> ((File) message.getPayload()).getName() + ".test"); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); @@ -181,14 +160,10 @@ public class FtpOutboundTests { Log logger = spy(TestUtils.getPropertyValue(handler, "remoteFileTemplate.logger", Log.class)); when(logger.isWarnEnabled()).thenReturn(true); final AtomicReference logged = new AtomicReference(); - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - logged.set((String) invocation.getArguments()[0]); - invocation.callRealMethod(); - return null; - } + doAnswer(invocation -> { + logged.set((String) invocation.getArguments()[0]); + invocation.callRealMethod(); + return null; }).when(logger).warn(Mockito.anyString()); RemoteFileTemplate template = TestUtils.getPropertyValue(handler, "remoteFileTemplate", RemoteFileTemplate.class); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java index 9ca71330a9..b541ff8e96 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java @@ -42,7 +42,6 @@ import java.io.IOException; import java.io.InputStream; import java.util.Arrays; import java.util.Calendar; -import java.util.Comparator; import java.util.List; import java.util.Set; import java.util.concurrent.BlockingQueue; @@ -55,8 +54,6 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mockito; -import org.mockito.invocation.InvocationOnMock; -import org.mockito.stubbing.Answer; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -70,7 +67,6 @@ import org.springframework.integration.file.filters.FileListFilter; import org.springframework.integration.file.remote.InputStreamCallback; import org.springframework.integration.file.remote.MessageSessionCallback; import org.springframework.integration.file.remote.RemoteFileTemplate; -import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.ftp.FtpTestSupport; @@ -304,23 +300,13 @@ public class FtpServerOutboundTests extends FtpTestSupport { template.setBeanFactory(mock(BeanFactory.class)); template.afterPropertiesSet(); final ByteArrayOutputStream baos1 = new ByteArrayOutputStream(); - assertTrue(template.get(new GenericMessage("ftpSource/ ftpSource1.txt"), new InputStreamCallback() { - - @Override - public void doWithInputStream(InputStream stream) throws IOException { - FileCopyUtils.copy(stream, baos1); - } - })); + assertTrue(template.get(new GenericMessage("ftpSource/ ftpSource1.txt"), + (InputStreamCallback) stream -> FileCopyUtils.copy(stream, baos1))); assertEquals("source1", new String(baos1.toByteArray())); final ByteArrayOutputStream baos2 = new ByteArrayOutputStream(); - assertTrue(template.get(new GenericMessage("ftpSource/ftpSource2.txt"), new InputStreamCallback() { - - @Override - public void doWithInputStream(InputStream stream) throws IOException { - FileCopyUtils.copy(stream, baos2); - } - })); + assertTrue(template.get(new GenericMessage("ftpSource/ftpSource2.txt"), + (InputStreamCallback) stream -> FileCopyUtils.copy(stream, baos2))); assertEquals("source2", new String(baos2.toByteArray())); } @@ -438,18 +424,14 @@ public class FtpServerOutboundTests extends FtpTestSupport { @Test public void testMgetPartial() throws Exception { Session session = spyOnSession(); - doAnswer(new Answer() { - - @Override - public FTPFile[] answer(InvocationOnMock invocation) throws Throwable { - FTPFile[] files = (FTPFile[]) invocation.callRealMethod(); - // add an extra file where the get will fail - files = Arrays.copyOf(files, files.length + 1); - FTPFile bogusFile = new FTPFile(); - bogusFile.setName("bogus.txt"); - files[files.length - 1] = bogusFile; - return files; - } + doAnswer(invocation -> { + FTPFile[] files = (FTPFile[]) invocation.callRealMethod(); + // add an extra file where the get will fail + files = Arrays.copyOf(files, files.length + 1); + FTPFile bogusFile = new FTPFile(); + bogusFile.setName("bogus.txt"); + files[files.length - 1] = bogusFile; + return files; }).when(session).list("ftpSource/subFtpSource/*"); String dir = "ftpSource/subFtpSource/"; try { @@ -468,19 +450,15 @@ public class FtpServerOutboundTests extends FtpTestSupport { @Test public void testMgetRecursivePartial() throws Exception { Session session = spyOnSession(); - doAnswer(new Answer() { - - @Override - public FTPFile[] answer(InvocationOnMock invocation) throws Throwable { - FTPFile[] files = (FTPFile[]) invocation.callRealMethod(); - // add an extra file where the get will fail - files = Arrays.copyOf(files, files.length + 1); - FTPFile bogusFile = new FTPFile(); - bogusFile.setName("bogus.txt"); - bogusFile.setTimestamp(Calendar.getInstance()); - files[files.length - 1] = bogusFile; - return files; - } + doAnswer(invocation -> { + FTPFile[] files = (FTPFile[]) invocation.callRealMethod(); + // add an extra file where the get will fail + files = Arrays.copyOf(files, files.length + 1); + FTPFile bogusFile = new FTPFile(); + bogusFile.setName("bogus.txt"); + bogusFile.setTimestamp(Calendar.getInstance()); + files[files.length - 1] = bogusFile; + return files; }).when(session).list("ftpSource/subFtpSource/"); String dir = "ftpSource/"; try { @@ -498,13 +476,8 @@ public class FtpServerOutboundTests extends FtpTestSupport { @Test public void testMputPartial() throws Exception { Session session = spyOnSession(); - doAnswer(new Answer() { - - @Override - public Void answer(InvocationOnMock invocation) throws Throwable { - throw new IOException("Failed to send localSource2"); - } - + doAnswer(invocation -> { + throw new IOException("Failed to send localSource2"); }).when(session).write(Mockito.any(InputStream.class), Mockito.contains("localSource2")); try { this.inboundMPut.send(new GenericMessage(getSourceLocalDirectory())); @@ -528,13 +501,8 @@ public class FtpServerOutboundTests extends FtpTestSupport { FileOutputStream writer = new FileOutputStream(extra); writer.write("foo".getBytes()); writer.close(); - doAnswer(new Answer() { - - @Override - public Void answer(InvocationOnMock invocation) throws Throwable { - throw new IOException("Failed to send subLocalSource2"); - } - + doAnswer(invocation -> { + throw new IOException("Failed to send subLocalSource2"); }).when(session).write(Mockito.any(InputStream.class), Mockito.contains("subLocalSource2")); try { this.inboundMPutRecursive.send(new GenericMessage(getSourceLocalDirectory())); @@ -569,13 +537,7 @@ public class FtpServerOutboundTests extends FtpTestSupport { } private void assertLength6(FtpRemoteFileTemplate template) { - FTPFile[] files = template.execute(new SessionCallback() { - - @Override - public FTPFile[] doInSession(Session session) throws IOException { - return session.list("ftpTarget/appending.txt"); - } - }); + FTPFile[] files = template.execute(session -> session.list("ftpTarget/appending.txt")); assertEquals(1, files.length); assertEquals(6, files[0].getSize()); } @@ -638,21 +600,16 @@ public class FtpServerOutboundTests extends FtpTestSupport { @Override public List filterFiles(File[] files) { File[] sorted = Arrays.copyOf(files, files.length); - Arrays.sort(sorted, new Comparator() { - - @Override - public int compare(File o1, File o2) { - if (o1.isDirectory() && !o2.isDirectory()) { - return 1; - } - else if (!o1.isDirectory() && o2.isDirectory()) { - return -1; - } - else { - return o1.getName().compareTo(o2.getName()); - } + Arrays.sort(sorted, (o1, o2) -> { + if (o1.isDirectory() && !o2.isDirectory()) { + return 1; + } + else if (!o1.isDirectory() && o2.isDirectory()) { + return -1; + } + else { + return o1.getName().compareTo(o2.getName()); } - }); return Arrays.asList(sorted); } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java index e2415ade42..76c50bb6bc 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java @@ -39,9 +39,7 @@ import org.springframework.context.annotation.Configuration; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.remote.ClientCallbackWithoutResult; -import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.SessionCallbackWithoutResult; -import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.ftp.FtpTestSupport; import org.springframework.messaging.MessagingException; @@ -69,41 +67,28 @@ public class FtpRemoteFileTemplateTests extends FtpTestSupport { template.setFileNameGenerator(fileNameGenerator); template.setRemoteDirectoryExpression(new LiteralExpression("foo/")); template.setUseTemporaryFileName(false); - template.execute(new SessionCallback() { - - @Override - public Boolean doInSession(Session session) throws IOException { - session.mkdir("foo/"); - return session.mkdir("foo/bar/"); - } - + template.execute(session -> { + session.mkdir("foo/"); + return session.mkdir("foo/bar/"); }); template.append(new GenericMessage("foo")); template.append(new GenericMessage("bar")); assertTrue(template.exists("foo/foobar.txt")); - template.executeWithClient(new ClientCallbackWithoutResult() { - - @Override - public void doWithClientWithoutResult(FTPClient client) { - try { - FTPFile[] files = client.listFiles("foo/foobar.txt"); - assertEquals(6, files[0].getSize()); - } - catch (IOException e) { - throw new RuntimeException(e); - } + template.executeWithClient((ClientCallbackWithoutResult) client -> { + try { + FTPFile[] files = client.listFiles("foo/foobar.txt"); + assertEquals(6, files[0].getSize()); + } + catch (IOException e) { + throw new RuntimeException(e); } }); - template.execute(new SessionCallbackWithoutResult() { - - @Override - public void doInSessionWithoutResult(Session session) throws IOException { - assertTrue(session.remove("foo/foobar.txt")); - assertTrue(session.rmdir("foo/bar/")); - FTPFile[] files = session.list("foo/"); - assertEquals(0, files.length); - assertTrue(session.rmdir("foo/")); - } + template.execute((SessionCallbackWithoutResult) session -> { + assertTrue(session.remove("foo/foobar.txt")); + assertTrue(session.rmdir("foo/bar/")); + FTPFile[] files = session.list("foo/"); + assertEquals(0, files.length); + assertTrue(session.rmdir("foo/")); }); assertFalse(template.getSession().exists("foo")); } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/SessionFactoryTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/SessionFactoryTests.java index 98eb86e26a..43453b8854 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/SessionFactoryTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/SessionFactoryTests.java @@ -191,18 +191,15 @@ public class SessionFactoryTests { final Random random = new Random(); final AtomicInteger failures = new AtomicInteger(); for (int i = 0; i < 30; i++) { - executor.execute(new Runnable() { - @Override - public void run() { - try { - Session session = factory.getSession(); - Thread.sleep(random.nextInt(5000)); - session.close(); - } - catch (Exception e) { - e.printStackTrace(); - failures.incrementAndGet(); - } + executor.execute(() -> { + try { + Session session = factory.getSession(); + Thread.sleep(random.nextInt(5000)); + session.close(); + } + catch (Exception e) { + e.printStackTrace(); + failures.incrementAndGet(); } }); } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/ForkUtil.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/ForkUtil.java index d5efb1f17d..82a59802c2 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/ForkUtil.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/ForkUtil.java @@ -103,21 +103,17 @@ public class ForkUtil { private static Thread copyStdXxx(final BufferedReader br, final AtomicBoolean run, final PrintStream out) { - Thread reader = new Thread(new Runnable() { - - @Override - public void run() { - try { - String line = null; - do { - while ((line = br.readLine()) != null) { - out.println("[FORK] " + line); - } - } while (run.get()); - } - catch (Exception ex) { - // ignore and exit - } + Thread reader = new Thread(() -> { + try { + String line = null; + do { + while ((line = br.readLine()) != null) { + out.println("[FORK] " + line); + } + } while (run.get()); + } + catch (Exception ex) { + // ignore and exit } }); return reader; diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java index bcff80678b..0993e3023e 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java @@ -90,11 +90,8 @@ public class GemfireInboundChannelAdapterTests { @Test public void testErrorChannel() { - channel3.subscribe(new MessageHandler() { - @Override - public void handleMessage(Message message) throws MessagingException { - throw new MessagingException("got an error"); - } + channel3.subscribe(message -> { + throw new MessagingException("got an error"); }); ErrorHandler errorHandler = new ErrorHandler(); errorChannel.subscribe(errorHandler); diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java index 81e21f92ac..4caba29de3 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java @@ -51,6 +51,7 @@ import org.springframework.messaging.support.GenericMessage; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.Region; import com.gemstone.gemfire.cache.Scope; + import junit.framework.AssertionFailedError; /** @@ -291,25 +292,19 @@ public class GemfireGroupStoreTests { for (int i = 0; i < 100; i++) { executor = Executors.newCachedThreadPool(); - executor.execute(new Runnable() { - @Override - public void run() { - MessageGroup group = store1.addMessageToGroup(1, message); - if (group.getMessages().size() != 1) { - failures.add("ADD"); - throw new AssertionFailedError("Failed on ADD"); - } + executor.execute(() -> { + MessageGroup group = store1.addMessageToGroup(1, message); + if (group.getMessages().size() != 1) { + failures.add("ADD"); + throw new AssertionFailedError("Failed on ADD"); } }); - executor.execute(new Runnable() { - @Override - public void run() { - store2.removeMessagesFromGroup(1, message); - MessageGroup group = store2.getMessageGroup(1); - if (group.getMessages().size() != 0) { - failures.add("REMOVE"); - throw new AssertionFailedError("Failed on Remove"); - } + executor.execute(() -> { + store2.removeMessagesFromGroup(1, message); + MessageGroup group = store2.getMessageGroup(1); + if (group.getMessages().size() != 0) { + failures.add("REMOVE"); + throw new AssertionFailedError("Failed on Remove"); } }); diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests.java index db0013375c..2c66642361 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests.java @@ -103,17 +103,13 @@ public class AggregatorWithGemfireLocksTests { public void testDistributedAggregator() throws Exception { this.releaseStrategy.reset(1); Executors.newSingleThreadExecutor().execute(asyncSend("foo", 1, 1)); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - try { - in2.send(new GenericMessage("bar", stubHeaders(2, 2, 1))); - } - catch (Exception e) { - e.printStackTrace(); - exception = e; - } + Executors.newSingleThreadExecutor().execute(() -> { + try { + in2.send(new GenericMessage("bar", stubHeaders(2, 2, 1))); + } + catch (Exception e) { + e.printStackTrace(); + exception = e; } }); assertTrue(this.releaseStrategy.latch2.await(10, TimeUnit.SECONDS)); @@ -124,17 +120,13 @@ public class AggregatorWithGemfireLocksTests { } private Runnable asyncSend(final String payload, final int sequence, final int correlation) { - return new Runnable() { - - @Override - public void run() { - try { - in.send(new GenericMessage(payload, stubHeaders(sequence, 2, correlation))); - } - catch (Exception e) { - e.printStackTrace(); - exception = e; - } + return () -> { + try { + in.send(new GenericMessage(payload, stubHeaders(sequence, 2, correlation))); + } + catch (Exception e) { + e.printStackTrace(); + exception = e; } }; } diff --git a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java index 9109c5cadb..7268a0c020 100644 --- a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java +++ b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java @@ -56,6 +56,7 @@ import groovy.transform.CompileStatic; * @author Oleg Zhurakousky * @author Stefan Reuter * @author Artem Bilan + * @author Gary Russell * @since 2.0 */ public class GroovyScriptExecutingMessageProcessor extends AbstractScriptExecutingMessageProcessor @@ -69,13 +70,7 @@ public class GroovyScriptExecutingMessageProcessor extends AbstractScriptExecuti private volatile ScriptSource scriptSource; private volatile GroovyClassLoader groovyClassLoader = AccessController.doPrivileged( - new PrivilegedAction() { - - public GroovyClassLoader run() { - return new GroovyClassLoader(ClassUtils.getDefaultClassLoader()); - } - - }); + (PrivilegedAction) () -> new GroovyClassLoader(ClassUtils.getDefaultClassLoader())); private volatile Class scriptClass; diff --git a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessorTests.java b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessorTests.java index 515870fc98..9af3324e5e 100644 --- a/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessorTests.java +++ b/spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessorTests.java @@ -83,6 +83,7 @@ public class GroovyScriptExecutingMessageProcessorTests { ScriptSource scriptSource = new ResourceScriptSource(resource); Object result = null; class CustomScriptVariableGenerator implements ScriptVariableGenerator { + @Override public Map generateScriptVariables(Message message) { Map variables = new HashMap(); variables.put("date", System.nanoTime()); @@ -197,24 +198,16 @@ public class GroovyScriptExecutingMessageProcessorTests { ScriptSource scriptSource = new StaticScriptSource(script, Script.class.getName()); final MessageProcessor processor = - new GroovyScriptExecutingMessageProcessor(scriptSource, new ScriptVariableGenerator() { - @Override - public Map generateScriptVariables(Message message) { - Map variables = new HashMap(2); - variables.put("var1", var1); - variables.put("var2", var2); - return variables; - } + new GroovyScriptExecutingMessageProcessor(scriptSource, message1 -> { + Map variables = new HashMap(2); + variables.put("var1", var1); + variables.put("var2", var2); + return variables; }); ExecutorService executor = Executors.newFixedThreadPool(10); for (int i = 0; i < 10; i++) { - executor.execute(new Runnable() { - @Override - public void run() { - processor.processMessage(message); - } - }); + executor.execute(() -> processor.processMessage(message)); } executor.shutdown(); assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); @@ -246,6 +239,7 @@ public class GroovyScriptExecutingMessageProcessorTests { this.script = script; } + @Override public String getDescription() { return "test"; } @@ -255,6 +249,7 @@ public class GroovyScriptExecutingMessageProcessorTests { return this.filename; } + @Override public InputStream getInputStream() throws IOException { return new ByteArrayInputStream(script.getBytes("UTF-8")); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java index f14db94adc..e12af0f2e3 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java @@ -23,7 +23,6 @@ import org.apache.sshd.SshServer; import org.apache.sshd.common.NamedFactory; import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory; import org.apache.sshd.server.Command; -import org.apache.sshd.server.PasswordAuthenticator; import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; import org.apache.sshd.server.sftp.SftpSubsystem; import org.junit.AfterClass; @@ -59,14 +58,7 @@ public class SftpTestSupport extends RemoteFileTestSupport { @BeforeClass public static void createServer() throws Exception { server = SshServer.setUpDefaultServer(); - server.setPasswordAuthenticator(new PasswordAuthenticator() { - - @Override - public boolean authenticate(String username, String password, - org.apache.sshd.server.session.ServerSession session) { - return true; - } - }); + server.setPasswordAuthenticator((username, password, session) -> true); server.setPort(0); server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser")); server.setSubsystemFactories(Collections.>singletonList(new SftpSubsystem.Factory())); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java index 30399e47cd..99d3dece19 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java @@ -122,16 +122,10 @@ public class SftpOutboundGatewayParserTests { //INT-3129 assertNotNull(TestUtils.getPropertyValue(gateway, "localFilenameGeneratorExpression")); final AtomicReference genMethod = new AtomicReference(); - ReflectionUtils.doWithMethods(SftpOutboundGateway.class, new ReflectionUtils.MethodCallback() { - - @Override - public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { - if ("generateLocalFileName".equals(method.getName())) { - method.setAccessible(true); - genMethod.set(method); - } - } - }); + ReflectionUtils.doWithMethods(SftpOutboundGateway.class, method -> { + method.setAccessible(true); + genMethod.set(method); + }, method -> "generateLocalFileName".equals(method.getName())); assertEquals("FOO.afoo", genMethod.get().invoke(gateway, new GenericMessage(""), "foo")); assertThat(TestUtils.getPropertyValue(gateway, "mputFilter"), Matchers.instanceOf(SimplePatternFileListFilter.class)); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java index bf26106a67..45e489947d 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java @@ -41,8 +41,6 @@ import java.util.concurrent.atomic.AtomicInteger; 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.beans.factory.BeanFactory; @@ -210,12 +208,9 @@ public class SftpOutboundTests { handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); final List madeDirs = new ArrayList(); - doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - madeDirs.add((String) invocation.getArguments()[0]); - return null; - } + doAnswer(invocation -> { + madeDirs.add((String) invocation.getArguments()[0]); + return null; }).when(session).mkdir(anyString()); handler.handleMessage(new GenericMessage("qux")); assertEquals(3, madeDirs.size()); @@ -227,7 +222,8 @@ public class SftpOutboundTests { @Test public void testSharedSession() throws Exception { JSch jsch = spy(new JSch()); - Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class, String.class, String.class, int.class); + Constructor ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class, + String.class, String.class, int.class); ctor.setAccessible(true); com.jcraft.jsch.Session jschSession1 = spy(ctor.newInstance(jsch, "foo", "host", 22)); com.jcraft.jsch.Session jschSession2 = spy(ctor.newInstance(jsch, "foo", "host", 22)); @@ -242,13 +238,7 @@ public class SftpOutboundTests { new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); // Can't use when(session.open()) with a spy final AtomicInteger n = new AtomicInteger(); - doAnswer(new Answer() { - - @Override - public ChannelSftp answer(InvocationOnMock invocation) throws Throwable { - return n.getAndIncrement() == 0 ? channel1 : channel2; - } - }).when(jschSession1).openChannel("sftp"); + doAnswer(invocation -> n.getAndIncrement() == 0 ? channel1 : channel2).when(jschSession1).openChannel("sftp"); DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, true); factory.setHost("host"); factory.setUser("foo"); @@ -313,20 +303,8 @@ public class SftpOutboundTests { new DirectFieldAccessor(channel2).setPropertyValue("session", jschSession1); // Can't use when(session.open()) with a spy final AtomicInteger n = new AtomicInteger(); - doAnswer(new Answer() { - - @Override - public ChannelSftp answer(InvocationOnMock invocation) throws Throwable { - return n.getAndIncrement() == 0 ? channel1 : channel2; - } - }).when(jschSession1).openChannel("sftp"); - doAnswer(new Answer() { - - @Override - public ChannelSftp answer(InvocationOnMock invocation) throws Throwable { - return n.getAndIncrement() < 3 ? channel3 : channel4; - } - }).when(jschSession2).openChannel("sftp"); + doAnswer(invocation -> n.getAndIncrement() == 0 ? channel1 : channel2).when(jschSession1).openChannel("sftp"); + doAnswer(invocation -> n.getAndIncrement() < 3 ? channel3 : channel4).when(jschSession2).openChannel("sftp"); DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(jsch, true); factory.setHost("host"); factory.setUser("foo"); @@ -367,13 +345,7 @@ public class SftpOutboundTests { } private void noopConnect(ChannelSftp channel1) throws JSchException { - doAnswer(new Answer() { - - @Override - public Object answer(InvocationOnMock invocation) throws Throwable { - return null; - } - }).when(channel1).connect(); + doAnswer(invocation -> null).when(channel1).connect(); } public static class TestSftpSessionFactory extends DefaultSftpSessionFactory { @@ -383,29 +355,19 @@ public class SftpOutboundTests { try { ChannelSftp channel = mock(ChannelSftp.class); - doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) - throws Throwable { - File file = new File((String) invocation.getArguments()[1]); - assertTrue(file.getName().endsWith(".writing")); - FileCopyUtils.copy((InputStream) invocation.getArguments()[0], new FileOutputStream(file)); - return null; - } - + doAnswer(invocation -> { + File file = new File((String) invocation.getArguments()[1]); + assertTrue(file.getName().endsWith(".writing")); + FileCopyUtils.copy((InputStream) invocation.getArguments()[0], new FileOutputStream(file)); + return null; }).when(channel).put(Mockito.any(InputStream.class), Mockito.anyString()); - doAnswer(new Answer() { - @Override - public Object answer(InvocationOnMock invocation) - throws Throwable { - File file = new File((String) invocation.getArguments()[0]); - assertTrue(file.getName().endsWith(".writing")); - File renameToFile = new File((String) invocation.getArguments()[1]); - file.renameTo(renameToFile); - return null; - } - + doAnswer(invocation -> { + File file = new File((String) invocation.getArguments()[0]); + assertTrue(file.getName().endsWith(".writing")); + File renameToFile = new File((String) invocation.getArguments()[1]); + file.renameTo(renameToFile); + return null; }).when(channel).rename(Mockito.anyString(), Mockito.anyString()); String[] files = new File("remote-test-dir").list(); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java index f8c2a1e5dd..5a86a08a52 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java @@ -50,7 +50,6 @@ import org.springframework.context.annotation.Bean; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.remote.MessageSessionCallback; -import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.sftp.SftpTestSupport; @@ -266,31 +265,23 @@ public class SftpServerOutboundTests extends SftpTestSupport { PipedOutputStream out2 = new PipedOutputStream(pipe2); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - try { - session1.write(pipe1, "foo.txt"); - } - catch (IOException e) { - e.printStackTrace(); - } - latch1.countDown(); + Executors.newSingleThreadExecutor().execute(() -> { + try { + session1.write(pipe1, "foo.txt"); } + catch (IOException e) { + e.printStackTrace(); + } + latch1.countDown(); }); - Executors.newSingleThreadExecutor().execute(new Runnable() { - - @Override - public void run() { - try { - session2.write(pipe2, "bar.txt"); - } - catch (IOException e) { - e.printStackTrace(); - } - latch2.countDown(); + Executors.newSingleThreadExecutor().execute(() -> { + try { + session2.write(pipe2, "bar.txt"); } + catch (IOException e) { + e.printStackTrace(); + } + latch2.countDown(); }); out1.write('a'); @@ -444,13 +435,7 @@ public class SftpServerOutboundTests extends SftpTestSupport { } private void assertLength6(SftpRemoteFileTemplate template) { - LsEntry[] files = template.execute(new SessionCallback() { - - @Override - public LsEntry[] doInSession(Session session) throws IOException { - return session.list("sftpTarget/appending.txt"); - } - }); + LsEntry[] files = template.execute(session -> session.list("sftpTarget/appending.txt")); assertEquals(1, files.length); assertEquals(6, files[0].getAttrs().getSize()); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java index 7a65edc141..0e0387fc06 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java @@ -20,8 +20,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; -import java.io.IOException; - import org.junit.Test; import org.junit.runner.RunWith; @@ -31,10 +29,8 @@ import org.springframework.context.annotation.Configuration; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.remote.ClientCallbackWithoutResult; -import org.springframework.integration.file.remote.SessionCallback; import org.springframework.integration.file.remote.SessionCallbackWithoutResult; import org.springframework.integration.file.remote.session.CachingSessionFactory; -import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.sftp.SftpTestSupport; import org.springframework.messaging.support.GenericMessage; @@ -68,41 +64,28 @@ public class SftpRemoteFileTemplateTests extends SftpTestSupport { template.setFileNameGenerator(fileNameGenerator); template.setRemoteDirectoryExpression(new LiteralExpression("foo/")); template.setUseTemporaryFileName(false); - template.execute(new SessionCallback() { - - @Override - public Boolean doInSession(Session session) throws IOException { - session.mkdir("foo/"); - return session.mkdir("foo/bar/"); - } - + template.execute(session -> { + session.mkdir("foo/"); + return session.mkdir("foo/bar/"); }); template.append(new GenericMessage("foo")); template.append(new GenericMessage("bar")); assertTrue(template.exists("foo/foobar.txt")); - template.executeWithClient(new ClientCallbackWithoutResult() { - - @Override - public void doWithClientWithoutResult(ChannelSftp client) { - try { - SftpATTRS file = client.lstat("foo/foobar.txt"); - assertEquals(6, file.getSize()); - } - catch (SftpException e) { - throw new RuntimeException(e); - } + template.executeWithClient((ClientCallbackWithoutResult) client -> { + try { + SftpATTRS file = client.lstat("foo/foobar.txt"); + assertEquals(6, file.getSize()); + } + catch (SftpException e) { + throw new RuntimeException(e); } }); - template.execute(new SessionCallbackWithoutResult() { - - @Override - public void doInSessionWithoutResult(Session session) throws IOException { - assertTrue(session.remove("foo/foobar.txt")); - assertTrue(session.rmdir("foo/bar/")); - LsEntry[] files = session.list("foo/"); - assertEquals(0, files.length); - assertTrue(session.rmdir("foo/")); - } + template.execute((SessionCallbackWithoutResult) session -> { + assertTrue(session.remove("foo/foobar.txt")); + assertTrue(session.rmdir("foo/bar/")); + LsEntry[] files = session.list("foo/"); + assertEquals(0, files.length); + assertTrue(session.rmdir("foo/")); }); assertFalse(template.exists("foo")); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java index b10284f987..c2720ea192 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java @@ -35,10 +35,7 @@ import org.apache.sshd.common.NamedFactory; import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory; import org.apache.sshd.common.util.Base64; import org.apache.sshd.server.Command; -import org.apache.sshd.server.PasswordAuthenticator; -import org.apache.sshd.server.PublickeyAuthenticator; import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; -import org.apache.sshd.server.session.ServerSession; import org.apache.sshd.server.sftp.SftpSubsystem; import org.junit.Test; @@ -63,13 +60,7 @@ public class SftpServerTests { public void testUcPw() throws Exception { SshServer server = SshServer.setUpDefaultServer(); try { - server.setPasswordAuthenticator(new PasswordAuthenticator() { - - @Override - public boolean authenticate(String arg0, String arg1, ServerSession arg2) { - return true; - } - }); + server.setPasswordAuthenticator((arg0, arg1, arg2) -> true); server.setPort(0); server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser")); server.setSubsystemFactories(Collections.>singletonList(new SftpSubsystem.Factory())); @@ -107,14 +98,7 @@ public class SftpServerTests { SshServer server = SshServer.setUpDefaultServer(); final PublicKey allowedKey = decodePublicKey(pubKey); try { - server.setPublickeyAuthenticator(new PublickeyAuthenticator() { - - @Override - public boolean authenticate(String username, PublicKey key, ServerSession session) { - return key.equals(allowedKey); - } - - }); + server.setPublickeyAuthenticator((username, key, session) -> key.equals(allowedKey)); server.setPort(0); server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser")); server.setSubsystemFactories(Collections.>singletonList(new SftpSubsystem.Factory())); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java index 1f376f179c..4a7a853bec 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java @@ -30,16 +30,12 @@ import static org.mockito.Mockito.when; import java.io.IOException; import java.net.ConnectException; -import java.security.PublicKey; import java.util.Collections; import org.apache.sshd.SshServer; import org.apache.sshd.common.NamedFactory; import org.apache.sshd.server.Command; -import org.apache.sshd.server.PasswordAuthenticator; -import org.apache.sshd.server.PublickeyAuthenticator; import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; -import org.apache.sshd.server.session.ServerSession; import org.apache.sshd.server.sftp.SftpSubsystem; import org.junit.Test; @@ -65,13 +61,7 @@ public class SftpSessionFactoryTests { public void testConnectFailSocketOpen() throws Exception { SshServer server = SshServer.setUpDefaultServer(); try { - server.setPasswordAuthenticator(new PasswordAuthenticator() { - - @Override - public boolean authenticate(String arg0, String arg1, ServerSession arg2) { - return true; - } - }); + server.setPasswordAuthenticator((arg0, arg1, arg2) -> true); server.setPort(0); server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser")); server.start(); @@ -219,15 +209,8 @@ public class SftpSessionFactoryTests { } } - @SuppressWarnings("unchecked") private DefaultSftpSessionFactory createServerAndClient(SshServer server) throws IOException { - server.setPublickeyAuthenticator(new PublickeyAuthenticator() { - - @Override - public boolean authenticate(String username, PublicKey key, ServerSession session) { - return true; - } - }); + server.setPublickeyAuthenticator((username, key, session) -> true); server.setPort(0); server.setSubsystemFactories(Collections.>singletonList(new SftpSubsystem.Factory())); server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser"));