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

Support triple echo and fix conn bug (#7292) 

* Add sync connect to avoid connect failed error

* Support echo service
This commit is contained in:
GuoHao 2021-03-01 11:25:58 +08:00 committed by GitHub
parent 0a8747a678
commit 445cc82642
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 13 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/api/Connection.java
View File

@ -66,6 +66,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou
private final InetSocketAddress remote;
private final AtomicBoolean closed = new AtomicBoolean(false);
private final AtomicReference<Channel> channel = new AtomicReference<>();
private final ChannelFuture initConnectFuture;
public Connection(URL url) {
url = ExecutorUtil.setThreadName(url, "DubboClientHandler");
@ -76,7 +77,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou
this.closeFuture = new DefaultPromise<>(GlobalEventExecutor.INSTANCE);
this.remote = getConnectAddress();
final Bootstrap bootstrap = open();
bootstrap.connect();
this.initConnectFuture = bootstrap.connect();
}
public static Connection getConnectionFromChannel(Channel channel) {
@ -124,7 +125,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou
@Override
public String toString() {
return "(Ref=" + ReferenceCountUtil.refCnt(this) + ",local="+(getChannel()==null?null:getChannel().localAddress())+",remote="+getRemote();
return "(Ref=" + ReferenceCountUtil.refCnt(this) + ",local=" + (getChannel() == null ? null : getChannel().localAddress()) + ",remote=" + getRemote();
}
public void onGoaway(Channel channel) {
@ -143,6 +144,9 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou
}
}
public void connectSync(){
this.initConnectFuture.awaitUninterruptibly();
}
public boolean isAvailable() {
final Channel channel = getChannel();
return channel != null && channel.isActive();

6
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerStream.java
View File

@ -34,6 +34,7 @@ import org.apache.dubbo.rpc.model.ProviderModel;
import org.apache.dubbo.rpc.model.ServiceDescriptor;
import org.apache.dubbo.rpc.model.ServiceRepository;
import org.apache.dubbo.rpc.protocol.tri.GrpcStatus.Code;
import org.apache.dubbo.rpc.service.EchoService;
import org.apache.dubbo.rpc.service.GenericService;
import org.apache.dubbo.triple.TripleWrapper;
@ -216,7 +217,10 @@ public class ServerStream extends AbstractStream implements Stream {
if (CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals(methodName)) {
this.methodDescriptor = repo.lookupMethod(GenericService.class.getName(), methodName);
setNeedWrap(true);
} else {
} else if("$echo".equals(methodName)) {
this.methodDescriptor=repo.lookupMethod(EchoService.class.getName(),methodName);
setNeedWrap(true);
}else{
if (methods == null || methods.isEmpty()) {
responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED)
.withDescription("Method not found:" + methodName + " of service:" + serviceDescriptor.getServiceName()));

2
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleInvoker.java
View File

@ -100,6 +100,8 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
AsyncRpcResult result = new AsyncRpcResult(respFuture, inv);
result.setExecutor(executor);
connection.connectSync();
if (!connection.isAvailable()) {
Response response = new Response(req.getId(), req.getVersion());
response.setStatus(Response.CHANNEL_INACTIVE);