diff --git a/dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java b/dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java index a3e35a7b5c..1177d45b83 100644 --- a/dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java +++ b/dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java @@ -22,7 +22,6 @@ import java.net.InetSocketAddress; import java.net.MulticastSocket; import java.util.ArrayList; import java.util.List; -import java.util.Map; import com.alibaba.dubbo.common.URL; import com.alibaba.dubbo.common.logger.Logger; @@ -41,10 +40,18 @@ public class MulticastRegistry extends AbstractRegistry { // 日志输出 private static final Logger logger = LoggerFactory.getLogger(MulticastRegistry.class); + private static final String REGISTER = "register"; + + private static final String UNREGISTER = "unregister"; + + private static final String SUBSCRIBE = "subscribe"; + + private static final String UNSUBSCRIBE = "unsubscribe"; + private InetAddress mutilcastAddress; private MulticastSocket mutilcastSocket; - + public MulticastRegistry(URL url) { super(url); if (! isMulticastAddress(url.getHost())) { @@ -89,11 +96,10 @@ public class MulticastRegistry extends AbstractRegistry { } private void receive(String msg, InetSocketAddress remoteAddress) { - if (msg.startsWith("register")) { - String[] parts = msg.split("\\s+"); - String service = parts[1]; + if (msg.startsWith(REGISTER)) { + URL url = URL.valueOf(msg.substring(REGISTER.length()).trim()); + String service = url.getServiceKey(); if (getSubscribed().containsKey(service)) { - URL url = URL.valueOf(parts[2]); List urls = new ArrayList(); List notified = getNotified().get(service); if (notified != null) { @@ -104,14 +110,26 @@ public class MulticastRegistry extends AbstractRegistry { } notify(service, urls); } - } else if (msg.startsWith("subscribe")) { - String[] parts = msg.split("\\s+"); - String service = parts[1]; + } else if (msg.startsWith(UNREGISTER)) { + URL url = URL.valueOf(msg.substring(UNREGISTER.length()).trim()); + String service = url.getServiceKey(); + if (getSubscribed().containsKey(service)) { + List urls = new ArrayList(); + List notified = getNotified().get(service); + if (notified != null) { + urls.addAll(notified); + } + urls.remove(url); + notify(service, urls); + } + } else if (msg.startsWith(SUBSCRIBE)) { + String service = URL.valueOf(msg.substring(SUBSCRIBE.length()).trim()).getServiceKey(); if (getRegistered().containsKey(service)) { for (URL url : getRegistered().get(service)) { - send("register " + service + " " + url.toString()); + send(REGISTER + " " + url.toFullString()); } } + } else if (msg.startsWith(UNSUBSCRIBE)) { } } @@ -123,19 +141,19 @@ public class MulticastRegistry extends AbstractRegistry { throw new IllegalStateException(e.getMessage(), e); } } - + public void register(String service, URL url) { super.register(service, url); - send("register " + service + " " + url); + send(REGISTER + " " + url.toFullString()); } public void unregister(String service, URL url) { - send("unregister " + service + " " + url); + send(UNREGISTER + " " + url.toFullString()); } - public void subscribe(String service, Map parameters, NotifyListener listener) { - super.subscribe(service, parameters, listener); - send("subscribe " + service + " " + StringUtils.toQueryString(parameters)); + public void subscribe(String service, URL url, NotifyListener listener) { + super.subscribe(service, url, listener); + send(SUBSCRIBE + " " + url.toFullString()); synchronized (this) { try { this.wait(5000); @@ -144,8 +162,8 @@ public class MulticastRegistry extends AbstractRegistry { } } - public void unsubscribe(String service, Map query, NotifyListener listener) { - send("unsubscribe " + service + " " + StringUtils.toQueryString(query)); + public void unsubscribe(String service, URL url, NotifyListener listener) { + send(UNSUBSCRIBE + " " + url.toFullString()); } public boolean isAvailable() { diff --git a/dubbo-registry-multicast/src/test/java/com/alibaba/dubbo/registry/multicast/MulticastRegistryTest.java b/dubbo-registry-multicast/src/test/java/com/alibaba/dubbo/registry/multicast/MulticastRegistryTest.java index 85e0727e84..75f7c53068 100644 --- a/dubbo-registry-multicast/src/test/java/com/alibaba/dubbo/registry/multicast/MulticastRegistryTest.java +++ b/dubbo-registry-multicast/src/test/java/com/alibaba/dubbo/registry/multicast/MulticastRegistryTest.java @@ -27,6 +27,7 @@ import org.junit.BeforeClass; import org.junit.Test; import com.alibaba.dubbo.common.URL; +import com.alibaba.dubbo.common.utils.NetUtils; import com.alibaba.dubbo.common.utils.StringUtils; import com.alibaba.dubbo.registry.NotifyListener; @@ -94,7 +95,7 @@ public class MulticastRegistryTest { final String subscribearg = "arg1=1&arg2=2"; // verify lisener. final AtomicReference> args = new AtomicReference>(); - registry.subscribe(service, StringUtils.parseQueryString(subscribearg), new NotifyListener() { + registry.subscribe(service, new URL("dubbo", NetUtils.getLocalHost(), 0, StringUtils.parseQueryString(subscribearg)), new NotifyListener() { public void notify(List urls) { // FIXME assertEquals(MulticastRegistry.this.service, service); diff --git a/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistryService.java b/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistryService.java index 1d0e6a219e..cf78ba9d56 100644 --- a/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistryService.java +++ b/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistryService.java @@ -73,14 +73,14 @@ public abstract class AbstractRegistryService implements RegistryService { if (logger.isInfoEnabled()) { logger.info("Subscribe service: " + url.getServiceKey() + ",url:" + url); } - subscribe(url.getServiceKey(), url.getParameters(), listener); + subscribe(url.getServiceKey(), url, listener); } public void unsubscribe(URL url, NotifyListener listener) { if (logger.isInfoEnabled()) { logger.info("Unsubscribe service: " + url.getServiceKey() + ",url:" + url); } - unsubscribe(url.getServiceKey(), url.getParameters(), listener); + unsubscribe(url.getServiceKey(), url, listener); } public List lookup(URL url) { @@ -117,17 +117,17 @@ public abstract class AbstractRegistryService implements RegistryService { } } - public void subscribe(String service, Map parameters, NotifyListener listener) { + public void subscribe(String service, URL url, NotifyListener listener) { if (service == null) { throw new IllegalArgumentException("service == null"); } - if (parameters == null) { + if (url == null) { throw new IllegalArgumentException("parameters == null"); } if (listener == null) { throw new IllegalArgumentException("listener == null"); } - subscribed.put(service, parameters); + subscribed.put(service, url.getParameters()); List listeners = notifyListeners.get(service); if (listeners == null) { notifyListeners.putIfAbsent(service, new CopyOnWriteArrayList()); @@ -136,11 +136,11 @@ public abstract class AbstractRegistryService implements RegistryService { listeners.add(listener); } - public void unsubscribe(String service, Map parameters, NotifyListener listener) { + public void unsubscribe(String service, URL url, NotifyListener listener) { if (service == null) { throw new IllegalArgumentException("service == null"); } - if (parameters == null) { + if (url == null) { throw new IllegalArgumentException("parameters == null"); } if (listener == null) { diff --git a/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/SimpleRegistryService.java b/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/SimpleRegistryService.java index e6a6946aa5..5f835b921f 100644 --- a/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/SimpleRegistryService.java +++ b/dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/SimpleRegistryService.java @@ -68,7 +68,7 @@ public class SimpleRegistryService extends AbstractRegistryService { } @Override - public void subscribe(String service, Map parameters, NotifyListener listener) { + public void subscribe(String service, URL url, NotifyListener listener) { String client = RpcContext.getContext().getRemoteAddressString(); if (logger.isInfoEnabled()){ logger.info("[subscribe] service: "+service + ",client:"+ client); @@ -80,15 +80,15 @@ public class SimpleRegistryService extends AbstractRegistryService { NetUtils.getLocalHost(), RpcContext.getContext().getLocalPort(), com.alibaba.dubbo.registry.RegistryService.class.getName(), - parameters)); + url.getParameters())); List rs = registries; if (rs != null && rs.size() > 0) { for (String registry : rs) { - register(service, UrlUtils.parseURL(registry, parameters)); + register(service, UrlUtils.parseURL(registry, url.getParameters())); } } } - super.subscribe(service, parameters, listener); + super.subscribe(service, url, listener); Map listeners = remoteListeners.get(client); if (listeners == null) { @@ -105,8 +105,8 @@ public class SimpleRegistryService extends AbstractRegistryService { } @Override - public void unsubscribe(String service, Map parameters, NotifyListener listener) { - super.unsubscribe(service, parameters, listener); + public void unsubscribe(String service, URL url, NotifyListener listener) { + super.unsubscribe(service, url, listener); String client = RpcContext.getContext().getRemoteAddressString(); Map listeners = remoteListeners.get(client); if (listeners != null && listeners.size() > 0) { @@ -133,7 +133,10 @@ public class SimpleRegistryService extends AbstractRegistryService { if (listeners != null && listeners.size() > 0) { for (Map.Entry entry : listeners.entrySet()) { String service = entry.getKey(); - super.unsubscribe(service, getSubscribed(service), entry.getValue()); + super.unsubscribe(service, new URL("subscribe", + RpcContext.getContext().getRemoteHost(), + RpcContext.getContext().getRemotePort(), + com.alibaba.dubbo.registry.RegistryService.class.getName(), getSubscribed(service)), entry.getValue()); } } }