getRegisteredServiceUrls();
}
diff --git a/motan-core/src/main/java/com/weibo/api/motan/registry/support/AbstractRegistry.java b/motan-core/src/main/java/com/weibo/api/motan/registry/support/AbstractRegistry.java
index ebcf751d0..b18236f61 100644
--- a/motan-core/src/main/java/com/weibo/api/motan/registry/support/AbstractRegistry.java
+++ b/motan-core/src/main/java/com/weibo/api/motan/registry/support/AbstractRegistry.java
@@ -16,18 +16,18 @@
package com.weibo.api.motan.registry.support;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
+import com.weibo.api.motan.common.MotanConstants;
import com.weibo.api.motan.common.URLParamType;
import com.weibo.api.motan.registry.NotifyListener;
import com.weibo.api.motan.registry.Registry;
import com.weibo.api.motan.rpc.URL;
+import com.weibo.api.motan.switcher.SwitcherListener;
+import com.weibo.api.motan.util.ConcurrentHashSet;
import com.weibo.api.motan.util.LoggerUtil;
+import com.weibo.api.motan.util.MotanSwitcherUtil;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
/**
*
@@ -47,10 +47,26 @@ public abstract class AbstractRegistry implements Registry {
new ConcurrentHashMap>>();
private URL registryUrl;
+ private Set registeredServiceUrls = new ConcurrentHashSet();
protected String registryClassName = this.getClass().getSimpleName();
public AbstractRegistry(URL url) {
this.registryUrl = url.createCopy();
+ // register a heartbeat switcher to perceive service state change and change available state
+ MotanSwitcherUtil.registerSwitcherListener(MotanConstants.REGISTRY_HEARTBEAT_SWITCHER, new SwitcherListener() {
+
+ @Override
+ public void onValueChanged(String key, Boolean value) {
+ if (key != null && value != null) {
+ if (value) {
+ available(null);
+ } else {
+ unavailable(null);
+ }
+ }
+
+ }
+ });
}
@Override
@@ -61,6 +77,7 @@ public void register(URL url) {
}
LoggerUtil.info("[{}] Url ({}) will register to Registry [{}]", registryClassName, url, registryUrl.getIdentity());
doRegister(removeUnnecessaryParmas(url.createCopy()));
+ registeredServiceUrls.add(url);
}
@Override
@@ -71,6 +88,7 @@ public void unregister(URL url) {
}
LoggerUtil.info("[{}] Url ({}) will unregister to Registry [{}]", registryClassName, url, registryUrl.getIdentity());
doUnregister(removeUnnecessaryParmas(url.createCopy()));
+ registeredServiceUrls.remove(url);
}
@Override
@@ -128,6 +146,34 @@ public URL getUrl() {
return registryUrl;
}
+ @Override
+ public Collection getRegisteredServiceUrls() {
+ return registeredServiceUrls;
+ }
+
+ @Override
+ public void available(URL url) {
+ LoggerUtil.info("[{}] Url ({}) will set to available to Registry [{}]", registryClassName, url, registryUrl.getIdentity());
+ if(url != null) {
+ doAvailable(removeUnnecessaryParmas(url.createCopy()));
+ } else {
+ doAvailable(null);
+ }
+ }
+
+
+
+ @Override
+ public void unavailable(URL url) {
+ LoggerUtil.info("[{}] Url ({}) will set to unavailable to Registry [{}]", registryClassName, url, registryUrl.getIdentity());
+ if(url == null) {
+ doUnavailable(removeUnnecessaryParmas(url.createCopy()));
+ } else {
+ doUnregister(null);
+ }
+
+ }
+
protected List getCachedUrls(URL url) {
Map> rsUrls = subscribedCategoryResponses.get(url);
if (rsUrls == null || rsUrls.size() == 0) {
@@ -194,4 +240,9 @@ private URL removeUnnecessaryParmas(URL url) {
protected abstract List doDiscover(URL url);
+ protected abstract void doAvailable(URL url);
+
+ protected abstract void doUnavailable(URL url);
+
+
}
diff --git a/motan-core/src/main/java/com/weibo/api/motan/registry/support/FailbackRegistry.java b/motan-core/src/main/java/com/weibo/api/motan/registry/support/FailbackRegistry.java
index 9b4b68fc1..c20fa3841 100644
--- a/motan-core/src/main/java/com/weibo/api/motan/registry/support/FailbackRegistry.java
+++ b/motan-core/src/main/java/com/weibo/api/motan/registry/support/FailbackRegistry.java
@@ -70,12 +70,12 @@ public void run() {
}
@Override
- public void doRegister(URL url) {
+ public void register(URL url) {
failedRegistered.remove(url);
failedUnregistered.remove(url);
try {
- concreteRegister(url);
+ super.register(url);
} catch (Exception e) {
if (isCheckingUrls(getUrl(), url)) {
throw new MotanFrameworkException(String.format("[%s] false to registery %s to %s", registryClassName, url, getUrl()), e);
@@ -85,12 +85,12 @@ public void doRegister(URL url) {
}
@Override
- public void doUnregister(URL url) {
+ public void unregister(URL url) {
failedRegistered.remove(url);
failedUnregistered.remove(url);
try {
- concreteUnregister(url);
+ super.unregister(url);
} catch (Exception e) {
if (isCheckingUrls(getUrl(), url)) {
throw new MotanFrameworkException(String.format("[%s] false to unregistery %s to %s", registryClassName, url, getUrl()), e);
@@ -100,11 +100,11 @@ public void doUnregister(URL url) {
}
@Override
- public void doSubscribe(URL url, NotifyListener listener) {
+ public void subscribe(URL url, NotifyListener listener) {
removeForFailedSubAndUnsub(url, listener);
try {
- concreteSubscribe(url, listener);
+ super.subscribe(url, listener);
} catch (Exception e) {
List cachedUrls = getCachedUrls(url);
if (cachedUrls != null && cachedUrls.size() > 0) {
@@ -118,11 +118,11 @@ public void doSubscribe(URL url, NotifyListener listener) {
}
@Override
- public void doUnsubscribe(URL url, NotifyListener listener) {
+ public void unsubscribe(URL url, NotifyListener listener) {
removeForFailedSubAndUnsub(url, listener);
try {
- concreteUnsubscribe(url, listener);
+ super.unsubscribe(url, listener);
} catch (Exception e) {
if (isCheckingUrls(getUrl(), url)) {
throw new MotanFrameworkException(String.format("[%s] false to unsubscribe %s from %s", registryClassName, url, getUrl()),
@@ -134,9 +134,9 @@ public void doUnsubscribe(URL url, NotifyListener listener) {
@Override
@SuppressWarnings("unchecked")
- protected List doDiscover(URL url) {
+ public List discover(URL url) {
try {
- return concreteDiscover(url);
+ return super.discover(url);
} catch (Exception e) {
// 如果discover失败，返回一个empty list吧，毕竟是个下行动作，
LoggerUtil.warn(String.format("Failed to discover url:%s in registry (%s)", url, getUrl()), e);
@@ -179,7 +179,7 @@ private void retry() {
LoggerUtil.info("[{}] Retry register {}", registryClassName, failed);
try {
for (URL url : failed) {
- concreteRegister(url);
+ super.register(url);
failedRegistered.remove(url);
}
} catch (Exception e) {
@@ -193,7 +193,7 @@ private void retry() {
LoggerUtil.info("[{}] Retry unregister {}", registryClassName, failed);
try {
for (URL url : failed) {
- concreteUnregister(url);
+ super.unregister(url);
failedUnregistered.remove(url);
}
} catch (Exception e) {
@@ -216,7 +216,7 @@ private void retry() {
URL url = entry.getKey();
Set listeners = entry.getValue();
for (NotifyListener listener : listeners) {
- concreteSubscribe(url, listener);
+ super.subscribe(url, listener);
listeners.remove(listener);
}
}
@@ -240,7 +240,7 @@ private void retry() {
URL url = entry.getKey();
Set listeners = entry.getValue();
for (NotifyListener listener : listeners) {
- concreteUnsubscribe(url, listener);
+ super.unsubscribe(url, listener);
listeners.remove(listener);
}
}
@@ -253,13 +253,4 @@ private void retry() {
}
- protected abstract void concreteRegister(URL url);
-
- protected abstract void concreteUnregister(URL url);
-
- protected abstract void concreteSubscribe(URL url, NotifyListener listener);
-
- protected abstract void concreteUnsubscribe(URL url, NotifyListener listener);
-
- protected abstract List concreteDiscover(URL url);
}
diff --git a/motan-core/src/main/java/com/weibo/api/motan/registry/support/LocalRegistryService.java b/motan-core/src/main/java/com/weibo/api/motan/registry/support/LocalRegistryService.java
index 313da1756..f1487a2b4 100644
--- a/motan-core/src/main/java/com/weibo/api/motan/registry/support/LocalRegistryService.java
+++ b/motan-core/src/main/java/com/weibo/api/motan/registry/support/LocalRegistryService.java
@@ -104,6 +104,16 @@ public List doDiscover(URL url) {
return registeredServices.get(getRegistryKey(url));
}
+ @Override
+ protected void doAvailable(URL url) {
+ //do nothing
+ }
+
+ @Override
+ protected void doUnavailable(URL url) {
+ //do nothing
+ }
+
@Override
public void doRegister(URL url) {
String registryKey = getRegistryKey(url);
diff --git a/motan-core/src/main/java/com/weibo/api/motan/registry/support/command/CommandFailbackRegistry.java b/motan-core/src/main/java/com/weibo/api/motan/registry/support/command/CommandFailbackRegistry.java
index 82eb8a3b7..ca96537b2 100644
--- a/motan-core/src/main/java/com/weibo/api/motan/registry/support/command/CommandFailbackRegistry.java
+++ b/motan-core/src/main/java/com/weibo/api/motan/registry/support/command/CommandFailbackRegistry.java
@@ -38,7 +38,7 @@ public CommandFailbackRegistry(URL url) {
}
@Override
- protected void concreteSubscribe(URL url, final NotifyListener listener) {
+ protected void doSubscribe(URL url, final NotifyListener listener) {
LoggerUtil.info("CommandFailbackRegistry subscribe. url: " + url.toSimpleString());
URL urlCopy = url.createCopy();
CommandServiceManager manager = getCommandServiceManager(urlCopy);
@@ -47,14 +47,14 @@ protected void concreteSubscribe(URL url, final NotifyListener listener) {
subscribeService(urlCopy, manager);
subscribeCommand(urlCopy, manager);
- List urls = concreteDiscover(urlCopy);
+ List urls = doDiscover(urlCopy);
if (urls != null && urls.size() > 0) {
this.notify(urlCopy, listener, urls);
}
}
@Override
- protected void concreteUnsubscribe(URL url, NotifyListener listener) {
+ protected void doUnsubscribe(URL url, NotifyListener listener) {
LoggerUtil.info("CommandFailbackRegistry unsubscribe. url: " + url.toSimpleString());
URL urlCopy = url.createCopy();
CommandServiceManager manager = commandManagerMap.get(urlCopy);
@@ -66,7 +66,7 @@ protected void concreteUnsubscribe(URL url, NotifyListener listener) {
}
@Override
- protected List concreteDiscover(URL url) {
+ protected List doDiscover(URL url) {
LoggerUtil.info("CommandFailbackRegistry discover. url: " + url.toSimpleString());
List finalResult;
diff --git a/motan-core/src/main/java/com/weibo/api/motan/registry/support/command/CommandServiceManager.java b/motan-core/src/main/java/com/weibo/api/motan/registry/support/command/CommandServiceManager.java
index 76388f674..8fdb6a2b2 100644
--- a/motan-core/src/main/java/com/weibo/api/motan/registry/support/command/CommandServiceManager.java
+++ b/motan-core/src/main/java/com/weibo/api/motan/registry/support/command/CommandServiceManager.java
@@ -97,7 +97,7 @@ public void notifyService(URL serviceUrl, URL registryUrl, List urls) {
public void notifyCommand(URL serviceUrl, String commandString) {
LoggerUtil.info("CommandServiceManager notify command. service:" + serviceUrl.toSimpleString() + ", command:" + commandString);
- if (!MotanSwitcherUtil.switcherIsOpen(MOTAN_COMMAND_SWITCHER) || commandString == null) {
+ if (!MotanSwitcherUtil.isOpen(MOTAN_COMMAND_SWITCHER) || commandString == null) {
LoggerUtil.info("command reset empty since swither is close.");
commandString = "";
}
diff --git a/motan-core/src/main/java/com/weibo/api/motan/switcher/LocalSwitcherService.java b/motan-core/src/main/java/com/weibo/api/motan/switcher/LocalSwitcherService.java
index 709831096..bbdddc4c9 100644
--- a/motan-core/src/main/java/com/weibo/api/motan/switcher/LocalSwitcherService.java
+++ b/motan-core/src/main/java/com/weibo/api/motan/switcher/LocalSwitcherService.java
@@ -16,15 +16,16 @@
package com.weibo.api.motan.switcher;
+import com.weibo.api.motan.core.extension.SpiMeta;
+import com.weibo.api.motan.exception.MotanFrameworkException;
+
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import com.weibo.api.motan.core.extension.SpiMeta;
-import com.weibo.api.motan.exception.MotanFrameworkException;
-import com.weibo.api.motan.util.LoggerUtil;
-
/**
* @author maijunsheng
* @version 创建时间：2013-6-14
@@ -35,6 +36,8 @@ public class LocalSwitcherService implements SwitcherService {
private static ConcurrentMap switchers = new ConcurrentHashMap();
+ private Map> listenerMap = new ConcurrentHashMap();
+
@Override
public Switcher getSwitcher(String name) {
return switchers.get(name);
@@ -45,15 +48,7 @@ public List getAllSwitchers() {
return new ArrayList(switchers.values());
}
- public static Switcher getSwitcherStatic(String name) {
- return switchers.get(name);
- }
-
- public static List getAllSwitchersStatic() {
- return new ArrayList(switchers.values());
- }
-
- public static void putSwitcher(Switcher switcher) {
+ private void putSwitcher(Switcher switcher) {
if (switcher == null) {
throw new MotanFrameworkException("LocalSwitcherService addSwitcher Error: switcher is null");
}
@@ -61,49 +56,19 @@ public static void putSwitcher(Switcher switcher) {
switchers.put(switcher.getName(), switcher);
}
- public static void putSwitcher(String name, boolean on) {
- if (switchers.get(name) != null) {
- LoggerUtil.warn("LocalSwitcherService addSwitcher Error: switcher exists");
- return;
- }
- Switcher switcher = new Switcher(name, on);
-
- switchers.putIfAbsent(name, switcher);
- }
-
- public static void onSwitcher(String name) {
- Switcher switcher = switchers.get(name);
- if (switcher == null) {
- switcher = new Switcher(name, true);
- putSwitcher(switcher);
- }
-
- switcher.onSwitcher();
- }
-
- public static void offSwitcher(String name) {
- Switcher switcher = switchers.get(name);
- if (switcher == null) {
- switcher = new Switcher(name, false);
- putSwitcher(switcher);
- }
-
- switcher.offSwitcher();
- }
-
@Override
public void initSwitcher(String switcherName, boolean initialValue) {
- putSwitcher(switcherName, initialValue);
+ setValue(switcherName, initialValue);
}
@Override
- public boolean switcherIsOpen(String switcherName) {
+ public boolean isOpen(String switcherName) {
Switcher switcher = switchers.get(switcherName);
return switcher != null && switcher.isOn();
}
@Override
- public boolean switcherIsOpen(String switcherName, boolean defaultValue) {
+ public boolean isOpen(String switcherName, boolean defaultValue) {
Switcher switcher = switchers.get(switcherName);
if (switcher == null) {
switchers.putIfAbsent(switcherName, new Switcher(switcherName, defaultValue));
@@ -113,8 +78,43 @@ public boolean switcherIsOpen(String switcherName, boolean defaultValue) {
}
@Override
- public void setSwitcher(String switcherName, boolean value) {
+ public void setValue(String switcherName, boolean value) {
putSwitcher(new Switcher(switcherName, value));
+
+ List listeners = listenerMap.get(switcherName);
+ if(listeners != null) {
+ for (SwitcherListener listener : listeners) {
+ listener.onValueChanged(switcherName, value);
+ }
+ }
+ }
+
+ @Override
+ public void registerListener(String switcherName, SwitcherListener listener) {
+ synchronized (listenerMap) {
+ if (listenerMap.get(switcherName) == null) {
+ List listeners = Collections.synchronizedList(new ArrayList());
+ listeners.add(listener);
+ listenerMap.put(switcherName, listeners);
+ } else {
+ List listeners = listenerMap.get(switcherName);
+ if (!listeners.contains(listener)) {
+ listeners.add(listener);
+ }
+ }
+ }
+ }
+
+ @Override
+ public void unRegisterListener(String switcherName, SwitcherListener listener) {
+ synchronized (listenerMap) {
+ if (listener == null) {
+ listenerMap.remove(switcherName);
+ } else {
+ List listeners = listenerMap.get(switcherName);
+ listeners.remove(listener);
+ }
+ }
}
}
diff --git a/motan-core/src/main/java/com/weibo/api/motan/switcher/SwitcherListener.java b/motan-core/src/main/java/com/weibo/api/motan/switcher/SwitcherListener.java
new file mode 100644
index 000000000..f2e22d9d0
--- /dev/null
+++ b/motan-core/src/main/java/com/weibo/api/motan/switcher/SwitcherListener.java
@@ -0,0 +1,9 @@
+package com.weibo.api.motan.switcher;
+
+/**
+ * Created by axb on 16/4/25.
+ */
+public interface SwitcherListener {
+
+ void onValueChanged(String key,Boolean value);
+}
diff --git a/motan-core/src/main/java/com/weibo/api/motan/switcher/SwitcherService.java b/motan-core/src/main/java/com/weibo/api/motan/switcher/SwitcherService.java
index 8461e33d0..8916139d4 100644
--- a/motan-core/src/main/java/com/weibo/api/motan/switcher/SwitcherService.java
+++ b/motan-core/src/main/java/com/weibo/api/motan/switcher/SwitcherService.java
@@ -16,11 +16,11 @@
package com.weibo.api.motan.switcher;
-import java.util.List;
-
import com.weibo.api.motan.core.extension.Scope;
import com.weibo.api.motan.core.extension.Spi;
+import java.util.List;
+
/**
*
* @author maijunsheng
@@ -59,7 +59,7 @@ public interface SwitcherService {
* @param switcherName
* @return true ：设置来开关，并且开关值为true false：未设置开关或开关为false
*/
- boolean switcherIsOpen(String switcherName);
+ boolean isOpen(String switcherName);
/**
* 检查开关是否开启，如果开关不存在则将开关置默认值，并返回。
@@ -68,7 +68,7 @@ public interface SwitcherService {
* @param defaultValue
* @return 开关存在时返回开关值，开关不存在时设置开关为默认值，并返回默认值。
*/
- boolean switcherIsOpen(String switcherName, boolean defaultValue);
+ boolean isOpen(String switcherName, boolean defaultValue);
/**
* 设置开关状态。
@@ -76,7 +76,22 @@ public interface SwitcherService {
* @param switcherName
* @param value
*/
- void setSwitcher(String switcherName, boolean value);
+ void setValue(String switcherName, boolean value);
+ /**
+ * register a listener for switcher value change, register a listener twice will only fire once
+ *
+ * @param switcherName
+ * @param listener
+ */
+ void registerListener(String switcherName, SwitcherListener listener);
+
+ /**
+ * unregister a listener
+ *
+ * @param switcherName
+ * @param listener the listener to be unregistered, null for all listeners for this switcherName
+ */
+ void unRegisterListener(String switcherName, SwitcherListener listener);
}
diff --git a/motan-core/src/main/java/com/weibo/api/motan/util/MotanSwitcherUtil.java b/motan-core/src/main/java/com/weibo/api/motan/util/MotanSwitcherUtil.java
index 9b8eb71f1..d71f78b57 100644
--- a/motan-core/src/main/java/com/weibo/api/motan/util/MotanSwitcherUtil.java
+++ b/motan-core/src/main/java/com/weibo/api/motan/util/MotanSwitcherUtil.java
@@ -17,6 +17,7 @@
package com.weibo.api.motan.util;
import com.weibo.api.motan.switcher.LocalSwitcherService;
+import com.weibo.api.motan.switcher.SwitcherListener;
import com.weibo.api.motan.switcher.SwitcherService;
@@ -40,8 +41,8 @@ public static void initSwitcher(String switcherName, boolean initialValue) {
* @param switcherName
* @return true ：设置了开关，并且开关值为true false：未设置开关或开关为false
*/
- public static boolean switcherIsOpen(String switcherName) {
- return switcherService.switcherIsOpen(switcherName);
+ public static boolean isOpen(String switcherName) {
+ return switcherService.isOpen(switcherName);
}
/**
@@ -52,7 +53,7 @@ public static boolean switcherIsOpen(String switcherName) {
* @return 开关存在时返回开关值，开关不存在时设置开关为默认值，并返回默认值。
*/
public static boolean switcherIsOpenWithDefault(String switcherName, boolean defaultValue) {
- return switcherService.switcherIsOpen(switcherName, defaultValue);
+ return switcherService.isOpen(switcherName, defaultValue);
}
/**
@@ -62,8 +63,8 @@ public static boolean switcherIsOpenWithDefault(String switcherName, boolean def
* @param value
* @return
*/
- public static void setSwitcher(String switcherName, boolean value) {
- switcherService.setSwitcher(switcherName, value);
+ public static void setSwitcherValue(String switcherName, boolean value) {
+ switcherService.setValue(switcherName, value);
}
public static SwitcherService getSwitcherService() {
@@ -74,4 +75,8 @@ public static void setSwitcherService(SwitcherService switcherService) {
MotanSwitcherUtil.switcherService = switcherService;
}
+ public static void registerSwitcherListener(String switcherName, SwitcherListener listener) {
+ switcherService.registerListener(switcherName, listener);
+ }
+
}
diff --git a/motan-core/src/test/java/com/weibo/api/motan/filter/SwitcherFilterTest.java b/motan-core/src/test/java/com/weibo/api/motan/filter/SwitcherFilterTest.java
index fdd7599e0..29ca9f073 100644
--- a/motan-core/src/test/java/com/weibo/api/motan/filter/SwitcherFilterTest.java
+++ b/motan-core/src/test/java/com/weibo/api/motan/filter/SwitcherFilterTest.java
@@ -25,7 +25,7 @@
import com.weibo.api.motan.rpc.Request;
import com.weibo.api.motan.rpc.Response;
import com.weibo.api.motan.rpc.URL;
-import com.weibo.api.motan.switcher.LocalSwitcherService;
+import com.weibo.api.motan.util.MotanSwitcherUtil;
import com.weibo.api.motan.util.NetUtils;
import org.jmock.Expectations;
@@ -82,7 +82,7 @@ public void testFilter() {
}
public void testPutSwitcher() {
- LocalSwitcherService.putSwitcher("mock_class_name", true);
+ MotanSwitcherUtil.setSwitcherValue("mock_class_name", true);
mockery.checking(new Expectations() {
{
@@ -108,7 +108,7 @@ public void testPutSwitcher() {
}
public void testOnSwitcher() {
- LocalSwitcherService.onSwitcher("mock_class_name");
+ MotanSwitcherUtil.setSwitcherValue("mock_class_name",true);
mockery.checking(new Expectations() {
{
@@ -135,7 +135,7 @@ public void testOnSwitcher() {
assertNotNull(resultOnSwitcher);
assertTrue(resultOnSwitcher.getException().getMessage().contains("Request false for switcher is on"));
- LocalSwitcherService.offSwitcher("mock_class_name");
+ MotanSwitcherUtil.setSwitcherValue("mock_class_name",false);
Response resultOffSwitcher = filter.filter(caller, request);
diff --git a/motan-core/src/test/java/com/weibo/api/motan/protocol/rpc/CompressRpcCodecTest.java b/motan-core/src/test/java/com/weibo/api/motan/protocol/rpc/CompressRpcCodecTest.java
index 5e742eec5..c160def43 100644
--- a/motan-core/src/test/java/com/weibo/api/motan/protocol/rpc/CompressRpcCodecTest.java
+++ b/motan-core/src/test/java/com/weibo/api/motan/protocol/rpc/CompressRpcCodecTest.java
@@ -51,12 +51,12 @@ public class CompressRpcCodecTest extends DefaultRpcCodecTest {
@Before
public void setUp() throws Exception {
rpcCodec = new CompressRpcCodec();
- MotanSwitcherUtil.setSwitcher(CompressRpcCodec.CODEC_VERSION_SWITCHER, false);
+ MotanSwitcherUtil.setSwitcherValue(CompressRpcCodec.CODEC_VERSION_SWITCHER, false);
boolean isopen =
MotanSwitcherUtil.switcherIsOpenWithDefault(CompressRpcCodec.GROUP_CODEC_VERSION_SWITCHER + URLParamType.group.getValue(),
false);
if (isopen) {
- MotanSwitcherUtil.setSwitcher(CompressRpcCodec.GROUP_CODEC_VERSION_SWITCHER + URLParamType.group.getValue(), false);
+ MotanSwitcherUtil.setSwitcherValue(CompressRpcCodec.GROUP_CODEC_VERSION_SWITCHER + URLParamType.group.getValue(), false);
}
}
@@ -71,12 +71,12 @@ public void testSwitcher() throws IOException {
byte[] bytes = rpcCodec.encode(channel, request);
assertTrue(isCompressVersion(bytes));
// 整体开关测试
- MotanSwitcherUtil.setSwitcher(CompressRpcCodec.CODEC_VERSION_SWITCHER, true);
+ MotanSwitcherUtil.setSwitcherValue(CompressRpcCodec.CODEC_VERSION_SWITCHER, true);
bytes = rpcCodec.encode(channel, request);
assertTrue(isV1Version(bytes));
// 分组开关测试
- MotanSwitcherUtil.setSwitcher(CompressRpcCodec.CODEC_VERSION_SWITCHER, false);
- MotanSwitcherUtil.setSwitcher(CompressRpcCodec.GROUP_CODEC_VERSION_SWITCHER + URLParamType.group.getValue(), true);
+ MotanSwitcherUtil.setSwitcherValue(CompressRpcCodec.CODEC_VERSION_SWITCHER, false);
+ MotanSwitcherUtil.setSwitcherValue(CompressRpcCodec.GROUP_CODEC_VERSION_SWITCHER + URLParamType.group.getValue(), true);
bytes = rpcCodec.encode(channel, request);
assertTrue(isV1Version(bytes));
diff --git a/motan-core/src/test/java/com/weibo/api/motan/rpc/LocalSwitcherServiceTest.java b/motan-core/src/test/java/com/weibo/api/motan/rpc/LocalSwitcherServiceTest.java
index e1e810f5c..870feafee 100644
--- a/motan-core/src/test/java/com/weibo/api/motan/rpc/LocalSwitcherServiceTest.java
+++ b/motan-core/src/test/java/com/weibo/api/motan/rpc/LocalSwitcherServiceTest.java
@@ -37,16 +37,17 @@ public void testProtocolSwitcher() {
String protocolSwitcher = MotanConstants.PROTOCOL_SWITCHER_PREFIX + "motan";
- LocalSwitcherService.putSwitcher(new Switcher(protocolSwitcher, false));
+ LocalSwitcherService localSwitcherService = new LocalSwitcherService();
+ localSwitcherService.setValue(protocolSwitcher, false);
- Switcher switcher = LocalSwitcherService.getSwitcherStatic(protocolSwitcher);
+ Switcher switcher = localSwitcherService.getSwitcher(protocolSwitcher);
Assert.assertNotNull(switcher);
Assert.assertFalse(switcher.isOn());
- LocalSwitcherService.putSwitcher(new Switcher(protocolSwitcher, true));
+ localSwitcherService.setValue(protocolSwitcher, true);
- switcher = LocalSwitcherService.getSwitcherStatic(protocolSwitcher);
+ switcher = localSwitcherService.getSwitcher(protocolSwitcher);
Assert.assertNotNull(switcher);
Assert.assertTrue(switcher.isOn());
diff --git a/motan-demo/motan-demo-server/src/main/java/com/weibo/motan/demo/server/DemoRpcServer.java b/motan-demo/motan-demo-server/src/main/java/com/weibo/motan/demo/server/DemoRpcServer.java
index a2d709e4a..99cae2c39 100644
--- a/motan-demo/motan-demo-server/src/main/java/com/weibo/motan/demo/server/DemoRpcServer.java
+++ b/motan-demo/motan-demo-server/src/main/java/com/weibo/motan/demo/server/DemoRpcServer.java
@@ -16,7 +16,7 @@
package com.weibo.motan.demo.server;
-import com.weibo.api.motan.registry.consul.ConsulConstants;
+import com.weibo.api.motan.common.MotanConstants;
import com.weibo.api.motan.util.MotanSwitcherUtil;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -24,11 +24,10 @@
public class DemoRpcServer {
public static void main(String[] args) throws InterruptedException {
- ApplicationContext applicationContext = new ClassPathXmlApplicationContext(
- new String[]{"classpath*:motan_demo_server.xml"});
- MotanSwitcherUtil.setSwitcher(ConsulConstants.NAMING_PROCESS_HEARTBEAT_SWITCHER, true);
+ ApplicationContext applicationContext = new ClassPathXmlApplicationContext(new String[] {"classpath*:motan_demo_server.xml"});
+ MotanSwitcherUtil.setSwitcherValue(MotanConstants.REGISTRY_HEARTBEAT_SWITCHER, true);
System.out.println("server start...");
- Thread.sleep(Long.MAX_VALUE);
+
}
}
diff --git a/motan-demo/motan-demo-server/src/main/resources/motan_demo_server.xml b/motan-demo/motan-demo-server/src/main/resources/motan_demo_server.xml
index 16c8641c1..81a5f6549 100755
--- a/motan-demo/motan-demo-server/src/main/resources/motan_demo_server.xml
+++ b/motan-demo/motan-demo-server/src/main/resources/motan_demo_server.xml
@@ -31,22 +31,20 @@
-
-
+
+ ref="motanDemoServiceImpl" export="demoMotan:8001" basicService="serviceBasicConfig">
+ ref="motanDemoServiceImpl" export="demoMotan:8002" basicService="serviceBasicConfig">
diff --git a/motan-registry-consul/src/main/java/com/weibo/api/motan/registry/consul/ConsulHeartbeatManager.java b/motan-registry-consul/src/main/java/com/weibo/api/motan/registry/consul/ConsulHeartbeatManager.java
index c7a670fa3..cc2ccd0bb 100644
--- a/motan-registry-consul/src/main/java/com/weibo/api/motan/registry/consul/ConsulHeartbeatManager.java
+++ b/motan-registry-consul/src/main/java/com/weibo/api/motan/registry/consul/ConsulHeartbeatManager.java
@@ -30,6 +30,7 @@ public class ConsulHeartbeatManager {
private ScheduledExecutorService heartbeatExecutor;
// 上一次心跳开关的状态
private boolean lastHeartBeatSwitcherStatus = false;
+ private volatile boolean currentHeartBeatSwitcherStatus = false;
// 开关检查次数。
private int switcherCheckTimes = 0;
@@ -134,8 +135,11 @@ public void removeHeartbeatServiceId(String serviceid) {
// 检查心跳开关是否打开
private boolean isHeartbeatOpen() {
- return MotanSwitcherUtil
- .switcherIsOpen(ConsulConstants.NAMING_PROCESS_HEARTBEAT_SWITCHER);
+ return currentHeartBeatSwitcherStatus;
+ }
+
+ public void setHeartbeatOpen(boolean open) {
+ currentHeartBeatSwitcherStatus = open;
}
class HeartbeatJob implements Runnable {
@@ -170,4 +174,5 @@ public void setClient(MotanConsulClient client) {
this.client = client;
}
+
}
diff --git a/motan-registry-consul/src/main/java/com/weibo/api/motan/registry/consul/ConsulRegistry.java b/motan-registry-consul/src/main/java/com/weibo/api/motan/registry/consul/ConsulRegistry.java
index 0038a33df..be2bb50f6 100644
--- a/motan-registry-consul/src/main/java/com/weibo/api/motan/registry/consul/ConsulRegistry.java
+++ b/motan-registry-consul/src/main/java/com/weibo/api/motan/registry/consul/ConsulRegistry.java
@@ -1,5 +1,13 @@
package com.weibo.api.motan.registry.consul;
+import com.weibo.api.motan.common.MotanConstants;
+import com.weibo.api.motan.common.URLParamType;
+import com.weibo.api.motan.registry.NotifyListener;
+import com.weibo.api.motan.registry.consul.client.MotanConsulClient;
+import com.weibo.api.motan.registry.support.FailbackRegistry;
+import com.weibo.api.motan.rpc.URL;
+import com.weibo.api.motan.util.LoggerUtil;
+
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -10,22 +18,10 @@
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
-import org.apache.commons.lang3.StringUtils;
-
-import com.weibo.api.motan.common.MotanConstants;
-import com.weibo.api.motan.common.URLParamType;
-import com.weibo.api.motan.registry.NotifyListener;
-import com.weibo.api.motan.registry.consul.client.ConsulEcwidClient;
-import com.weibo.api.motan.registry.consul.client.MotanConsulClient;
-import com.weibo.api.motan.registry.support.FailbackRegistry;
-import com.weibo.api.motan.rpc.URL;
-import com.weibo.api.motan.util.LoggerUtil;
-
public class ConsulRegistry extends FailbackRegistry {
private MotanConsulClient client;
private ConsulHeartbeatManager heartbeatManager;
private int lookupInterval;
- private static ConcurrentHashMap registeredUrls = new ConcurrentHashMap();
/**
* consul service的本地缓存。 key为group（consul中的service），value为接口名与对应的url list
@@ -61,23 +57,23 @@ public ConsulRegistry(URL url, MotanConsulClient client) {
@Override
- protected void concreteRegister(URL url) {
+ protected void doRegister(URL url) {
ConsulService service = buildService(url);
client.registerService(service);
heartbeatManager.addHeartbeatServcieId(service.getId());
- registeredUrls.putIfAbsent(url, client);
+ getRegisteredServiceUrls().add(url);
}
@Override
- protected void concreteUnregister(URL url) {
+ protected void doUnregister(URL url) {
ConsulService service = buildService(url);
client.deregisterService(service.getId());
heartbeatManager.removeHeartbeatServiceId(service.getId());
- registeredUrls.remove(url);
+ getRegisteredServiceUrls().remove(url);
}
@Override
- protected void concreteSubscribe(URL url, NotifyListener listener) {
+ protected void doSubscribe(URL url, NotifyListener listener) {
addSubscribeListeners(url, listener);
startListenerThreadIfNewService(url);
List urls = discover(url); // 订阅后先查询一次对应的服务，供refer初始化使用。
@@ -126,7 +122,7 @@ private void addSubscribeListeners(URL url, NotifyListener listener) {
}
@Override
- protected void concreteUnsubscribe(URL url, NotifyListener listener) {
+ protected void doUnsubscribe(URL url, NotifyListener listener) {
HashMap clusterListeners = subscribeListeners
.get(getUrlClusterInfo(url));
@@ -138,7 +134,7 @@ protected void concreteUnsubscribe(URL url, NotifyListener listener) {
}
@Override
- protected List concreteDiscover(URL url) {
+ protected List doDiscover(URL url) {
String cluster = getUrlClusterInfo(url);
String group = url.getGroup();
List clusterUrl = new ArrayList();
@@ -161,9 +157,27 @@ protected List concreteDiscover(URL url) {
return clusterUrl;
}
- @Override
+ @Override
+ protected void doAvailable(URL url) {
+ if (url == null) {
+ heartbeatManager.setHeartbeatOpen(true);
+ } else {
+ throw new UnsupportedOperationException("consul registry not support available by urls yet");
+ }
+ }
+
+ @Override
+ protected void doUnavailable(URL url) {
+ if (url == null) {
+ heartbeatManager.setHeartbeatOpen(false);
+ } else {
+ throw new UnsupportedOperationException("consul registry not support unavailable by urls yet");
+ }
+ }
+
+ @Override
public List discover(URL url) {
- return concreteDiscover(url);
+ return doDiscover(url);
}
@Override
@@ -313,7 +327,7 @@ private static ConsulService buildService(URL url) {
/**
* 根据service生成motan使用的
*
- * @param healthService
+ * @param service
* @return
*/
private URL buildUrl(ConsulService service) {
@@ -401,11 +415,11 @@ public void setConsulClient(MotanConsulClient client) {
}
//重新对jvm中注册过的consul service进行注册。因本地agent异常导致已注册server丢失的情况使用。
- public static void reRegister(){
- for(Entry entry : registeredUrls.entrySet()){
- ConsulService service = buildService(entry.getKey());
- entry.getValue().registerService(service);
- }
+ public void reRegister(){
+ for(URL url:getRegisteredServiceUrls()) {
+ ConsulService service = buildService(url);
+ client.registerService(service);
+ }
}
}
diff --git a/motan-registry-consul/src/test/java/com/weibo/api/motan/registry/consul/ConsulHeartbeatManagerTest.java b/motan-registry-consul/src/test/java/com/weibo/api/motan/registry/consul/ConsulHeartbeatManagerTest.java
index 093989217..369064b02 100644
--- a/motan-registry-consul/src/test/java/com/weibo/api/motan/registry/consul/ConsulHeartbeatManagerTest.java
+++ b/motan-registry-consul/src/test/java/com/weibo/api/motan/registry/consul/ConsulHeartbeatManagerTest.java
@@ -78,8 +78,8 @@ private void checkHeartbeat(Map services, boolean start, int times
}
public void setHeartbeatSwitcher(boolean value) {
- MotanSwitcherUtil.setSwitcher(ConsulConstants.NAMING_PROCESS_HEARTBEAT_SWITCHER, value);
-
+ heartbeatManager.setHeartbeatOpen(value);
+
}
}
diff --git a/motan-registry-consul/src/test/java/com/weibo/api/motan/registry/consul/ConsulRegistryTest.java b/motan-registry-consul/src/test/java/com/weibo/api/motan/registry/consul/ConsulRegistryTest.java
index 696d65ef3..dbc1bb5e3 100644
--- a/motan-registry-consul/src/test/java/com/weibo/api/motan/registry/consul/ConsulRegistryTest.java
+++ b/motan-registry-consul/src/test/java/com/weibo/api/motan/registry/consul/ConsulRegistryTest.java
@@ -42,38 +42,38 @@ public void tearDown() throws Exception {
}
@Test
- public void testConcreteDiscover() {
+ public void testdoDiscover() {
URL referUrl = MockUtils.getMockUrl(0);
- List serviceUrls = consulRegistry.concreteDiscover(referUrl);
+ List serviceUrls = consulRegistry.doDiscover(referUrl);
assertEquals(serviceUrls.size(), client.getMockServiceNum());
}
@Test
- public void testConcreteRegisterAndUnregister() {
+ public void testdoRegisterAndUnregister() {
// register
URL serviceUrl1 = MockUtils.getMockUrl(123);
- consulRegistry.concreteRegister(serviceUrl1);
+ consulRegistry.doRegister(serviceUrl1);
String serviceid1 = ConsulUtils.convertConsulSerivceId(serviceUrl1);
assertTrue(client.isServiceRegister(serviceid1));
URL serviceUrl2 = MockUtils.getMockUrl(456);
String serviceid2 = ConsulUtils.convertConsulSerivceId(serviceUrl2);
- consulRegistry.concreteRegister(serviceUrl2);
+ consulRegistry.doRegister(serviceUrl2);
assertTrue(client.isServiceRegister(serviceid2));
// unregister
- consulRegistry.concreteUnregister(serviceUrl1);
+ consulRegistry.doUnregister(serviceUrl1);
assertFalse(client.isServiceRegister(serviceid1));
assertTrue(client.isServiceRegister(serviceid2));
- consulRegistry.concreteUnregister(serviceUrl2);
+ consulRegistry.doUnregister(serviceUrl2);
assertFalse(client.isServiceRegister(serviceid2));
}
@Test
- public void testConcreteSubscribeAndUnsubscribe() throws InterruptedException {
+ public void testdoSubscribeAndUnsubscribe() throws InterruptedException {
URL referUrl = MockUtils.getMockUrl(0);
NotifyListener listener = new NotifyListener() {
@@ -85,7 +85,7 @@ public void notify(URL registryUrl, List urls) {
}
};
- consulRegistry.concreteSubscribe(referUrl, listener);
+ consulRegistry.doSubscribe(referUrl, listener);
Thread.sleep(200);
long lastIndex = client.getMockIndex();
@@ -103,20 +103,20 @@ public void notify(URL registryUrl, List urls) {
@Test
public void testReRegister() {
URL serviceUrl1 = MockUtils.getMockUrl(123);
- consulRegistry.concreteRegister(serviceUrl1);
+ consulRegistry.doRegister(serviceUrl1);
String serviceid1 = ConsulUtils.convertConsulSerivceId(serviceUrl1);
assertTrue(client.isServiceRegister(serviceid1));
URL serviceUrl2 = MockUtils.getMockUrl(456);
String serviceid2 = ConsulUtils.convertConsulSerivceId(serviceUrl2);
- consulRegistry.concreteRegister(serviceUrl2);
+ consulRegistry.doRegister(serviceUrl2);
assertTrue(client.isServiceRegister(serviceid2));
client.removeService(serviceid1);
client.removeService(serviceid2);
assertFalse(client.isServiceRegister(serviceid1));
assertFalse(client.isServiceRegister(serviceid2));
- ConsulRegistry.reRegister();
+ consulRegistry.reRegister();
assertTrue(client.isServiceRegister(serviceid1));
assertTrue(client.isServiceRegister(serviceid2));
diff --git a/motan-registry-zookeeper/src/main/java/com/weibo/api/motan/registry/zookeeper/ZookeeperRegistry.java b/motan-registry-zookeeper/src/main/java/com/weibo/api/motan/registry/zookeeper/ZookeeperRegistry.java
index 41d9b0e2c..6b6b4a8a4 100644
--- a/motan-registry-zookeeper/src/main/java/com/weibo/api/motan/registry/zookeeper/ZookeeperRegistry.java
+++ b/motan-registry-zookeeper/src/main/java/com/weibo/api/motan/registry/zookeeper/ZookeeperRegistry.java
@@ -69,7 +69,7 @@ public ConcurrentHashMap childChangeListeners = urlListeners.get(url);
if (childChangeListeners == null) {
urlListeners.putIfAbsent(url, new ConcurrentHashMap());
@@ -132,7 +132,7 @@ public void handleChildChange(String parentPath, List currentChilds) thr
}
@Override
- protected void concreteUnsubscribe(URL url, NotifyListener notifyListener) {
+ protected void doUnsubscribe(URL url, NotifyListener notifyListener) {
Map childChangeListeners = urlListeners.get(url);
if (childChangeListeners != null) {
IZkChildListener zkChildListener = childChangeListeners.get(notifyListener);
@@ -144,10 +144,22 @@ protected void concreteUnsubscribe(URL url, NotifyListener notifyListener) {
}
@Override
- protected List concreteDiscover(URL url) {
+ protected List doDiscover(URL url) {
return nodeChildsToUrls(toServerTypePath(url));
}
+ @Override
+ protected void doAvailable(URL url) {
+ //TODO implement for zk
+
+ }
+
+ @Override
+ protected void doUnavailable(URL url) {
+ //TODO implement for zk
+
+ }
+
private List nodeChildsToUrls(String parentPath, List currentChilds) {
List urls = new ArrayList();
for (String node : currentChilds) {
diff --git a/motan-registry-zookeeper/src/test/java/com/weibo/api/motan/registry/zookeeper/ZookeeperRegistryTest.java b/motan-registry-zookeeper/src/test/java/com/weibo/api/motan/registry/zookeeper/ZookeeperRegistryTest.java
index 24096d9b9..cc6d7d79d 100644
--- a/motan-registry-zookeeper/src/test/java/com/weibo/api/motan/registry/zookeeper/ZookeeperRegistryTest.java
+++ b/motan-registry-zookeeper/src/test/java/com/weibo/api/motan/registry/zookeeper/ZookeeperRegistryTest.java
@@ -76,7 +76,7 @@ public void setUp() throws Exception {
}
@Test
- public void testConcreteRegister() {
+ public void testDoRegister() {
URL url = new URL(MotanConstants.PROTOCOL_MOTAN, "127.0.0.1", 8001, "com.weibo.motan.demo.service.MotanDemoService");
registry.register(url);
@@ -85,15 +85,15 @@ public void testConcreteRegister() {
}
@Test
- public void testConcreteUnregister() {
+ public void testDoUnregister() {
URL url = new URL(MotanConstants.PROTOCOL_MOTAN, "127.0.0.1", 8001, "com.weibo.motan.demo.service.MotanDemoService");
- registry.concreteUnregister(url);
+ registry.doUnregister(url);
Set registeredUrls = registry.getRegisteredUrls();
assertFalse(registeredUrls.contains(url));
}
@Test
- public void testConcreteSubscribe() {
+ public void testDoSubscribe() {
URL url = new URL(MotanConstants.PROTOCOL_MOTAN, "127.0.0.1", 0, "com.weibo.motan.demo.service.MotanDemoService");
NotifyListener notifyListener = new NotifyListener() {
@Override
@@ -101,13 +101,13 @@ public void notify(URL registryUrl, List urls) {
}
};
- registry.concreteSubscribe(url, notifyListener);
+ registry.doSubscribe(url, notifyListener);
ConcurrentHashMap> urlListeners = registry.getUrlListeners();
assertTrue(urlListeners.containsKey(url));
}
@Test
- public void testConcreteUnsubscribe() {
+ public void testDoUnsubscribe() {
URL url = new URL(MotanConstants.PROTOCOL_MOTAN, "127.0.0.1", 0, "com.weibo.motan.demo.service.MotanDemoService");
NotifyListener notifyListener = new NotifyListener() {
@Override
@@ -115,15 +115,15 @@ public void notify(URL registryUrl, List urls) {
}
};
- registry.concreteUnsubscribe(url, notifyListener);
+ registry.doUnsubscribe(url, notifyListener);
ConcurrentHashMap> urlListeners = registry.getUrlListeners();
assertFalse(urlListeners.containsKey(url));
}
@Test
- public void testConcreteDiscover() {
+ public void testDoDiscover() {
URL url = new URL(MotanConstants.PROTOCOL_MOTAN, "127.0.0.1", 0, "com.weibo.motan.demo.service.MotanDemoService");
- registry.concreteDiscover(url);
+ registry.doDiscover(url);
}
diff --git a/motan-springsupport/src/test/java/com/weibo/api/motan/config/springsupport/MockRegistryFactory.java b/motan-springsupport/src/test/java/com/weibo/api/motan/config/springsupport/MockRegistryFactory.java
index 67560853c..fe8faa958 100644
--- a/motan-springsupport/src/test/java/com/weibo/api/motan/config/springsupport/MockRegistryFactory.java
+++ b/motan-springsupport/src/test/java/com/weibo/api/motan/config/springsupport/MockRegistryFactory.java
@@ -17,6 +17,7 @@
package com.weibo.api.motan.config.springsupport;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.List;
import com.weibo.api.motan.core.extension.SpiMeta;
@@ -39,6 +40,21 @@ public void register(URL url) {}
@Override
public void unregister(URL url) {}
+ @Override
+ public void available(URL url) {
+
+ }
+
+ @Override
+ public void unavailable(URL url) {
+
+ }
+
+ @Override
+ public Collection getRegisteredServiceUrls() {
+ return null;
+ }
+
@Override
public void subscribe(URL url, NotifyListener listener) {}