Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4366: Fix MulticastSendingMessageHandler

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4366

Fix race condition in the `MulticastSendingMessageHandler` around
`multicastSocket` and super `socket` properties.

* Synchronize around `this` and check for the `multicastSocket == null`.
This let the `MulticastSendingMessageHandler` to fully configure and
prepare the socket for use.
* Remove `socket.setInterface(whichNic)` since it is populated by the
`InetSocketAddress` ctor before

**Cherry-pick to 4.3.x**
This commit is contained in:
Artem Bilan
2018-01-16 17:44:20 -05:00
committed by Gary Russell
parent 3e99ae38f1
commit 3cccf9b526
1 changed files with 32 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2001-2016 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.
@@ -35,6 +35,8 @@ import org.springframework.messaging.MessageHandler;
* determine success.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*/
public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler {
@@ -101,49 +103,45 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
@Override
protected DatagramSocket getSocket() throws IOException {
if (this.getTheSocket() == null) {
if (this.multicastSocket == null) {
synchronized (this) {
createSocket();
if (this.multicastSocket == 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 (isAcknowledge()) {
int ackPort = 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(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;
}
@@ -153,7 +151,7 @@ public class MulticastSendingMessageHandler extends UnicastSendingMessageHandler
* @param minAcksForSuccess The minimum number of acks that will represent success.
*/
public void setMinAcksForSuccess(int minAcksForSuccess) {
this.setAckCounter(minAcksForSuccess);
setAckCounter(minAcksForSuccess);
}
/**