Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Multicast ipv6 support for branch 2.6.x #3430

Merged
merged 3 commits into from Mar 8, 2019
Merged

Multicast ipv6 support for branch 2.6.x #3430

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
58a59fd
Multicast demo fails with message "Can't assign requested address
chickenlj Feb 3, 2019
605b6c0
remove useless code
chickenlj Mar 8, 2019
ab40335
Fix multicast registry ut
chickenlj Mar 8, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
33 changes: 33 additions & 0 deletions dubbo-common/src/main/java/com/alibaba/dubbo/common/utils/NetUtils.java
View file
Open in desktop
Expand Up @@ -21,8 +21,11 @@
import com.alibaba.dubbo.common.logger.LoggerFactory;

import java.io.IOException;
import java.net.Inet4Address;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
import java.net.NetworkInterface;
import java.net.ServerSocket;
import java.net.UnknownHostException;
Expand Down Expand Up @@ -284,4 +287,34 @@ public static String toURL(String protocol, String host, int port, String path)
return sb.toString();
}

public static void joinMulticastGroup(MulticastSocket multicastSocket, InetAddress multicastAddress) throws IOException {
setInterface(multicastSocket, multicastAddress instanceof Inet6Address);
multicastSocket.setLoopbackMode(false);
multicastSocket.joinGroup(multicastAddress);
}

public static void setInterface(MulticastSocket multicastSocket, boolean preferIpv6) throws IOException {
boolean interfaceSet = false;
Enumeration interfaces = NetworkInterface.getNetworkInterfaces();
while (interfaces.hasMoreElements()) {
NetworkInterface i = (NetworkInterface) interfaces.nextElement();
Enumeration addresses = i.getInetAddresses();
while (addresses.hasMoreElements()) {
InetAddress address = (InetAddress) addresses.nextElement();
if (preferIpv6 && address instanceof Inet6Address) {
multicastSocket.setInterface(address);
interfaceSet = true;
break;
} else if (!preferIpv6 && address instanceof Inet4Address) {
multicastSocket.setInterface(address);
interfaceSet = true;
break;
}
}
if (interfaceSet) {
break;
}
}
}

}
57 changes: 36 additions & 21 deletions ...istry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java
View file
Open in desktop
Expand Up @@ -31,6 +31,7 @@

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.MulticastSocket;
Expand Down Expand Up @@ -58,11 +59,11 @@ public class MulticastRegistry extends FailbackRegistry {

private static final int DEFAULT_MULTICAST_PORT = 1234;

private final InetAddress mutilcastAddress;
private final InetAddress multicastAddress;

private final MulticastSocket mutilcastSocket;
private final MulticastSocket multicastSocket;

private final int mutilcastPort;
private final int multicastPort;

private final ConcurrentMap<URL, Set<URL>> received = new ConcurrentHashMap<URL, Set<URL>>();

Expand All @@ -83,19 +84,20 @@ public MulticastRegistry(URL url) {
throw new IllegalArgumentException("Invalid multicast address " + url.getHost() + ", scope: 224.0.0.0 - 239.255.255.255");
chickenlj marked this conversation as resolved.
Show resolved Hide resolved
}
try {
mutilcastAddress = InetAddress.getByName(url.getHost());
mutilcastPort = url.getPort() <= 0 ? DEFAULT_MULTICAST_PORT : url.getPort();
mutilcastSocket = new MulticastSocket(mutilcastPort);
mutilcastSocket.setLoopbackMode(false);
mutilcastSocket.joinGroup(mutilcastAddress);
multicastAddress = InetAddress.getByName(url.getHost());
checkMulticastAddress(multicastAddress);

multicastPort = url.getPort() <= 0 ? DEFAULT_MULTICAST_PORT : url.getPort();
multicastSocket = new MulticastSocket(multicastPort);
NetUtils.joinMulticastGroup(multicastSocket, multicastAddress);
Thread thread = new Thread(new Runnable() {
@Override
public void run() {
byte[] buf = new byte[2048];
DatagramPacket recv = new DatagramPacket(buf, buf.length);
while (!mutilcastSocket.isClosed()) {
while (!multicastSocket.isClosed()) {
try {
mutilcastSocket.receive(recv);
multicastSocket.receive(recv);
String msg = new String(recv.getData()).trim();
int i = msg.indexOf('\n');
if (i > 0) {
Expand All @@ -104,7 +106,7 @@ public void run() {
MulticastRegistry.this.receive(msg, (InetSocketAddress) recv.getSocketAddress());
Arrays.fill(buf, (byte) 0);
} catch (Throwable e) {
if (!mutilcastSocket.isClosed()) {
if (!multicastSocket.isClosed()) {
logger.error(e.getMessage(), e);
}
}
Expand Down Expand Up @@ -133,6 +135,19 @@ public void run() {
}
}

private void checkMulticastAddress(InetAddress multicastAddress) {
if (!multicastAddress.isMulticastAddress()) {
String message = "Invalid multicast address " + multicastAddress;
if (!(multicastAddress instanceof Inet4Address)) {
throw new IllegalArgumentException(message + ", " +
"ipv4 multicast address scope: 224.0.0.0 - 239.255.255.255.");
} else {
throw new IllegalArgumentException(message + ", " + "ipv6 multicast address must start with ff, " +
"for example: ff01::1");
}
}
}

private static boolean isMulticastAddress(String ip) {
int i = ip.indexOf('.');
if (i > 0) {
Expand Down Expand Up @@ -233,25 +248,25 @@ private void receive(String msg, InetSocketAddress remoteAddress) {

private void broadcast(String msg) {
if (logger.isInfoEnabled()) {
logger.info("Send broadcast message: " + msg + " to " + mutilcastAddress + ":" + mutilcastPort);
logger.info("Send broadcast message: " + msg + " to " + multicastAddress + ":" + multicastPort);
}
try {
byte[] data = (msg + "\n").getBytes();
DatagramPacket hi = new DatagramPacket(data, data.length, mutilcastAddress, mutilcastPort);
mutilcastSocket.send(hi);
DatagramPacket hi = new DatagramPacket(data, data.length, multicastAddress, multicastPort);
multicastSocket.send(hi);
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
}

private void unicast(String msg, String host) {
if (logger.isInfoEnabled()) {
logger.info("Send unicast message: " + msg + " to " + host + ":" + mutilcastPort);
logger.info("Send unicast message: " + msg + " to " + host + ":" + multicastPort);
}
try {
byte[] data = (msg + "\n").getBytes();
DatagramPacket hi = new DatagramPacket(data, data.length, InetAddress.getByName(host), mutilcastPort);
mutilcastSocket.send(hi);
DatagramPacket hi = new DatagramPacket(data, data.length, InetAddress.getByName(host), multicastPort);
multicastSocket.send(hi);
} catch (Exception e) {
throw new IllegalStateException(e.getMessage(), e);
}
Expand Down Expand Up @@ -293,7 +308,7 @@ protected void doUnsubscribe(URL url, NotifyListener listener) {
@Override
public boolean isAvailable() {
try {
return mutilcastSocket != null;
return multicastSocket != null;
} catch (Throwable t) {
return false;
}
Expand All @@ -310,8 +325,8 @@ public void destroy() {
logger.warn(t.getMessage(), t);
}
try {
mutilcastSocket.leaveGroup(mutilcastAddress);
mutilcastSocket.close();
multicastSocket.leaveGroup(multicastAddress);
multicastSocket.close();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
Expand Down Expand Up @@ -434,7 +449,7 @@ public List<URL> lookup(URL url) {
}

public MulticastSocket getMutilcastSocket() {
return mutilcastSocket;
return multicastSocket;
}

public Map<URL, Set<URL>> getReceived() {
Expand Down
42 changes: 41 additions & 1 deletion ...y-multicast/src/test/java/com/alibaba/dubbo/registry/multicast/MulticastRegistryTest.java
View file
Open in desktop
Expand Up @@ -23,16 +23,17 @@
import org.junit.Before;
import org.junit.Test;

import java.net.InetAddress;
import java.net.MulticastSocket;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;

import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertEquals;

public class MulticastRegistryTest {

Expand Down Expand Up @@ -124,4 +125,43 @@ public void testDefaultPort() {
}
}

@Test
public void testMulticastAddress() {
InetAddress multicastAddress = null;
MulticastSocket multicastSocket = null;
try {
// ipv4 multicast address
multicastAddress = InetAddress.getByName("224.55.66.77");
multicastSocket = new MulticastSocket(2345);
multicastSocket.setLoopbackMode(false);
NetUtils.setInterface(multicastSocket, false);
multicastSocket.joinGroup(multicastAddress);
} catch (Exception e) {
e.printStackTrace();
Assert.fail(e.getMessage());
} finally {
if (multicastSocket != null) {
multicastSocket.close();
}
}

// multicast ipv6 address,
try {
multicastAddress = InetAddress.getByName("ff01::1");

multicastSocket = new MulticastSocket();
multicastSocket.setLoopbackMode(false);
NetUtils.setInterface(multicastSocket, true);
multicastSocket.joinGroup(multicastAddress);
} catch (Throwable t) {
t.printStackTrace();
} finally {
if (multicastSocket != null) {
multicastSocket.close();
}
}

}


}