INT-3952: Local IP Address Header and SocketInfo

JIRA: https://jira.spring.io/browse/INT-3952

spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java

@@ -84,6 +84,12 @@ public final class IpHeaders {
 	 */
 	public static final String ACTUAL_CONNECTION_ID = IP + "actualConnectionId";
 
+	/**
+	 * The local address (InetAddress) that the socket is connected to.
+	 * @since 4.2.5.
+	 */
+	public static final String LOCAL_ADDRESS = IP + "localInetAddress";
+
 	private IpHeaders() {}
 
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -306,6 +306,11 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
 			return this.connectionId + ":" + epoch;
 		}
 
+		@Override
+		public SocketInfo getSocketInfo() {
+			return this.delegate.getSocketInfo();
+		}
+
 		@Override
 		public boolean isServer() {
 			return this.delegate.isServer();

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/SocketInfo.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.ip.tcp.connection;
+
+import java.net.InetAddress;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.nio.channels.SocketChannel;
+
+import org.springframework.util.Assert;
+
+/**
+ * Simple wrapper around {@link Socket} providing access to getters (except
+ * input/output streams).
+ *
+ * @author Gary Russell
+ * @since 4.2.5
+ *
+ */
+public class SocketInfo {
+
+	private final Socket socket;
+
+	public SocketInfo(Socket socket) {
+		Assert.notNull(socket, "'socket' cannot be null");
+		this.socket = socket;
+	}
+
+	public InetAddress getInetAddress() {
+		return this.socket.getInetAddress();
+	}
+
+	public InetAddress getLocalAddress() {
+		return this.socket.getLocalAddress();
+	}
+
+	public int getPort() {
+		return this.socket.getPort();
+	}
+
+	public int getLocalPort() {
+		return this.socket.getLocalPort();
+	}
+
+	public SocketAddress getRemoteSocketAddress() {
+		return this.socket.getRemoteSocketAddress();
+	}
+
+	public SocketAddress getLocalSocketAddress() {
+		return this.socket.getLocalSocketAddress();
+	}
+
+	public SocketChannel getChannel() {
+		return this.socket.getChannel();
+	}
+
+	public boolean getTcpNoDelay() throws SocketException {
+		return this.socket.getTcpNoDelay();
+	}
+
+	public int getSoLinger() throws SocketException {
+		return this.socket.getSoLinger();
+	}
+
+	public boolean getOOBInline() throws SocketException {
+		return this.socket.getOOBInline();
+	}
+
+	public int getSoTimeout() throws SocketException {
+		return this.socket.getSoTimeout();
+	}
+
+	public int getSendBufferSize() throws SocketException {
+		return this.socket.getSendBufferSize();
+	}
+
+	public int getReceiveBufferSize() throws SocketException {
+		return this.socket.getReceiveBufferSize();
+	}
+
+	public boolean getKeepAlive() throws SocketException {
+		return this.socket.getKeepAlive();
+	}
+
+	public int getTrafficClass() throws SocketException {
+		return this.socket.getTrafficClass();
+	}
+
+	public boolean getReuseAddress() throws SocketException {
+		return this.socket.getReuseAddress();
+	}
+
+	@Override
+	public String toString() {
+		return this.socket.toString();
+	}
+
+	public boolean isConnected() {
+		return this.socket.isConnected();
+	}
+
+	public boolean isClosed() {
+		return this.socket.isClosed();
+	}
+
+	public boolean isInputShutdown() {
+		return this.socket.isInputShutdown();
+	}
+
+	public boolean isOutputShutdown() {
+		return this.socket.isOutputShutdown();
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2015 the original author or authors.
+ * Copyright 2001-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -125,4 +125,12 @@ public interface TcpConnection extends Runnable {
 	 */
 	SSLSession getSslSession();
 
+	/**
+	 * Provides getters for {@link Socket} properties.
+	 * @return the socketInfo - may be null, for example in interceptors; interceptors
+	 * should override and delegate to the actual TcpConnection.
+	 * @since 4.3
+	 */
+	SocketInfo getSocketInfo();
+
 }

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -101,6 +101,11 @@ public abstract class TcpConnectionInterceptorSupport extends TcpConnectionSuppo
 		return this.theConnection.getConnectionId();
 	}
 
+	@Override
+	public SocketInfo getSocketInfo() {
+		return this.theConnection.getSocketInfo();
+	}
+
 	@Override
 	public void run() {
 		this.theConnection.run();

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2015 the original author or authors.
+ * Copyright 2001-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -54,6 +54,18 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 
 	private final CountDownLatch listenerRegisteredLatch = new CountDownLatch(1);
 
+	private final boolean server;
+
+	private final AtomicLong sequence = new AtomicLong();
+
+	private final ApplicationEventPublisher applicationEventPublisher;
+
+	private final AtomicBoolean closePublished = new AtomicBoolean();
+
+	private final AtomicBoolean exceptionSent = new AtomicBoolean();
+
+	private final SocketInfo socketInfo;
+
 	@SuppressWarnings("rawtypes")
 	private volatile Deserializer deserializer;
 
@@ -66,24 +78,14 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 
 	private volatile TcpSender sender;
 
-	private final boolean server;
-
 	private volatile String connectionId;
 
-	private final AtomicLong sequence = new AtomicLong();
-
 	private volatile String hostName = "unknown";
 
 	private volatile String hostAddress = "unknown";
 
 	private volatile String connectionFactoryName = "unknown";
 
-	private final ApplicationEventPublisher applicationEventPublisher;
-
-	private final AtomicBoolean closePublished = new AtomicBoolean();
-
-	private final AtomicBoolean exceptionSent = new AtomicBoolean();
-
 	private volatile boolean noReadErrorOnClose;
 
 	private volatile boolean manualListenerRegistration;
@@ -95,6 +97,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	public TcpConnectionSupport(ApplicationEventPublisher applicationEventPublisher) {
 		this.server = false;
 		this.applicationEventPublisher = applicationEventPublisher;
+		this.socketInfo = null;
 	}
 
 	/**
@@ -112,6 +115,7 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 	public TcpConnectionSupport(Socket socket, boolean server, boolean lookupHost,
 			ApplicationEventPublisher applicationEventPublisher,
 			String connectionFactoryName) {
+		this.socketInfo = new SocketInfo(socket);
 		this.server = server;
 		InetAddress inetAddress = socket.getInetAddress();
 		if (inetAddress != null) {
@@ -324,6 +328,14 @@ public abstract class TcpConnectionSupport implements TcpConnection {
 		return this.connectionId;
 	}
 
+	/**
+	 * @since 4.2.5
+	 */
+	@Override
+	public SocketInfo getSocketInfo() {
+		return this.socketInfo;
+	}
+
 	protected boolean isNoReadErrorOnClose() {
 		return noReadErrorOnClose;
 	}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -132,6 +132,10 @@ public class TcpMessageMapper implements
 			.setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress())
 			.setHeader(IpHeaders.REMOTE_PORT, connection.getPort())
 			.setHeader(IpHeaders.CONNECTION_ID, connectionId);
+		SocketInfo socketInfo = connection.getSocketInfo();
+		if (socketInfo != null) {
+			messageBuilder.setHeader(IpHeaders.LOCAL_ADDRESS, socketInfo.getLocalAddress());
+		}
 		if (this.applySequence) {
 			messageBuilder
 				.setCorrelationId(connectionId)

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapperTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2015 the original author or authors.
+ * Copyright 2002-2016 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,11 +17,13 @@ package org.springframework.integration.ip.tcp.connection;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.net.InetAddress;
 import java.net.Socket;
 import java.util.Collections;
 import java.util.HashMap;
@@ -69,18 +71,21 @@ public class TcpMessageMapperTests {
 
 		TcpMessageMapper mapper = new TcpMessageMapper();
 		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));
+		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));
 	}
 
 	@Test

src/reference/asciidoc/ip.adoc

@@ -1422,6 +1422,9 @@ If `lookupHost` is `false`, this will contain the ip address.
 | ip_port
 | PORT
 | The remote port for a UDP packet.
+| ip_localInetAddress
+| IP_LOCAL_ADDRESS
+| The local `InetAddress` to which the socket is connected (since _version 4.2.5_).
 | ip_ackTo
 | ACKADDRESS
 | The remote ip address to which UDP application-level acks will be sent.