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

Merge branch 'int1486'

Browse Source
This commit is contained in:
Mark Fisher
2010-09-28 16:15:28 -04:00
parent d59b40ace7 729a37fb62
commit 43e25630fd
42 changed files with 399 additions and 386 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParser.java
View File

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

2
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParser.java
View File

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

10
spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadDeserializingTransformer.java
View File

@@ -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<byte[], Object> {
@Override
public void setConverter(Converter<byte[], Object> converter) {
this.converter = converter;
public void setDeserializer(Deserializer<Object> 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);
}

10
spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadSerializingTransformer.java
View File

@@ -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<Object, byte[]> {
@Override
public void setConverter(Converter<Object, byte[]> converter) {
this.converter = converter;
public void setSerializer(Serializer<Object> 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);
}

18
spring-integration-core/src/main/java/org/springframework/integration/transformer/PayloadTypeConvertingTransformer.java
View File

@@ -25,20 +25,13 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
* @since 2.0
*
*/
public class PayloadTypeConvertingTransformer<T, U> extends AbstractPayloadTransformer<T, U> {
protected Converter<T, U> converter;
@Override
protected U transformPayload(T payload) throws Exception {
Assert.notNull(this.converter, this.getClass().getName() + " needs a Converter<Object, Object>");
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<T, U> extends AbstractPayloadTrans
this.converter = converter;
}
@Override
protected U transformPayload(T payload) throws Exception {
Assert.notNull(this.converter, this.getClass().getName() + " requires a Converter<Object, Object>");
return this.converter.convert(payload);
}
}

16
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd
View File

@@ -1650,15 +1650,15 @@
<xsd:choice minOccurs="0" maxOccurs="unbounded">
<xsd:element ref="poller" />
</xsd:choice>
<xsd:attribute name="converter" use="optional">
<xsd:attribute name="serializer" use="optional">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.core.convert.converter.Converter" />
<tool:expected-type type="org.springframework.commons.serializer.Serializer" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -1684,15 +1684,15 @@
<xsd:choice minOccurs="0" maxOccurs="unbounded">
<xsd:element ref="poller" />
</xsd:choice>
<xsd:attribute name="converter" use="optional">
<xsd:attribute name="deserializer" use="optional">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.core.convert.converter.Converter" />
<tool:expected-type type="org.springframework.commons.serializer.Deserializer" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>

6
spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests-context.xml
View File

@@ -13,7 +13,7 @@
<queue capacity="1"/>
</channel>
<channel id="customConverterInput"/>
<channel id="customDeserializerInput"/>
<channel id="output">
<queue capacity="1"/>
@@ -25,8 +25,8 @@
<poller fixed-delay="10000"/>
</payload-deserializing-transformer>
<payload-deserializing-transformer input-channel="customConverterInput" output-channel="output" converter="customConverter"/>
<payload-deserializing-transformer input-channel="customDeserializerInput" output-channel="output" deserializer="customDeserializer"/>
<beans:bean id="customConverter" class="org.springframework.integration.config.xml.PayloadDeserializingTransformerParserTests$TestDeserializingConverter"/>
<beans:bean id="customDeserializer" class="org.springframework.integration.config.xml.PayloadDeserializingTransformerParserTests$TestDeserializer"/>
</beans:beans>

24
spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadDeserializingTransformerParserTests.java
View File

@@ -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<byte[]>("test".getBytes("UTF-8")));
public void customDeserializer() throws Exception {
customDeserializerInput.send(new GenericMessage<byte[]>("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<byte[], Object> {
public static class TestDeserializer implements Deserializer<Object> {
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();
}
}

6
spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests-context.xml
View File

@@ -13,7 +13,7 @@
<queue capacity="1"/>
</channel>
<channel id="customConverterInput"/>
<channel id="customSerializerInput"/>
<channel id="output">
<queue capacity="1"/>
@@ -25,8 +25,8 @@
<poller fixed-delay="10000"/>
</payload-serializing-transformer>
<payload-serializing-transformer input-channel="customConverterInput" output-channel="output" converter="customConverter"/>
<payload-serializing-transformer input-channel="customSerializerInput" output-channel="output" serializer="customSerializer"/>
<beans:bean id="customConverter" class="org.springframework.integration.config.xml.PayloadSerializingTransformerParserTests$TestSerializingConverter"/>
<beans:bean id="customSerializer" class="org.springframework.integration.config.xml.PayloadSerializingTransformerParserTests$TestSerializer"/>
</beans:beans>

24
spring-integration-core/src/test/java/org/springframework/integration/config/xml/PayloadSerializingTransformerParserTests.java
View File

@@ -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<String>("test"));
public void customSerializer() throws Exception {
customSerializerInput.send(new GenericMessage<String>("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<Object, byte[]> {
public static class TestSerializer implements Serializer<Object> {
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();
}
}

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java
View File

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

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java
View File

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

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java
View File

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

21
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -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;
}
/**

17
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java
View File

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

41
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java
View File

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

24
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnectionInterceptor.java
View File

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

20
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java
View File

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

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java
View File

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

4
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

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

9
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java
View File

@@ -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<byte[]>,
OutputStreamingConverter<byte[]> {
Serializer<byte[]>,
Deserializer<byte[]> {
protected int maxMessageSize = 2048;

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java
View File

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

8
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java
View File

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

5
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java
View File

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

20
spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd
View File

@@ -290,31 +290,31 @@ the factory, the connection will be closed after a response is received.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="input-converter" type="xsd:string" >
<xsd:attribute name="serializer" type="xsd:string" >
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.commons.serializer.InputStreamingConverter"/>
<tool:expected-type type="org.springframework.commons.serializer.Serializer"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="output-converter" type="xsd:string" >
<xsd:attribute name="deserializer" type="xsd:string" >
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.commons.serializer.OutputStreamingConverter"/>
<tool:expected-type type="org.springframework.commons.serializer.Deserializer"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

20
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml
View File

@@ -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 @@
<bean id="interceptors" class="org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain" />
<bean id="serial" class="org.springframework.commons.serializer.java.JavaStreamingConverter" />
<bean id="defaultSerializer" class="org.springframework.commons.serializer.DefaultSerializer" />
<bean id="defaultDeserializer" class="org.springframework.commons.serializer.DefaultDeserializer" />
<ip:tcp-outbound-channel-adapter id="tcpNewOut1"
channel="tcpChannel"
connection-factory="client1" />

29
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

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

13
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/InterceptedSharedConnectionTests-context.xml
View File

@@ -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">
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
<bean id="serializer" class="org.springframework.commons.serializer.java.JavaStreamingConverter" />
<bean id="serializer" class="org.springframework.commons.serializer.DefaultSerializer" />
<bean id="deserializer" class="org.springframework.commons.serializer.DefaultDeserializer" />
<bean id="helloWorldInterceptors" class="org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain">
<property name="interceptors">
@@ -27,8 +28,8 @@
<int-ip:tcp-connection-factory id="server"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(10100)}"
input-converter="serializer"
output-converter="serializer"
serializer="serializer"
deserializer="deserializer"
using-nio="true"
single-use="true"
interceptor-factory-chain="helloWorldInterceptors"
@@ -41,8 +42,8 @@
single-use="true"
so-timeout="100000"
using-nio="true"
input-converter="serializer"
output-converter="serializer"
serializer="serializer"
deserializer="deserializer"
interceptor-factory-chain="helloWorldInterceptors"
/>

12
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml
View File

@@ -11,13 +11,15 @@
<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketUtils" />
<bean id="serializer" class="org.springframework.commons.serializer.java.JavaStreamingConverter" />
<bean id="serializer" class="org.springframework.commons.serializer.DefaultSerializer" />
<bean id="deserializer" class="org.springframework.commons.serializer.DefaultDeserializer" />
<int-ip:tcp-connection-factory id="server"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(10200)}"
input-converter="serializer"
output-converter="serializer"
serializer="serializer"
deserializer="deserializer"
using-nio="true"
single-use="true"
/>
@@ -28,8 +30,8 @@
port="#{server.port}"
single-use="true"
so-timeout="10000"
input-converter="serializer"
output-converter="serializer"
serializer="serializer"
deserializer="deserializer"
/>
<int:channel id="input" />

1
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java
View File

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

21
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpOutboundGatewayTests.java
View File

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

68
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java
View File

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

81
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -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[]

68
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/common-context.xml
View File

@@ -15,119 +15,119 @@
<bean id="crLfConverter" class="org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter" />
<bean id="stxEtxConverter" class="org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter" />
<bean id="lengthHeaderConverter" class="org.springframework.integration.ip.tcp.converter.ByteArrayLengthHeaderConverter" />
<bean id="javaConverter" class="org.springframework.commons.serializer.java.JavaStreamingConverter" />
<bean id="javaSerializer" class="org.springframework.commons.serializer.DefaultSerializer" />
<bean id="javaDeserializer" class="org.springframework.commons.serializer.DefaultDeserializer" />
<ip:tcp-connection-factory id="crLfServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7000)}"
input-converter="crLfConverter"
output-converter="crLfConverter"/>
serializer="crLfConverter"
deserializer="crLfConverter"/>
<ip:tcp-connection-factory id="stxEtxServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7100)}"
input-converter="stxEtxConverter"
output-converter="stxEtxConverter"/>
serializer="stxEtxConverter"
deserializer="stxEtxConverter"/>
<ip:tcp-connection-factory id="lengthHeaderServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7200)}"
input-converter="lengthHeaderConverter"
output-converter="lengthHeaderConverter"/>
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"/>
<ip:tcp-connection-factory id="javaSerialServer"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7300)}"
input-converter="javaConverter"
output-converter="javaConverter"/>
serializer="javaSerializer"
deserializer="javaDeserializer"/>
<ip:tcp-connection-factory id="crLfClient"
type="client"
host="localhost"
port="#{crLfServer.port}"
input-converter="crLfConverter"
output-converter="crLfConverter"/>
serializer="crLfConverter"
deserializer="crLfConverter"/>
<ip:tcp-connection-factory id="stxEtxClient"
type="client"
host="localhost"
port="#{stxEtxServer.port}"
input-converter="stxEtxConverter"
output-converter="stxEtxConverter"/>
serializer="stxEtxConverter"
deserializer="stxEtxConverter"/>
<ip:tcp-connection-factory id="lengthHeaderClient"
type="client"
host="localhost"
port="#{lengthHeaderServer.port}"
input-converter="lengthHeaderConverter"
output-converter="lengthHeaderConverter"/>
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"/>
<ip:tcp-connection-factory id="javaSerialClient"
type="client"
host="localhost"
port="#{javaSerialServer.port}"
input-converter="javaConverter"
output-converter="javaConverter"/>
serializer="javaSerializer"
deserializer="javaDeserializer"/>
<ip:tcp-connection-factory id="crLfServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7400)}"
input-converter="crLfConverter"
output-converter="crLfConverter"
serializer="crLfConverter"
deserializer="crLfConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="stxEtxServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7500)}"
input-converter="stxEtxConverter"
output-converter="stxEtxConverter"
serializer="stxEtxConverter"
deserializer="stxEtxConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="lengthHeaderServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7600)}"
input-converter="lengthHeaderConverter"
output-converter="lengthHeaderConverter"
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="javaSerialServerNio"
type="server"
port="#{tcpIpUtils.findAvailableServerSocket(7700)}"
input-converter="javaConverter"
output-converter="javaConverter"
serializer="javaSerializer"
deserializer="javaDeserializer"
using-nio="true"/>
<ip:tcp-connection-factory id="crLfClientNio"
type="client"
host="localhost"
port="#{crLfServer.port}"
input-converter="crLfConverter"
output-converter="crLfConverter"
serializer="crLfConverter"
deserializer="crLfConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="stxEtxClientNio"
type="client"
host="localhost"
port="#{stxEtxServer.port}"
input-converter="stxEtxConverter"
output-converter="stxEtxConverter"
serializer="stxEtxConverter"
deserializer="stxEtxConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="lengthHeaderClientNio"
type="client"
host="localhost"
port="#{lengthHeaderServer.port}"
input-converter="lengthHeaderConverter"
output-converter="lengthHeaderConverter"
serializer="lengthHeaderConverter"
deserializer="lengthHeaderConverter"
using-nio="true"/>
<ip:tcp-connection-factory id="javaSerialClientNio"
type="client"
host="localhost"
port="#{javaSerialServer.port}"
input-converter="javaConverter"
output-converter="javaConverter"
serializer="javaSerializer"
deserializer="javaDeserializer"
using-nio="true"/>
</beans>

8
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

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

8
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionWriteTests.java
View File

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

38
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/InputConverterTests.java → spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/DeserializationTests.java
View File

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

20
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/OutputConverterTests.java → spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/converter/SerializationTests.java
View File

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

42
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -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 <code>org/springframework/integration/jdbc/schema-*.sql</code>, where <code>*</code> 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<? super Message<?>> serializer) {
this.serializer = new SerializingConverter((OutputStreamingConverter<Object>) serializer);
public void setSerializer(Serializer/*<? super Message<?>>*/ 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<? super Message<?>> deserializer) {
this.deserializer = new DeserializingConverter((InputStreamingConverter<Object>) deserializer);
public void setDeserializer(Deserializer/*<? super Message<?>>*/ deserializer) {
this.deserializer = new DeserializingConverter(deserializer);
}
/**

15
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -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<Message<?>>() {
public void convert(Message<?> object, OutputStream outputStream) throws IOException {
outputStream.write(object.getPayload().toString().getBytes());
messageStore.setSerializer(new Serializer/*<Message<?>><Message<?>>*/() {
public void serialize(/*Message<?>*/ Object object, OutputStream outputStream) throws IOException {
outputStream.write(((Message<?>) object).getPayload().toString().getBytes());
outputStream.flush();
}
});
messageStore.setDeserializer(new InputStreamingConverter<Message<?>>() {
public Message<?> convert(InputStream inputStream) throws IOException {
messageStore.setDeserializer(new Deserializer/*<Message<?>>*/() {
public Message<?> deserialize(InputStream inputStream) throws IOException {
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
return new GenericMessage<String>(reader.readLine());
}

29
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java
View File

@@ -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<Object>, Deserializer<Object> {
private final Serializer<Object> targetSerializer = new DefaultSerializer();
private final Deserializer<Object> 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);
}
}

2
spring-integration-parent/pom.xml
View File

@@ -209,7 +209,7 @@
<dependency>
<groupId>org.springframework.commons</groupId>
<artifactId>spring-commons-serializer</artifactId>
<version>1.0.0.M1</version>
<version>1.0.0.BUILD-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.security</groupId>