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

INT-3886: TCP Fix Socket Timeout: Raw Deserializer

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3886

An SO user reported a short message delivery when an NIO connection was timed out.

See JIRA for link.

We could not produce the problem with his deserializer but it did identify a problem
with the standard `ByteArrayRawSerializer`.

With NIO, socket timeouts were reported to the deserializer as a normal EOF (-1). This
caused the raw serializer to emit a short message - it's signal for end of message is the
socket closure.

We can't treat a timeout as a normal EOF.

If the socket is forcibly timed out due to no recent data, throw a `SocketTimeoutException` to
the deserializer.

Just in case the old behavior is being relied upon, a boolean has been added to restore that
behavior. This is not recommended, as using timeout to delimit messages is not reliable.

* Fix typos
* Increase `MongoDbInboundChannelAdapterIntegrationTests` timeouts
* Some code polishing
This commit is contained in:
Gary Russell
2015-11-12 19:37:59 -05:00
committed by Artem Bilan
parent f0d0772397
commit e5269bb265
5 changed files with 283 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,22 +21,23 @@ import static org.junit.Assert.assertNull;
import java.util.List;
import com.mongodb.DBObject;
import com.mongodb.util.JSON;
import org.junit.Test;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.BasicQuery;
import org.springframework.messaging.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
import org.springframework.messaging.Message;
import com.mongodb.DBObject;
import com.mongodb.util.JSON;
/**
* @author Oleg Zhurakousky
* @since 2.2
@@ -45,171 +46,186 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab
@Test
@MongoDbAvailable
public void testWithDefaultMongoFactory() throws Exception{
public void testWithDefaultMongoFactory() throws Exception {
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
MongoTemplate template = new MongoTemplate(mongoDbFactory);
template.save(this.createPerson("Bob"), "data");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
ConfigurableApplicationContext context =
new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapter", SourcePollingChannelAdapter.class);
QueueChannel replyChannel = context.getBean("replyChannel", QueueChannel.class);
spca.start();
@SuppressWarnings("unchecked")
Message<List<Person>> message = (Message<List<Person>>) replyChannel.receive(1000);
Message<List<Person>> message = (Message<List<Person>>) replyChannel.receive(10000);
assertNotNull(message);
assertEquals("Bob", message.getPayload().get(0).getName());
assertNotNull(replyChannel.receive(1000));
spca.stop();
assertNotNull(replyChannel.receive(10000));
context.close();
}
@Test
@MongoDbAvailable
public void testWithNamedMongoFactory() throws Exception{
public void testWithNamedMongoFactory() throws Exception {
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
MongoTemplate template = new MongoTemplate(mongoDbFactory);
template.save(this.createPerson("Bob"), "data");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterNamedFactory", SourcePollingChannelAdapter.class);
ConfigurableApplicationContext context =
new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterNamedFactory",
SourcePollingChannelAdapter.class);
QueueChannel replyChannel = context.getBean("replyChannel", QueueChannel.class);
spca.start();
@SuppressWarnings("unchecked")
Message<List<DBObject>> message = (Message<List<DBObject>>) replyChannel.receive(1000);
Message<List<DBObject>> message = (Message<List<DBObject>>) replyChannel.receive(10000);
assertNotNull(message);
assertEquals("Bob", message.getPayload().get(0).get("name"));
spca.stop();
context.close();
}
@Test
@MongoDbAvailable
public void testWithMongoTemplate() throws Exception{
public void testWithMongoTemplate() throws Exception {
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
MongoTemplate template = new MongoTemplate(mongoDbFactory);
template.save(this.createPerson("Bob"), "data");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterWithTemplate", SourcePollingChannelAdapter.class);
ConfigurableApplicationContext context =
new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterWithTemplate",
SourcePollingChannelAdapter.class);
QueueChannel replyChannel = context.getBean("replyChannel", QueueChannel.class);
spca.start();
@SuppressWarnings("unchecked")
Message<Person> message = (Message<Person>) replyChannel.receive(1000);
Message<Person> message = (Message<Person>) replyChannel.receive(10000);
assertNotNull(message);
assertEquals("Bob", message.getPayload().getName());
spca.stop();
context.close();
}
@Test
@MongoDbAvailable
public void testWithNamedCollection() throws Exception{
public void testWithNamedCollection() throws Exception {
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
MongoTemplate template = new MongoTemplate(mongoDbFactory);
template.save(this.createPerson("Bob"), "foo");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterWithNamedCollection", SourcePollingChannelAdapter.class);
ConfigurableApplicationContext context =
new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterWithNamedCollection",
SourcePollingChannelAdapter.class);
QueueChannel replyChannel = context.getBean("replyChannel", QueueChannel.class);
spca.start();
@SuppressWarnings("unchecked")
Message<List<Person>> message = (Message<List<Person>>) replyChannel.receive(1000);
Message<List<Person>> message = (Message<List<Person>>) replyChannel.receive(10000);
assertNotNull(message);
assertEquals("Bob", message.getPayload().get(0).getName());
spca.stop();
context.close();
}
@Test
@MongoDbAvailable
public void testWithNamedCollectionExpression() throws Exception{
public void testWithNamedCollectionExpression() throws Exception {
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
MongoTemplate template = new MongoTemplate(mongoDbFactory);
template.save(this.createPerson("Bob"), "foo");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterWithNamedCollectionExpression", SourcePollingChannelAdapter.class);
ConfigurableApplicationContext context =
new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterWithNamedCollectionExpression",
SourcePollingChannelAdapter.class);
QueueChannel replyChannel = context.getBean("replyChannel", QueueChannel.class);
spca.start();
@SuppressWarnings("unchecked")
Message<List<Person>> message = (Message<List<Person>>) replyChannel.receive(1000);
Message<List<Person>> message = (Message<List<Person>>) replyChannel.receive(10000);
assertNotNull(message);
assertEquals("Bob", message.getPayload().get(0).getName());
spca.stop();
context.close();
}
@Test
@MongoDbAvailable
public void testWithOnSuccessDisposition() throws Exception{
public void testWithOnSuccessDisposition() throws Exception {
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
MongoTemplate template = new MongoTemplate(mongoDbFactory);
template.save(this.createPerson("Bob"), "data");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("inboundAdapterWithOnSuccessDisposition", SourcePollingChannelAdapter.class);
ConfigurableApplicationContext context =
new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("inboundAdapterWithOnSuccessDisposition",
SourcePollingChannelAdapter.class);
QueueChannel replyChannel = context.getBean("replyChannel", QueueChannel.class);
spca.start();
assertNotNull(replyChannel.receive(1000));
assertNotNull(replyChannel.receive(10000));
Thread.sleep(300);
assertNull(replyChannel.receive(1000));
spca.stop();
assertNull(replyChannel.receive(100));
context.close();
}
@Test
@MongoDbAvailable
public void testWithMongoConverter() throws Exception{
public void testWithMongoConverter() throws Exception {
MongoDbFactory mongoDbFactory = this.prepareMongoFactory();
MongoTemplate template = new MongoTemplate(mongoDbFactory);
template.save(this.createPerson("Bob"), "data");
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterWithConverter", SourcePollingChannelAdapter.class);
ConfigurableApplicationContext context =
new ClassPathXmlApplicationContext("inbound-adapter-config.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("mongoInboundAdapterWithConverter",
SourcePollingChannelAdapter.class);
QueueChannel replyChannel = context.getBean("replyChannel", QueueChannel.class);
spca.start();
@SuppressWarnings("unchecked")
Message<List<Person>> message = (Message<List<Person>>) replyChannel.receive(1000);
Message<List<Person>> message = (Message<List<Person>>) replyChannel.receive(10000);
assertNotNull(message);
assertEquals("Bob", message.getPayload().get(0).getName());
assertNotNull(replyChannel.receive(1000));
spca.stop();
assertNotNull(replyChannel.receive(10000));
context.close();
}
@Test(expected=BeanDefinitionParsingException.class)
@Test(expected = BeanDefinitionParsingException.class)
@MongoDbAvailable
public void testFailureWithQueryAndQueryExpression() throws Exception{
new ClassPathXmlApplicationContext("inbound-fail-q-qex.xml", this.getClass());
public void testFailureWithQueryAndQueryExpression() throws Exception {
new ClassPathXmlApplicationContext("inbound-fail-q-qex.xml", this.getClass()).close();
}
@Test(expected=BeanDefinitionParsingException.class)
@Test(expected = BeanDefinitionParsingException.class)
@MongoDbAvailable
public void testFailureWithFactoryAndTemplate() throws Exception{
new ClassPathXmlApplicationContext("inbound-fail-factory-template.xml", this.getClass());
public void testFailureWithFactoryAndTemplate() throws Exception {
new ClassPathXmlApplicationContext("inbound-fail-factory-template.xml", this.getClass()).close();
}
@Test(expected=BeanDefinitionParsingException.class)
@Test(expected = BeanDefinitionParsingException.class)
@MongoDbAvailable
public void testFailureWithCollectionAndCollectioinExpression() throws Exception{
new ClassPathXmlApplicationContext("inbound-fail-c-cex.xml", this.getClass());
public void testFailureWithCollectionAndCollectioinExpression() throws Exception {
new ClassPathXmlApplicationContext("inbound-fail-c-cex.xml", this.getClass()).close();
}
@Test(expected=BeanDefinitionParsingException.class)
@Test(expected = BeanDefinitionParsingException.class)
@MongoDbAvailable
public void testFailureWithTemplateAndConverter() throws Exception{
new ClassPathXmlApplicationContext("inbound-fail-converter-template.xml", this.getClass());
public void testFailureWithTemplateAndConverter() throws Exception {
new ClassPathXmlApplicationContext("inbound-fail-converter-template.xml", this.getClass()).close();
}
public static class DocumentCleaner {
public void remove(MongoOperations mongoOperations, Object target, String collectionName) {
if (target instanceof List<?>){
if (target instanceof List<?>) {
List<?> documents = (List<?>) target;
for (Object document : documents) {
mongoOperations.remove(new BasicQuery(JSON.serialize(document)), collectionName);
}
}
}
}
}