diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java index b07633d4c5..bb8c6f38cb 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/MulticastSendingMessageHandler.java @@ -100,56 +100,51 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler } @Override - protected DatagramSocket getSocket() throws IOException { - if (this.getTheSocket() == null) { - synchronized (this) { - createSocket(); - } + protected synchronized DatagramSocket getSocket() throws IOException { + if (getTheSocket() == null) { + createSocket(); } - return this.getTheSocket(); + return getTheSocket(); } private void createSocket() throws IOException { - 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(); + MulticastSocket socket; + if (this.isAcknowledge()) { + int ackPort = this.getAckPort(); + if (this.localAddress == null) { + socket = ackPort == 0 ? new MulticastSocket() : new MulticastSocket(ackPort); } else { - socket = new MulticastSocket(); - setSocket(socket); - } - if (this.timeToLive >= 0) { - socket.setTimeToLive(this.timeToLive); - } - setSocketAttributes(socket); - if (this.localAddress != null) { InetAddress whichNic = InetAddress.getByName(this.localAddress); - socket.setInterface(whichNic); + socket = new MulticastSocket(new InetSocketAddress(whichNic, ackPort)); } - this.multicastSocket = socket; + if (getSoReceiveBufferSize() > 0) { + socket.setReceiveBufferSize(this.getSoReceiveBufferSize()); + } + if (logger.isDebugEnabled()) { + logger.debug("Listening for acks on port: " + socket.getLocalPort()); + } + setSocket(socket); + updateAckAddress(); + } + else { + socket = new MulticastSocket(); + setSocket(socket); + } + if (this.timeToLive >= 0) { + socket.setTimeToLive(this.timeToLive); + } + setSocketAttributes(socket); + this.multicastSocket = socket; + if (this.localAddress != null) { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + socket.setInterface(whichNic); } } /** * If acknowledge = true; how many acks needed for success. - * * @param minAcksForSuccess The minimum number of acks that will represent success. */ public void setMinAcksForSuccess(int minAcksForSuccess) { @@ -158,7 +153,6 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler /** * Set the underlying {@link MulticastSocket} time to live property. - * * @param timeToLive {@link MulticastSocket#setTimeToLive(int)} */ public void setTimeToLive(int timeToLive) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java index c3a45ed70e..d467cfe04f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2001-2017 the original author or authors. + * Copyright 2001-2018 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. @@ -83,8 +83,8 @@ public class UnicastSendingMessageHandler extends private volatile int ackCounter = 1; - private volatile Map ackControl = Collections - .synchronizedMap(new HashMap()); + private volatile Map ackControl = + Collections.synchronizedMap(new HashMap()); private volatile int soReceiveBufferSize = -1; @@ -282,11 +282,7 @@ public class UnicastSendingMessageHandler extends throw e; } catch (Exception e) { - try { - this.socket.close(); - } - catch (Exception e1) { } - this.socket = null; + closeSocketIfNeeded(); throw new MessageHandlingException(message, "failed to send UDP packet", e); } finally { @@ -511,7 +507,7 @@ public class UnicastSendingMessageHandler extends } catch (IOException e) { if (this.socket != null && !this.socket.isClosed()) { - logger.error("Error on UDP Acknowledge thread:" + e.getMessage()); + logger.error("Error on UDP Acknowledge thread: " + e.getMessage()); } } finally { @@ -529,7 +525,11 @@ public class UnicastSendingMessageHandler extends private void closeSocketIfNeeded() { if (this.socket != null) { - this.socket.close(); + try { + this.socket.close(); + } + catch (Exception e) { + } this.socket = null; } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java index c638da589b..7feb872689 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMulticastSendingHandlerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -26,7 +26,7 @@ import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.MulticastSocket; import java.util.concurrent.CountDownLatch; -import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -45,6 +45,8 @@ import org.springframework.messaging.Message; /** * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class DatagramPacketMulticastSendingHandlerTests { @@ -100,7 +102,7 @@ public class DatagramPacketMulticastSendingHandlerTests { } } }; - Executor executor = Executors.newFixedThreadPool(2); + ExecutorService executor = Executors.newFixedThreadPool(2); executor.execute(catcher); executor.execute(catcher); assertTrue(listening.await(10000, TimeUnit.MILLISECONDS)); @@ -112,6 +114,7 @@ public class DatagramPacketMulticastSendingHandlerTests { assertTrue(received.await(10000, TimeUnit.MILLISECONDS)); handler.stop(); socket.close(); + executor.shutdownNow(); } @Test @@ -177,7 +180,7 @@ public class DatagramPacketMulticastSendingHandlerTests { } } }; - Executor executor = Executors.newFixedThreadPool(2); + ExecutorService executor = Executors.newFixedThreadPool(2); executor.execute(catcher); executor.execute(catcher); assertTrue(listening.await(10000, TimeUnit.MILLISECONDS)); @@ -195,6 +198,7 @@ public class DatagramPacketMulticastSendingHandlerTests { assertTrue(ackSent.await(10000, TimeUnit.MILLISECONDS)); handler.stop(); socket.close(); + executor.shutdownNow(); } public void waitAckListening(UnicastSendingMessageHandler handler) throws InterruptedException { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index eeeea7f0e5..e673bdb4e3 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -299,6 +299,7 @@ public class UdpChannelAdapterTests { assertNotNull(receivedMessage); assertEquals(new String(message.getPayload()), new String(receivedMessage.getPayload())); adapter.stop(); + handler.stop(); } @Test @@ -309,6 +310,8 @@ public class UdpChannelAdapterTests { // SocketUtils.setLocalNicIfPossible(adapter); adapter.setOutputChannel(channel); ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService()); + handler.setBeanFactory(mock(BeanFactory.class)); + handler.afterPropertiesSet(); channel.subscribe(handler); QueueChannel errorChannel = new QueueChannel(); adapter.setErrorChannel(errorChannel);