DUBBO-350 注册中心客户端服务列表通知问题

git-svn-id: http://code.alibabatech.com/svn/dubbo/trunk@1690 1a56cb94-b969-4eaa-88fa-be21384802f2

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/AbstractClusterInvoker.java

@@ -216,7 +216,7 @@ public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
 
         LoadBalance loadbalance;
         
-        List<Invoker<T>> invokers = directory.list(invocation);
+        List<Invoker<T>> invokers = list(invocation);
         if (invokers != null && invokers.size() > 0) {
             loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
                     .getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
@@ -230,9 +230,9 @@ public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
     protected void checkWheatherDestoried() {
 
         if(destroyed){
-            throw new RpcException("Rpc invoker for " + getInterface() + " on consumer " + NetUtils.getLocalHost()
+            throw new RpcException("Rpc cluster invoker for " + getInterface() + " on consumer " + NetUtils.getLocalHost()
                     + " use dubbo version " + Version.getVersion()
-                    + " is not destroyed! Can not invoke any more.");
+                    + " is now destroyed! Can not invoke any more.");
         }
     }
 
@@ -256,4 +256,8 @@ public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
     protected abstract Result doInvoke(Invocation invocation, List<Invoker<T>> invokers,
                                        LoadBalance loadbalance) throws RpcException;
     
+    protected  List<Invoker<T>> list(Invocation invocation) throws RpcException {
+    	List<Invoker<T>> invokers = directory.list(invocation);
+    	return invokers;
+    }
 }
\ No newline at end of file

dubbo-cluster/src/main/java/com/alibaba/dubbo/rpc/cluster/support/FailoverClusterInvoker.java

@@ -39,6 +39,7 @@ import com.alibaba.dubbo.rpc.cluster.LoadBalance;
  * <a href="http://en.wikipedia.org/wiki/Failover">Failover</a>
  * 
  * @author william.liangf
+ * @author chao.liuc
  */
 public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
 
@@ -49,18 +50,27 @@ public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
     }
 
     @SuppressWarnings({ "unchecked", "rawtypes" })
-    public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
-        checkInvokers(invokers, invocation);
+    public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
+    	List<Invoker<T>> copyinvokers = invokers;
+    	checkInvokers(copyinvokers, invocation);
         int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1;
         if (len <= 0) {
             len = 1;
         }
         // retry loop.
         RpcException le = null; // last exception.
-        List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(invokers.size()); // invoked invokers.
+        List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
         Set<String> providers = new HashSet<String>(len);
         for (int i = 0; i < len; i++) {
-            Invoker<T> invoker = select(loadbalance, invocation, invokers, invoked);
+        	//重试时，进行重新选择，避免重试时invoker列表已发生变化.
+        	//注意：如果列表发生了变化，那么invoked判断会失效，因为invoker示例已经改变
+        	if (i > 0) {
+        		checkWheatherDestoried();
+        		copyinvokers = list(invocation);
+        		//重新检查一下
+        		checkInvokers(copyinvokers, invocation);
+        	}
+            Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
             invoked.add(invoker);
             RpcContext.getContext().setInvokers((List)invoked);
             try {
@@ -70,7 +80,7 @@ public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
                             + " in the service " + getInterface().getName()
                             + " was successful by the provider " + invoker.getUrl().getAddress()
                             + ", but there have been failed providers " + providers 
-                            + " (" + providers.size() + "/" + invokers.size()
+                            + " (" + providers.size() + "/" + copyinvokers.size()
                             + ") from the registry " + directory.getUrl().getAddress()
                             + " on the consumer " + NetUtils.getLocalHost()
                             + " using the dubbo version " + Version.getVersion() + ". Last error is: "
@@ -91,7 +101,7 @@ public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
         throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method "
                 + invocation.getMethodName() + " in the service " + getInterface().getName() 
                 + ". Tried " + len + " times of the providers " + providers 
-                + " (" + providers.size() + "/" + invokers.size() 
+                + " (" + providers.size() + "/" + copyinvokers.size() 
                 + ") from the registry " + directory.getUrl().getAddress()
                 + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version "
                 + Version.getVersion() + ". Last error is: "

dubbo-cluster/src/test/java/com/alibaba/dubbo/rpc/cluster/support/FailoverClusterInvokerTest.java

@@ -23,6 +23,7 @@ import static org.junit.Assert.fail;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Callable;
 
 import org.easymock.EasyMock;
 import org.junit.Before;
@@ -36,6 +37,8 @@ import com.alibaba.dubbo.rpc.RpcException;
 import com.alibaba.dubbo.rpc.RpcInvocation;
 import com.alibaba.dubbo.rpc.RpcResult;
 import com.alibaba.dubbo.rpc.cluster.Directory;
+import com.alibaba.dubbo.rpc.cluster.directory.StaticDirectory;
+import com.alibaba.dubbo.rpc.protocol.AbstractInvoker;
 
 /**
  * FailoverClusterInvokerTest
@@ -172,4 +175,95 @@ public class FailoverClusterInvokerTest {
         }
     }
     
+    /**
+     * 测试在调用重试过程中，directory列表变更，invoke重试时重新进行list选择 
+     */
+    @Test
+    public void testInvokerDestoryAndReList(){
+    	final URL url = URL.valueOf("test://localhost/"+ Demo.class.getName() + "?loadbalance=roundrobin&retries="+retries);
+    	RpcException exception = new RpcException(RpcException.TIMEOUT_EXCEPTION);
+    	MockInvoker<Demo> invoker1 = new MockInvoker<Demo>(Demo.class, url);
+    	invoker1.setException(exception);
+    	
+    	MockInvoker<Demo> invoker2 = new MockInvoker<Demo>(Demo.class, url);
+    	invoker2.setException(exception);
+    	
+        final List<Invoker<Demo>> invokers = new ArrayList<Invoker<Demo>>();
+        invokers.add(invoker1);
+        invokers.add(invoker2);
+        
+        Callable<Object> callable = new Callable<Object>() {
+			public Object call() throws Exception {
+				//模拟invoker全部被destroy掉
+				for (Invoker<Demo> invoker:invokers){
+					invoker.destroy();
+				}
+				invokers.clear();
+				MockInvoker<Demo> invoker3  = new MockInvoker<Demo>(Demo.class, url);
+				invokers.add(invoker3);
+				return null;
+			}
+		};
+		invoker1.setCallable(callable);
+        invoker2.setCallable(callable);
+        
+        RpcInvocation inv = new RpcInvocation();
+        inv.setMethodName("test");
+        
+        Directory<Demo> dic = new MockDirectory<Demo>(url, invokers);
+        
+        FailoverClusterInvoker<Demo> clusterinvoker = new FailoverClusterInvoker<Demo>(dic);
+        clusterinvoker.invoke(inv);
+    }
+    
+    public static interface Demo{}
+    
+    public static class MockInvoker<T> extends AbstractInvoker<T> {
+    	URL url;
+    	boolean available = true;
+    	boolean destoryed = false;
+    	Result result ;
+    	RpcException exception;
+    	Callable<?> callable;
+    	
+		public MockInvoker(Class<T> type, URL url) {
+			super(type, url);
+		}
+
+		public void setResult(Result result) {
+			this.result = result;
+		}
+		public void setException(RpcException exception) {
+			this.exception = exception;
+		}
+		public void setCallable(Callable<?> callable) {
+			this.callable = callable;
+		}
+
+		@Override
+		protected Result doInvoke(Invocation invocation) throws Throwable {
+			if (callable != null) {
+				try {
+					callable.call();
+				} catch (Exception e) {
+					throw new RpcException(e);
+				}
+			}
+			if (exception != null) {
+				throw exception;
+			} else {
+				return result;
+			}
+		}
+    }
+    
+    public class MockDirectory<T> extends StaticDirectory<T> {
+		public MockDirectory(URL url , List<Invoker<T>> invokers) {
+			super(url, invokers);
+		}
+		@Override
+		protected List<Invoker<T>> doList(Invocation invocation) throws RpcException {
+			return new ArrayList<Invoker<T>>(super.doList(invocation));
+		}
+    }
 }
\ No newline at end of file