diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java index e961579d7e..1319f11902 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java @@ -34,6 +34,10 @@ import org.springframework.integration.support.MessageBuilderFactory; import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.MessageHeaders; +import org.springframework.util.Assert; +import org.springframework.util.InvalidMimeTypeException; +import org.springframework.util.MimeType; /** * Maps incoming data from a {@link TcpConnection} to a {@link Message}. @@ -68,9 +72,14 @@ public class TcpMessageMapper implements private volatile boolean messageBuilderFactorySet; + private volatile String contentType = "application/octet-stream;charset=" + this.charset; + + private volatile boolean addContentTypeHeader; + private BeanFactory beanFactory; /** + * Set the charset to use when converting outbound String messages to {@code byte[]}. * @param charset the charset to set */ public void setCharset(String charset) { @@ -93,6 +102,37 @@ public class TcpMessageMapper implements this.applySequence = applySequence; } + /** + * Set the content type header value to add to inbound messages when + * {@link #setAddContentTypeHeader(boolean) addContentTypeHeader} is true. + * Default {@code application/octet-stream;charset=UTF-8}. This default is not + * modified by {@link #setCharset(String)}. + * @param contentType the content type header value to set. + * @since 4.3 + * @see #setAddContentTypeHeader(boolean) + * @see TcpMessageMapper#setCharset(String) + */ + public void setContentType(String contentType) { + Assert.notNull(contentType, "'contentType' cannot be null"); + try { + MimeType.valueOf(contentType); + } + catch (InvalidMimeTypeException e) { + throw new IllegalArgumentException("'contentType' could not be parsed", e); + } + this.contentType = contentType; + } + + /** + * Set to true to add a content type header; default false. + * @param addContentTypeHeader true to add a content type header. + * @since 4.3 + * @see #setContentType(String) + */ + public void setAddContentTypeHeader(boolean addContentTypeHeader) { + this.addContentTypeHeader = addContentTypeHeader; + } + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { this.beanFactory = beanFactory; @@ -126,7 +166,8 @@ public class TcpMessageMapper implements return message; } - protected final void addStandardHeaders(TcpConnection connection, AbstractIntegrationMessageBuilder messageBuilder) { + protected final void addStandardHeaders(TcpConnection connection, + AbstractIntegrationMessageBuilder messageBuilder) { String connectionId = connection.getConnectionId(); messageBuilder .setHeader(IpHeaders.HOSTNAME, connection.getHostName()) @@ -142,9 +183,13 @@ public class TcpMessageMapper implements .setCorrelationId(connectionId) .setSequenceNumber((int) connection.incrementAndGetConnectionSequence()); } + if (this.addContentTypeHeader) { + messageBuilder.setHeader(MessageHeaders.CONTENT_TYPE, this.contentType); + } } - protected final void addCustomHeaders(TcpConnection connection, AbstractIntegrationMessageBuilder messageBuilder) { + protected final void addCustomHeaders(TcpConnection connection, + AbstractIntegrationMessageBuilder messageBuilder) { Map customHeaders = this.supplyCustomHeaders(connection); if (customHeaders != null) { messageBuilder.copyHeadersIfAbsent(customHeaders); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java index 4ad6c6da6a..5a9d97380b 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java @@ -16,9 +16,12 @@ package org.springframework.integration.ip.tcp.connection; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -48,7 +51,9 @@ import org.springframework.integration.ip.tcp.serializer.MapJsonSerializer; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.converter.MapMessageConverter; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.MessageConverter; +import org.springframework.util.MimeType; /** * @author Gary Russell @@ -69,7 +74,6 @@ public class TcpMessageMapperTests { @Test public void testToMessage() throws Exception { - TcpMessageMapper mapper = new TcpMessageMapper(); TcpConnection connection = mock(TcpConnection.class); Socket socket = mock(Socket.class); @@ -87,11 +91,75 @@ public class TcpMessageMapperTests { assertEquals("1.1.1.1", message.getHeaders().get(IpHeaders.IP_ADDRESS)); assertEquals(1234, message.getHeaders().get(IpHeaders.REMOTE_PORT)); assertSame(local, message.getHeaders().get(IpHeaders.LOCAL_ADDRESS)); + assertNull(message.getHeaders().get(MessageHeaders.CONTENT_TYPE)); + } + + @Test + public void testToMessageWithContentType() throws Exception { + TcpMessageMapper mapper = new TcpMessageMapper(); + mapper.setAddContentTypeHeader(true); + TcpConnection connection = mock(TcpConnection.class); + Socket socket = mock(Socket.class); + InetAddress local = mock(InetAddress.class); + SocketInfo info = new SocketInfo(socket); + when(socket.getLocalAddress()).thenReturn(local); + when(connection.getPayload()).thenReturn(TEST_PAYLOAD.getBytes()); + when(connection.getHostName()).thenReturn("MyHost"); + when(connection.getHostAddress()).thenReturn("1.1.1.1"); + when(connection.getPort()).thenReturn(1234); + when(connection.getSocketInfo()).thenReturn(info); + Message message = mapper.toMessage(connection); + assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload())); + assertEquals("MyHost", message.getHeaders().get(IpHeaders.HOSTNAME)); + assertEquals("1.1.1.1", message.getHeaders().get(IpHeaders.IP_ADDRESS)); + assertEquals(1234, message.getHeaders().get(IpHeaders.REMOTE_PORT)); + assertSame(local, message.getHeaders().get(IpHeaders.LOCAL_ADDRESS)); + assertEquals("application/octet-stream;charset=UTF-8", message.getHeaders().get(MessageHeaders.CONTENT_TYPE)); + MimeType parseOk = MimeType.valueOf((String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE)); + assertEquals(message.getHeaders().get(MessageHeaders.CONTENT_TYPE), parseOk.toString()); + } + + @Test + public void testToMessageWithCustomContentType() throws Exception { + TcpMessageMapper mapper = new TcpMessageMapper(); + mapper.setAddContentTypeHeader(true); + mapper.setContentType("application/octet-stream;charset=ISO-8859-1"); + TcpConnection connection = mock(TcpConnection.class); + Socket socket = mock(Socket.class); + InetAddress local = mock(InetAddress.class); + SocketInfo info = new SocketInfo(socket); + when(socket.getLocalAddress()).thenReturn(local); + when(connection.getPayload()).thenReturn(TEST_PAYLOAD.getBytes()); + when(connection.getHostName()).thenReturn("MyHost"); + when(connection.getHostAddress()).thenReturn("1.1.1.1"); + when(connection.getPort()).thenReturn(1234); + when(connection.getSocketInfo()).thenReturn(info); + Message message = mapper.toMessage(connection); + assertEquals(TEST_PAYLOAD, new String((byte[]) message.getPayload())); + assertEquals("MyHost", message.getHeaders().get(IpHeaders.HOSTNAME)); + assertEquals("1.1.1.1", message.getHeaders().get(IpHeaders.IP_ADDRESS)); + assertEquals(1234, message.getHeaders().get(IpHeaders.REMOTE_PORT)); + assertSame(local, message.getHeaders().get(IpHeaders.LOCAL_ADDRESS)); + assertEquals("application/octet-stream;charset=ISO-8859-1", message.getHeaders().get(MessageHeaders.CONTENT_TYPE)); + MimeType parseOk = MimeType.valueOf((String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE)); + assertEquals(message.getHeaders().get(MessageHeaders.CONTENT_TYPE), parseOk.toString()); + } + + @Test(expected = IllegalArgumentException.class) + public void testToMessageWithBadContentType() throws Exception { + TcpMessageMapper mapper = new TcpMessageMapper(); + mapper.setAddContentTypeHeader(true); + try { + mapper.setContentType(""); + } + catch (IllegalArgumentException e) { + assertThat(e.getMessage(), containsString("'contentType' could not be parsed")); + throw e; + } } @Test public void testToMessageSequence() throws Exception { - TcpMessageMapper mapper = new TcpMessageMapper(); Socket socket = SocketFactory.getDefault().createSocket(); TcpConnection connection = new TcpConnectionSupport(socket, false, false, null, null) { diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index f645d26890..ce3175a451 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/src/reference/asciidoc/ip.adoc @@ -1456,15 +1456,25 @@ The framework includes acknowledgment information in the data packet. | ip_actualConnectionId | ACTUAL_ CONNECTION_ID | For information only - when using a cached or failover client connection factory, contains the actual underlying connection id. +| contentType +| MessageHeaders.CONTENT_TYPE +| An optional content type for inbound messages; see below. |=== For inbound messages, `ip_hostname`, `ip_address`, `ip_tcp_remotePort` and `ip_connectionId` are mapped by the default `TcpHeaderMapper`. -Users can add additional headers by subclassing `TcpHeaderMapper`, overriding the method `supplyCustomHeaders`, and -providing an instance to the connection factory using the `mapper` property. +Set the mapper's `addContentTypeHeader` property to `true` and the mapper will set the `contentType` header (`application/octet-stream;charset="UTF-8"`) by default. +You can change the default by setting the `contentType` property. +Users can add additional headers by subclassing `TcpHeaderMapper` and overriding the method `supplyCustomHeaders`. For example, when using SSL, properties of the `SSLSession` can be added by obtaining the session object from the `TcpConnection` object which is provided as an argument to the `supplyCustomHeaders` method. +For outbound messages, `String` payloads are converted to `byte[]` using the default (`UTF-8`) charset. +Set the `charset` property to change the default. + +When customizing the mapper properties, or subclassing, declare the mapper as a bean and provide an instance to the connection factory using the `mapper` property + + [[ip-annotation]] === Annotation-Based Configuration diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e316643ffb..d1332fa71d 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -120,6 +120,11 @@ pooling of the raw buffer into which the data is received, rather than creating message. See <> for more information. +===== TCP Message Mapper + +The message mapper now, optionally, sets a configured content type header. +See <> for more information. + ==== File Changes ===== Destination Directory Creation