Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

(S)FTP Lambdas

Browse Source 
AMQP Lambdas

Event Lambdas

Feed/File Lambdas

Gemfile/Groovy Lambdas

* Ensure that JavaDocs are checked via `	check.dependsOn javadoc`
* Add `@return` tag to the `MessageBuilder.readOnlyHeaders()`
This commit is contained in:
Gary Russell
2016-09-22 12:32:47 -04:00
committed by Artem Bilan
parent 3402a86fb5
commit e5ae7d886d
48 changed files with 569 additions and 1148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
build.gradle
View File

@@ -244,6 +244,7 @@ subprojects { subproject ->
}
}
check.dependsOn javadoc
build.dependsOn jacocoTestReport
}

26
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpInboundChannelAdapter.java
View File

@@ -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<String, Object> 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<String, Object> 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();

14
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java
View File

@@ -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;
});
}
}

16
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/ChannelTests.java
View File

@@ -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<String>("foo"));

30
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/channel/DispatcherHasNoSubscribersTests.java
View File

@@ -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<Channel>() {
@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<Channel>() {
@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<String> logList = new ArrayList<String>();
doAnswer(new Answer<Object>() {
@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();

32
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpInboundGatewayParserTests.java
View File

@@ -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);

42
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundChannelAdapterParserTests.java
View File

@@ -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<Object>() {
@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),

110
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpOutboundGatewayParserTests.java
View File

@@ -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<org.springframework.amqp.core.Message>() {
@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));

9
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/OutboundGatewayTests.java
View File

@@ -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<Object>() {
@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())))

71
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -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<Channel>() {
@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<Channel>() {
@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<Channel>() {
@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<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
org.springframework.amqp.core.Message message =
(org.springframework.amqp.core.Message) invocation.getArguments()[2];
Map<String, Object> 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<String, Object> 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));

59
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AsyncAmqpGatewayTests.java
View File

@@ -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<String, String>) 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<Void>() {
@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();

22
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java
View File

@@ -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<Message> amqpMessage =
new AtomicReference<Message>();
doAnswer(new Answer<Object>() {
@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<Message> amqpMessage =
new AtomicReference<Message>();
doAnswer(new Answer<Object>() {
@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")

13
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/JsonConverterCompatibilityTests.java
View File

@@ -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<Void>() {
@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);

1
spring-integration-core/src/main/java/org/springframework/integration/support/MessageBuilder.java
View File

@@ -279,6 +279,7 @@ public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T
* and prohibited from being populated in the message.
* @param readOnlyHeaders the list of headers for {@code readOnly} mode.
* Defaults to {@link MessageHeaders#ID} and {@link MessageHeaders#TIMESTAMP}.
* @return the current {@link MessageBuilder}
* @since 4.3.2
* @see IntegrationMessageHeaderAccessor#isReadOnly(String)
*/

83
spring-integration-event/src/test/java/org/springframework/integration/event/config/EventOutboundChannelAdapterParserTests.java
View File

@@ -25,7 +25,6 @@ import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.PayloadApplicationEvent;
@@ -73,18 +72,13 @@ public class EventOutboundChannelAdapterParserTests {
@Test
public void validateUsage() {
ApplicationListener<?> listener = new ApplicationListener<ApplicationEvent>() {
@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<ApplicationEvent>() {
@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<ApplicationEvent>() {
@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<ApplicationEvent>() {
@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);

8
spring-integration-feed/src/test/java/org/springframework/integration/feed/config/FeedInboundChannelAdapterParserTests.java
View File

@@ -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);

40
spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java
View File

@@ -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<F> extends AbstractReply
return doMput(requestMessage);
}
}
return this.remoteFileTemplate.execute(new SessionCallback<F, Object>() {
@Override
public Object doInSession(Session<F> 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<F> extends AbstractReply
dir += this.remoteFileTemplate.getRemoteFileSeparator();
}
final String fullDir = dir;
List<?> payload = this.remoteFileTemplate.execute(new SessionCallback<F, List<?>>() {
@Override
public List<?> doInSession(Session<F> 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<F> extends AbstractReply
}
}
else {
payload = this.remoteFileTemplate.execute(new SessionCallback<F, File>() {
@Override
public File doInSession(Session<F> 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<F> extends AbstractReply
final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
final String remoteFilename = getRemoteFilename(remoteFilePath);
final String remoteDir = getRemoteDirectory(remoteFilePath, remoteFilename);
List<File> payload = this.remoteFileTemplate.execute(new SessionCallback<F, List<File>>() {
@Override
public List<File> doInSession(Session<F> session) throws IOException {
return mGet(requestMessage, session, remoteDir, remoteFilename);
}
});
List<File> 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)

76
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
View File

@@ -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<F>
}
final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class);
try {
int transferred = this.remoteFileTemplate.execute(new SessionCallback<F, Integer>() {
@Override
public Integer doInSession(Session<F> session) throws IOException {
F[] files = session.list(remoteDirectory);
if (!ObjectUtils.isEmpty(files)) {
List<F> filteredFiles = filterFiles(files);
if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) {
rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
List<F> 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<F> filteredFiles = filterFiles(files);
if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) {
rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
List<F> 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<F> filteredFiles, F file) {
if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
((ReversibleFileListFilter<F>) 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<F>
}
}
protected void rollbackFromFileToListEnd(List<F> filteredFiles, F file) {
if (this.filter instanceof ReversibleFileListFilter) {
((ReversibleFileListFilter<F>) this.filter)
.rollback(file, filteredFiles);
}
}
protected void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory,
Session<F> session) throws IOException {
String remoteFileName = this.getFilename(remoteFile);

119
spring-integration-file/src/main/java/org/springframework/integration/file/tail/OSDelegatingFileTailingMessageProducer.java
View File

@@ -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);
}
}
}

25
spring-integration-file/src/test/java/org/springframework/integration/file/FileInboundTransactionTests.java
View File

@@ -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");

28
spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourceIntegrationTests.java
View File

@@ -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<File> received = pollableFileSource.receive();
while (received == null) {
Thread.yield();
received = pollableFileSource.receive();
}
Runnable successfulConsumer = () -> {
Message<File> received = pollableFileSource.receive();
while (received == null) {
Thread.yield();
received = pollableFileSource.receive();
}
};
Runnable failingConsumer = new Runnable() {
@Override
public void run() {
Message<File> received = pollableFileSource.receive();
if (received != null) {
pollableFileSource.onFailure(received);
}
Runnable failingConsumer = () -> {
Message<File> received = pollableFileSource.receive();
if (received != null) {
pollableFileSource.onFailure(received);
}
};
CountDownLatch successfulDone = doConcurrently(3, successfulConsumer, go);
CountDownLatch failingDone = doConcurrently(10, failingConsumer, go);

40
spring-integration-file/src/test/java/org/springframework/integration/file/FileWritingMessageHandlerTests.java
View File

@@ -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<InputStream>(new ByteArrayInputStream("box".getBytes())));
handler.trigger(new GenericMessage<String>("foo"));
@@ -503,14 +484,9 @@ public class FileWritingMessageHandlerTests {
handler.handleMessage(new GenericMessage<InputStream>(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());

34
spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterExternalStoreTests.java
View File

@@ -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<Object>() {
@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<Integer> result = Executors.newSingleThreadExecutor().submit(new Callable<Integer>() {
@Override
public Integer call() throws Exception {
return filter.filterFiles(new File[] {file}).size();
}
});
Future<Integer> result = Executors.newSingleThreadExecutor()
.submit(() -> filter.filterFiles(new File[] { file }).size());
assertTrue(latch2.await(10, TimeUnit.SECONDS));
store.put("foo:" + file.getAbsolutePath(), "43");
latch1.countDown();

68
spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java
View File

@@ -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<String> args = new AtomicReference<String>();
doAnswer(new Answer<Object>() {
@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<String> requestMessage = MessageBuilder.withPayload("foo")
@@ -287,14 +281,10 @@ public class RemoteFileOutboundGatewayTests {
gw.afterPropertiesSet();
Session<?> session = mock(Session.class);
final AtomicReference<String> args = new AtomicReference<String>();
doAnswer(new Answer<Object>() {
@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<String>("foo"));
@@ -312,23 +302,15 @@ public class RemoteFileOutboundGatewayTests {
gw.afterPropertiesSet();
Session<?> session = mock(Session.class);
final AtomicReference<String> args = new AtomicReference<String>();
doAnswer(new Answer<Object>() {
@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<String> madeDirs = new ArrayList<String>();
doAnswer(new Answer<Object>() {
@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<String> requestMessage = MessageBuilder.withPayload("foo")
@@ -926,13 +908,9 @@ public class RemoteFileOutboundGatewayTests {
gw.afterPropertiesSet();
when(sessionFactory.getSession()).thenReturn(session);
final AtomicReference<String> written = new AtomicReference<String>();
doAnswer(new Answer<Object>() {
@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<String> written = new AtomicReference<String>();
doAnswer(new Answer<Object>() {
@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");

34
spring-integration-file/src/test/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandlerTests.java
View File

@@ -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<F> session = mock(Session.class);
when(sf.getSession()).thenReturn(session);
doAnswer(new Answer<Object>() {
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<F> handler = new FileTransferringMessageHandler<F>(sf);
@@ -90,12 +86,10 @@ public class FileTransferringMessageHandlerTests {
final AtomicReference<String> temporaryPath = new AtomicReference<String>();
final AtomicReference<String> finalPath = new AtomicReference<String>();
when(sf.getSession()).thenReturn(session);
doAnswer(new Answer<Object>() {
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<F> handler = new FileTransferringMessageHandler<F>(sf);
handler.setRemoteDirectoryExpression(new LiteralExpression("foo"));
@@ -115,12 +109,10 @@ public class FileTransferringMessageHandlerTests {
Session<F> session = mock(Session.class);
when(sf.getSession()).thenReturn(session);
doAnswer(new Answer<Object>() {
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<F> handler = new FileTransferringMessageHandler<F>(sf);

8
spring-integration-file/src/test/java/org/springframework/integration/file/remote/session/CachingSessionFactoryTests.java
View File

@@ -92,12 +92,8 @@ public class CachingSessionFactoryTests {
template.setBeanFactory(mock(BeanFactory.class));
template.afterPropertiesSet();
try {
template.get(new GenericMessage<String>("foo"), new InputStreamCallback() {
@Override
public void doWithInputStream(InputStream stream) throws IOException {
throw new RuntimeException("bar");
}
template.get(new GenericMessage<String>("foo"), (InputStreamCallback) stream -> {
throw new RuntimeException("bar");
});
fail("Expected exception");
}

20
spring-integration-file/src/test/java/org/springframework/integration/file/tail/FileTailingMessageProducerTests.java
View File

@@ -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<FileTailingEvent> events = new ArrayList<FileTailingEvent>();
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();

33
spring-integration-file/src/test/java/org/springframework/integration/file/tail/TailRule.java
View File

@@ -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<Integer> c = new AtomicReference<Integer>();
final CountDownLatch latch = new CountDownLatch(1);
Future<Process> future = Executors.newSingleThreadExecutor().submit(new Callable<Process>() {
@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<Process> 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);

37
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplate.java
View File

@@ -80,33 +80,28 @@ public class FtpRemoteFileTemplate extends RemoteFileTemplate<FTPFile> {
*/
@Override
public boolean exists(final String path) {
return doExecuteWithClient(new ClientCallback<FTPClient, Boolean>() {
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);
}
});
}

15
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
View File

@@ -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<Method> genMethod = new AtomicReference<Method>();
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());
}

14
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java
View File

@@ -136,16 +136,10 @@ public class FtpOutboundGatewayParserTests {
//INT-3129
assertNotNull(TestUtils.getPropertyValue(gateway, "localFilenameGeneratorExpression"));
final AtomicReference<Method> genMethod = new AtomicReference<Method>();
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<String>(""), "foo"));
assertThat(TestUtils.getPropertyValue(gateway, "mputFilter"), Matchers.instanceOf(SimplePatternFileListFilter.class));
}

41
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java
View File

@@ -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<FTPFile> handler = new FileTransferringMessageHandler<FTPFile>(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>("String data"));
@@ -119,12 +113,7 @@ public class FtpOutboundTests {
assertFalse(file.exists());
FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<FTPFile>(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[]>("byte[] data".getBytes()));
@@ -141,12 +130,7 @@ public class FtpOutboundTests {
FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<FTPFile>(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<FTPFile> handler = new FileTransferringMessageHandler<FTPFile>(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<String> logged = new AtomicReference<String>();
doAnswer(new Answer<Object>() {
@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);

113
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java
View File

@@ -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<String>("ftpSource/ ftpSource1.txt"), new InputStreamCallback() {
@Override
public void doWithInputStream(InputStream stream) throws IOException {
FileCopyUtils.copy(stream, baos1);
}
}));
assertTrue(template.get(new GenericMessage<String>("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<String>("ftpSource/ftpSource2.txt"), new InputStreamCallback() {
@Override
public void doWithInputStream(InputStream stream) throws IOException {
FileCopyUtils.copy(stream, baos2);
}
}));
assertTrue(template.get(new GenericMessage<String>("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<FTPFile> session = spyOnSession();
doAnswer(new Answer<FTPFile[]>() {
@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<FTPFile> session = spyOnSession();
doAnswer(new Answer<FTPFile[]>() {
@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<FTPFile> session = spyOnSession();
doAnswer(new Answer<Void>() {
@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<File>(getSourceLocalDirectory()));
@@ -528,13 +501,8 @@ public class FtpServerOutboundTests extends FtpTestSupport {
FileOutputStream writer = new FileOutputStream(extra);
writer.write("foo".getBytes());
writer.close();
doAnswer(new Answer<Void>() {
@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<File>(getSourceLocalDirectory()));
@@ -569,13 +537,7 @@ public class FtpServerOutboundTests extends FtpTestSupport {
}
private void assertLength6(FtpRemoteFileTemplate template) {
FTPFile[] files = template.execute(new SessionCallback<FTPFile, FTPFile[]>() {
@Override
public FTPFile[] doInSession(Session<FTPFile> 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<File> filterFiles(File[] files) {
File[] sorted = Arrays.copyOf(files, files.length);
Arrays.sort(sorted, new Comparator<File>() {
@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);
}

47
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/FtpRemoteFileTemplateTests.java
View File

@@ -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<FTPFile, Boolean>() {
@Override
public Boolean doInSession(Session<FTPFile> 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<String>("foo"));
template.append(new GenericMessage<String>("bar"));
assertTrue(template.exists("foo/foobar.txt"));
template.executeWithClient(new ClientCallbackWithoutResult<FTPClient>() {
@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<FTPClient>) 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<FTPFile>() {
@Override
public void doInSessionWithoutResult(Session<FTPFile> 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<FTPFile>) 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"));
}

21
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/session/SessionFactoryTests.java
View File

@@ -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();
}
});
}

26
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/ForkUtil.java
View File

@@ -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;

7
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java
View File

@@ -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);

29
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/GemfireGroupStoreTests.java
View File

@@ -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");
}
});

36
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/util/AggregatorWithGemfireLocksTests.java
View File

@@ -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<String>("bar", stubHeaders(2, 2, 1)));
}
catch (Exception e) {
e.printStackTrace();
exception = e;
}
Executors.newSingleThreadExecutor().execute(() -> {
try {
in2.send(new GenericMessage<String>("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<String>(payload, stubHeaders(sequence, 2, correlation)));
}
catch (Exception e) {
e.printStackTrace();
exception = e;
}
return () -> {
try {
in.send(new GenericMessage<String>(payload, stubHeaders(sequence, 2, correlation)));
}
catch (Exception e) {
e.printStackTrace();
exception = e;
}
};
}

9
spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java
View File

@@ -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<Object>
@@ -69,13 +70,7 @@ public class GroovyScriptExecutingMessageProcessor extends AbstractScriptExecuti
private volatile ScriptSource scriptSource;
private volatile GroovyClassLoader groovyClassLoader = AccessController.doPrivileged(
new PrivilegedAction<GroovyClassLoader>() {
public GroovyClassLoader run() {
return new GroovyClassLoader(ClassUtils.getDefaultClassLoader());
}
});
(PrivilegedAction<GroovyClassLoader>) () -> new GroovyClassLoader(ClassUtils.getDefaultClassLoader()));
private volatile Class<?> scriptClass;

23
spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessorTests.java
View File

@@ -83,6 +83,7 @@ public class GroovyScriptExecutingMessageProcessorTests {
ScriptSource scriptSource = new ResourceScriptSource(resource);
Object result = null;
class CustomScriptVariableGenerator implements ScriptVariableGenerator {
@Override
public Map<String, Object> generateScriptVariables(Message<?> message) {
Map<String, Object> variables = new HashMap<String, Object>();
variables.put("date", System.nanoTime());
@@ -197,24 +198,16 @@ public class GroovyScriptExecutingMessageProcessorTests {
ScriptSource scriptSource = new StaticScriptSource(script, Script.class.getName());
final MessageProcessor<Object> processor =
new GroovyScriptExecutingMessageProcessor(scriptSource, new ScriptVariableGenerator() {
@Override
public Map<String, Object> generateScriptVariables(Message<?> message) {
Map<String, Object> variables = new HashMap<String, Object>(2);
variables.put("var1", var1);
variables.put("var2", var2);
return variables;
}
new GroovyScriptExecutingMessageProcessor(scriptSource, message1 -> {
Map<String, Object> variables = new HashMap<String, Object>(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"));
}

10
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/SftpTestSupport.java
View File

@@ -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.<NamedFactory<Command>>singletonList(new SftpSubsystem.Factory()));

14
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java
View File

@@ -122,16 +122,10 @@ public class SftpOutboundGatewayParserTests {
//INT-3129
assertNotNull(TestUtils.getPropertyValue(gateway, "localFilenameGeneratorExpression"));
final AtomicReference<Method> genMethod = new AtomicReference<Method>();
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<String>(""), "foo"));
assertThat(TestUtils.getPropertyValue(gateway, "mputFilter"), Matchers.instanceOf(SimplePatternFileListFilter.class));
}

78
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java
View File

@@ -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<String> madeDirs = new ArrayList<String>();
doAnswer(new Answer<Object>() {
@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<String>("qux"));
assertEquals(3, madeDirs.size());
@@ -227,7 +222,8 @@ public class SftpOutboundTests {
@Test
public void testSharedSession() throws Exception {
JSch jsch = spy(new JSch());
Constructor<com.jcraft.jsch.Session> ctor = com.jcraft.jsch.Session.class.getDeclaredConstructor(JSch.class, String.class, String.class, int.class);
Constructor<com.jcraft.jsch.Session> 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<ChannelSftp>() {
@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<ChannelSftp>() {
@Override
public ChannelSftp answer(InvocationOnMock invocation) throws Throwable {
return n.getAndIncrement() == 0 ? channel1 : channel2;
}
}).when(jschSession1).openChannel("sftp");
doAnswer(new Answer<ChannelSftp>() {
@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<Object>() {
@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<Object>() {
@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<Object>() {
@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();

45
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpServerOutboundTests.java
View File

@@ -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<LsEntry, LsEntry[]>() {
@Override
public LsEntry[] doInSession(Session<LsEntry> 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());
}

49
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpRemoteFileTemplateTests.java
View File

@@ -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<LsEntry, Boolean>() {
@Override
public Boolean doInSession(Session<LsEntry> 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<String>("foo"));
template.append(new GenericMessage<String>("bar"));
assertTrue(template.exists("foo/foobar.txt"));
template.executeWithClient(new ClientCallbackWithoutResult<ChannelSftp>() {
@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<ChannelSftp>) client -> {
try {
SftpATTRS file = client.lstat("foo/foobar.txt");
assertEquals(6, file.getSize());
}
catch (SftpException e) {
throw new RuntimeException(e);
}
});
template.execute(new SessionCallbackWithoutResult<LsEntry>() {
@Override
public void doInSessionWithoutResult(Session<LsEntry> 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<LsEntry>) 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"));
}

20
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpServerTests.java
View File

@@ -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.<NamedFactory<Command>>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.<NamedFactory<Command>>singletonList(new SftpSubsystem.Factory()));

21
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java
View File

@@ -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.<NamedFactory<Command>>singletonList(new SftpSubsystem.Factory()));
server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider("hostkey.ser"));