Revert "INT-4366: Fix MulticastSendingMH

This reverts commit c3b64dc.

Author: garyrussell

spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2001-2018 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.
@@ -38,8 +38,6 @@
  * determine success.
  *
  * @author Gary Russell
- * @author Artem Bilan
- *
  * @since 2.0
  */
 public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler {
@@ -128,45 +126,49 @@ public MulticastSendingMessageHandler(String destinationExpression) {
 
 	@Override
 	protected DatagramSocket getSocket() throws IOException {
-		if (this.multicastSocket == null) {
+		if (this.getTheSocket() == null) {
 			synchronized (this) {
-				if (this.multicastSocket == null) {
-					createSocket();
-				}
+				createSocket();
 			}
 		}
-		return getTheSocket();
+		return this.getTheSocket();
 	}
 
 	private void createSocket() throws IOException {
-		MulticastSocket socket;
-		if (isAcknowledge()) {
-			int ackPort = getAckPort();
-			if (this.localAddress == null) {
-				socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort);
+		if (this.getTheSocket() == null) {
+			MulticastSocket socket;
+			if (this.isAcknowledge()) {
+				int ackPort = this.getAckPort();
+				if (this.localAddress == null) {
+					socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort);
+				}
+				else {
+					InetAddress whichNic = InetAddress.getByName(this.localAddress);
+					socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort));
+				}
+				if (getSoReceiveBufferSize() > 0) {
+					socket.setReceiveBufferSize(this.getSoReceiveBufferSize());
+				}
+				if (logger.isDebugEnabled()) {
+					logger.debug("Listening for acks on port: " + socket.getLocalPort());
+				}
+				setSocket(socket);
+				updateAckAddress();
 			}
 			else {
-				InetAddress whichNic = InetAddress.getByName(this.localAddress);
-				socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort));
+				socket = new MulticastSocket();
+				setSocket(socket);
 			}
-			if (getSoReceiveBufferSize() > 0) {
-				socket.setReceiveBufferSize(getSoReceiveBufferSize());
+			if (this.timeToLive >= 0) {
+				socket.setTimeToLive(this.timeToLive);
 			}
-			if (logger.isDebugEnabled()) {
-				logger.debug("Listening for acks on port: " + socket.getLocalPort());
+			setSocketAttributes(socket);
+			if (this.localAddress != null) {
+				InetAddress whichNic = InetAddress.getByName(this.localAddress);
+				socket.setInterface(whichNic);
 			}
-			setSocket(socket);
-			updateAckAddress();
-		}
-		else {
-			socket = new MulticastSocket();
-			setSocket(socket);
-		}
-		if (this.timeToLive >= 0) {
-			socket.setTimeToLive(this.timeToLive);
+			this.multicastSocket = socket;
 		}
-		setSocketAttributes(socket);
-		this.multicastSocket = socket;
 	}
 
 
@@ -176,7 +178,7 @@ private void createSocket() throws IOException {
 	 * @param minAcksForSuccess The minimum number of acks that will represent success.
 	 */
 	public void setMinAcksForSuccess(int minAcksForSuccess) {
-		setAckCounter(minAcksForSuccess);
+		this.setAckCounter(minAcksForSuccess);
 	}
 
 	/**

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 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.
@@ -28,6 +28,7 @@
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -38,7 +39,6 @@
 import org.junit.Test;
 
 import org.springframework.beans.factory.BeanFactory;
-import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.channel.DirectChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.handler.ServiceActivatingHandler;
@@ -56,8 +56,6 @@
 
 /**
  * @author Gary Russell
- * @author Artem Bilan
- *
  * @since 2.0
  */
 public class TcpInboundGatewayTests {
@@ -121,31 +119,30 @@ public void testNetClientMode() throws Exception {
 		final CountDownLatch latch2 = new CountDownLatch(1);
 		final CountDownLatch latch3 = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
-		new SimpleAsyncTaskExecutor()
-				.execute(() -> {
-					try {
-						ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
-						port.set(server.getLocalPort());
-						latch1.countDown();
-						Socket socket = server.accept();
-						socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes());
-						byte[] bytes = new byte[12];
-						readFully(socket.getInputStream(), bytes);
-						assertEquals("Echo:Test1\r\n", new String(bytes));
-						readFully(socket.getInputStream(), bytes);
-						assertEquals("Echo:Test2\r\n", new String(bytes));
-						latch2.await();
-						socket.close();
-						server.close();
-						done.set(true);
-						latch3.countDown();
-					}
-					catch (Exception e) {
-						if (!done.get()) {
-							e.printStackTrace();
-						}
-					}
-				});
+		Executors.newSingleThreadExecutor().execute(() -> {
+			try {
+				ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
+				port.set(server.getLocalPort());
+				latch1.countDown();
+				Socket socket = server.accept();
+				socket.getOutputStream().write("Test1\r\nTest2\r\n".getBytes());
+				byte[] bytes = new byte[12];
+				readFully(socket.getInputStream(), bytes);
+				assertEquals("Echo:Test1\r\n", new String(bytes));
+				readFully(socket.getInputStream(), bytes);
+				assertEquals("Echo:Test2\r\n", new String(bytes));
+				latch2.await();
+				socket.close();
+				server.close();
+				done.set(true);
+				latch3.countDown();
+			}
+			catch (Exception e) {
+				if (!done.get()) {
+					e.printStackTrace();
+				}
+			}
+		});
 		assertTrue(latch1.await(10, TimeUnit.SECONDS));
 		AbstractClientConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port.get());
 		ccf.setSingleUse(false);

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

@@ -43,6 +43,7 @@
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -60,8 +61,6 @@
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.serializer.DefaultDeserializer;
 import org.springframework.core.serializer.DefaultSerializer;
-import org.springframework.core.task.AsyncTaskExecutor;
-import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -91,8 +90,6 @@ public class TcpOutboundGatewayTests {
 
 	private static final Log logger = LogFactory.getLog(TcpOutboundGatewayTests.class);
 
-	private AsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
-
 	@ClassRule
 	public static LongRunningIntegrationTest longTests = new LongRunningIntegrationTest();
 
@@ -104,13 +101,13 @@ public class TcpOutboundGatewayTests {
 	public void testGoodNetSingle() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
-		final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
-		this.executor.execute(() -> {
+		final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
+		Executors.newSingleThreadExecutor().execute(() -> {
 			try {
 				ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 100);
 				serverSocket.set(server);
 				latch.countDown();
-				List<Socket> sockets = new ArrayList<>();
+				List<Socket> sockets = new ArrayList<Socket>();
 				int i = 0;
 				while (true) {
 					Socket socket = server.accept();
@@ -168,8 +165,8 @@ public void testGoodNetSingle() throws Exception {
 	public void testGoodNetMultiplex() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
-		final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
-		this.executor.execute(() -> {
+		final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
+		Executors.newSingleThreadExecutor().execute(() -> {
 			try {
 				ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0, 10);
 				serverSocket.set(server);
@@ -223,8 +220,8 @@ public void testGoodNetMultiplex() throws Exception {
 	public void testGoodNetTimeout() throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
-		final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
-		this.executor.execute(() -> {
+		final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
+		Executors.newSingleThreadExecutor().execute(() -> {
 			try {
 				ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
 				serverSocket.set(server);
@@ -263,12 +260,12 @@ public void testGoodNetTimeout() throws Exception {
 		Future<Integer>[] results = (Future<Integer>[]) new Future<?>[2];
 		for (int i = 0; i < 2; i++) {
 			final int j = i;
-			results[j] = (this.executor.submit(() -> {
+			results[j] = (Executors.newSingleThreadExecutor().submit(() -> {
 				gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
 				return 0;
 			}));
 		}
-		Set<String> replies = new HashSet<>();
+		Set<String> replies = new HashSet<String>();
 		int timeouts = 0;
 		for (int i = 0; i < 2; i++) {
 			try {
@@ -347,7 +344,7 @@ private void testGoodNetGWTimeoutGuts(final int port, AbstractClientConnectionFa
 		final AtomicReference<String> lastReceived = new AtomicReference<String>();
 		final CountDownLatch serverLatch = new CountDownLatch(2);
 
-		this.executor.execute(() -> {
+		Executors.newSingleThreadExecutor().execute(() -> {
 			try {
 				latch.countDown();
 				int i = 0;
@@ -401,7 +398,7 @@ private void testGoodNetGWTimeoutGuts(final int port, AbstractClientConnectionFa
 
 		for (int i = 0; i < 2; i++) {
 			final int j = i;
-			results[j] = (this.executor.submit(() -> {
+			results[j] = (Executors.newSingleThreadExecutor().submit(() -> {
 				gateway.handleMessage(MessageBuilder.withPayload("Test" + j).build());
 				return j;
 			}));
@@ -445,7 +442,7 @@ public void testCachingFailover() throws Exception {
 		final AtomicBoolean done = new AtomicBoolean();
 		final CountDownLatch serverLatch = new CountDownLatch(1);
 
-		this.executor.execute(() -> {
+		Executors.newSingleThreadExecutor().execute(() -> {
 			try {
 				ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
 				serverSocket.set(server);
@@ -520,12 +517,12 @@ public void testCachingFailover() throws Exception {
 
 	@Test
 	public void testFailoverCached() throws Exception {
-		final AtomicReference<ServerSocket> serverSocket = new AtomicReference<>();
+		final AtomicReference<ServerSocket> serverSocket = new AtomicReference<ServerSocket>();
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 		final CountDownLatch serverLatch = new CountDownLatch(1);
 
-		this.executor.execute(() -> {
+		Executors.newSingleThreadExecutor().execute(() -> {
 			try {
 				ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(0);
 				serverSocket.set(server);
@@ -670,11 +667,11 @@ private void testGWPropagatesSocketCloseGuts(final int port, AbstractClientConne
 			final ServerSocket server) throws Exception {
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
-		final AtomicReference<String> lastReceived = new AtomicReference<>();
+		final AtomicReference<String> lastReceived = new AtomicReference<String>();
 		final CountDownLatch serverLatch = new CountDownLatch(1);
 
-		this.executor.execute(() -> {
-			List<Socket> sockets = new ArrayList<>();
+		Executors.newSingleThreadExecutor().execute(() -> {
+			List<Socket> sockets = new ArrayList<Socket>();
 			try {
 				latch.countDown();
 				while (!done.get()) {
@@ -796,8 +793,8 @@ private void testGWPropagatesSocketTimeoutGuts(final int port, AbstractClientCon
 		final CountDownLatch latch = new CountDownLatch(1);
 		final AtomicBoolean done = new AtomicBoolean();
 
-		this.executor.execute(() -> {
-			List<Socket> sockets = new ArrayList<>();
+		Executors.newSingleThreadExecutor().execute(() -> {
+			List<Socket> sockets = new ArrayList<Socket>();
 			try {
 				latch.countDown();
 				while (!done.get()) {