diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index d4e370fbfd..8a2bee94d2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -86,6 +86,8 @@ public class TcpNioConnection extends TcpConnectionSupport { private volatile long pipeTimeout = DEFAULT_PIPE_TIMEOUT; + private volatile boolean timedOut; + /** * Constructs a TcpNetConnection for the SocketChannel. * @param socketChannel The socketChannel. @@ -497,6 +499,7 @@ public class TcpNioConnection extends TcpConnectionSupport { * Close the socket due to timeout. */ void timeout() { + this.timedOut = true; this.closeConnection(true); } @@ -644,10 +647,10 @@ public class TcpNioConnection extends TcpConnectionSupport { public int read(byte[] b, int off, int len) throws IOException { Assert.notNull(b, "byte[] cannot be null"); if (off < 0 || len < 0 || len > b.length - off) { - throw new IndexOutOfBoundsException(); + throw new IndexOutOfBoundsException(); } else if (len == 0) { - return 0; + return 0; } int n = 0; @@ -670,12 +673,18 @@ public class TcpNioConnection extends TcpConnectionSupport { @Override public synchronized int read() throws IOException { if (this.isClosed && available.get() == 0) { + if (TcpNioConnection.this.timedOut) { + throw new SocketTimeoutException("Connection has timed out"); + } return -1; } if (this.currentBuffer == null) { this.currentBuffer = getNextBuffer(); this.currentOffset = 0; if (this.currentBuffer == null) { + if (TcpNioConnection.this.timedOut) { + throw new SocketTimeoutException("Connection has timed out"); + } return -1; } } @@ -744,4 +753,5 @@ public class TcpNioConnection extends TcpConnectionSupport { } } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java index b212f2f8af..6a3642a2d2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayRawSerializer.java @@ -18,6 +18,7 @@ package org.springframework.integration.ip.tcp.serializer; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; +import java.net.SocketTimeoutException; /** * A byte array (de)serializer that does nothing with the payload; sends it raw. @@ -28,6 +29,12 @@ import java.io.OutputStream; * Because the socket must be closed to indicate message end, this (de)serializer * can only be used by uni-directional (non-collaborating) channel adapters, and * not by gateways. + *

+ * Prior to 4.2.2, when using NIO, a timeout caused whatever had been partially + * received to be emitted as a message. + *

+ * Now, a {@link SocketTimeoutException} is thrown. To revert to the previous + * behavior, set the {@code treatTimeoutAsEndOfMessage} constructor argument to true. * * @author Gary Russell * @since 2.0.3 @@ -35,6 +42,22 @@ import java.io.OutputStream; */ public class ByteArrayRawSerializer extends AbstractByteArraySerializer { + private final boolean treatTimeoutAsEndOfMessage; + + public ByteArrayRawSerializer() { + this(false); + } + + /** + * Treat socket timeouts as a normal EOF and emit the (possibly partial) + * message. + * @param treatTimeoutAsEndOfMessage true to emit a message after a timeout. + * @since 4.2.2 + */ + public ByteArrayRawSerializer(boolean treatTimeoutAsEndOfMessage) { + this.treatTimeoutAsEndOfMessage = treatTimeoutAsEndOfMessage; + } + @Override public void serialize(byte[] bytes, OutputStream outputStream) throws IOException { @@ -51,7 +74,15 @@ public class ByteArrayRawSerializer extends AbstractByteArraySerializer { } try { while (bite >= 0) { - bite = inputStream.read(); + try { + bite = inputStream.read(); + } + catch (SocketTimeoutException e) { + if (!this.treatTimeoutAsEndOfMessage) { + throw e; + } + bite = -1; + } if (bite < 0) { if (n == 0) { throw new SoftEndOfStreamException("Stream closed between payloads"); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java index de51c64dff..f05330c928 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/serializer/DeserializationTests.java @@ -19,25 +19,45 @@ package org.springframework.integration.ip.tcp.serializer; import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThat; import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; import java.io.ByteArrayInputStream; import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; import java.net.ServerSocket; import java.net.Socket; +import java.nio.ByteBuffer; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicReference; import javax.net.ServerSocketFactory; +import org.junit.Rule; import org.junit.Test; +import org.springframework.beans.factory.BeanFactory; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.serializer.DefaultDeserializer; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.ip.tcp.TcpInboundGateway; +import org.springframework.integration.ip.tcp.TcpOutboundGateway; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; import org.springframework.integration.ip.util.SocketTestUtils; +import org.springframework.integration.ip.util.TestingUtilities; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.support.LongRunningIntegrationTest; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; /** * @author Gary Russell @@ -46,6 +66,9 @@ import org.springframework.integration.ip.util.SocketTestUtils; */ public class DeserializationTests { + @Rule + public LongRunningIntegrationTest longRunningIntegrationTest = new LongRunningIntegrationTest(); + @Test public void testReadLength() throws Exception { ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0); @@ -324,4 +347,144 @@ public class DeserializationTests { return event.get(); } + @Test + public void testTimeoutWithCustomDeserializer() throws Exception { + testTimeoutWhileDecoding(new CustomDeserializer(), "\u0000\u0002\u0000\u0005reply"); + } + + @Test + public void testTimeoutWithRawDeserializer() throws Exception { + testTimeoutWhileDecoding(new ByteArrayRawSerializer(), "reply"); + } + + public void testTimeoutWhileDecoding(AbstractByteArraySerializer deserializer, String reply) throws Exception { + ByteArrayRawSerializer serializer = new ByteArrayRawSerializer(); + TcpNioServerConnectionFactory serverNio = new TcpNioServerConnectionFactory(0); + ByteArrayLengthHeaderSerializer lengthHeaderSerializer = new ByteArrayLengthHeaderSerializer(1); + serverNio.setDeserializer(lengthHeaderSerializer); + serverNio.setSerializer(serializer); + serverNio.afterPropertiesSet(); + TcpInboundGateway in = new TcpInboundGateway(); + in.setConnectionFactory(serverNio); + QueueChannel serverSideChannel = new QueueChannel(); + in.setRequestChannel(serverSideChannel); + in.setBeanFactory(mock(BeanFactory.class)); + in.afterPropertiesSet(); + in.start(); + TestingUtilities.waitListening(serverNio, null); + TcpNioClientConnectionFactory clientNio = new TcpNioClientConnectionFactory("localhost", serverNio.getPort()); + clientNio.setSerializer(serializer); + clientNio.setDeserializer(deserializer); + clientNio.setSoTimeout(1000); + clientNio.afterPropertiesSet(); + final TcpOutboundGateway out = new TcpOutboundGateway(); + out.setConnectionFactory(clientNio); + QueueChannel outputChannel = new QueueChannel(); + out.setOutputChannel(outputChannel); + out.setRemoteTimeout(60000); + out.setBeanFactory(mock(BeanFactory.class)); + out.afterPropertiesSet(); + out.start(); + Runnable command = new Runnable() { + + @Override + public void run() { + try { + out.handleMessage(MessageBuilder.withPayload("\u0004Test").build()); + } + catch (Exception e) { + // eat SocketTimeoutException. Doesn't matter for this test + } + } + + }; + ExecutorService exec = Executors.newSingleThreadExecutor(); + + Message message; + + // short reply should not be received. + exec.execute(command); + message = serverSideChannel.receive(10000); + assertNotNull(message); + assertEquals("Test", new String((byte[]) message.getPayload())); + String shortReply = reply.substring(0, reply.length() - 1); + ((MessageChannel) message.getHeaders().getReplyChannel()).send(new GenericMessage(shortReply)); + message = outputChannel.receive(6000); + assertNull(message); + + // good message should be received + if ((deserializer instanceof ByteArrayRawSerializer)) { // restore old behavior + clientNio.setDeserializer(new ByteArrayRawSerializer(true)); + } + exec.execute(command); + message = serverSideChannel.receive(10000); + assertNotNull(message); + assertEquals("Test", new String((byte[]) message.getPayload())); + ((MessageChannel) message.getHeaders().getReplyChannel()).send(new GenericMessage(reply)); + message = outputChannel.receive(10000); + assertNotNull(message); + assertEquals(reply, new String(((byte[]) message.getPayload()))); + } + + private static class CustomDeserializer extends AbstractByteArraySerializer { + + @Override + public byte[] deserialize(InputStream inputStream) throws IOException { + if (logger.isDebugEnabled()) { + logger.debug("Available to read:" + inputStream.available()); + } + + byte[] header = new byte[2]; + header[0] = (byte) inputStream.read(); + if (header[0] < 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); + } + + header[1] = (byte) inputStream.read(); + if (header[1] < 0) { + checkClosure(-1); + } + + ByteBuffer headerBB = ByteBuffer.wrap(header); + int val = headerBB.getShort(); + + byte[] length = new byte[val]; + for (int i = 0; i < val; i++) { + length[i] = (byte) inputStream.read(); + } + + ByteBuffer lengthBB = ByteBuffer.wrap(length); + int messageLength; + if (val == 2) { + messageLength = lengthBB.getShort(); + } + else if (val == 4) { + messageLength = lengthBB.getInt(); + } + else { + throw new IOException("Unexpected count of bytes that holds message length"); + } + + byte[] answer = new byte[messageLength]; + for (int i = 0; i < messageLength; i++) { + int bite = inputStream.read(); + if (bite < 0) { + checkClosure(-1); + } + answer[i] = (byte) bite; + } + + ByteBuffer b = ByteBuffer.allocate(2 + val + messageLength); + b.put(header); + b.put(length); + b.put(answer); + return b.array(); + } + + @Override + public void serialize(byte[] object, OutputStream outputStream) throws IOException { + } + + } + } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java index 32d143b9d0..8fd2e4ce07 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java @@ -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> message = (Message>) replyChannel.receive(1000); + Message> message = (Message>) 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> message = (Message>) replyChannel.receive(1000); + Message> message = (Message>) 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 message = (Message) replyChannel.receive(1000); + Message message = (Message) 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> message = (Message>) replyChannel.receive(1000); + Message> message = (Message>) 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> message = (Message>) replyChannel.receive(1000); + Message> message = (Message>) 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> message = (Message>) replyChannel.receive(1000); + Message> message = (Message>) 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); } } } + } } diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 5e118ad183..733379c2b9 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -223,6 +223,12 @@ When using this serializer, message reception will hang until the client closes When this serializer is being used, and the client is a Spring Integration application, the client must use a connection factory that is configured with single-use=true - this causes the adapter to close the socket after sending the message; the serializer will not, itself, close the connection. This serializer should only be used with connection factories used by channel adapters (not gateways), and the connection factories should be used by either an inbound or outbound adapter, and not both. +NOTE: Before version 4.2.2, when using NIO, this serializer treated a timeout (during read) as an end of file and the +data read so far was emitted as a message. +This is unreliable and should not be used to delimit messages; it now treats such conditions as an exception. +In the unlikely event you are using it this way, the previous behavior can be restored by setting the +`treatTimeoutAsEndOfMessage` constructor argument to `true`. + Each of these is a subclass of `AbstractByteArraySerializer` which implements both `org.springframework.core.serializer.Serializer` and `org.springframework.core.serializer.Deserializer`. For backwards compatibility, connections using any subclass of `AbstractByteArraySerializer` for serialization will also accept a String which will be converted to a byte array first. Each of these (de)serializers converts an input stream containing the corresponding format to a byte array payload.