DUBBO-71 还原错误的提交
git-svn-id: http://code.alibabatech.com/svn/dubbo/trunk@434 1a56cb94-b969-4eaa-88fa-be21384802f2
This commit is contained in:
chao.liuc
parent
0648274ff2
commit
92334b08b5
|
|
@ -0,0 +1,138 @@
|
|||
/*
|
||||
* Copyright 1999-2011 Alibaba Group.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package com.alibaba.dubbo.rpc.protocol.dubbo;
|
||||
|
||||
import com.alibaba.dubbo.common.Constants;
|
||||
import com.alibaba.dubbo.common.URL;
|
||||
import com.alibaba.dubbo.common.utils.AtomicPositiveInteger;
|
||||
import com.alibaba.dubbo.remoting.RemotingException;
|
||||
import com.alibaba.dubbo.remoting.TimeoutException;
|
||||
import com.alibaba.dubbo.remoting.exchange.ExchangeClient;
|
||||
import com.alibaba.dubbo.remoting.exchange.ResponseFuture;
|
||||
import com.alibaba.dubbo.rpc.Invocation;
|
||||
import com.alibaba.dubbo.rpc.Result;
|
||||
import com.alibaba.dubbo.rpc.RpcConstants;
|
||||
import com.alibaba.dubbo.rpc.RpcContext;
|
||||
import com.alibaba.dubbo.rpc.RpcException;
|
||||
import com.alibaba.dubbo.rpc.RpcInvocation;
|
||||
import com.alibaba.dubbo.rpc.RpcResult;
|
||||
import com.alibaba.dubbo.rpc.protocol.AbstractInvoker;
|
||||
|
||||
/**
|
||||
* DubboInvoker
|
||||
*
|
||||
* @author william.liangf
|
||||
* @author chao.liuc
|
||||
*/
|
||||
public class DubboInvoker<T> extends AbstractInvoker<T> {
|
||||
|
||||
private final ExchangeClient[] clients;
|
||||
|
||||
private final AtomicPositiveInteger index = new AtomicPositiveInteger();
|
||||
|
||||
private final String version;
|
||||
|
||||
public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients){
|
||||
super(serviceType, url, new String[] {Constants.GROUP_KEY, Constants.TOKEN_KEY, Constants.TIMEOUT_KEY});
|
||||
this.clients = clients;
|
||||
// get version.
|
||||
this.version = url.getParameter(Constants.VERSION_KEY, "0.0.0");
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Result doInvoke(final Invocation invocation) throws Throwable {
|
||||
RpcInvocation inv = null;
|
||||
final String methodName ;
|
||||
if(Constants.$INVOKE.equals(invocation.getMethodName()) && invocation.getArguments() !=null && invocation.getArguments().length >0 && invocation.getArguments()[0] != null){
|
||||
inv = (RpcInvocation) invocation;
|
||||
//the frist argument must be real method name;
|
||||
methodName = invocation.getArguments()[0].toString();
|
||||
}else {
|
||||
inv = new RpcInvocation(invocation.getMethodName(), invocation.getParameterTypes(),
|
||||
invocation.getArguments(), invocation.getAttachments());
|
||||
methodName = invocation.getMethodName();
|
||||
}
|
||||
inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
|
||||
inv.setAttachment(Constants.VERSION_KEY, version);
|
||||
|
||||
ExchangeClient currentClient;
|
||||
if (clients.length == 1) {
|
||||
currentClient = clients[0];
|
||||
} else {
|
||||
currentClient = clients[index.getAndIncrement() % clients.length];
|
||||
}
|
||||
try {
|
||||
// 不可靠异步
|
||||
boolean isAsync = getUrl().getMethodParameter(methodName, Constants.ASYNC_KEY, false);
|
||||
int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT);
|
||||
if (isAsync) {
|
||||
boolean isReturn = getUrl().getMethodParameter(methodName, RpcConstants.RETURN_KEY, true);
|
||||
if (isReturn) {
|
||||
ResponseFuture future = currentClient.request(inv, timeout) ;
|
||||
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
|
||||
} else {
|
||||
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
|
||||
currentClient.send(inv, isSent);
|
||||
RpcContext.getContext().setFuture(null);
|
||||
}
|
||||
return new RpcResult();
|
||||
}
|
||||
RpcContext.getContext().setFuture(null);
|
||||
return (Result) currentClient.request(inv, timeout).get();
|
||||
} catch (TimeoutException e) {
|
||||
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, e.getMessage(), e);
|
||||
} catch (RemotingException e) {
|
||||
throw new RpcException(RpcException.NETWORK_EXCEPTION, e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isAvailable() {
|
||||
if (!super.isAvailable())
|
||||
return false;
|
||||
for (ExchangeClient client : clients){
|
||||
if (client.isConnected()){
|
||||
boolean isLazy = client.getUrl().getParameter(RpcConstants.LAZY_CONNECT_KEY, false);
|
||||
//cannot write == not Available ?
|
||||
if (! isLazy ) {
|
||||
return ! client.hasAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY);
|
||||
} else if (client instanceof LazyConnectExchangeClient) {
|
||||
LazyConnectExchangeClient lazyClient = (LazyConnectExchangeClient) client;
|
||||
if (lazyClient.isInited() && lazyClient.hasAttribute(Constants.CHANNEL_ATTRIBUTE_READONLY_KEY)){
|
||||
return false;
|
||||
} else {
|
||||
return true;
|
||||
}
|
||||
} else {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
public void destroy() {
|
||||
super.destroy();
|
||||
for (ExchangeClient client : clients) {
|
||||
try {
|
||||
client.close();
|
||||
} catch (Throwable t) {
|
||||
logger.warn(t.getMessage(), t);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -37,7 +37,6 @@ import com.alibaba.dubbo.rpc.RpcConstants;
|
|||
*
|
||||
* @author chao.liuc
|
||||
*/
|
||||
@SuppressWarnings("deprecation")
|
||||
final class LazyConnectExchangeClient implements ExchangeClient {
|
||||
|
||||
private final static Logger logger = LoggerFactory.getLogger(LazyConnectExchangeClient.class);
|
||||
|
|
@ -109,7 +108,8 @@ final class LazyConnectExchangeClient implements ExchangeClient {
|
|||
}
|
||||
|
||||
public ExchangeHandler getExchangeHandler() {
|
||||
return requestHandler;
|
||||
checkClient();
|
||||
return client.getExchangeHandler();
|
||||
}
|
||||
|
||||
public void send(Object message) throws RemotingException {
|
||||
|
|
@ -155,11 +155,8 @@ final class LazyConnectExchangeClient implements ExchangeClient {
|
|||
}
|
||||
|
||||
public Object getAttribute(String key) {
|
||||
if (client == null){
|
||||
return null;
|
||||
} else {
|
||||
return client.getAttribute(key);
|
||||
}
|
||||
checkClient();
|
||||
return client.getAttribute(key);
|
||||
}
|
||||
|
||||
public void setAttribute(String key, Object value) {
|
||||
|
|
@ -173,11 +170,8 @@ final class LazyConnectExchangeClient implements ExchangeClient {
|
|||
}
|
||||
|
||||
public boolean hasAttribute(String key) {
|
||||
if (client == null){
|
||||
return false;
|
||||
} else {
|
||||
return client.hasAttribute(key);
|
||||
}
|
||||
checkClient();
|
||||
return client.hasAttribute(key);
|
||||
}
|
||||
|
||||
private void checkClient() {
|
||||
|
|
@ -186,4 +180,8 @@ final class LazyConnectExchangeClient implements ExchangeClient {
|
|||
"LazyConnectExchangeClient state error. the client has not be init .url:" + url);
|
||||
}
|
||||
}
|
||||
|
||||
public boolean isInited(){
|
||||
return client != null ;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,160 @@
|
|||
/*
|
||||
* Copyright 1999-2011 Alibaba Group.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package com.alibaba.dubbo.rpc.protocol;
|
||||
|
||||
import java.lang.reflect.InvocationTargetException;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import com.alibaba.dubbo.common.URL;
|
||||
import com.alibaba.dubbo.common.Version;
|
||||
import com.alibaba.dubbo.common.logger.Logger;
|
||||
import com.alibaba.dubbo.common.logger.LoggerFactory;
|
||||
import com.alibaba.dubbo.common.utils.NetUtils;
|
||||
import com.alibaba.dubbo.rpc.Invoker;
|
||||
import com.alibaba.dubbo.rpc.RpcContext;
|
||||
import com.alibaba.dubbo.rpc.RpcException;
|
||||
import com.alibaba.dubbo.rpc.Invocation;
|
||||
import com.alibaba.dubbo.rpc.Result;
|
||||
import com.alibaba.dubbo.rpc.RpcInvocation;
|
||||
import com.alibaba.dubbo.rpc.RpcResult;
|
||||
|
||||
/**
|
||||
* AbstractInvoker.
|
||||
*
|
||||
* @author qian.lei
|
||||
* @author william.liangf
|
||||
*/
|
||||
public abstract class AbstractInvoker<T> implements Invoker<T> {
|
||||
|
||||
protected final Logger logger = LoggerFactory.getLogger(getClass());
|
||||
|
||||
private final Class<T> type;
|
||||
|
||||
private final URL url;
|
||||
|
||||
private final Map<String, String> attachment;
|
||||
|
||||
private volatile boolean available = true;
|
||||
|
||||
private volatile boolean destroyed = false;
|
||||
|
||||
public AbstractInvoker(Class<T> type, URL url){
|
||||
this(type, url, (Map<String, String>) null);
|
||||
}
|
||||
|
||||
public AbstractInvoker(Class<T> type, URL url, String[] keys) {
|
||||
this(type, url, convertAttachment(url, keys));
|
||||
}
|
||||
|
||||
public AbstractInvoker(Class<T> type, URL url, Map<String, String> attachment) {
|
||||
if (type == null)
|
||||
throw new IllegalArgumentException("service type == null");
|
||||
if (url == null)
|
||||
throw new IllegalArgumentException("service url == null");
|
||||
this.type = type;
|
||||
this.url = url;
|
||||
this.attachment = attachment == null ? null : Collections.unmodifiableMap(attachment);
|
||||
}
|
||||
|
||||
private static Map<String, String> convertAttachment(URL url, String[] keys) {
|
||||
if (keys == null || keys.length == 0) {
|
||||
return null;
|
||||
}
|
||||
Map<String, String> attachment = new HashMap<String, String>();
|
||||
for (String key : keys) {
|
||||
String value = url.getParameter(key);
|
||||
if (value != null && value.length() > 0) {
|
||||
attachment.put(key, value);
|
||||
}
|
||||
}
|
||||
return attachment;
|
||||
}
|
||||
|
||||
public Class<T> getInterface() {
|
||||
return type;
|
||||
}
|
||||
|
||||
public URL getUrl() {
|
||||
return url;
|
||||
}
|
||||
|
||||
public boolean isAvailable() {
|
||||
return available;
|
||||
}
|
||||
|
||||
protected void setAvailable(boolean available) {
|
||||
this.available = available;
|
||||
}
|
||||
|
||||
public void destroy() {
|
||||
if (destroyed) {
|
||||
return;
|
||||
}
|
||||
destroyed = true;
|
||||
setAvailable(false);
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return getInterface() + " -> " + getUrl()==null?" ":getUrl().toString();
|
||||
}
|
||||
|
||||
public Result invoke(Invocation inv) throws RpcException {
|
||||
if(destroyed) {
|
||||
throw new RpcException("Rpc invoker for service " + this + " on consumer " + NetUtils.getLocalHost()
|
||||
+ " use dubbo version " + Version.getVersion()
|
||||
+ " is DESTROYED, can not be invoked any more!");
|
||||
}
|
||||
RpcInvocation invocation = (RpcInvocation) inv;
|
||||
Map<String, String> attachments = new HashMap<String, String>();
|
||||
if (attachment != null && attachment.size() > 0) {
|
||||
attachments.putAll(attachment);
|
||||
}
|
||||
Map<String, String> context = RpcContext.getContext().getAttachments();
|
||||
if (context != null) {
|
||||
attachments.putAll(context);
|
||||
}
|
||||
if (invocation.getAttachments() != null) {
|
||||
attachments.putAll(invocation.getAttachments());
|
||||
}
|
||||
invocation.setAttachments(attachments);
|
||||
try {
|
||||
return doInvoke(invocation);
|
||||
} catch (InvocationTargetException e) { // biz exception
|
||||
Throwable te = e.getTargetException();
|
||||
if (te == null) {
|
||||
return new RpcResult(e);
|
||||
} else {
|
||||
if (te instanceof RpcException) {
|
||||
((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
|
||||
}
|
||||
return new RpcResult(te);
|
||||
}
|
||||
} catch (RpcException e) {
|
||||
if (e.isBiz()) {
|
||||
return new RpcResult(e);
|
||||
} else {
|
||||
throw e;
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
return new RpcResult(e);
|
||||
}
|
||||
}
|
||||
|
||||
protected abstract Result doInvoke(Invocation invocation) throws Throwable;
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue