natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修改注册中心基类 

git-svn-id: http://code.alibabatech.com/svn/dubbo/trunk@74 1a56cb94-b969-4eaa-88fa-be21384802f2
This commit is contained in:
william.liangf 2011-10-26 07:48:53 +00:00
parent eb62512fae
commit 7c7d9b6ce1
4 changed files with 55 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
dubbo-registry-multicast/src/main/java/com/alibaba/dubbo/registry/multicast/MulticastRegistry.java
View File

@ -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<URL> urls = new ArrayList<URL>();
List<URL> 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<URL> urls = new ArrayList<URL>();
List<URL> 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<String, String> 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<String, String> 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() {

3
dubbo-registry-multicast/src/test/java/com/alibaba/dubbo/registry/multicast/MulticastRegistryTest.java
View File

@ -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<Map<String, String>> args = new AtomicReference<Map<String, String>>();
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<URL> urls) {
// FIXME assertEquals(MulticastRegistry.this.service, service);

14
dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/AbstractRegistryService.java
View File

@ -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<URL> lookup(URL url) {
@ -117,17 +117,17 @@ public abstract class AbstractRegistryService implements RegistryService {
}
}
public void subscribe(String service, Map<String, String> 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<NotifyListener> listeners = notifyListeners.get(service);
if (listeners == null) {
notifyListeners.putIfAbsent(service, new CopyOnWriteArrayList<NotifyListener>());
@ -136,11 +136,11 @@ public abstract class AbstractRegistryService implements RegistryService {
listeners.add(listener);
}
public void unsubscribe(String service, Map<String, String> 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) {

17
dubbo-registry/src/main/java/com/alibaba/dubbo/registry/support/SimpleRegistryService.java
View File

@ -68,7 +68,7 @@ public class SimpleRegistryService extends AbstractRegistryService {
}
@Override
public void subscribe(String service, Map<String, String> 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<String> 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<String, NotifyListener> listeners = remoteListeners.get(client);
if (listeners == null) {
@ -105,8 +105,8 @@ public class SimpleRegistryService extends AbstractRegistryService {
}
@Override
public void unsubscribe(String service, Map<String, String> 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<String, NotifyListener> 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<String, NotifyListener> 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());
}
}
}