diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java index 86d538aec4..b15d95d05f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java +++ b/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() {} } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java index 43a752051a..023d24506c 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/FailoverClientConnectionFactory.java +++ b/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(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/SocketInfo.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/SocketInfo.java new file mode 100644 index 0000000000..a535cb1b90 --- /dev/null +++ b/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(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java index a51de10aea..bbf0681976 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -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(); + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java index 81ec832c4e..ff22493ef2 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionInterceptorSupport.java +++ b/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(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java index 9c86f48882..2cb919c66e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java +++ b/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; } 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 ef66ab40c1..f7bfe9a94d 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 @@ -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) 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 e84321672c..781fcfc226 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 @@ -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 diff --git a/src/reference/asciidoc/ip.adoc b/src/reference/asciidoc/ip.adoc index 9d0c134f99..5df4019e40 100644 --- a/src/reference/asciidoc/ip.adoc +++ b/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.