diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParser.java index d089dc4545..38339907b6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParser.java @@ -35,7 +35,7 @@ public class PayloadDeserializingTransformerParser extends AbstractTransformerPa @Override protected void parseTransformer(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "converter"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "deserializer"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParser.java index 9c65d52713..edaaa2b0ea 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParser.java @@ -36,7 +36,7 @@ public class PayloadSerializingTransformerParser extends AbstractTransformerPars @Override protected void parseTransformer(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "converter"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer"); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadDeserializingTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadDeserializingTransformer.java index faf9638dd6..006564ed02 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadDeserializingTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadDeserializingTransformer.java @@ -16,9 +16,8 @@ package org.springframework.integration.transformer; +import org.springframework.commons.serializer.Deserializer; import org.springframework.commons.serializer.DeserializingConverter; -import org.springframework.commons.serializer.java.JavaStreamingConverter; -import org.springframework.core.convert.converter.Converter; /** * Transformer that deserializes the inbound byte array payload to an object by delegating to a @@ -33,15 +32,14 @@ import org.springframework.core.convert.converter.Converter; */ public class PayloadDeserializingTransformer extends PayloadTypeConvertingTransformer { - @Override - public void setConverter(Converter converter) { - this.converter = converter; + public void setDeserializer(Deserializer deserializer) { + this.setConverter(new DeserializingConverter(deserializer)); } @Override protected Object transformPayload(byte[] payload) throws Exception { if (this.converter == null) { - this.converter = new DeserializingConverter(new JavaStreamingConverter()); + this.setConverter(new DeserializingConverter()); } return this.converter.convert(payload); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadSerializingTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadSerializingTransformer.java index 0d2edb1c82..2704842720 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadSerializingTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadSerializingTransformer.java @@ -16,9 +16,8 @@ package org.springframework.integration.transformer; +import org.springframework.commons.serializer.Serializer; import org.springframework.commons.serializer.SerializingConverter; -import org.springframework.commons.serializer.java.JavaStreamingConverter; -import org.springframework.core.convert.converter.Converter; /** * Transformer that serializes the inbound payload into a byte array by delegating to a @@ -33,15 +32,14 @@ import org.springframework.core.convert.converter.Converter; */ public class PayloadSerializingTransformer extends PayloadTypeConvertingTransformer { - @Override - public void setConverter(Converter converter) { - this.converter = converter; + public void setSerializer(Serializer serializer) { + this.setConverter(new SerializingConverter(serializer)); } @Override protected byte[] transformPayload(Object payload) throws Exception { if (this.converter == null) { - this.converter = new SerializingConverter(new JavaStreamingConverter()); + this.setConverter(new SerializingConverter()); } return this.converter.convert(payload); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java index f93f41949a..008f9d4e6c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java @@ -25,20 +25,13 @@ import org.springframework.util.Assert; * * @author Gary Russell * @since 2.0 - * */ public class PayloadTypeConvertingTransformer extends AbstractPayloadTransformer { - + protected Converter converter; - @Override - protected U transformPayload(T payload) throws Exception { - Assert.notNull(this.converter, this.getClass().getName() + " needs a Converter"); - return converter.convert(payload); - } - /** - * Sets the converter to be used for Serialization. + * Specify the converter to use. * * @param converter The Converter. */ @@ -46,5 +39,10 @@ public class PayloadTypeConvertingTransformer extends AbstractPayloadTrans this.converter = converter; } - + @Override + protected U transformPayload(T payload) throws Exception { + Assert.notNull(this.converter, this.getClass().getName() + " requires a Converter"); + return this.converter.convert(payload); + } + } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 874ca978d4..783d3b80da 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -1650,15 +1650,15 @@ - + - Reference to a Converter instance that converts from an object to a byte array. - This is optional. The default Converter will use standard Java serialization. + Reference to a Serializer instance to convert from an object to a byte array. + This is optional. The default will use standard Java serialization. - + @@ -1684,15 +1684,15 @@ - + - Reference to a Converter instance that converts from a byte array to an object. - This is optional. The default Converter will use standard Java deserialization. + Reference to a Deserializer instance to convert from a byte array to an object. + This is optional. The default will use standard Java deserialization. - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests-context.xml index 4f16993314..5ba8ba0b39 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests-context.xml @@ -13,7 +13,7 @@ - + @@ -25,8 +25,8 @@ - + - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java index 692e7b7b2f..165a6a7096 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java @@ -21,15 +21,17 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; import java.io.ObjectOutputStream; import java.io.Serializable; -import java.io.UnsupportedEncodingException; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.core.convert.converter.Converter; +import org.springframework.commons.serializer.Deserializer; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.core.PollableChannel; @@ -37,6 +39,7 @@ import org.springframework.integration.message.GenericMessage; import org.springframework.integration.transformer.MessageTransformationException; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.FileCopyUtils; /** * @author Mark Fisher @@ -52,7 +55,7 @@ public class PayloadDeserializingTransformerParserTests { private MessageChannel queueInput; @Autowired - private MessageChannel customConverterInput; + private MessageChannel customDeserializerInput; @Autowired private PollableChannel output; @@ -105,8 +108,8 @@ public class PayloadDeserializingTransformerParserTests { } @Test - public void customConverter() throws Exception { - customConverterInput.send(new GenericMessage("test".getBytes("UTF-8"))); + public void customDeserializer() throws Exception { + customDeserializerInput.send(new GenericMessage("test".getBytes("UTF-8"))); Message result = output.receive(3000); assertNotNull(result); assertEquals(String.class, result.getPayload().getClass()); @@ -130,15 +133,10 @@ public class PayloadDeserializingTransformerParserTests { } - public static class TestDeserializingConverter implements Converter { + public static class TestDeserializer implements Deserializer { - public Object convert(byte[] source) { - try { - return new String(source, "UTF-8").toUpperCase(); - } - catch (UnsupportedEncodingException e) { - throw new MessageTransformationException("failed to convert payload", e); - } + public Object deserialize(InputStream source) throws IOException { + return FileCopyUtils.copyToString(new InputStreamReader(source, "UTF-8")).toUpperCase(); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests-context.xml index 61d4eb9d3d..a47d7d6af8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests-context.xml @@ -13,7 +13,7 @@ - + @@ -25,8 +25,8 @@ - + - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests.java index 3ec6df8bd0..a79fc69d4f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests.java @@ -21,15 +21,16 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import java.io.ByteArrayInputStream; +import java.io.IOException; import java.io.ObjectInputStream; +import java.io.OutputStream; import java.io.Serializable; -import java.io.UnsupportedEncodingException; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.core.convert.converter.Converter; +import org.springframework.commons.serializer.Serializer; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.core.PollableChannel; @@ -52,7 +53,7 @@ public class PayloadSerializingTransformerParserTests { private MessageChannel queueInput; @Autowired - private MessageChannel customConverterInput; + private MessageChannel customSerializerInput; @Autowired private PollableChannel output; @@ -103,8 +104,8 @@ public class PayloadSerializingTransformerParserTests { } @Test - public void customConverter() throws Exception { - customConverterInput.send(new GenericMessage("test")); + public void customSerializer() throws Exception { + customSerializerInput.send(new GenericMessage("test")); Message result = output.receive(3000); assertNotNull(result); assertEquals(byte[].class, result.getPayload().getClass()); @@ -127,15 +128,12 @@ public class PayloadSerializingTransformerParserTests { } - public static class TestSerializingConverter implements Converter { + public static class TestSerializer implements Serializer { - public byte[] convert(Object source) { - try { - return source.toString().toUpperCase().getBytes("UTF-8"); - } - catch (UnsupportedEncodingException e) { - throw new MessageTransformationException("failed to convert payload", e); - } + public void serialize(Object source, OutputStream outputStream) throws IOException { + outputStream.write(source.toString().toUpperCase().getBytes("UTF-8")); + outputStream.flush(); + outputStream.close(); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index b55ab51349..20a2ae9804 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -89,9 +89,9 @@ public abstract class IpAdapterParserUtils { static final String TCP_CONNECTION_TYPE = "type"; - static final String INPUT_CONVERTER = "input-converter"; + static final String SERIALIZER = "serializer"; - static final String OUTPUT_CONVERTER = "output-converter"; + static final String DESERIALIZER = "deserializer"; static final String SINGLE_USE = "single-use"; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java index 449237129d..e42f86f670 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java @@ -85,9 +85,9 @@ public class TcpConnectionParser extends AbstractBeanDefinitionParser { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, IpAdapterParserUtils.TASK_EXECUTOR); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - IpAdapterParserUtils.INPUT_CONVERTER); + IpAdapterParserUtils.SERIALIZER); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, - IpAdapterParserUtils.OUTPUT_CONVERTER); + IpAdapterParserUtils.DESERIALIZER); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.SINGLE_USE); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java index b5d4766f51..f8e476a3eb 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -70,8 +70,9 @@ public abstract class AbstractClientConnectionFactory extends AbstractConnection } } connection.setMapper(this.mapper); - connection.setInputConverter(this.inputConverter); - connection.setOutputConverter(this.outputConverter); + connection.setDeserializer(this.deserializer); + connection.setSerializer(this.serializer); connection.setSingleUse(this.singleUse); } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java index efb4b86bb6..8a45713703 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -23,8 +23,9 @@ import java.util.concurrent.Executors; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.commons.serializer.InputStreamingConverter; -import org.springframework.commons.serializer.OutputStreamingConverter; + +import org.springframework.commons.serializer.Deserializer; +import org.springframework.commons.serializer.Serializer; import org.springframework.context.SmartLifecycle; import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter; import org.springframework.util.Assert; @@ -67,9 +68,9 @@ public abstract class AbstractConnectionFactory protected Executor taskExecutor; - protected InputStreamingConverter inputConverter = new ByteArrayCrLfConverter(); + protected Deserializer deserializer = new ByteArrayCrLfConverter(); - protected OutputStreamingConverter outputConverter = new ByteArrayCrLfConverter(); + protected Serializer serializer = new ByteArrayCrLfConverter(); protected TcpMessageMapper mapper = new TcpMessageMapper(); @@ -250,18 +251,18 @@ public abstract class AbstractConnectionFactory /** * - * @param converter the inputConverter to set + * @param deserializer the deserializer to set */ - public void setInputConverter(InputStreamingConverter converter) { - this.inputConverter = converter; + public void setDeserializer(Deserializer deserializer) { + this.deserializer = deserializer; } /** * - * @param outputConverter the outputConverter to set + * @param serializer the serializer to set */ - public void setOutputConverter(OutputStreamingConverter outputConverter) { - this.outputConverter = outputConverter; + public void setSerializer(Serializer serializer) { + this.serializer = serializer; } /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index af8c6be7c4..d434b26b03 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -19,21 +19,21 @@ package org.springframework.integration.ip.tcp.connection; import java.net.Socket; import java.net.SocketException; - - /** * Base class for all server connection factories. Server connection factories * listen on a port for incoming connections and create new TcpConnection objects - * for each new connection. + * for each new connection. + * * @author Gary Russell - * + * @since 2.0 */ public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory { protected boolean listening; protected String localAddress; - + + /** * The port on which the factory will listen. * @param port @@ -41,7 +41,8 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection public AbstractServerConnectionFactory(int port) { this.port = port; } - + + /** * Not supported because the factory manages multiple connections and this * method cannot discriminate. @@ -70,8 +71,8 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection } connection.registerSender(this.sender); connection.setMapper(this.mapper); - connection.setInputConverter(this.inputConverter); - connection.setOutputConverter(this.outputConverter); + connection.setDeserializer(this.deserializer); + connection.setSerializer(this.serializer); connection.setSingleUse(this.singleUse); /* * If we have a collaborating outbound channel adapter and we are configured diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java index 45fd6d4f61..4b0d914267 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java @@ -20,8 +20,9 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.commons.serializer.InputStreamingConverter; -import org.springframework.commons.serializer.OutputStreamingConverter; + +import org.springframework.commons.serializer.Deserializer; +import org.springframework.commons.serializer.Serializer; import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter; import org.springframework.util.Assert; @@ -39,10 +40,10 @@ public abstract class AbstractTcpConnection implements TcpConnection { protected Log logger = LogFactory.getLog(this.getClass()); @SuppressWarnings("rawtypes") - protected InputStreamingConverter inputConverter; + protected Deserializer deserializer; @SuppressWarnings("rawtypes") - protected OutputStreamingConverter outputConverter; + protected Serializer serializer; protected TcpMessageMapper mapper; @@ -84,42 +85,42 @@ public abstract class AbstractTcpConnection implements TcpConnection { public void setMapper(TcpMessageMapper mapper) { Assert.notNull(mapper, this.getClass().getName() + " Mapper may not be null"); this.mapper = mapper; - if (this.outputConverter != null && - !(this.outputConverter instanceof AbstractByteArrayStreamingConverter)) { + if (this.serializer != null && + !(this.serializer instanceof AbstractByteArrayStreamingConverter)) { mapper.setStringToBytes(false); } } /** * - * @return the input converter + * @return the deserializer */ - public InputStreamingConverter getInputConverter() { - return inputConverter; + public Deserializer getDeserializer() { + return this.deserializer; } /** - * @param inputConverter the input converter to set + * @param deserializer the deserializer to set */ - public void setInputConverter(InputStreamingConverter inputConverter) { - this.inputConverter = inputConverter; + public void setDeserializer(Deserializer deserializer) { + this.deserializer = deserializer; } /** * - * @return the output converter + * @return the serializer */ - public OutputStreamingConverter getOutputConverter() { - return outputConverter; + public Serializer getSerializer() { + return this.serializer; } /** - * @param outputConverter the output converter to set + * @param serializer the serializer to set */ - public void setOutputConverter(OutputStreamingConverter outputConverter) { - this.outputConverter = outputConverter; - if (!(outputConverter instanceof AbstractByteArrayStreamingConverter)) { - mapper.setStringToBytes(false); + public void setSerializer(Serializer serializer) { + this.serializer = serializer; + if (!(serializer instanceof AbstractByteArrayStreamingConverter)) { + this.mapper.setStringToBytes(false); } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java index 30e7143f2c..81ef26ac74 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java @@ -13,11 +13,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.tcp.connection; - -import org.springframework.commons.serializer.InputStreamingConverter; -import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.commons.serializer.Deserializer; +import org.springframework.commons.serializer.Serializer; import org.springframework.integration.Message; /** @@ -26,7 +26,6 @@ import org.springframework.integration.Message; * * @author Gary Russell * @since 2.0 - * */ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionInterceptor { @@ -36,6 +35,7 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI private TcpSender tcpSender; + public void close() { this.theConnection.close(); } @@ -90,20 +90,20 @@ public abstract class AbstractTcpConnectionInterceptor implements TcpConnectionI this.theConnection.setMapper(mapper); } - public InputStreamingConverter getInputConverter() { - return this.theConnection.getInputConverter(); + public Deserializer getDeserializer() { + return this.theConnection.getDeserializer(); } - public void setInputConverter(InputStreamingConverter inputConverter) { - this.theConnection.setInputConverter(inputConverter); + public void setDeserializer(Deserializer deserializer) { + this.theConnection.setDeserializer(deserializer); } - public OutputStreamingConverter getOutputConverter() { - return this.theConnection.getOutputConverter(); + public Serializer getSerializer() { + return this.theConnection.getSerializer(); } - public void setOutputConverter(OutputStreamingConverter outputConverter) { - this.theConnection.setOutputConverter(outputConverter); + public void setSerializer(Serializer serializer) { + this.theConnection.setSerializer(serializer); } public boolean isServer() { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java index d6e126e33c..46de065050 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -19,8 +19,8 @@ package org.springframework.integration.ip.tcp.connection; import java.net.Socket; import java.nio.channels.SocketChannel; -import org.springframework.commons.serializer.InputStreamingConverter; -import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.commons.serializer.Deserializer; +import org.springframework.commons.serializer.Serializer; import org.springframework.integration.Message; /** @@ -119,25 +119,25 @@ public interface TcpConnection extends Runnable { /** * - * @return the input converter + * @return the deserializer */ - public InputStreamingConverter getInputConverter(); + public Deserializer getDeserializer(); /** - * @param inputConverter the inputConverter to set + * @param deserializer the deserializer to set */ - public void setInputConverter(InputStreamingConverter inputConverter); + public void setDeserializer(Deserializer deserializer); /** * - * @return the output converter + * @return the serializer */ - public OutputStreamingConverter getOutputConverter(); + public Serializer getSerializer(); /** - * @param outputConverter the outputConverter to set + * @param serializer the serializer to set */ - public void setOutputConverter(OutputStreamingConverter outputConverter); + public void setSerializer(Serializer serializer); /** * @return this connection's listener diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index 66e1ef38c5..2b4e78b16d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -19,7 +19,6 @@ package org.springframework.integration.ip.tcp.connection; import java.net.Socket; import java.net.SocketTimeoutException; -import org.springframework.commons.serializer.InputStreamingConverter; import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.SocketIoUtils; import org.springframework.integration.ip.tcp.converter.SoftEndOfStreamException; @@ -64,7 +63,7 @@ public class TcpNetConnection extends AbstractTcpConnection { @SuppressWarnings("unchecked") public synchronized void send(Message message) throws Exception { Object object = mapper.fromMessage(message); - this.outputConverter.convert(object, this.socket.getOutputStream()); + this.serializer.serialize(object, this.socket.getOutputStream()); if (logger.isDebugEnabled()) logger.debug("Message sent " + message); } @@ -78,7 +77,7 @@ public class TcpNetConnection extends AbstractTcpConnection { } public Object getPayload() throws Exception { - return this.inputConverter.convert(this.socket.getInputStream()); + return this.deserializer.deserialize(this.socket.getInputStream()); } public int getPort() { 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 b751a95e0c..fd310d2a18 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 @@ -106,7 +106,7 @@ public class TcpNioConnection extends AbstractTcpConnection { public void send(Message message) throws Exception { synchronized(mapper) { Object object = mapper.fromMessage(message); - this.outputConverter.convert(object, this.channelOutputStream); + this.serializer.serialize(object, this.channelOutputStream); } } @@ -119,7 +119,7 @@ public class TcpNioConnection extends AbstractTcpConnection { } public Object getPayload() throws Exception { - return this.inputConverter.convert(pipedInputStream); + return this.deserializer.deserialize(pipedInputStream); } public int getPort() { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java index 5ee928e6ad..9a5037abd8 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java @@ -20,8 +20,9 @@ import java.io.IOException; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.commons.serializer.InputStreamingConverter; -import org.springframework.commons.serializer.OutputStreamingConverter; + +import org.springframework.commons.serializer.Deserializer; +import org.springframework.commons.serializer.Serializer; /** * Base class for streaming converters that convert to/from a byte array. @@ -31,8 +32,8 @@ import org.springframework.commons.serializer.OutputStreamingConverter; * */ public abstract class AbstractByteArrayStreamingConverter implements - InputStreamingConverter, - OutputStreamingConverter { + Serializer, + Deserializer { protected int maxMessageSize = 2048; diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java index ca89503e33..7e45571ad1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java @@ -27,7 +27,6 @@ import java.io.OutputStream; * * @author Gary Russell * @since 2.0 - * */ public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter { @@ -37,7 +36,7 @@ public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter * is closed immediately after the \r\n (i.e. no data is in the process of * being read). */ - public byte[] convert(InputStream inputStream) throws IOException { + public byte[] deserialize(InputStream inputStream) throws IOException { byte[] buffer = new byte[this.maxMessageSize]; int n = 0; int bite; @@ -66,7 +65,7 @@ public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter /** * Writes the byte[] to the stream and appends \r\n. */ - public void convert(byte[] bytes, OutputStream outputStream) throws IOException { + public void serialize(byte[] bytes, OutputStream outputStream) throws IOException { outputStream.write(bytes); outputStream.write('\r'); outputStream.write('\n'); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java index 4bcd39b255..d95cdedd5d 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java @@ -32,8 +32,9 @@ import org.apache.commons.logging.LogFactory; * The length field contains the length of data following the length * field. * (network byte order). + * * @author Gary Russell - * + * @since 2.0 */ public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingConverter { @@ -46,7 +47,7 @@ public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingCo * Throws a {@link SoftEndOfStreamException} if the stream * is closed between messages. */ - public byte[] convert(InputStream inputStream) throws IOException { + public byte[] deserialize(InputStream inputStream) throws IOException { byte[] lengthPart = new byte[4]; int status = read(inputStream, lengthPart, true); if (status < 0) { @@ -69,8 +70,7 @@ public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingCo * Writes the byte[] to the output stream, preceded by a 4 byte * length in network byte order (big endian). */ - public void convert(byte[] bytes, OutputStream outputStream) - throws IOException { + public void serialize(byte[] bytes, OutputStream outputStream) throws IOException { ByteBuffer lengthPart = ByteBuffer.allocate(4); lengthPart.putInt(bytes.length); outputStream.write(lengthPart.array()); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java index 54886a8d2e..0029989fef 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java @@ -29,7 +29,6 @@ import org.springframework.integration.mapping.MessageMappingException; * * @author Gary Russell * @since 2.0 - * */ public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverter { @@ -45,7 +44,7 @@ public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverte * being read). * */ - public byte[] convert(InputStream inputStream) throws IOException { + public byte[] deserialize(InputStream inputStream) throws IOException { int bite = inputStream.read(); if (bite < 0) { throw new SoftEndOfStreamException("Stream closed between payloads"); @@ -71,7 +70,7 @@ public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverte * Writes the byte[] to the stream, prefixed by an ASCII STX character and * terminated with an ASCII ETX character. */ - public void convert(byte[] bytes, OutputStream outputStream) throws IOException { + public void serialize(byte[] bytes, OutputStream outputStream) throws IOException { outputStream.write(STX); outputStream.write(bytes); outputStream.write(ETX); diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index a02586f7e5..63b513d5f5 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -290,31 +290,31 @@ the factory, the connection will be closed after a response is received. - + - + -An InputStreamingConverter that converts message payloads to/from output streams/input streams -associated with the connection. Default is ByteArrayCrLfConverter. Input and output converters -would normally be the same but this is not required. +A Serializer that converts message payloads to/from output streams/input streams +associated with the connection. Default is ByteArrayCrLfConverter. Serializer and Deserializer +would normally be the same but this is not required. - + - + -An OutputStreamingConverter that converts message payloads to/from output streams/input streams -associated with the connection. Default is ByteArrayCrLfConverter. Input and output converters -would normally be the same but this is not required. +A Deserializer that converts message payloads to/from output streams/input streams +associated with the connection. Default is ByteArrayCrLfConverter. Serializer and Deserializer +would normally be the same but this is not required. diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 30ff2356fd..f665a16205 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -131,8 +131,8 @@ type="client" host="localhost" port="#{tcpIpUtils.findAvailableServerSocket(6000)}" - input-converter="serial" - output-converter="serial" + serializer="defaultSerializer" + deserializer="defaultDeserializer" so-keep-alive="true" so-linger="54" so-receive-buffer-size="1234" @@ -153,8 +153,8 @@ type="server" port="#{client1.port}" local-address="127.0.0.1" - input-converter="serial" - output-converter="serial" + serializer="defaultSerializer" + deserializer="defaultDeserializer" so-keep-alive="true" so-linger="55" so-receive-buffer-size="1234" @@ -175,8 +175,8 @@ type="client" host="localhost" port="#{tcpIpUtils.findAvailableServerSocket(6100)}" - input-converter="serial" - output-converter="serial" + serializer="defaultSerializer" + deserializer="defaultDeserializer" so-keep-alive="true" so-linger="54" so-receive-buffer-size="1234" @@ -196,8 +196,8 @@ type="server" port="#{client1.port}" local-address="127.0.0.1" - input-converter="serial" - output-converter="serial" + serializer="defaultSerializer" + deserializer="defaultDeserializer" so-keep-alive="true" so-linger="55" so-receive-buffer-size="1234" @@ -214,8 +214,10 @@ - + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 8b41c044b0..c8db3148e7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.config; import static org.junit.Assert.assertEquals; @@ -23,10 +24,12 @@ import static org.junit.Assert.assertTrue; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.commons.serializer.Deserializer; +import org.springframework.commons.serializer.Serializer; import org.springframework.context.ApplicationContext; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.ip.tcp.TcpInboundGateway; @@ -46,7 +49,6 @@ import org.springframework.integration.ip.udp.UnicastSendingMessageHandler; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; - /** * @author Gary Russell * @since 2.0 @@ -111,8 +113,11 @@ public class ParserUnitTests { AbstractConnectionFactory cfC2; @Autowired - InputStreamingConverter converter; - + Serializer serializer; + + @Autowired + Deserializer deserializer; + @Autowired @Qualifier(value="server1") AbstractConnectionFactory server1; @@ -263,8 +268,8 @@ public class ParserUnitTests { assertEquals(1236, client1.getSoTimeout()); assertEquals(12, client1.getSoTrafficClass()); DirectFieldAccessor dfa = new DirectFieldAccessor(client1); - assertSame(converter, dfa.getPropertyValue("inputConverter")); - assertSame(converter, dfa.getPropertyValue("outputConverter")); + assertSame(serializer, dfa.getPropertyValue("serializer")); + assertSame(deserializer, dfa.getPropertyValue("deserializer")); assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); @@ -283,8 +288,8 @@ public class ParserUnitTests { assertEquals(1236, server1.getSoTimeout()); assertEquals(12, server1.getSoTrafficClass()); DirectFieldAccessor dfa = new DirectFieldAccessor(server1); - assertSame(converter, dfa.getPropertyValue("inputConverter")); - assertSame(converter, dfa.getPropertyValue("outputConverter")); + assertSame(serializer, dfa.getPropertyValue("serializer")); + assertSame(deserializer, dfa.getPropertyValue("deserializer")); assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); @@ -304,8 +309,8 @@ public class ParserUnitTests { assertEquals(1236, client1.getSoTimeout()); assertEquals(12, client1.getSoTrafficClass()); DirectFieldAccessor dfa = new DirectFieldAccessor(client1); - assertSame(converter, dfa.getPropertyValue("inputConverter")); - assertSame(converter, dfa.getPropertyValue("outputConverter")); + assertSame(serializer, dfa.getPropertyValue("serializer")); + assertSame(deserializer, dfa.getPropertyValue("deserializer")); assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); @@ -323,8 +328,8 @@ public class ParserUnitTests { assertEquals(1236, server1.getSoTimeout()); assertEquals(12, server1.getSoTrafficClass()); DirectFieldAccessor dfa = new DirectFieldAccessor(server1); - assertSame(converter, dfa.getPropertyValue("inputConverter")); - assertSame(converter, dfa.getPropertyValue("outputConverter")); + assertSame(serializer, dfa.getPropertyValue("serializer")); + assertSame(deserializer, dfa.getPropertyValue("deserializer")); assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); assertEquals(true, dfa.getPropertyValue("singleUse")); assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml index 77ba91ebe3..a708f3c058 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml @@ -7,10 +7,11 @@ http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd"> - - + + + @@ -27,8 +28,8 @@ diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml index efb160c822..81646ba7bd 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml @@ -11,13 +11,15 @@ - + + + @@ -28,8 +30,8 @@ port="#{server.port}" single-use="true" so-timeout="10000" - input-converter="serializer" - output-converter="serializer" + serializer="serializer" + deserializer="deserializer" /> diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java index d6ecbd36d1..01eb84e257 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java @@ -16,7 +16,6 @@ package org.springframework.integration.ip.tcp; -import static junit.framework.Assert.assertTrue; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.fail; diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java index 0df2bbc0e0..7afab7bd62 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java @@ -38,7 +38,8 @@ import javax.net.ServerSocketFactory; import org.junit.Test; -import org.springframework.commons.serializer.java.JavaStreamingConverter; +import org.springframework.commons.serializer.DefaultDeserializer; +import org.springframework.commons.serializer.DefaultSerializer; import org.springframework.integration.Message; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.channel.QueueChannel; @@ -50,7 +51,6 @@ import org.springframework.integration.support.MessageBuilder; /** * @author Gary Russell * @since 2.0 - * */ public class TcpOutboundGatewayTests { @@ -69,7 +69,7 @@ public class TcpOutboundGatewayTests { while (true) { Socket socket = server.accept(); ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); - Object in = ois.readObject(); + ois.readObject(); ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); oos.writeObject("Reply" + (i++)); } @@ -80,9 +80,8 @@ public class TcpOutboundGatewayTests { } } }); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(true); ccf.setPoolSize(10); @@ -135,9 +134,8 @@ public class TcpOutboundGatewayTests { } }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(false); ccf.start(); @@ -188,9 +186,8 @@ public class TcpOutboundGatewayTests { } }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.setSingleUse(false); ccf.start(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index 8f820c8e6d..21dc3faaa7 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; @@ -35,7 +36,9 @@ import java.util.concurrent.Executors; import javax.net.SocketFactory; import org.junit.Test; -import org.springframework.commons.serializer.java.JavaStreamingConverter; + +import org.springframework.commons.serializer.DefaultDeserializer; +import org.springframework.commons.serializer.DefaultSerializer; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; @@ -49,7 +52,6 @@ import org.springframework.integration.ip.util.SocketUtils; /** * @author Gary Russell - * */ public class TcpReceivingChannelAdapterTests { @@ -58,8 +60,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); scf.start(); @@ -88,8 +90,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); scf.setSoTimeout(5000); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); @@ -123,8 +125,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(scf); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); @@ -161,8 +163,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(scf); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); @@ -199,8 +201,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); scf.setSingleUse(true); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); @@ -235,8 +237,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); scf.setSingleUse(true); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); @@ -281,8 +283,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); scf.setSingleUse(true); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(scf); @@ -322,8 +324,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); scf.setSingleUse(true); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(scf); @@ -363,8 +365,8 @@ public class TcpReceivingChannelAdapterTests { final int port = SocketUtils.findAvailableServerSocket(); TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); scf.setSingleUse(true); scf.setPoolSize(100); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -444,11 +446,9 @@ public class TcpReceivingChannelAdapterTests { singleSharedInterceptorsGuts(port, scf); } - private void interceptorsGuts(final int port, - AbstractServerConnectionFactory scf) throws Exception { - JavaStreamingConverter converter = new JavaStreamingConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + private void interceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception { + scf.setSerializer(new DefaultSerializer()); + scf.setDeserializer(new DefaultDeserializer()); scf.setSingleUse(false); TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); adapter.setConnectionFactory(scf); @@ -487,11 +487,9 @@ public class TcpReceivingChannelAdapterTests { assertTrue(results.contains("Test2")); } - private void singleNoOutboundInterceptorsGuts(final int port, - AbstractServerConnectionFactory scf) throws Exception { - JavaStreamingConverter converter = new JavaStreamingConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + private void singleNoOutboundInterceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception { + scf.setSerializer(new DefaultSerializer()); + scf.setDeserializer(new DefaultDeserializer()); scf.setSingleUse(true); scf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); @@ -536,11 +534,9 @@ public class TcpReceivingChannelAdapterTests { assertTrue(results.contains("Test2")); } - private void singleSharedInterceptorsGuts(final int port, - AbstractServerConnectionFactory scf) throws Exception { - JavaStreamingConverter converter = new JavaStreamingConverter(); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + private void singleSharedInterceptorsGuts(final int port, AbstractServerConnectionFactory scf) throws Exception { + scf.setSerializer(new DefaultSerializer()); + scf.setDeserializer(new DefaultDeserializer()); scf.setSingleUse(true); scf.setSoTimeout(60000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); @@ -588,7 +584,5 @@ public class TcpReceivingChannelAdapterTests { assertEquals("Test1", new ObjectInputStream(socket1.getInputStream()).readObject()); assertEquals("Test2", new ObjectInputStream(socket2.getInputStream()).readObject()); } - - } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index 455afc9cec..a5ebd338a3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; @@ -38,7 +39,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import javax.net.ServerSocketFactory; import org.junit.Test; -import org.springframework.commons.serializer.java.JavaStreamingConverter; + +import org.springframework.commons.serializer.DefaultDeserializer; +import org.springframework.commons.serializer.DefaultSerializer; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; @@ -53,11 +56,9 @@ import org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter import org.springframework.integration.ip.util.SocketUtils; import org.springframework.integration.support.MessageBuilder; - /** * @author Gary Russell * @since 2.0 - * */ public class TcpSendingMessageHandlerTests { @@ -94,8 +95,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); handler.setConnectionFactory(ccf); @@ -142,8 +143,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -194,8 +195,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -243,8 +244,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -298,8 +299,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -350,8 +351,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -401,9 +402,8 @@ public class TcpSendingMessageHandlerTests { } }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -450,9 +450,8 @@ public class TcpSendingMessageHandlerTests { } }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); ccf.start(); TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); @@ -503,8 +502,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); ccf.setSingleUse(true); @@ -544,8 +543,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); ccf.setSingleUse(true); @@ -587,8 +586,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); ccf.setSingleUse(true); @@ -642,8 +641,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.start(); ccf.setSingleUse(true); @@ -697,8 +696,8 @@ public class TcpSendingMessageHandlerTests { }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.setSingleUse(true); ccf.setTaskExecutor(Executors.newFixedThreadPool(100)); @@ -771,9 +770,8 @@ public class TcpSendingMessageHandlerTests { } }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); fc.setInterceptors(new TcpConnectionInterceptorFactory[] @@ -834,9 +832,8 @@ public class TcpSendingMessageHandlerTests { } }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); fc.setInterceptors(new TcpConnectionInterceptorFactory[] {new HelloWorldInterceptorFactory()}); @@ -904,9 +901,8 @@ public class TcpSendingMessageHandlerTests { } }); AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); fc.setInterceptors(new TcpConnectionInterceptorFactory[] @@ -961,9 +957,8 @@ public class TcpSendingMessageHandlerTests { } }); AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(new DefaultSerializer()); + ccf.setDeserializer(new DefaultDeserializer()); ccf.setSoTimeout(10000); TcpConnectionInterceptorFactoryChain fc = new TcpConnectionInterceptorFactoryChain(); fc.setInterceptors(new TcpConnectionInterceptorFactory[] diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml index 2b2b539a7a..15eb1eeafb 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml @@ -15,119 +15,119 @@ - + + + serializer="crLfConverter" + deserializer="crLfConverter"/> + serializer="stxEtxConverter" + deserializer="stxEtxConverter"/> + serializer="lengthHeaderConverter" + deserializer="lengthHeaderConverter"/> + serializer="javaSerializer" + deserializer="javaDeserializer"/> + serializer="crLfConverter" + deserializer="crLfConverter"/> + serializer="stxEtxConverter" + deserializer="stxEtxConverter"/> + serializer="lengthHeaderConverter" + deserializer="lengthHeaderConverter"/> + serializer="javaSerializer" + deserializer="javaDeserializer"/> - diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index 3e4b926211..cddae3139b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.tcp.connection; import static org.junit.Assert.assertEquals; @@ -31,6 +32,7 @@ import java.util.concurrent.TimeUnit; import javax.net.SocketFactory; import org.junit.Test; + import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter; import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter; @@ -40,7 +42,7 @@ import org.springframework.integration.ip.util.SocketUtils; /** * @author Gary Russell - * + * @since 2.0 */ public class TcpNioConnectionReadTests { @@ -54,8 +56,8 @@ public class TcpNioConnectionReadTests { private AbstractServerConnectionFactory getConnectionFactory(int port, AbstractByteArrayStreamingConverter converter, TcpListener listener, TcpSender sender) throws Exception { AbstractServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); - scf.setInputConverter(converter); - scf.setOutputConverter(converter); + scf.setSerializer(converter); + scf.setDeserializer(converter); scf.registerListener(listener); if (sender != null) { scf.registerSender(sender); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java index ccbf97ef85..9bf22f7d46 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.tcp.connection; import static org.junit.Assert.assertEquals; @@ -26,6 +27,7 @@ import java.nio.ByteBuffer; import javax.net.ServerSocketFactory; import org.junit.Test; + import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter; import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter; import org.springframework.integration.ip.tcp.converter.ByteArrayLengthHeaderConverter; @@ -35,15 +37,15 @@ import org.springframework.integration.support.MessageBuilder; /** * @author Gary Russell - * + * @since 2.0 */ public class TcpNioConnectionWriteTests { private AbstractConnectionFactory getClientConnectionFactory(boolean direct, final int port, AbstractByteArrayStreamingConverter converter) { TcpNioClientConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); - ccf.setInputConverter(converter); - ccf.setOutputConverter(converter); + ccf.setSerializer(converter); + ccf.setDeserializer(converter); ccf.setSoTimeout(10000); ccf.setUsingDirectBuffers(direct); ccf.start(); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/InputConverterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java similarity index 87% rename from spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/InputConverterTests.java rename to spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java index edf3e42729..fa3abce0fd 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/InputConverterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.tcp.converter; import static org.junit.Assert.assertEquals; @@ -25,14 +26,15 @@ import java.net.Socket; import javax.net.ServerSocketFactory; import org.junit.Test; -import org.springframework.commons.serializer.java.JavaStreamingConverter; + +import org.springframework.commons.serializer.DefaultDeserializer; import org.springframework.integration.ip.util.SocketUtils; /** * @author Gary Russell - * + * @since 2.0 */ -public class InputConverterTests { +public class DeserializationTests { @Test public void testReadLength() throws Exception { @@ -43,10 +45,10 @@ public class InputConverterTests { Socket socket = server.accept(); socket.setSoTimeout(5000); ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter(); - byte[] out = converter.convert(socket.getInputStream()); + byte[] out = converter.deserialize(socket.getInputStream()); assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(out)); - out = converter.convert(socket.getInputStream()); + out = converter.deserialize(socket.getInputStream()); assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(out)); server.close(); @@ -61,10 +63,10 @@ public class InputConverterTests { Socket socket = server.accept(); socket.setSoTimeout(5000); ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter(); - byte[] out = converter.convert(socket.getInputStream()); + byte[] out = converter.deserialize(socket.getInputStream()); assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(out)); - out = converter.convert(socket.getInputStream()); + out = converter.deserialize(socket.getInputStream()); assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(out)); server.close(); @@ -79,10 +81,10 @@ public class InputConverterTests { Socket socket = server.accept(); socket.setSoTimeout(5000); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - byte[] out = converter.convert(socket.getInputStream()); + byte[] out = converter.deserialize(socket.getInputStream()); assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(out)); - out = converter.convert(socket.getInputStream()); + out = converter.deserialize(socket.getInputStream()); assertEquals("Data", SocketUtils.TEST_STRING + SocketUtils.TEST_STRING, new String(out)); server.close(); @@ -96,10 +98,10 @@ public class InputConverterTests { SocketUtils.testSendSerialized(port); Socket socket = server.accept(); socket.setSoTimeout(5000); - JavaStreamingConverter converter = new JavaStreamingConverter(); - Object out = converter.convert(socket.getInputStream()); + DefaultDeserializer deserializer = new DefaultDeserializer(); + Object out = deserializer.deserialize(socket.getInputStream()); assertEquals("Data", SocketUtils.TEST_STRING, out); - out = converter.convert(socket.getInputStream()); + out = deserializer.deserialize(socket.getInputStream()); assertEquals("Data", SocketUtils.TEST_STRING, out); server.close(); } @@ -114,7 +116,7 @@ public class InputConverterTests { socket.setSoTimeout(5000); ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter(); try { - converter.convert(socket.getInputStream()); + converter.deserialize(socket.getInputStream()); fail("Expected message length exceeded exception"); } catch (IOException e) { if (!e.getMessage().startsWith("Message length")) { @@ -135,7 +137,7 @@ public class InputConverterTests { socket.setSoTimeout(500); ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter(); try { - converter.convert(socket.getInputStream()); + converter.deserialize(socket.getInputStream()); fail("Expected timeout exception"); } catch (IOException e) { if (!e.getMessage().startsWith("Read timed out")) { @@ -157,7 +159,7 @@ public class InputConverterTests { ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter(); converter.setMaxMessageSize(1024); try { - converter.convert(socket.getInputStream()); + converter.deserialize(socket.getInputStream()); fail("Expected message length exceeded exception"); } catch (IOException e) { if (!e.getMessage().startsWith("ETX not found")) { @@ -178,7 +180,7 @@ public class InputConverterTests { socket.setSoTimeout(500); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); try { - converter.convert(socket.getInputStream()); + converter.deserialize(socket.getInputStream()); fail("Expected timout exception"); } catch (IOException e) { if (!e.getMessage().startsWith("Read timed out")) { @@ -200,7 +202,7 @@ public class InputConverterTests { ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); converter.setMaxMessageSize(1024); try { - converter.convert(socket.getInputStream()); + converter.deserialize(socket.getInputStream()); fail("Expected message length exceeded exception"); } catch (IOException e) { if (!e.getMessage().startsWith("CRLF not found")) { @@ -210,5 +212,5 @@ public class InputConverterTests { } server.close(); } - + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/OutputConverterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java similarity index 91% rename from spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/OutputConverterTests.java rename to spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java index 843709810a..e6ada7fe9b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/OutputConverterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.ip.tcp.converter; import static org.junit.Assert.assertEquals; @@ -28,14 +29,15 @@ import javax.net.ServerSocketFactory; import javax.net.SocketFactory; import org.junit.Test; -import org.springframework.commons.serializer.java.JavaStreamingConverter; + +import org.springframework.commons.serializer.DefaultSerializer; import org.springframework.integration.ip.util.SocketUtils; /** * @author Gary Russell - * + * @since 2.0 */ -public class OutputConverterTests { +public class SerializationTests { @Test public void testWriteLengthHeader() throws Exception { @@ -50,7 +52,7 @@ public class OutputConverterTests { ByteBuffer buffer = ByteBuffer.allocate(testString.length()); buffer.put(testString.getBytes()); ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter(); - converter.convert(buffer.array(), socket.getOutputStream()); + converter.serialize(buffer.array(), socket.getOutputStream()); Thread.sleep(1000000000L); } catch (Exception e) { e.printStackTrace(); @@ -83,7 +85,7 @@ public class OutputConverterTests { ByteBuffer buffer = ByteBuffer.allocate(testString.length()); buffer.put(testString.getBytes()); ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter(); - converter.convert(buffer.array(), socket.getOutputStream()); + converter.serialize(buffer.array(), socket.getOutputStream()); Thread.sleep(1000000000L); } catch (Exception e) { e.printStackTrace(); @@ -116,7 +118,7 @@ public class OutputConverterTests { ByteBuffer buffer = ByteBuffer.allocate(testString.length()); buffer.put(testString.getBytes()); ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); - converter.convert(buffer.array(), socket.getOutputStream()); + converter.serialize(buffer.array(), socket.getOutputStream()); Thread.sleep(1000000000L); } catch (Exception e) { e.printStackTrace(); @@ -146,9 +148,9 @@ public class OutputConverterTests { public void run() { try { Socket socket = SocketFactory.getDefault().createSocket("localhost", port); - JavaStreamingConverter converter = new JavaStreamingConverter(); - converter.convert(testString, socket.getOutputStream()); - converter.convert(testString, socket.getOutputStream()); + DefaultSerializer serializer = new DefaultSerializer(); + serializer.serialize(testString, socket.getOutputStream()); + serializer.serialize(testString, socket.getOutputStream()); Thread.sleep(1000000000L); } catch (Exception e) { e.printStackTrace(); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index b34dee8763..6ef9206afe 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -13,13 +13,24 @@ package org.springframework.integration.jdbc; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Timestamp; +import java.sql.Types; +import java.util.Iterator; +import java.util.List; +import java.util.UUID; + +import javax.sql.DataSource; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + +import org.springframework.commons.serializer.Deserializer; import org.springframework.commons.serializer.DeserializingConverter; -import org.springframework.commons.serializer.InputStreamingConverter; -import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.commons.serializer.Serializer; import org.springframework.commons.serializer.SerializingConverter; -import org.springframework.commons.serializer.java.JavaStreamingConverter; import org.springframework.integration.Message; import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; @@ -27,18 +38,16 @@ import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.util.UUIDConverter; -import org.springframework.jdbc.core.*; +import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.PreparedStatementSetter; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.core.SingleColumnRowMapper; import org.springframework.jdbc.support.lob.DefaultLobHandler; import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.util.Assert; import org.springframework.util.StringUtils; -import javax.sql.DataSource; -import java.sql.*; -import java.util.Iterator; -import java.util.List; -import java.util.UUID; - /** * Implementation of {@link MessageStore} using a relational database via JDBC. SQL scripts to create the necessary * tables are packaged as org/springframework/integration/jdbc/schema-*.sql, where * is the @@ -113,9 +122,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa * Convenient constructor for configuration use. */ public JdbcMessageStore() { - JavaStreamingConverter converter = new JavaStreamingConverter(); - deserializer = new DeserializingConverter(converter); - serializer = new SerializingConverter(converter); + deserializer = new DeserializingConverter(); + serializer = new SerializingConverter(); } /** @@ -194,8 +202,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa * @param serializer the serializer to set */ @SuppressWarnings("unchecked") - public void setSerializer(OutputStreamingConverter> serializer) { - this.serializer = new SerializingConverter((OutputStreamingConverter) serializer); + public void setSerializer(Serializer/*>*/ serializer) { + this.serializer = new SerializingConverter(serializer); } /** @@ -204,8 +212,8 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa * @param deserializer the deserializer to set */ @SuppressWarnings("unchecked") - public void setDeserializer(InputStreamingConverter> deserializer) { - this.deserializer = new DeserializingConverter((InputStreamingConverter) deserializer); + public void setDeserializer(Deserializer/*>*/ deserializer) { + this.deserializer = new DeserializingConverter(deserializer); } /** diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index 9eca461a5d..0aab2a4e2a 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -37,9 +37,10 @@ import javax.sql.DataSource; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.commons.serializer.InputStreamingConverter; -import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.commons.serializer.Deserializer; +import org.springframework.commons.serializer.Serializer; import org.springframework.integration.Message; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.store.MessageGroup; @@ -88,14 +89,14 @@ public class JdbcMessageStoreTests { @Transactional public void testSerializer() throws Exception { // N.B. these serializers are not realistic (just for test purposes) - messageStore.setSerializer(new OutputStreamingConverter>() { - public void convert(Message object, OutputStream outputStream) throws IOException { - outputStream.write(object.getPayload().toString().getBytes()); + messageStore.setSerializer(new Serializer/*>>*/() { + public void serialize(/*Message*/ Object object, OutputStream outputStream) throws IOException { + outputStream.write(((Message) object).getPayload().toString().getBytes()); outputStream.flush(); } }); - messageStore.setDeserializer(new InputStreamingConverter>() { - public Message convert(InputStream inputStream) throws IOException { + messageStore.setDeserializer(new Deserializer/*>*/() { + public Message deserialize(InputStream inputStream) throws IOException { BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream)); return new GenericMessage(reader.readLine()); } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java index e63637ef7c..e683f87892 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java @@ -9,7 +9,11 @@ import java.io.OutputStream; import org.junit.After; import org.junit.Test; -import org.springframework.commons.serializer.java.JavaStreamingConverter; + +import org.springframework.commons.serializer.DefaultDeserializer; +import org.springframework.commons.serializer.DefaultSerializer; +import org.springframework.commons.serializer.Deserializer; +import org.springframework.commons.serializer.Serializer; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; import org.springframework.integration.jdbc.JdbcMessageStore; @@ -41,9 +45,9 @@ public class JdbcMessageStoreParserTests { public void testSimpleMessageStoreWithSerializer() { setUp("serializerJdbcMessageStore.xml", getClass()); MessageStore store = context.getBean("messageStore", MessageStore.class); - Object serializer = TestUtils.getPropertyValue(store, "serializer.streamingConverter"); + Object serializer = TestUtils.getPropertyValue(store, "serializer.serializer"); assertTrue(serializer instanceof EnhancedSerializer); - Object deserializer = TestUtils.getPropertyValue(store, "deserializer.streamingConverter"); + Object deserializer = TestUtils.getPropertyValue(store, "deserializer.deserializer"); assertTrue(deserializer instanceof EnhancedSerializer); } @@ -67,17 +71,22 @@ public class JdbcMessageStoreParserTests { context = new ClassPathXmlApplicationContext(name, cls); } - public static class EnhancedSerializer extends JavaStreamingConverter { - @Override - public Object convert(InputStream inputStream) throws IOException { - Message message = (Message) super.convert(inputStream); + + public static class EnhancedSerializer implements Serializer, Deserializer { + + private final Serializer targetSerializer = new DefaultSerializer(); + + private final Deserializer targetDeserializer = new DefaultDeserializer(); + + public Object deserialize(InputStream inputStream) throws IOException { + Message message = (Message) targetDeserializer.deserialize(inputStream); return message; } - @Override - public void convert(Object object, OutputStream outputStream) throws IOException { + public void serialize(Object object, OutputStream outputStream) throws IOException { Message message = (Message) object; - super.convert(MessageBuilder.fromMessage(message).setHeader("serializer", "CUSTOM").build(), outputStream); + message = MessageBuilder.fromMessage(message).setHeader("serializer", "CUSTOM").build(); + targetSerializer.serialize(message, outputStream); } } diff --git a/spring-integration-parent/pom.xml b/spring-integration-parent/pom.xml index 137d3c1f99..96ddbeb988 100644 --- a/spring-integration-parent/pom.xml +++ b/spring-integration-parent/pom.xml @@ -209,7 +209,7 @@ org.springframework.commons spring-commons-serializer - 1.0.0.M1 + 1.0.0.BUILD-SNAPSHOT org.springframework.security