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

[WIP] Stream base3.0 (#7423) 

* import stream

* import stream

* fix

* fix

* fix

* fix

* fix

* fix comment

* suport client

* fix

* fix

* fix

* import client stream

* Add copyright

* add lisence

* fix lisence

* fix comment

* fix client decode

* fix stream client

* Reactor stream

* Add transport observer

* Update stream interface

* Update stream interface

* Add server stream support

* Temp backup

* Init client stream

* support stream client

* format files

* support wrapper

* fix commont

* Add empty handler

* Add handler comment

* Remove unused methods

* Reacdtor client stream logic

* Add test

* add client callback executor

* Support method overwrite for wrapper

* Support client error

* Format code

* Interop Bugfix

* Interop Bugfix

* Add single thread callback executor for client

* Fix format

* Update copyright

Co-authored-by: guohao <guohaoice@gmail.com>
This commit is contained in:
panxiaojun233 2021-04-23 17:24:17 +08:00 committed by GitHub
parent e93338749a
commit 78ffad2cdb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
41 changed files with 1947 additions and 863 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
dubbo-common/pom.xml
View File

@ -72,6 +72,10 @@
<groupId>javax.annotation</groupId>
<artifactId>javax.annotation-api</artifactId>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
</dependency>
<dependency>
<groupId>org.eclipse.collections</groupId>
<artifactId>eclipse-collections</artifactId>

4
dubbo-common/src/main/java/org/apache/dubbo/common/constants/CommonConstants.java
View File

@ -225,6 +225,10 @@ public interface CommonConstants {
String $INVOKE_ASYNC = "$invokeAsync";
String GENERIC_PARAMETER_DESC = "Ljava/lang/String;[Ljava/lang/String;[Ljava/lang/Object;";
/**
* echo call
*/
String $ECHO = "$echo";
/**
* package version in the manifest
*/

37
dubbo-common/src/main/java/org/apache/dubbo/common/stream/StreamObserver.java Normal file
View File

@ -0,0 +1,37 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.common.stream;
public interface StreamObserver<T> {
/**
* onNext
* @param data
*/
void onNext(T data);
/**
* onError
* @param throwable
*/
void onError(Throwable throwable);
/**
* onCompleted
*/
void onCompleted();
}

69
dubbo-common/src/main/java/org/apache/dubbo/rpc/model/MethodDescriptor.java
View File

@ -16,12 +16,18 @@
*/
package org.apache.dubbo.rpc.model;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.common.utils.ReflectUtils;
import com.google.protobuf.Message;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.stream.Stream;
import static org.apache.dubbo.common.constants.CommonConstants.$ECHO;
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE;
import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE_ASYNC;
@ -31,7 +37,7 @@ import static org.apache.dubbo.common.constants.CommonConstants.$INVOKE_ASYNC;
public class MethodDescriptor {
private final Method method;
// private final boolean isCallBack;
// private final boolean isFuture;
// private final boolean isFuture;
private final String paramDesc;
// duplicate filed as paramDesc, but with different format.
private final String[] compatibleParamSignatures;
@ -40,11 +46,30 @@ public class MethodDescriptor {
private final Type[] returnTypes;
private final String methodName;
private final boolean generic;
private final RpcType rpcType;
public MethodDescriptor(Method method) {
this.method = method;
this.parameterClasses = method.getParameterTypes();
this.returnClass = method.getReturnType();
Class<?>[] parameterTypes = method.getParameterTypes();
if (parameterTypes.length == 1 && isStreamType(parameterTypes[0])) {
this.parameterClasses = new Class<?>[]{
(Class<?>) ((ParameterizedType) method.getGenericReturnType()).getActualTypeArguments()[0]};
this.returnClass = (Class<?>) ((ParameterizedType) method.getGenericParameterTypes()[0])
.getActualTypeArguments()[0];
if (needWrap()) {
rpcType = RpcType.STREAM_WRAP;
} else {
rpcType = RpcType.STREAM_UNWRAP;
}
} else {
this.parameterClasses = method.getParameterTypes();
this.returnClass = method.getReturnType();
if (needWrap()) {
rpcType = RpcType.UNARY_WRAP;
} else {
rpcType = RpcType.UNARY_UNWRAP;
}
}
this.returnTypes = ReflectUtils.getReturnTypes(method);
this.paramDesc = ReflectUtils.getDesc(parameterClasses);
this.compatibleParamSignatures = Stream.of(parameterClasses)
@ -54,7 +79,36 @@ public class MethodDescriptor {
this.generic = (methodName.equals($INVOKE) || methodName.equals($INVOKE_ASYNC)) && parameterClasses.length == 3;
}
public boolean matchParams (String params) {
private static boolean isStreamType(Class<?> clz) {
return StreamObserver.class.isAssignableFrom(clz);
}
public boolean isStream() {
return rpcType.equals(RpcType.STREAM_WRAP) || rpcType.equals(RpcType.STREAM_UNWRAP);
}
public boolean isUnary() {
return rpcType.equals(RpcType.UNARY_WRAP) || rpcType.equals(RpcType.UNARY_UNWRAP);
}
public boolean isNeedWrap() {
return rpcType.equals(RpcType.UNARY_WRAP) || rpcType.equals(RpcType.STREAM_WRAP);
}
private boolean needWrap() {
if (CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals(methodName)) {
return true;
} else if ($ECHO.equals(methodName)) {
return true;
} else {
if (parameterClasses.length != 1) {
return true;
}
return !Message.class.isAssignableFrom(parameterClasses[0]);
}
}
public boolean matchParams(String params) {
return paramDesc.equalsIgnoreCase(params);
}
@ -90,4 +144,11 @@ public class MethodDescriptor {
return generic;
}
public enum RpcType {
UNARY_WRAP,
UNARY_UNWRAP,
STREAM_WRAP,
STREAM_UNWRAP;
}
}

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

@ -168,6 +168,7 @@ public class Connection extends AbstractReferenceCounted implements ReferenceCou
return closed.get();
}
//TODO replace channelFuture with intermediate future
public ChannelFuture write(Object request) throws RemotingException {
if (!isAvailable()) {
throw new RemotingException(null, null, "Failed to send request " + request + ", cause: The channel to " + remote + " is closed!");

92
dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/exchange/support/DefaultFuture2.java
View File

@ -30,8 +30,6 @@ import org.apache.dubbo.remoting.api.Connection;
import org.apache.dubbo.remoting.exchange.Request;
import org.apache.dubbo.remoting.exchange.Response;
import io.netty.channel.Channel;
import java.net.InetSocketAddress;
import java.text.SimpleDateFormat;
import java.util.Date;
@ -51,12 +49,10 @@ public class DefaultFuture2 extends CompletableFuture<Object> {
30,
TimeUnit.MILLISECONDS);
private static final Logger logger = LoggerFactory.getLogger(DefaultFuture2.class);
private static final Map<Long, Connection> CONNECTIONS = new ConcurrentHashMap<>();
private static final Map<Long, DefaultFuture2> FUTURES = new ConcurrentHashMap<>();
// invoke id.
private final Long id;
private final Connection connection;
private final Request request;
private final Connection connection;
private final int timeout;
private final long start = System.currentTimeMillis();
private volatile long sent;
@ -67,11 +63,9 @@ public class DefaultFuture2 extends CompletableFuture<Object> {
private DefaultFuture2(Connection client2, Request request, int timeout) {
this.connection = client2;
this.request = request;
this.id = request.getId();
this.timeout = timeout;
// put into waiting map.
FUTURES.put(id, this);
CONNECTIONS.put(id, connection);
FUTURES.put(request.getId(), this);
}
/**
@ -87,13 +81,13 @@ public class DefaultFuture2 extends CompletableFuture<Object> {
* 1.init a DefaultFuture
* 2.timeout check
*
* @param channel channel
* @param request the request
* @param timeout timeout
* @param connection connection
* @param request the request
* @param timeout timeout
* @return a new DefaultFuture
*/
public static DefaultFuture2 newFuture(Connection channel, Request request, int timeout, ExecutorService executor) {
final DefaultFuture2 future = new DefaultFuture2(channel, request, timeout);
public static DefaultFuture2 newFuture(Connection connection, Request request, int timeout, ExecutorService executor) {
final DefaultFuture2 future = new DefaultFuture2(connection, request, timeout);
future.setExecutor(executor);
// ThreadlessExecutor needs to hold the waiting future in case of circuit return.
if (executor instanceof ThreadlessExecutor) {
@ -108,10 +102,6 @@ public class DefaultFuture2 extends CompletableFuture<Object> {
return FUTURES.get(id);
}
public static boolean hasFuture(Channel channel) {
return CONNECTIONS.containsValue(channel);
}
public static void sent(Request request) {
DefaultFuture2 future = FUTURES.get(request.getId());
if (future != null) {
@ -119,57 +109,25 @@ public class DefaultFuture2 extends CompletableFuture<Object> {
}
}
/**
* close a channel when a channel is inactive
* directly return the unfinished requests.
*
* @param connection channel to close
*/
public static void closeChannel(Connection connection) {
for (Map.Entry<Long, Connection> entry : CONNECTIONS.entrySet()) {
if (connection.equals(entry.getValue())) {
DefaultFuture2 future = getFuture(entry.getKey());
if (future != null && !future.isDone()) {
ExecutorService futureExecutor = future.getExecutor();
if (futureExecutor != null && !futureExecutor.isTerminated()) {
futureExecutor.shutdownNow();
}
Response disconnectResponse = new Response(future.getId());
disconnectResponse.setStatus(Response.CHANNEL_INACTIVE);
disconnectResponse.setErrorMessage("Channel " +
connection +
" is inactive. Directly return the unFinished request : " +
future.getRequest());
DefaultFuture2.received(connection, disconnectResponse);
}
}
}
}
public static void received(Connection connection, Response response) {
received(connection, response, false);
}
public static void received(Connection connection, Response response, boolean timeout) {
try {
DefaultFuture2 future = FUTURES.remove(response.getId());
if (future != null) {
Timeout t = future.timeoutCheckTask;
if (!timeout) {
// decrease Time
t.cancel();
}
future.doReceived(response);
} else {
logger.warn("The timeout response finally returned at "
+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
+ ", response status is " + response.getStatus()
+ (connection == null ? "" : ", channel: " + connection.getChannel().localAddress()
+ " -> " + connection.getRemote()) + ", please check provider side for detailed result.");
DefaultFuture2 future = FUTURES.remove(response.getId());
if (future != null) {
Timeout t = future.timeoutCheckTask;
if (!timeout) {
// decrease Time
t.cancel();
}
} finally {
CONNECTIONS.remove(response.getId());
future.doReceived(response);
} else {
logger.warn("The timeout response finally returned at "
+ (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
+ ", response status is " + response.getStatus()
+ (connection == null ? "" : ", channel: " + connection.getChannel().localAddress()
+ " -> " + connection.getRemote()) + ", please check provider side for detailed result.");
}
}
@ -183,12 +141,11 @@ public class DefaultFuture2 extends CompletableFuture<Object> {
@Override
public boolean cancel(boolean mayInterruptIfRunning) {
Response errorResult = new Response(id);
Response errorResult = new Response(request.getId());
errorResult.setStatus(Response.CLIENT_ERROR);
errorResult.setErrorMessage("request future has been canceled.");
this.doReceived(errorResult);
FUTURES.remove(id);
CONNECTIONS.remove(id);
FUTURES.remove(request.getId());
return true;
}
@ -225,7 +182,7 @@ public class DefaultFuture2 extends CompletableFuture<Object> {
}
private long getId() {
return id;
return request.getId();
}
private Connection getConnection() {
@ -236,9 +193,6 @@ public class DefaultFuture2 extends CompletableFuture<Object> {
return sent > 0;
}
public Request getRequest() {
return request;
}
private int getTimeout() {
return timeout;

28
dubbo-rpc/dubbo-rpc-triple/pom.xml
View File

@ -1,18 +1,18 @@
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You 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.
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You 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.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">

188
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractClientStream.java Normal file
View File

@ -0,0 +1,188 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.remoting.api.Connection;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.model.ConsumerModel;
import org.apache.dubbo.triple.TripleWrapper;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
public abstract class AbstractClientStream extends AbstractStream implements Stream {
private ConsumerModel consumerModel;
private Connection connection;
protected AbstractClientStream(URL url) {
super(url);
}
protected AbstractClientStream(URL url, Executor executor) {
super(url, executor);
}
public static UnaryClientStream unary(URL url, Executor executor) {
return new UnaryClientStream(url, executor);
}
public static AbstractClientStream stream(URL url) {
return new ClientStream(url);
}
public AbstractClientStream service(ConsumerModel model) {
this.consumerModel = model;
return this;
}
public ConsumerModel getConsumerModel() {
return consumerModel;
}
public AbstractClientStream connection(Connection connection) {
this.connection = connection;
return this;
}
public Connection getConnection() {
return connection;
}
@Override
public void execute(Runnable runnable) {
try {
super.execute(runnable);
} catch (RejectedExecutionException e) {
LOGGER.error("Consumer's thread pool is full", e);
getStreamSubscriber().onError(GrpcStatus.fromCode(GrpcStatus.Code.RESOURCE_EXHAUSTED)
.withDescription("Consumer's thread pool is full").asException());
} catch (Throwable t) {
LOGGER.error("Consumer submit request to thread pool error ", t);
getStreamSubscriber().onError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withCause(t)
.withDescription("Consumer's error")
.asException());
}
}
protected byte[] encodeRequest(Object value) {
final byte[] out;
final Object obj;
if (getMethodDescriptor().isNeedWrap()) {
obj = getRequestWrapper(value);
} else {
obj = getRequestValue(value);
}
out = TripleUtil.pack(obj);
return out;
}
private TripleWrapper.TripleRequestWrapper getRequestWrapper(Object value) {
if (getMethodDescriptor().isStream()) {
String type = getMethodDescriptor().getParameterClasses()[0].getName();
return TripleUtil.wrapReq(getUrl(), getSerializeType(), value, type, getMultipleSerialization());
} else {
RpcInvocation invocation = (RpcInvocation) value;
return TripleUtil.wrapReq(getUrl(), invocation, getMultipleSerialization());
}
}
private Object getRequestValue(Object value) {
if (getMethodDescriptor().isUnary()) {
RpcInvocation invocation = (RpcInvocation) value;
return invocation.getArguments()[0];
}
return value;
}
protected Object deserializeResponse(byte[] data) {
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
try {
if (getConsumerModel() != null) {
ClassLoadUtil.switchContextLoader(getConsumerModel().getServiceInterfaceClass().getClassLoader());
}
if (getMethodDescriptor().isNeedWrap()) {
final TripleWrapper.TripleResponseWrapper wrapper = TripleUtil.unpack(data,
TripleWrapper.TripleResponseWrapper.class);
serialize(wrapper.getSerializeType());
return TripleUtil.unwrapResp(getUrl(), wrapper, getMultipleSerialization());
} else {
return TripleUtil.unpack(data, getMethodDescriptor().getReturnClass());
}
} finally {
ClassLoadUtil.switchContextLoader(tccl);
}
}
protected Metadata createRequestMeta(RpcInvocation inv) {
Metadata metadata = new DefaultMetadata();
metadata.put(TripleConstant.PATH_KEY, "/" + inv.getObjectAttachment(CommonConstants.PATH_KEY) + "/" + inv.getMethodName())
.put(TripleConstant.AUTHORITY_KEY, getUrl().getAddress())
.put(TripleConstant.CONTENT_TYPE_KEY, TripleConstant.CONTENT_PROTO)
.put(TripleConstant.TIMEOUT, inv.get(CommonConstants.TIMEOUT_KEY) + "m")
.put(HttpHeaderNames.TE, HttpHeaderValues.TRAILERS);
metadata.putIfNotNull(TripleConstant.SERVICE_VERSION, inv.getInvoker().getUrl().getVersion())
.putIfNotNull(TripleConstant.CONSUMER_APP_NAME_KEY,
(String) inv.getObjectAttachments().remove(CommonConstants.APPLICATION_KEY))
.putIfNotNull(TripleConstant.CONSUMER_APP_NAME_KEY,
(String) inv.getObjectAttachments().remove(CommonConstants.REMOTE_APPLICATION_KEY))
.putIfNotNull(TripleConstant.SERVICE_GROUP, inv.getInvoker().getUrl().getGroup());
inv.getObjectAttachments().remove(CommonConstants.GROUP_KEY);
inv.getObjectAttachments().remove(CommonConstants.INTERFACE_KEY);
inv.getObjectAttachments().remove(CommonConstants.PATH_KEY);
metadata.forEach(e -> metadata.put(e.getKey(), e.getValue()));
final Map<String, Object> attachments = inv.getObjectAttachments();
if (attachments != null) {
convertAttachment(metadata, attachments);
}
return metadata;
}
protected class ClientStreamObserver implements StreamObserver<Object> {
@Override
public void onNext(Object data) {
RpcInvocation invocation = (RpcInvocation) data;
final Metadata metadata = createRequestMeta(invocation);
getTransportSubscriber().tryOnMetadata(metadata, false);
final byte[] bytes = encodeRequest(invocation);
getTransportSubscriber().tryOnData(bytes, false);
}
@Override
public void onError(Throwable throwable) {
}
@Override
public void onCompleted() {
getTransportSubscriber().tryOnComplete();
}
}
}

211
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractServerStream.java Normal file
View File

@ -0,0 +1,211 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.MethodDescriptor;
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.triple.TripleWrapper;
import com.google.protobuf.Message;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
public abstract class AbstractServerStream extends AbstractStream implements Stream {
protected static final ExecutorRepository EXECUTOR_REPOSITORY =
ExtensionLoader.getExtensionLoader(ExecutorRepository.class).getDefaultExtension();
private final ProviderModel providerModel;
private List<MethodDescriptor> methodDescriptors;
private Invoker<?> invoker;
protected AbstractServerStream(URL url) {
this(url, lookupProviderModel(url));
}
protected AbstractServerStream(URL url, ProviderModel providerModel) {
this(url, lookupExecutor(url, providerModel), providerModel);
}
protected AbstractServerStream(URL url, Executor executor, ProviderModel providerModel) {
super(url, executor);
this.providerModel = providerModel;
}
private static Executor lookupExecutor(URL url, ProviderModel providerModel) {
ExecutorService executor = null;
if (providerModel != null) {
executor = (ExecutorService) providerModel.getServiceMetadata()
.getAttribute(CommonConstants.THREADPOOL_KEY);
}
if (executor == null) {
executor = EXECUTOR_REPOSITORY.getExecutor(url);
}
if (executor == null) {
executor = EXECUTOR_REPOSITORY.createExecutorIfAbsent(url);
}
return executor;
}
public static AbstractServerStream unary(URL url) {
return new UnaryServerStream(url);
}
public static AbstractServerStream stream(URL url) {
return new ServerStream(url);
}
private static ProviderModel lookupProviderModel(URL url) {
ServiceRepository repo = ApplicationModel.getServiceRepository();
final ProviderModel model = repo.lookupExportedService(url.getServiceKey());
if (model != null) {
ClassLoadUtil.switchContextLoader(model.getServiceInterfaceClass().getClassLoader());
}
return model;
}
public List<MethodDescriptor> getMethodDescriptors() {
return methodDescriptors;
}
public AbstractServerStream methods(List<MethodDescriptor> methods) {
this.methodDescriptors = methods;
return this;
}
public Invoker<?> getInvoker() {
return invoker;
}
public ProviderModel getProviderModel() {
return providerModel;
}
protected RpcInvocation buildInvocation(Metadata metadata) {
RpcInvocation inv = new RpcInvocation();
inv.setServiceName(getServiceDescriptor().getServiceName());
inv.setTargetServiceUniqueName(getUrl().getServiceKey());
inv.setMethodName(getMethodDescriptor().getMethodName());
inv.setParameterTypes(getMethodDescriptor().getParameterClasses());
inv.setReturnTypes(getMethodDescriptor().getReturnTypes());
final Map<String, Object> attachments = parseMetadataToMap(metadata);
attachments.remove("interface");
attachments.remove("serialization");
attachments.remove("te");
attachments.remove("path");
attachments.remove(TripleConstant.CONTENT_TYPE_KEY);
attachments.remove(TripleConstant.SERVICE_GROUP);
attachments.remove(TripleConstant.SERVICE_VERSION);
attachments.remove(TripleConstant.MESSAGE_KEY);
attachments.remove(TripleConstant.STATUS_KEY);
attachments.remove(TripleConstant.TIMEOUT);
inv.setObjectAttachments(attachments);
return inv;
}
protected Object[] deserializeRequest(byte[] data) {
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
try {
if (getProviderModel() != null) {
ClassLoadUtil.switchContextLoader(getProviderModel().getServiceInterfaceClass().getClassLoader());
}
if (getMethodDescriptor() == null || getMethodDescriptor().isNeedWrap()) {
final TripleWrapper.TripleRequestWrapper wrapper = TripleUtil.unpack(data,
TripleWrapper.TripleRequestWrapper.class);
serialize(wrapper.getSerializeType());
if (getMethodDescriptor() == null) {
final String[] paramTypes = wrapper.getArgTypesList().toArray(new String[wrapper.getArgsCount()]);
for (MethodDescriptor descriptor : getMethodDescriptors()) {
if (Arrays.equals(descriptor.getCompatibleParamSignatures(), paramTypes)) {
method(descriptor);
break;
}
}
if (getMethodDescriptor() == null) {
transportError(GrpcStatus.fromCode(GrpcStatus.Code.UNIMPLEMENTED)
.withDescription("Method :" + getMethodName() + "[" + Arrays.toString(paramTypes) + "] " +
"not found of service:" + getServiceDescriptor().getServiceName()));
return null;
}
}
return TripleUtil.unwrapReq(getUrl(), wrapper, getMultipleSerialization());
} else {
return new Object[]{TripleUtil.unpack(data, getMethodDescriptor().getParameterClasses()[0])};
}
} finally {
ClassLoadUtil.switchContextLoader(tccl);
}
}
protected byte[] encodeResponse(Object value) {
final com.google.protobuf.Message message;
if (getMethodDescriptor().isNeedWrap()) {
message = TripleUtil.wrapResp(getUrl(), getSerializeType(), value, getMethodDescriptor(),
getMultipleSerialization());
} else {
message = (Message) value;
}
return TripleUtil.pack(message);
}
@Override
public void execute(Runnable runnable) {
try {
super.execute(runnable);
} catch (RejectedExecutionException e) {
LOGGER.error("Provider's thread pool is full", e);
transportError(GrpcStatus.fromCode(GrpcStatus.Code.RESOURCE_EXHAUSTED)
.withDescription("Provider's thread pool is full"));
} catch (Throwable t) {
LOGGER.error("Provider submit request to thread pool error ", t);
transportError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withCause(t)
.withDescription("Provider's error"));
}
}
public AbstractServerStream service(ServiceDescriptor sd) {
setServiceDescriptor(sd);
return this;
}
public AbstractServerStream invoker(Invoker<?> invoker) {
this.invoker = invoker;
return this;
}
}

333
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/AbstractStream.java
View File

@ -14,130 +14,214 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.serialize.MultipleSerialization;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.common.threadlocal.NamedInternalThreadFactory;
import org.apache.dubbo.common.utils.ConfigUtils;
import org.apache.dubbo.config.Constants;
import org.apache.dubbo.remoting.exchange.Request;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.ServiceDescriptor;
import org.apache.dubbo.rpc.protocol.tri.GrpcStatus.Code;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http2.Http2Headers;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responseErr;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public abstract class AbstractStream implements Stream {
public static final boolean ENABLE_ATTACHMENT_WRAP = Boolean.parseBoolean(ConfigUtils.getProperty("triple.attachment", "false"));
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractStream.class);
private static final GrpcStatus TOO_MANY_DATA = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription("Too many data");
private final ChannelHandlerContext ctx;
private final URL url;
private MultipleSerialization multipleSerialization;
private Http2Headers headers;
private Http2Headers te;
private boolean needWrap;
private InputStream data;
private String serializeType;
public static final boolean ENABLE_ATTACHMENT_WRAP = Boolean.parseBoolean(
ConfigUtils.getProperty("triple.attachment", "false"));
protected static final String DUPLICATED_DATA = "Duplicated data";
private static final List<Executor> CALLBACK_EXECUTORS = new ArrayList<>(4);
protected AbstractStream(URL url, ChannelHandlerContext ctx) {
this(url, ctx, false);
}
protected AbstractStream(URL url, ChannelHandlerContext ctx, boolean needWrap) {
this.ctx = ctx;
this.url = url;
this.needWrap = needWrap;
if (needWrap) {
loadFromURL(url);
static {
ThreadFactory tripleTF = new NamedInternalThreadFactory("tri-callbcak", true);
for (int i = 0; i < 4; i++) {
final ThreadPoolExecutor tp = new ThreadPoolExecutor(1, 1, 0, TimeUnit.DAYS, new LinkedBlockingQueue<>(1024),
tripleTF, new ThreadPoolExecutor.AbortPolicy());
CALLBACK_EXECUTORS.add(tp);
}
}
protected void loadFromURL(URL url) {
final String value = url.getParameter(Constants.MULTI_SERIALIZATION_KEY, "default");
this.multipleSerialization = ExtensionLoader.getExtensionLoader(MultipleSerialization.class).getExtension(value);
private final URL url;
private final MultipleSerialization multipleSerialization;
private final StreamObserver<Object> streamObserver;
private final TransportObserver transportObserver;
private final Executor executor;
private ServiceDescriptor serviceDescriptor;
private MethodDescriptor methodDescriptor;
private String methodName;
private Request request;
private String serializeType;
private StreamObserver<Object> streamSubscriber;
private TransportObserver transportSubscriber;
protected AbstractStream(URL url) {
this(url, allocateCallbackExecutor());
}
public URL getUrl() {
return url;
protected AbstractStream(URL url, Executor executor) {
this.url = url;
this.executor = executor;
final String value = url.getParameter(Constants.MULTI_SERIALIZATION_KEY, CommonConstants.DEFAULT_KEY);
this.multipleSerialization = ExtensionLoader.getExtensionLoader(MultipleSerialization.class)
.getExtension(value);
this.streamObserver = createStreamObserver();
this.transportObserver = createTransportObserver();
}
private static Executor allocateCallbackExecutor() {
return CALLBACK_EXECUTORS.get(ThreadLocalRandom.current().nextInt(4));
}
public Request getRequest() {
return request;
}
public AbstractStream request(Request request) {
this.request = request;
return this;
}
public Executor getExecutor() {
return executor;
}
@Override
public void execute(Runnable runnable) {
executor.execute(runnable);
}
public String getMethodName() {
return methodName;
}
public AbstractStream methodName(String methodName) {
this.methodName = methodName;
return this;
}
public AbstractStream method(MethodDescriptor md) {
this.methodDescriptor = md;
return this;
}
protected abstract StreamObserver<Object> createStreamObserver();
protected abstract TransportObserver createTransportObserver();
public String getSerializeType() {
return serializeType;
}
protected void setSerializeType(String serializeType) {
public AbstractStream serialize(String serializeType) {
if (serializeType.equals("hessian4")) {
serializeType = "hessian2";
}
this.serializeType = serializeType;
}
protected boolean isNeedWrap() {
return needWrap;
}
protected void setNeedWrap(boolean needWrap) {
this.needWrap = needWrap;
}
public ChannelHandlerContext getCtx() {
return ctx;
}
public Http2Headers getHeaders() {
return headers;
}
public Http2Headers getTe() {
return te;
}
public InputStream getData() {
return data;
return this;
}
public MultipleSerialization getMultipleSerialization() {
return multipleSerialization;
}
public StreamObserver<Object> getStreamSubscriber() {
return streamSubscriber;
}
public TransportObserver getTransportSubscriber() {
return transportSubscriber;
}
public MethodDescriptor getMethodDescriptor() {
return methodDescriptor;
}
public ServiceDescriptor getServiceDescriptor() {
return serviceDescriptor;
}
public void setServiceDescriptor(ServiceDescriptor serviceDescriptor) {
this.serviceDescriptor = serviceDescriptor;
}
public URL getUrl() {
return url;
}
@Override
public void onData(InputStream in) {
if (data != null) {
responseErr(ctx, TOO_MANY_DATA);
return;
}
this.data = in;
public void subscribe(StreamObserver<Object> observer) {
this.streamSubscriber = observer;
}
public void onHeaders(Http2Headers headers) {
if (this.headers == null) {
this.headers = headers;
} else if (te == null) {
this.te = headers;
@Override
public void subscribe(TransportObserver observer) {
this.transportSubscriber = observer;
}
@Override
public StreamObserver<Object> asStreamObserver() {
return streamObserver;
}
@Override
public TransportObserver asTransportObserver() {
return transportObserver;
}
protected void transportError(GrpcStatus status) {
Metadata metadata = new DefaultMetadata();
metadata.put(TripleConstant.STATUS_KEY, Integer.toString(status.code.code));
metadata.put(TripleConstant.MESSAGE_KEY, status.toMessage());
getTransportSubscriber().tryOnMetadata(metadata, true);
if (LOGGER.isErrorEnabled()) {
LOGGER.error("[Triple-Server-Error] " + status.toMessage());
}
}
protected Map<String, Object> parseHeadersToMap(Http2Headers headers) {
protected void transportError(Throwable throwable) {
Metadata metadata = new DefaultMetadata();
metadata.put(TripleConstant.STATUS_KEY, Integer.toString(Code.UNKNOWN.code));
metadata.put(TripleConstant.MESSAGE_KEY, throwable.getMessage());
getTransportSubscriber().tryOnMetadata(metadata, true);
if (LOGGER.isErrorEnabled()) {
LOGGER.error("[Triple-Server-Error] service=" + getServiceDescriptor().getServiceName()
+ " method=" + getMethodName(), throwable);
}
}
protected Map<String, Object> parseMetadataToMap(Metadata metadata) {
Map<String, Object> attachments = new HashMap<>();
for (Map.Entry<CharSequence, CharSequence> header : headers) {
for (Map.Entry<CharSequence, CharSequence> header : metadata) {
String key = header.getKey().toString();
if(Http2Headers.PseudoHeaderName.isPseudoHeader(key)){
if (Http2Headers.PseudoHeaderName.isPseudoHeader(key)) {
continue;
}
if (ENABLE_ATTACHMENT_WRAP) {
if (key.endsWith("-tw-bin") && key.length() > 7) {
try {
attachments.put(key.substring(0, key.length() - 7), TripleUtil.decodeObjFromHeader(getUrl(), header.getValue(), getMultipleSerialization()));
attachments.put(key.substring(0, key.length() - 7),
TripleUtil.decodeObjFromHeader(url, header.getValue(), multipleSerialization));
} catch (Exception e) {
LOGGER.error("Failed to parse response attachment key=" + key, e);
}
@ -156,27 +240,114 @@ public abstract class AbstractStream implements Stream {
return attachments;
}
protected void convertAttachment(Http2Headers trailers, Map<String, Object> attachments) throws IOException {
protected void convertAttachment(Metadata metadata, Map<String, Object> attachments) {
for (Map.Entry<String, Object> entry : attachments.entrySet()) {
final String key = entry.getKey().toLowerCase(Locale.ROOT);
if(Http2Headers.PseudoHeaderName.isPseudoHeader(key)){
if (Http2Headers.PseudoHeaderName.isPseudoHeader(key)) {
continue;
}
final Object v = entry.getValue();
convertSingleAttachment(metadata, key, v);
}
}
private void convertSingleAttachment(Metadata metadata, String key, Object v) {
try {
if (!ENABLE_ATTACHMENT_WRAP) {
if (v instanceof String) {
trailers.addObject(key, v);
metadata.put(key, (String) v);
} else if (v instanceof byte[]) {
trailers.add(key + "-bin", TripleUtil.encodeBase64ASCII((byte[]) v));
metadata.put(key + "-bin", TripleUtil.encodeBase64ASCII((byte[]) v));
}
} else {
if (v instanceof String || serializeType == null) {
trailers.addObject(key, v);
metadata.put(key, v.toString());
} else {
String encoded = TripleUtil.encodeWrapper(url, v, this.serializeType, getMultipleSerialization());
trailers.add(key + "-tw-bin", encoded);
metadata.put(key + "-tw-bin", encoded);
}
}
} catch (IOException e) {
LOGGER.warn("Meet exception when convert single attachment key:" + key, e);
}
}
protected static abstract class AbstractTransportObserver implements TransportObserver {
private Metadata headers;
private Metadata trailers;
public Metadata getHeaders() {
return headers;
}
public Metadata getTrailers() {
return trailers;
}
@Override
public void onMetadata(Metadata metadata, boolean endStream, OperationHandler handler) {
if (headers == null) {
headers = metadata;
} else {
trailers = metadata;
}
}
protected GrpcStatus extractStatusFromMeta(Metadata metadata) {
if (metadata.contains(TripleConstant.STATUS_KEY)) {
final int code = Integer.parseInt(metadata.get(TripleConstant.STATUS_KEY).toString());
if (!GrpcStatus.Code.isOk(code)) {
GrpcStatus status = GrpcStatus.fromCode(code);
if (metadata.contains(TripleConstant.MESSAGE_KEY)) {
final String raw = metadata.get(TripleConstant.MESSAGE_KEY).toString();
status = status.withDescription(GrpcStatus.fromMessage(raw));
}
return status;
}
return GrpcStatus.fromCode(Code.OK);
}
return GrpcStatus.fromCode(Code.OK);
}
}
protected abstract static class UnaryTransportObserver extends AbstractTransportObserver implements TransportObserver {
private byte[] data;
public byte[] getData() {
return data;
}
protected abstract void onError(GrpcStatus status);
@Override
public void onComplete(OperationHandler handler) {
Metadata metadata;
if (getTrailers() == null) {
metadata = getHeaders();
} else {
metadata = getTrailers();
}
final GrpcStatus status = extractStatusFromMeta(metadata);
if (GrpcStatus.Code.isOk(status.code.code)) {
doOnComplete(handler);
} else {
onError(status);
}
}
protected abstract void doOnComplete(OperationHandler handler);
@Override
public void onData(byte[] in, boolean endStream, OperationHandler handler) {
if (data == null) {
this.data = in;
} else {
handler.operationDone(OperationResult.FAILURE, GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription(DUPLICATED_DATA).asException());
}
}
}
}

251
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientStream.java
View File

@ -14,221 +14,72 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.api.Connection;
import org.apache.dubbo.remoting.exchange.Request;
import org.apache.dubbo.remoting.exchange.Response;
import org.apache.dubbo.remoting.exchange.support.DefaultFuture2;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ConsumerModel;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.ServiceRepository;
import org.apache.dubbo.triple.TripleWrapper;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2NoMoreStreamIdsException;
import io.netty.handler.codec.http2.Http2StreamChannel;
import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
import io.netty.util.AsciiString;
public class ClientStream extends AbstractClientStream implements Stream {
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.concurrent.Executor;
import static org.apache.dubbo.rpc.Constants.CONSUMER_MODEL;
public class ClientStream extends AbstractStream implements Stream {
private static final GrpcStatus MISSING_RESP = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription("Missing Response");
private static final AsciiString SCHEME = AsciiString.of("http");
private final String authority;
private final Request request;
private final RpcInvocation invocation;
private final Executor callback;
public ClientStream(URL url, ChannelHandlerContext ctx, boolean needWrap, Request request, Executor callback) {
super(url, ctx, needWrap);
this.callback = callback;
if (needWrap) {
setSerializeType((String) ((RpcInvocation) (request.getData())).getObjectAttachment(Constants.SERIALIZATION_KEY));
}
this.authority = url.getAddress();
this.request = request;
this.invocation = (RpcInvocation) request.getData();
}
public static ConsumerModel getConsumerModel(Invocation invocation) {
Object o = invocation.get(CONSUMER_MODEL);
if (o instanceof ConsumerModel) {
return (ConsumerModel) o;
}
String serviceKey = invocation.getInvoker().getUrl().getServiceKey();
return ApplicationModel.getConsumerModel(serviceKey);
protected ClientStream(URL url) {
super(url);
}
@Override
public void onError(GrpcStatus status) {
Response response = new Response(request.getId(), request.getVersion());
if (status.description != null) {
response.setErrorMessage(status.description);
} else {
response.setErrorMessage(status.cause.getMessage());
}
final byte code = GrpcStatus.toDubboStatus(status.code);
response.setStatus(code);
DefaultFuture2.received(Connection.getConnectionFromChannel(getCtx().channel()), response);
protected StreamObserver<Object> createStreamObserver() {
return new ClientStreamObserver() {
boolean metaSent;
@Override
public void onNext(Object data) {
if (!metaSent) {
metaSent = true;
final Metadata metadata = createRequestMeta((RpcInvocation) getRequest().getData());
getTransportSubscriber().tryOnMetadata(metadata, false);
}
final byte[] bytes = encodeRequest(data);
getTransportSubscriber().tryOnData(bytes, false);
}
@Override
public void onError(Throwable throwable) {
transportError(throwable);
}
};
}
@Override
public void write(Object obj, ChannelPromise promise) throws IOException {
final Http2StreamChannelBootstrap streamChannelBootstrap = new Http2StreamChannelBootstrap(getCtx().channel());
final Http2StreamChannel streamChannel = streamChannelBootstrap.open().syncUninterruptibly().getNow();
protected TransportObserver createTransportObserver() {
return new AbstractTransportObserver() {
Http2Headers headers = new DefaultHttp2Headers()
.authority(authority)
.scheme(SCHEME)
.method(HttpMethod.POST.asciiName())
.path("/" + invocation.getObjectAttachment(CommonConstants.PATH_KEY) + "/" + invocation.getMethodName())
.set(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO)
.set(TripleConstant.TIMEOUT, invocation.get(CommonConstants.TIMEOUT_KEY) + "m")
.set(HttpHeaderNames.TE, HttpHeaderValues.TRAILERS);
final String version = invocation.getInvoker().getUrl().getVersion();
if (version != null) {
headers.set(TripleConstant.SERVICE_VERSION, version);
}
final String app = (String) invocation.getObjectAttachment(CommonConstants.APPLICATION_KEY);
if (app != null) {
headers.set(TripleConstant.CONSUMER_APP_NAME_KEY, app);
invocation.getObjectAttachments().remove(CommonConstants.APPLICATION_KEY);
}
final String group = invocation.getInvoker().getUrl().getGroup();
if (group != null) {
headers.set(TripleConstant.SERVICE_GROUP, group);
invocation.getObjectAttachments().remove(CommonConstants.GROUP_KEY);
}
final Map<String, Object> attachments = invocation.getObjectAttachments();
if (attachments != null) {
convertAttachment(headers, attachments);
}
headers.remove("path");
headers.remove("interface");
DefaultHttp2HeadersFrame frame = new DefaultHttp2HeadersFrame(headers);
final TripleHttp2ClientResponseHandler responseHandler = new TripleHttp2ClientResponseHandler();
TripleUtil.setClientStream(streamChannel, this);
streamChannel.pipeline().addLast(responseHandler)
.addLast(new GrpcDataDecoder(Integer.MAX_VALUE))
.addLast(new TripleClientInboundHandler());
streamChannel.write(frame).addListener(future -> {
if (!future.isSuccess()) {
if (future.cause() instanceof Http2NoMoreStreamIdsException) {
getCtx().close();
}
promise.setFailure(future.cause());
@Override
public void onData(byte[] data, boolean endStream, OperationHandler handler) {
execute(() -> {
final Object resp = deserializeResponse(data);
getStreamSubscriber().onNext(resp);
});
}
});
final ByteBuf out;
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
try {
final ConsumerModel model = getConsumerModel(invocation);
if (model != null) {
ClassLoadUtil.switchContextLoader(model.getClassLoader());
@Override
public void onComplete(OperationHandler handler) {
execute(() -> {
Metadata metadata;
if (getTrailers() == null) {
metadata = getHeaders();
} else {
metadata = getTrailers();
}
final GrpcStatus status = extractStatusFromMeta(metadata);
if (GrpcStatus.Code.isOk(status.code.code)) {
getStreamSubscriber().onCompleted();
} else {
getStreamSubscriber().onError(status.asException());
}
});
}
if (isNeedWrap()) {
final TripleWrapper.TripleRequestWrapper wrap = TripleUtil.wrapReq(getUrl(), invocation, getMultipleSerialization());
out = TripleUtil.pack(getCtx(), wrap);
} else {
out = TripleUtil.pack(getCtx(), invocation.getArguments()[0]);
}
} finally {
ClassLoadUtil.switchContextLoader(tccl);
}
final DefaultHttp2DataFrame data = new DefaultHttp2DataFrame(out, true);
streamChannel.write(data).addListener(f -> {
if (f.isSuccess()) {
promise.trySuccess();
} else {
promise.tryFailure(f.cause());
}
});
};
}
public void halfClose() {
final int httpCode = HttpResponseStatus.parseLine(getHeaders().status()).code();
if (HttpResponseStatus.OK.code() != httpCode) {
final Integer code = getHeaders().getInt(TripleConstant.STATUS_KEY);
final GrpcStatus status = GrpcStatus.fromCode(code)
.withDescription(TripleUtil.percentDecode(getHeaders().get(TripleConstant.MESSAGE_KEY)));
onError(status);
return;
}
final Http2Headers te = getTe() == null ? getHeaders() : getTe();
final Integer code = te.getInt(TripleConstant.STATUS_KEY);
if (!GrpcStatus.Code.isOk(code)) {
final GrpcStatus status = GrpcStatus.fromCode(code)
.withDescription(TripleUtil.percentDecode(getHeaders().get(TripleConstant.MESSAGE_KEY)));
onError(status);
return;
}
final InputStream data = getData();
if (data == null) {
onError(MISSING_RESP);
return;
}
callback.execute(() -> {
final Invocation invocation = (Invocation) (request.getData());
ServiceRepository repo = ApplicationModel.getServiceRepository();
MethodDescriptor methodDescriptor = repo.lookupMethod(invocation.getServiceName(), invocation.getMethodName());
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
try {
final Object resp;
final ConsumerModel model = getConsumerModel(invocation);
if (model != null) {
ClassLoadUtil.switchContextLoader(model.getClassLoader());
}
if (isNeedWrap()) {
final TripleWrapper.TripleResponseWrapper message = TripleUtil.unpack(data, TripleWrapper.TripleResponseWrapper.class);
resp = TripleUtil.unwrapResp(getUrl(), message, getMultipleSerialization());
} else {
resp = TripleUtil.unpack(data, methodDescriptor.getReturnClass());
}
Response response = new Response(request.getId(), request.getVersion());
final AppResponse result = new AppResponse(resp);
result.setObjectAttachments(parseHeadersToMap(te));
response.setResult(result);
DefaultFuture2.received(Connection.getConnectionFromChannel(getCtx().channel()), response);
} catch (Exception e) {
final GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withCause(e)
.withDescription("Failed to deserialize response");
onError(status);
} finally {
ClassLoadUtil.switchContextLoader(tccl);
}
});
}
}

102
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ClientTransportObserver.java Normal file
View File

@ -0,0 +1,102 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2StreamChannel;
import io.netty.handler.codec.http2.Http2StreamChannelBootstrap;
import io.netty.util.AsciiString;
public class ClientTransportObserver implements TransportObserver {
private static final AsciiString SCHEME = AsciiString.of("http");
private final ChannelHandlerContext ctx;
private final Http2StreamChannel streamChannel;
private final ChannelPromise promise;
private boolean headerSent = false;
private boolean endStreamSent = false;
public ClientTransportObserver(ChannelHandlerContext ctx, AbstractClientStream stream, ChannelPromise promise) {
this.ctx = ctx;
this.promise = promise;
final Http2StreamChannelBootstrap streamChannelBootstrap = new Http2StreamChannelBootstrap(ctx.channel());
streamChannel = streamChannelBootstrap.open().syncUninterruptibly().getNow();
final TripleHttp2ClientResponseHandler responseHandler = new TripleHttp2ClientResponseHandler();
streamChannel.pipeline().addLast(responseHandler)
.addLast(new GrpcDataDecoder(Integer.MAX_VALUE))
.addLast(new TripleClientInboundHandler());
streamChannel.attr(TripleUtil.CLIENT_STREAM_KEY).set(stream);
}
@Override
public void onMetadata(Metadata metadata, boolean endStream, Stream.OperationHandler handler) {
if (!headerSent) {
final Http2Headers headers = new DefaultHttp2Headers(true)
.path(metadata.get(TripleConstant.PATH_KEY))
.authority(metadata.get(TripleConstant.AUTHORITY_KEY))
.scheme(SCHEME)
.method(HttpMethod.POST.asciiName());
metadata.forEach(e -> headers.set(e.getKey(), e.getValue()));
headerSent = true;
streamChannel.writeAndFlush(new DefaultHttp2HeadersFrame(headers, endStream))
.addListener(future -> {
if (!future.isSuccess()) {
promise.tryFailure(future.cause());
}
});
}
}
@Override
public void onData(byte[] data, boolean endStream, Stream.OperationHandler handler) {
ByteBuf buf = ctx.alloc().buffer();
buf.writeByte(0);
buf.writeInt(data.length);
buf.writeBytes(data);
streamChannel.writeAndFlush(new DefaultHttp2DataFrame(buf, endStream))
.addListener(future -> {
if (!future.isSuccess()) {
promise.tryFailure(future.cause());
}
});
}
@Override
public void onComplete(Stream.OperationHandler handler) {
if (!endStreamSent) {
endStreamSent = true;
streamChannel.writeAndFlush(new DefaultHttp2DataFrame(true))
.addListener(future -> {
if (future.isSuccess()) {
promise.trySuccess();
} else {
promise.tryFailure(future.cause());
}
});
}
}
}

59
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/DefaultMetadata.java Normal file
View File

@ -0,0 +1,59 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Spliterator;
import java.util.function.Consumer;
public class DefaultMetadata implements Metadata {
private final Map<CharSequence, CharSequence> innerMap = new HashMap<>();
@Override
public CharSequence get(CharSequence key) {
return innerMap.get(key);
}
@Override
public Metadata put(CharSequence key, CharSequence value) {
innerMap.put(key, value);
return this;
}
@Override
public Iterator<Map.Entry<CharSequence, CharSequence>> iterator() {
return innerMap.entrySet().iterator();
}
@Override
public void forEach(Consumer<? super Map.Entry<CharSequence, CharSequence>> action) {
innerMap.entrySet().forEach(action);
}
@Override
public Spliterator<Map.Entry<CharSequence, CharSequence>> spliterator() {
return innerMap.entrySet().spliterator();
}
public boolean contains(CharSequence key) {
return innerMap.containsKey(key);
}
}

5
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GracefulShutdown.java
View File

@ -33,8 +33,8 @@ public class GracefulShutdown {
private static final long GRACEFUL_SHUTDOWN_PING_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(10);
private final ChannelHandlerContext ctx;
private final ChannelPromise originPromise;
private boolean pingAckedOrTimeout;
private final String goAwayMessage;
private boolean pingAckedOrTimeout;
private Future<?> pingFuture;
public GracefulShutdown(ChannelHandlerContext ctx, String goAwayMessage, ChannelPromise originPromise) {
@ -66,7 +66,8 @@ public class GracefulShutdown {
pingFuture.cancel(false);
try {
Http2GoAwayFrame goAwayFrame = new DefaultHttp2GoAwayFrame(Http2Error.NO_ERROR, ByteBufUtil.writeAscii(this.ctx.alloc(), this.goAwayMessage));
Http2GoAwayFrame goAwayFrame = new DefaultHttp2GoAwayFrame(Http2Error.NO_ERROR,
ByteBufUtil.writeAscii(this.ctx.alloc(), this.goAwayMessage));
ctx.write(goAwayFrame);
ctx.flush();
//TODO support customize graceful shutdown timeout mills

8
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcDataDecoder.java
View File

@ -63,8 +63,9 @@ public class GrpcDataDecoder extends ReplayingDecoder<GrpcDataDecoder.GrpcDecode
}
checkpoint(GrpcDecodeState.PAYLOAD);
case PAYLOAD:
ByteBuf buf = in.readRetainedSlice(len);
out.add(buf);
byte[] dst = new byte[len];
in.readBytes(dst);
out.add(dst);
checkpoint(GrpcDecodeState.HEADER);
break;
default:
@ -73,6 +74,7 @@ public class GrpcDataDecoder extends ReplayingDecoder<GrpcDataDecoder.GrpcDecode
}
enum GrpcDecodeState {
HEADER, PAYLOAD
HEADER,
PAYLOAD
}
}

38
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatus.java
View File

@ -16,8 +16,12 @@
*/
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.exchange.Response;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.QueryStringEncoder;
/**
* See https://github.com/grpc/grpc/blob/master/doc/statuscodes.md
*/
@ -76,6 +80,21 @@ public class GrpcStatus {
return status;
}
public static String limitSizeTo4KB(String desc) {
if (desc.length() < 4096) {
return desc;
} else {
return desc.substring(0, 4086);
}
}
public static String fromMessage(String raw) {
if (raw == null || raw.isEmpty()) {
return "";
}
return QueryStringDecoder.decodeComponent(raw);
}
public GrpcStatus withCause(Throwable cause) {
return new GrpcStatus(this.code, cause, this.description);
}
@ -88,6 +107,24 @@ public class GrpcStatus {
return new TripleRpcException(this);
}
public String toMessage() {
final String msg;
if (cause == null) {
msg = description;
} else {
String placeHolder = description == null ? "" : description;
msg = StringUtils.toString(placeHolder, cause);
}
if (msg == null) {
return "";
}
String output = limitSizeTo4KB(msg);
QueryStringEncoder encoder = new QueryStringEncoder("");
encoder.addParam("", output);
// ?=
return encoder.toString().substring(2);
}
enum Code {
OK(0),
CANCELLED(1),
@ -116,7 +153,6 @@ public class GrpcStatus {
return status == OK.code;
}
public static Code fromCode(int code) {
for (Code value : Code.values()) {
if (value.code == code) {

52
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/Http2HeaderMeta.java Normal file
View File

@ -0,0 +1,52 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import io.netty.handler.codec.http2.Http2Headers;
import java.util.Iterator;
import java.util.Map;
public class Http2HeaderMeta implements Metadata {
private final Http2Headers headers;
public Http2HeaderMeta(Http2Headers headers) {
this.headers = headers;
}
@Override
public Metadata put(CharSequence key, CharSequence value) {
headers.set(key, value);
return this;
}
@Override
public CharSequence get(CharSequence key) {
return headers.get(key);
}
@Override
public boolean contains(CharSequence key) {
return headers.contains(key);
}
@Override
public Iterator<Map.Entry<CharSequence, CharSequence>> iterator() {
return headers.iterator();
}
}

37
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/Metadata.java Normal file
View File

@ -0,0 +1,37 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import java.util.Map;
public interface Metadata extends Iterable<Map.Entry<CharSequence, CharSequence>> {
Metadata put(CharSequence key, CharSequence value);
default Metadata putIfNotNull(CharSequence key, CharSequence value) {
if (value != null) {
put(key, value);
}
return this;
}
CharSequence get(CharSequence key);
boolean contains(CharSequence key);
}

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

@ -14,282 +14,92 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
import org.apache.dubbo.common.utils.ExecutorUtil;
import org.apache.dubbo.remoting.TimeoutException;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.MethodDescriptor;
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;
import com.google.protobuf.Message;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
import io.netty.handler.codec.http2.Http2Headers;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.function.BiConsumer;
import java.util.function.Function;
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responseErr;
public class ServerStream extends AbstractStream implements Stream {
private static final Logger LOGGER = LoggerFactory.getLogger(ServerStream.class);
private static final String TOO_MANY_REQ = "Too many requests";
private static final String MISSING_REQ = "Missing request";
private static final ExecutorRepository EXECUTOR_REPOSITORY =
ExtensionLoader.getExtensionLoader(ExecutorRepository.class).getDefaultExtension();
private final Invoker<?> invoker;
private final ChannelHandlerContext ctx;
private final ServiceDescriptor serviceDescriptor;
private final ProviderModel providerModel;
private final String methodName;
private MethodDescriptor methodDescriptor;
public ServerStream(Invoker<?> invoker, ServiceDescriptor serviceDescriptor, String methodName, ChannelHandlerContext ctx) {
super(ExecutorUtil.setThreadName(invoker.getUrl(), "DubboPUServerHandler"), ctx);
this.invoker = invoker;
ServiceRepository repo = ApplicationModel.getServiceRepository();
this.providerModel = repo.lookupExportedService(getUrl().getServiceKey());
this.methodName = methodName;
this.serviceDescriptor = serviceDescriptor;
this.ctx = ctx;
}
@Override
public void onError(GrpcStatus status) {
public class ServerStream extends AbstractServerStream implements Stream {
protected ServerStream(URL url) {
super(url);
}
@Override
public void write(Object obj, ChannelPromise promise) throws Exception {
protected StreamObserver<Object> createStreamObserver() {
return new ServerStreamObserver();
}
public void halfClose() throws Exception {
if (getData() == null) {
responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription(MISSING_REQ));
return;
}
ExecutorService executor = null;
if (providerModel != null) {
executor = (ExecutorService) providerModel.getServiceMetadata().getAttribute(CommonConstants.THREADPOOL_KEY);
}
if (executor == null) {
executor = EXECUTOR_REPOSITORY.getExecutor(getUrl());
}
if (executor == null) {
executor = EXECUTOR_REPOSITORY.createExecutorIfAbsent(getUrl());
@Override
protected TransportObserver createTransportObserver() {
return new StreamTransportObserver();
}
private class ServerStreamObserver implements StreamObserver<Object> {
private boolean headersSent;
@Override
public void onNext(Object data) {
if (!headersSent) {
getTransportSubscriber().tryOnMetadata(new DefaultMetadata(), false);
headersSent = true;
}
final byte[] bytes = encodeResponse(data);
getTransportSubscriber().tryOnData(bytes, false);
}
try {
executor.execute(this::unaryInvoke);
} catch (RejectedExecutionException e) {
LOGGER.error("Provider's thread pool is full", e);
responseErr(ctx, GrpcStatus.fromCode(Code.RESOURCE_EXHAUSTED)
.withDescription("Provider's thread pool is full"));
} catch (Throwable t) {
LOGGER.error("Provider submit request to thread pool error ", t);
responseErr(ctx, GrpcStatus.fromCode(Code.INTERNAL)
.withCause(t)
.withDescription("Provider's error"));
@Override
public void onError(Throwable throwable) {
final GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withCause(throwable)
.withDescription("Biz exception");
transportError(status);
}
@Override
public void onCompleted() {
Metadata metadata = new DefaultMetadata();
metadata.put(TripleConstant.MESSAGE_KEY, "OK");
metadata.put(TripleConstant.STATUS_KEY, Integer.toString(GrpcStatus.Code.OK.code));
getTransportSubscriber().tryOnMetadata(metadata, true);
}
}
private void unaryInvoke() {
private class StreamTransportObserver extends AbstractTransportObserver implements TransportObserver {
Invocation invocation;
try {
invocation = buildInvocation();
} catch (Throwable t) {
LOGGER.warn("Exception processing triple message", t);
responseErr(ctx, GrpcStatus.fromCode(Code.INTERNAL).withDescription("Decode request failed:" + t.getMessage()));
return;
}
if (invocation == null) {
return;
}
final Result result = this.invoker.invoke(invocation);
CompletionStage<Object> future = result.thenApply(Function.identity());
BiConsumer<Object, Throwable> onComplete = (appResult, t) -> {
@Override
public void onMetadata(Metadata metadata, boolean endStream, OperationHandler handler) {
super.onMetadata(metadata, endStream, handler);
final RpcInvocation inv = buildInvocation(metadata);
inv.setArguments(new Object[]{asStreamObserver()});
final Result result = getInvoker().invoke(inv);
try {
if (t != null) {
if (t instanceof TimeoutException) {
responseErr(ctx, GrpcStatus.fromCode(Code.DEADLINE_EXCEEDED).withCause(t));
} else {
responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN).withCause(t));
}
return;
}
AppResponse response = (AppResponse) appResult;
if (response.hasException()) {
final Throwable exception = response.getException();
if (exception instanceof TripleRpcException) {
responseErr(ctx, ((TripleRpcException) exception).getStatus());
} else {
responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN)
.withCause(exception));
}
return;
}
Http2Headers http2Headers = new DefaultHttp2Headers()
.status(OK.codeAsText())
.set(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO);
final Message message;
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
final ByteBuf buf;
try {
ClassLoadUtil.switchContextLoader(providerModel.getServiceInterfaceClass().getClassLoader());
if (isNeedWrap()) {
message = TripleUtil.wrapResp(getUrl(), getSerializeType(), response.getValue(), methodDescriptor, getMultipleSerialization());
} else {
message = (Message) response.getValue();
}
buf = TripleUtil.pack(ctx, message);
} finally {
ClassLoadUtil.switchContextLoader(tccl);
}
final Http2Headers trailers = new DefaultHttp2Headers();
final Map<String, Object> attachments = response.getObjectAttachments();
if (attachments != null) {
convertAttachment(trailers, attachments);
}
trailers.setInt(TripleConstant.STATUS_KEY, GrpcStatus.Code.OK.code);
ctx.write(new DefaultHttp2HeadersFrame(http2Headers));
final DefaultHttp2DataFrame data = new DefaultHttp2DataFrame(buf);
ctx.write(data);
ctx.writeAndFlush(new DefaultHttp2HeadersFrame(trailers, true));
} catch (Throwable e) {
LOGGER.warn("Exception processing triple message", e);
if (e instanceof TripleRpcException) {
responseErr(ctx, ((TripleRpcException) e).getStatus());
} else {
responseErr(ctx, GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN)
.withDescription("Exception occurred in provider's execution:" + e.getMessage())
.withCause(e));
}
}
};
future.whenComplete(onComplete);
RpcContext.removeContext();
}
private Invocation buildInvocation() {
RpcInvocation inv = new RpcInvocation();
ClassLoader tccl = Thread.currentThread().getContextClassLoader();
ServiceRepository repo = ApplicationModel.getServiceRepository();
final List<MethodDescriptor> methods = serviceDescriptor.getMethods(methodName);
if (CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals(methodName)) {
this.methodDescriptor = repo.lookupMethod(GenericService.class.getName(), methodName);
setNeedWrap(true);
} 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()));
return null;
}
if (methods.size() == 1) {
this.methodDescriptor = methods.get(0);
setNeedWrap(TripleUtil.needWrapper(this.methodDescriptor.getParameterClasses()));
} else {
// can not determine which one to invoke when same protobuf method name is used, force wrap it
setNeedWrap(true);
subscribe((StreamObserver<Object>) result.getValue());
} catch (Throwable t) {
transportError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription("Failed to create server's observer"));
}
}
if (isNeedWrap()) {
loadFromURL(getUrl());
}
try {
if (providerModel != null) {
ClassLoadUtil.switchContextLoader(providerModel.getServiceInterfaceClass().getClassLoader());
}
if (isNeedWrap()) {
final TripleWrapper.TripleRequestWrapper req = TripleUtil.unpack(getData(), TripleWrapper.TripleRequestWrapper.class);
setSerializeType(req.getSerializeType());
if (this.methodDescriptor == null) {
String[] paramTypes = req.getArgTypesList().toArray(new String[req.getArgsCount()]);
for (MethodDescriptor method : methods) {
if (Arrays.equals(method.getCompatibleParamSignatures(), paramTypes)) {
this.methodDescriptor = method;
break;
}
}
if (this.methodDescriptor == null) {
responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED)
.withDescription("Method not found:" + methodName +
" args:" + Arrays.toString(paramTypes) + " of service:" + serviceDescriptor.getServiceName()));
return null;
}
@Override
public void onData(byte[] in, boolean endStream, OperationHandler handler) {
try {
final Object[] arguments = deserializeRequest(in);
if (arguments != null) {
getStreamSubscriber().onNext(arguments[0]);
}
final Object[] arguments = TripleUtil.unwrapReq(getUrl(), req, getMultipleSerialization());
inv.setArguments(arguments);
} else {
final Object req = TripleUtil.unpack(getData(), methodDescriptor.getParameterClasses()[0]);
inv.setArguments(new Object[]{req});
} catch (Throwable t) {
transportError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription("Deserialize request failed")
.withCause(t));
}
} finally {
ClassLoadUtil.switchContextLoader(tccl);
}
inv.setMethodName(methodDescriptor.getMethodName());
inv.setServiceName(serviceDescriptor.getServiceName());
inv.setTargetServiceUniqueName(getUrl().getServiceKey());
inv.setParameterTypes(methodDescriptor.getParameterClasses());
inv.setReturnTypes(methodDescriptor.getReturnTypes());
final Map<String, Object> attachments = parseHeadersToMap(getHeaders());
attachments.remove("interface");
attachments.remove("serialization");
attachments.remove("te");
attachments.remove("path");
attachments.remove(TripleConstant.CONTENT_TYPE_KEY);
attachments.remove(TripleConstant.SERVICE_GROUP);
attachments.remove(TripleConstant.SERVICE_VERSION);
attachments.remove(TripleConstant.MESSAGE_KEY);
attachments.remove(TripleConstant.STATUS_KEY);
attachments.remove(TripleConstant.TIMEOUT);
inv.setObjectAttachments(attachments);
return inv;
@Override
public void onComplete(OperationHandler handler) {
getStreamSubscriber().onCompleted();
}
}
}

64
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/ServerTransportObserver.java Normal file
View File

@ -0,0 +1,64 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
public class ServerTransportObserver implements TransportObserver {
private final ChannelHandlerContext ctx;
private boolean headerSent = false;
public ServerTransportObserver(ChannelHandlerContext ctx) {
this.ctx = ctx;
}
@Override
public void onMetadata(Metadata metadata, boolean endStream, Stream.OperationHandler handler) {
final DefaultHttp2Headers headers = new DefaultHttp2Headers(true);
metadata.forEach(e -> {
headers.set(e.getKey(), e.getValue());
});
if (!headerSent) {
headerSent = true;
headers.status(OK.codeAsText());
headers.set(TripleConstant.CONTENT_TYPE_KEY, TripleConstant.CONTENT_PROTO);
ctx.writeAndFlush(new DefaultHttp2HeadersFrame(headers, endStream));
} else {
ctx.writeAndFlush(new DefaultHttp2HeadersFrame(headers, endStream));
}
}
@Override
public void onData(byte[] data, boolean endStream, Stream.OperationHandler handler) {
ByteBuf buf = ctx.alloc().buffer();
buf.writeByte(0);
buf.writeInt(data.length);
buf.writeBytes(data);
ctx.writeAndFlush(new DefaultHttp2DataFrame(buf, false));
}
@Override
public void onComplete(Stream.OperationHandler handler) {
}
}

36
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/SingleProtobufSerialization.java
View File

@ -16,8 +16,6 @@
*/
package org.apache.dubbo.rpc.protocol.tri;
import com.google.protobuf.CodedInputStream;
import com.google.protobuf.ExtensionRegistryLite;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;
@ -66,54 +64,26 @@ public class SingleProtobufSerialization {
}
}
public int serialize(Object obj, OutputStream os) throws IOException {
public void serialize(Object obj, OutputStream os) throws IOException {
final MessageLite msg = (MessageLite) obj;
msg.writeTo(os);
return msg.getSerializedSize();
}
private SingleMessageMarshaller<?> getMarshaller(Class<?> clz) {
return marshallers.computeIfAbsent(clz, k -> new SingleMessageMarshaller(k));
}
private SingleMessageMarshaller<?> getMarshaller(Object obj) {
return getMarshaller(obj.getClass());
}
public static final class SingleMessageMarshaller<T extends MessageLite> {
private final Parser<T> parser;
private final T defaultInstance;
@SuppressWarnings("unchecked")
SingleMessageMarshaller(Class<T> clz) {
this.defaultInstance = (T) defaultInst(clz);
this.parser = (Parser<T>) defaultInstance.getParserForType();
}
@SuppressWarnings("unchecked")
public Class<T> getMessageClass() {
// Precisely T since protobuf doesn't let messages extend other messages.
return (Class<T>) defaultInstance.getClass();
}
public T getMessagePrototype() {
return defaultInstance;
final T inst = (T) defaultInst(clz);
this.parser = (Parser<T>) inst.getParserForType();
}
public T parse(InputStream stream) throws InvalidProtocolBufferException {
return parser.parseFrom(stream, globalRegistry);
}
private T parseFrom(CodedInputStream stream) throws InvalidProtocolBufferException {
T message = parser.parseFrom(stream, globalRegistry);
try {
stream.checkLastTagWas(0);
return message;
} catch (InvalidProtocolBufferException e) {
e.setUnfinishedMessage(message);
throw e;
}
}
}
}

34
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/Stream.java
View File

@ -16,20 +16,36 @@
*/
package org.apache.dubbo.rpc.protocol.tri;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http2.Http2Headers;
import java.io.InputStream;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.common.stream.StreamObserver;
public interface Stream {
void onHeaders(Http2Headers headers);
Logger LOGGER = LoggerFactory.getLogger(Stream.class);
void onData(InputStream in);
void subscribe(TransportObserver observer);
void onError(GrpcStatus status);
TransportObserver asTransportObserver();
void write(Object obj, ChannelPromise promise) throws Exception;
void subscribe(StreamObserver<Object> observer);
void halfClose() throws Exception;
StreamObserver<Object> asStreamObserver();
void execute(Runnable runnable);
enum OperationResult {
OK,
FAILURE,
NETWORK_FAIL
}
interface OperationHandler {
/**
* @param result operation's result
* @param cause null if the operation succeed
*/
void operationDone(OperationResult result, Throwable cause);
}
}

42
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TransportObserver.java Normal file
View File

@ -0,0 +1,42 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
public interface TransportObserver {
Stream.OperationHandler EMPTY_HANDLER = (result, cause) -> {
};
default void tryOnMetadata(Metadata metadata, boolean endStream) {
onMetadata(metadata, endStream, EMPTY_HANDLER);
}
default void tryOnData(byte[] data, boolean endStream) {
onData(data, endStream, EMPTY_HANDLER);
}
default void tryOnComplete() {
onComplete(EMPTY_HANDLER);
}
void onMetadata(Metadata metadata, boolean endStream, Stream.OperationHandler handler);
void onData(byte[] data, boolean endStream, Stream.OperationHandler handler);
void onComplete(Stream.OperationHandler handler);
}

49
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientHandler.java
View File

@ -17,9 +17,19 @@
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.remoting.api.Connection;
import org.apache.dubbo.remoting.api.ConnectionHandler;
import org.apache.dubbo.remoting.exchange.Request;
import org.apache.dubbo.remoting.exchange.Response;
import org.apache.dubbo.remoting.exchange.support.DefaultFuture2;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.ConsumerModel;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.ServiceRepository;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
@ -28,7 +38,6 @@ import io.netty.handler.codec.http2.Http2GoAwayFrame;
import io.netty.handler.codec.http2.Http2SettingsFrame;
import io.netty.util.ReferenceCountUtil;
import java.io.IOException;
import java.util.concurrent.Executor;
public class TripleClientHandler extends ChannelDuplexHandler {
@ -55,12 +64,40 @@ public class TripleClientHandler extends ChannelDuplexHandler {
}
}
private void writeRequest(ChannelHandlerContext ctx, final Request req, ChannelPromise promise) throws IOException {
private void writeRequest(ChannelHandlerContext ctx, final Request req, final ChannelPromise promise) {
final RpcInvocation inv = (RpcInvocation) req.getData();
final boolean needWrapper = TripleUtil.needWrapper(inv.getParameterTypes());
final URL url = inv.getInvoker().getUrl();
final Executor callback = (Executor) inv.getAttributes().remove("callback.executor");
ClientStream clientStream = new ClientStream(url, ctx, needWrapper, req,callback);
clientStream.write(req, promise);
ServiceRepository repo = ApplicationModel.getServiceRepository();
MethodDescriptor methodDescriptor = repo.lookupMethod(inv.getServiceName(), inv.getMethodName());
final ConsumerModel service = repo.lookupReferredService(url.getServiceKey());
if (service != null) {
ClassLoadUtil.switchContextLoader(service.getServiceInterfaceClass().getClassLoader());
}
final Executor executor = (Executor) inv.getAttributes().remove("callback.executor");
AbstractClientStream stream;
if (methodDescriptor.isUnary()) {
stream = AbstractClientStream.unary(url, executor);
} else {
stream = AbstractClientStream.stream(url);
}
stream.service(service)
.connection(Connection.getConnectionFromChannel(ctx.channel()))
.method(methodDescriptor)
.methodName(methodDescriptor.getMethodName())
.request(req)
.serialize((String) inv.getObjectAttachment(Constants.SERIALIZATION_KEY))
.subscribe(new ClientTransportObserver(ctx, stream, promise));
if (methodDescriptor.isUnary()) {
stream.asStreamObserver().onNext(inv);
stream.asStreamObserver().onCompleted();
} else {
final StreamObserver<Object> streamObserver = (StreamObserver<Object>) inv.getArguments()[0];
stream.subscribe(streamObserver);
Response response = new Response(req.getId(), req.getVersion());
final AppResponse result = new AppResponse(stream.asStreamObserver());
response.setResult(result);
DefaultFuture2.received(stream.getConnection(), response);
}
}
}

12
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleClientInboundHandler.java
View File

@ -16,18 +16,18 @@
*/
package org.apache.dubbo.rpc.protocol.tri;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
public class TripleClientInboundHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
final ClientStream invoker = TripleUtil.getClientStream(ctx);
final ByteBuf buffer = (ByteBuf) msg;
if (invoker != null) {
invoker.onData(new ByteBufInputStream(buffer, buffer.readableBytes(),true));
final AbstractClientStream clientStream = TripleUtil.getClientStream(ctx);
final byte[] data = (byte[]) msg;
if (clientStream != null) {
clientStream.asTransportObserver()
.tryOnData(data, false);
}
}
}

4
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleConstant.java
View File

@ -17,6 +17,9 @@
package org.apache.dubbo.rpc.protocol.tri;
public interface TripleConstant {
String AUTHORITY_KEY = ":authority";
String PATH_KEY = ":path";
String HTTP_STATUS_KEY = "http-status";
String STATUS_KEY = "grpc-status";
String MESSAGE_KEY = "grpc-message";
String TIMEOUT = "grpc-timeout";
@ -29,5 +32,6 @@ public interface TripleConstant {
String UNIT_INFO_KEY = "tri-unit-info";
String SERVICE_VERSION = "tri-service-version";
String SERVICE_GROUP = "tri-service-group";
String TRI_VERSION = "1.0.0";
}

26
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2ClientResponseHandler.java
View File

@ -23,11 +23,12 @@ import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http2.Http2DataFrame;
import io.netty.handler.codec.http2.Http2GoAwayFrame;
import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2HeadersFrame;
import io.netty.handler.codec.http2.Http2StreamFrame;
public final class TripleHttp2ClientResponseHandler extends SimpleChannelInboundHandler<Http2StreamFrame> {
private static final Logger LOGGER = LoggerFactory.getLogger(TripleHttp2ClientResponseHandler.class);
private static final Logger logger = LoggerFactory.getLogger(TripleHttp2ClientResponseHandler.class);
public TripleHttp2ClientResponseHandler() {
super(false);
@ -37,7 +38,10 @@ public final class TripleHttp2ClientResponseHandler extends SimpleChannelInbound
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
super.userEventTriggered(ctx, evt);
if (evt instanceof Http2GoAwayFrame) {
Http2GoAwayFrame event = (Http2GoAwayFrame) evt;
ctx.close();
logger.debug(
"Event triggered, event name is: " + event.name() + ", last stream id is: " + event.lastStreamId());
}
}
@ -53,29 +57,35 @@ public final class TripleHttp2ClientResponseHandler extends SimpleChannelInbound
}
private void onHeadersRead(ChannelHandlerContext ctx, Http2HeadersFrame msg) {
TripleUtil.getClientStream(ctx).onHeaders(msg.headers());
Http2Headers headers = msg.headers();
AbstractClientStream clientStream = TripleUtil.getClientStream(ctx);
final TransportObserver observer = clientStream.asTransportObserver();
observer.tryOnMetadata(new Http2HeaderMeta(headers), false);
if (msg.isEndStream()) {
final ClientStream clientStream = TripleUtil.getClientStream(ctx);
clientStream.halfClose();
observer.tryOnComplete();
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
final ClientStream clientStream = TripleUtil.getClientStream(ctx);
final AbstractClientStream clientStream = TripleUtil.getClientStream(ctx);
final GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withCause(cause);
clientStream.onError(status);
Metadata metadata = new DefaultMetadata();
metadata.put(TripleConstant.STATUS_KEY, Integer.toString(status.code.code));
metadata.put(TripleConstant.MESSAGE_KEY, status.toMessage());
logger.warn("Meet Exception on ClientResponseHandler, status code is: " + status.code, cause);
clientStream.asStreamObserver().onError(status.asException());
ctx.close();
}
public void onDataRead(ChannelHandlerContext ctx, Http2DataFrame msg) throws Exception {
super.channelRead(ctx, msg.content());
if (msg.isEndStream()) {
final ClientStream clientStream = TripleUtil.getClientStream(ctx);
final AbstractClientStream clientStream = TripleUtil.getClientStream(ctx);
// stream already closed;
if (clientStream != null) {
clientStream.halfClose();
clientStream.asTransportObserver().tryOnComplete();
}
}
}

102
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2FrameServerHandler.java
View File

@ -17,14 +17,18 @@
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.model.ApplicationModel;
import org.apache.dubbo.rpc.model.MethodDescriptor;
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 io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
@ -38,12 +42,16 @@ import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2HeadersFrame;
import io.netty.util.ReferenceCountUtil;
import java.util.List;
import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responseErr;
import static org.apache.dubbo.rpc.protocol.tri.TripleUtil.responsePlainTextError;
public class TripleHttp2FrameServerHandler extends ChannelDuplexHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(TripleHttp2FrameServerHandler.class);
private static final PathResolver PATH_RESOLVER = ExtensionLoader.getExtensionLoader(PathResolver.class).getDefaultExtension();
private static final PathResolver PATH_RESOLVER = ExtensionLoader.getExtensionLoader(PathResolver.class)
.getDefaultExtension();
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
@ -74,18 +82,20 @@ public class TripleHttp2FrameServerHandler extends ChannelDuplexHandler {
public void onDataRead(ChannelHandlerContext ctx, Http2DataFrame msg) throws Exception {
super.channelRead(ctx, msg.content());
if (msg.isEndStream()) {
final ServerStream serverStream = TripleUtil.getServerStream(ctx);
// stream already closed;
final AbstractServerStream serverStream = TripleUtil.getServerStream(ctx);
if (serverStream != null) {
serverStream.halfClose();
serverStream.asTransportObserver().tryOnComplete();
}
}
}
private Invoker<?> getInvoker(Http2Headers headers, String serviceName) {
final String version = headers.contains(TripleConstant.SERVICE_VERSION) ? headers.get(TripleConstant.SERVICE_VERSION).toString() : null;
final String group = headers.contains(TripleConstant.SERVICE_GROUP) ? headers.get(TripleConstant.SERVICE_GROUP).toString() : null;
final String version = headers.contains(TripleConstant.SERVICE_VERSION) ? headers.get(
TripleConstant.SERVICE_VERSION).toString() : null;
final String group = headers.contains(TripleConstant.SERVICE_GROUP) ? headers.get(TripleConstant.SERVICE_GROUP)
.toString() : null;
final String key = URL.buildKey(serviceName, group, version);
Invoker<?> invoker = PATH_RESOLVER.resolve(key);
if (invoker == null) {
@ -98,34 +108,39 @@ public class TripleHttp2FrameServerHandler extends ChannelDuplexHandler {
final Http2Headers headers = msg.headers();
if (!HttpMethod.POST.asciiName().contentEquals(headers.method())) {
responsePlainTextError(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED.code(), GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription(String.format("Method '%s' is not supported", headers.method())));
responsePlainTextError(ctx, HttpResponseStatus.METHOD_NOT_ALLOWED.code(),
GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription(String.format("Method '%s' is not supported", headers.method())));
return;
}
if (headers.path() == null) {
responsePlainTextError(ctx, HttpResponseStatus.NOT_FOUND.code(), GrpcStatus.fromCode(Code.UNIMPLEMENTED.code).withDescription("Expected path but is missing"));
responsePlainTextError(ctx, HttpResponseStatus.NOT_FOUND.code(),
GrpcStatus.fromCode(Code.UNIMPLEMENTED.code).withDescription("Expected path but is missing"));
return;
}
final String path = headers.path().toString();
if (path.charAt(0) != '/') {
responsePlainTextError(ctx, HttpResponseStatus.NOT_FOUND.code(), GrpcStatus.fromCode(Code.UNIMPLEMENTED.code)
.withDescription(String.format("Expected path to start with /: %s", path)));
responsePlainTextError(ctx, HttpResponseStatus.NOT_FOUND.code(),
GrpcStatus.fromCode(Code.UNIMPLEMENTED.code)
.withDescription(String.format("Expected path to start with /: %s", path)));
return;
}
final CharSequence contentType = HttpUtil.getMimeType(headers.get(HttpHeaderNames.CONTENT_TYPE));
if (contentType == null) {
responsePlainTextError(ctx, HttpResponseStatus.UNSUPPORTED_MEDIA_TYPE.code(), GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL.code)
.withDescription("Content-Type is missing from the request"));
responsePlainTextError(ctx, HttpResponseStatus.UNSUPPORTED_MEDIA_TYPE.code(),
GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL.code)
.withDescription("Content-Type is missing from the request"));
return;
}
final String contentString = contentType.toString();
if (!TripleUtil.supportContentType(contentString)) {
responsePlainTextError(ctx, HttpResponseStatus.UNSUPPORTED_MEDIA_TYPE.code(), GrpcStatus.fromCode(Code.INTERNAL.code)
.withDescription(String.format("Content-Type '%s' is not supported", contentString)));
responsePlainTextError(ctx, HttpResponseStatus.UNSUPPORTED_MEDIA_TYPE.code(),
GrpcStatus.fromCode(Code.INTERNAL.code)
.withDescription(String.format("Content-Type '%s' is not supported", contentString)));
return;
}
@ -138,24 +153,59 @@ public class TripleHttp2FrameServerHandler extends ChannelDuplexHandler {
String originalMethodName = parts[2];
String methodName = Character.toLowerCase(originalMethodName.charAt(0)) + originalMethodName.substring(1);
final Invoker<?> delegateInvoker = getInvoker(headers, serviceName);
if (delegateInvoker == null) {
responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED).withDescription("Service not found:" + serviceName));
final Invoker<?> invoker = getInvoker(headers, serviceName);
if (invoker == null) {
responseErr(ctx,
GrpcStatus.fromCode(Code.UNIMPLEMENTED).withDescription("Service not found:" + serviceName));
return;
}
ServiceRepository repo = ApplicationModel.getServiceRepository();
final ServiceDescriptor descriptor = repo.lookupService(delegateInvoker.getUrl().getServiceKey());
if (descriptor == null) {
responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED).withDescription("Service not found:" + serviceName));
final ServiceDescriptor serviceDescriptor = repo.lookupService(invoker.getUrl().getServiceKey());
if (serviceDescriptor == null) {
responseErr(ctx,
GrpcStatus.fromCode(Code.UNIMPLEMENTED).withDescription("Service not found:" + serviceName));
return;
}
MethodDescriptor methodDescriptor = null;
List<MethodDescriptor> methodDescriptors = null;
final ServerStream serverStream = new ServerStream(delegateInvoker, descriptor, methodName, ctx);
serverStream.onHeaders(headers);
ctx.channel().attr(TripleUtil.SERVER_STREAM_KEY).set(serverStream);
if (msg.isEndStream()) {
serverStream.halfClose();
if (CommonConstants.$INVOKE.equals(methodName) || CommonConstants.$INVOKE_ASYNC.equals(methodName)) {
methodDescriptor = repo.lookupMethod(GenericService.class.getName(), methodName);
} else if (CommonConstants.$ECHO.equals(methodName)) {
methodDescriptor = repo.lookupMethod(EchoService.class.getName(), methodName);
} else {
methodDescriptors = serviceDescriptor.getMethods(methodName);
if (methodDescriptors == null || methodDescriptors.isEmpty()) {
responseErr(ctx, GrpcStatus.fromCode(Code.UNIMPLEMENTED)
.withDescription("Method :" + methodName + " not found of service:" + serviceName));
return;
}
if (methodDescriptors.size() == 1) {
methodDescriptor = methodDescriptors.get(0);
}
}
final AbstractServerStream stream;
if (methodDescriptor != null && methodDescriptor.isStream()) {
stream = AbstractServerStream.stream(invoker.getUrl());
} else {
stream = AbstractServerStream.unary(invoker.getUrl());
}
stream.service(serviceDescriptor)
.invoker(invoker)
.methodName(methodName)
.subscribe(new ServerTransportObserver(ctx));
if (methodDescriptor != null) {
stream.method(methodDescriptor);
} else {
stream.methods(methodDescriptors);
}
final TransportObserver observer = stream.asTransportObserver();
observer.tryOnMetadata(new Http2HeaderMeta(headers), false);
if (msg.isEndStream()) {
observer.tryOnComplete();
}
ctx.channel().attr(TripleUtil.SERVER_STREAM_KEY).set(stream);
}
}

15
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleHttp2Protocol.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.remoting.api.Http2WireProtocol;
@ -32,7 +31,6 @@ import io.netty.handler.ssl.SslContext;
@Activate
public class TripleHttp2Protocol extends Http2WireProtocol {
@Override
public void close() {
super.close();
@ -50,12 +48,13 @@ public class TripleHttp2Protocol extends Http2WireProtocol {
.frameLogger(SERVER_LOGGER)
.build();
final Http2MultiplexHandler handler = new Http2MultiplexHandler(new TripleServerInitializer());
pipeline.addLast(codec, new TripleServerConnectionHandler(), handler, new SimpleChannelInboundHandler<Object>() {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
// empty
}
});
pipeline.addLast(codec, new TripleServerConnectionHandler(), handler,
new SimpleChannelInboundHandler<Object>() {
@Override
protected void channelRead0(ChannelHandlerContext ctx, Object msg) {
// empty
}
});
}
@Override

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

@ -64,13 +64,13 @@ import static org.apache.dubbo.rpc.Constants.TOKEN_KEY;
*/
public class TripleInvoker<T> extends AbstractInvoker<T> {
private static final ConnectionManager CONNECTION_MANAGER = ExtensionLoader.getExtensionLoader(ConnectionManager.class).getExtension("multiple");
private static final ConnectionManager CONNECTION_MANAGER = ExtensionLoader.getExtensionLoader(
ConnectionManager.class).getExtension("multiple");
private final Connection connection;
private final ReentrantLock destroyLock = new ReentrantLock();
private final Set<Invoker<?>> invokers;
public TripleInvoker(Class<T> serviceType, URL url, Set<Invoker<?>> invokers) throws RemotingException {
super(serviceType, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY});
this.invokers = invokers;
@ -82,7 +82,8 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
RpcInvocation inv = (RpcInvocation) invocation;
final String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment(PATH_KEY, getUrl().getPath());
inv.setAttachment(Constants.SERIALIZATION_KEY, getUrl().getParameter(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION));
inv.setAttachment(Constants.SERIALIZATION_KEY,
getUrl().getParameter(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION));
try {
int timeout = calculateTimeout(invocation, methodName);
invocation.put(TIMEOUT_KEY, timeout);
@ -101,8 +102,7 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
FutureContext.getContext().setCompatibleFuture(respFuture);
AsyncRpcResult result = new AsyncRpcResult(respFuture, inv);
result.setExecutor(executor);
inv.put("callback.executor",executor );
inv.put("callback.executor", executor);
if (!connection.isAvailable()) {
Response response = new Response(req.getId(), req.getVersion());
@ -125,9 +125,13 @@ public class TripleInvoker<T> extends AbstractInvoker<T> {
return result;
} catch (TimeoutException e) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
throw new RpcException(RpcException.TIMEOUT_EXCEPTION,
"Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl()
+ ", cause: " + e.getMessage(), e);
} catch (RemotingException e) {
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
throw new RpcException(RpcException.NETWORK_EXCEPTION,
"Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl()
+ ", cause: " + e.getMessage(), e);
}
}

1
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TriplePathResolver.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.rpc.Invoker;

9
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleProtocol.java
View File

@ -42,9 +42,10 @@ import static org.apache.dubbo.common.constants.CommonConstants.THREADPOOL_KEY;
public class TripleProtocol extends AbstractProtocol implements Protocol {
private static final Logger logger = LoggerFactory.getLogger(TripleProtocol.class);
private final PathResolver pathResolver = ExtensionLoader.getExtensionLoader(PathResolver.class).getDefaultExtension();
private final ExecutorRepository executorRepository = ExtensionLoader.getExtensionLoader(ExecutorRepository.class).getDefaultExtension();
private final PathResolver pathResolver = ExtensionLoader.getExtensionLoader(PathResolver.class)
.getDefaultExtension();
private final ExecutorRepository executorRepository = ExtensionLoader.getExtensionLoader(ExecutorRepository.class)
.getDefaultExtension();
@Override
public int getDefaultPort() {
@ -79,7 +80,7 @@ public class TripleProtocol extends AbstractProtocol implements Protocol {
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
TripleInvoker<T> invoker;
try {
url = ExecutorUtil.setThreadName(url,"DubboClientHandler");
url = ExecutorUtil.setThreadName(url, "DubboClientHandler");
url = url.addParameterIfAbsent(THREADPOOL_KEY, DEFAULT_CLIENT_THREADPOOL);
executorRepository.createExecutorIfAbsent(url);
invoker = new TripleInvoker<>(type, url, invokers);

1
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerConnectionHandler.java
View File

@ -16,7 +16,6 @@
*/
package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.logger.Logger;
import org.apache.dubbo.common.logger.LoggerFactory;

9
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleServerInboundHandler.java
View File

@ -16,18 +16,17 @@
*/
package org.apache.dubbo.rpc.protocol.tri;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
public class TripleServerInboundHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
final ServerStream serverStream = TripleUtil.getServerStream(ctx);
final ByteBuf buffer = (ByteBuf) msg;
final AbstractServerStream serverStream = TripleUtil.getServerStream(ctx);
final byte[] data = (byte[]) msg;
if (serverStream != null) {
serverStream.onData(new ByteBufInputStream(buffer,buffer.readableBytes(),true));
serverStream.asTransportObserver()
.tryOnData(data, false);
}
}
}

162
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleUtil.java
View File

@ -18,7 +18,6 @@ package org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.serialize.MultipleSerialization;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.Constants;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.model.MethodDescriptor;
@ -26,15 +25,10 @@ import org.apache.dubbo.triple.TripleWrapper;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.buffer.ByteBufUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.QueryStringDecoder;
import io.netty.handler.codec.http.QueryStringEncoder;
import io.netty.handler.codec.http2.DefaultHttp2DataFrame;
import io.netty.handler.codec.http2.DefaultHttp2Headers;
import io.netty.handler.codec.http2.DefaultHttp2HeadersFrame;
@ -48,29 +42,24 @@ import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
import java.util.List;
import java.util.Map;
import static io.netty.handler.codec.http.HttpResponseStatus.OK;
public class TripleUtil {
public static final AttributeKey<ServerStream> SERVER_STREAM_KEY = AttributeKey.newInstance("tri_server_stream");
public static final AttributeKey<ClientStream> CLIENT_STREAM_KEY = AttributeKey.newInstance("tri_client_stream");
public static final AttributeKey<AbstractServerStream> SERVER_STREAM_KEY = AttributeKey.newInstance(
"tri_server_stream");
public static final AttributeKey<AbstractClientStream> CLIENT_STREAM_KEY = AttributeKey.newInstance(
"tri_client_stream");
private static final SingleProtobufSerialization pbSerialization = new SingleProtobufSerialization();
private static final Base64.Decoder BASE64_DECODER = Base64.getDecoder();
private static final Base64.Encoder BASE64_ENCODER = Base64.getEncoder().withoutPadding();
public static ServerStream getServerStream(ChannelHandlerContext ctx) {
public static AbstractServerStream getServerStream(ChannelHandlerContext ctx) {
return ctx.channel().attr(TripleUtil.SERVER_STREAM_KEY).get();
}
public static void setClientStream(Channel channel, ClientStream clientStream) {
channel.attr(TripleUtil.CLIENT_STREAM_KEY).set(clientStream);
}
public static ClientStream getClientStream(ChannelHandlerContext ctx) {
public static AbstractClientStream getClientStream(ChannelHandlerContext ctx) {
return ctx.channel().attr(TripleUtil.CLIENT_STREAM_KEY).get();
}
@ -89,51 +78,10 @@ public class TripleUtil {
.status(OK.codeAsText())
.set(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO)
.setInt(TripleConstant.STATUS_KEY, status.code.code)
.set(TripleConstant.MESSAGE_KEY, getErrorMsg(status));
.set(TripleConstant.MESSAGE_KEY, status.toMessage());
ctx.writeAndFlush(new DefaultHttp2HeadersFrame(trailers, true));
}
public static String getErrorMsg(GrpcStatus status) {
final String msg;
if (status.cause == null) {
msg = status.description;
} else {
String placeHolder = status.description == null ? "" : status.description;
msg = StringUtils.toString(placeHolder, status.cause);
}
return percentEncode(msg);
}
public static String limitSizeTo4KB(String desc) {
if (desc.length() < 4096) {
return desc;
} else {
return desc.substring(0, 4086);
}
}
public static String percentDecode(CharSequence corpus) {
if (corpus == null) {
return "";
}
QueryStringDecoder decoder = new QueryStringDecoder("?=" + corpus);
for (Map.Entry<String, List<String>> e : decoder.parameters().entrySet()) {
return e.getKey();
}
return "";
}
public static String percentEncode(String corpus) {
if (corpus == null) {
return "";
}
corpus = limitSizeTo4KB(corpus);
QueryStringEncoder encoder = new QueryStringEncoder("");
encoder.addParam("", corpus);
// ?=
return encoder.toString().substring(2);
}
public static void responsePlainTextError(ChannelHandlerContext ctx, int code, GrpcStatus status) {
Http2Headers headers = new DefaultHttp2Headers(true)
.status("" + code)
@ -145,14 +93,8 @@ public class TripleUtil {
ctx.write(new DefaultHttp2DataFrame(buf, true));
}
public static boolean needWrapper(Class<?>[] parameterTypes) {
if (parameterTypes.length != 1) {
return true;
}
return !Message.class.isAssignableFrom(parameterTypes[0]);
}
public static Object unwrapResp(URL url, TripleWrapper.TripleResponseWrapper wrap, MultipleSerialization serialization) {
public static Object unwrapResp(URL url, TripleWrapper.TripleResponseWrapper wrap,
MultipleSerialization serialization) {
String serializeType = convertHessianFromWrapper(wrap.getSerializeType());
try {
final ByteArrayInputStream bais = new ByteArrayInputStream(wrap.getData().toByteArray());
@ -164,7 +106,8 @@ public class TripleUtil {
}
}
public static Object[] unwrapReq(URL url, TripleWrapper.TripleRequestWrapper wrap, MultipleSerialization multipleSerialization) {
public static Object[] unwrapReq(URL url, TripleWrapper.TripleRequestWrapper wrap,
MultipleSerialization multipleSerialization) {
String serializeType = convertHessianFromWrapper(wrap.getSerializeType());
try {
Object[] arguments = new Object[wrap.getArgsCount()];
@ -180,7 +123,8 @@ public class TripleUtil {
}
}
public static TripleWrapper.TripleResponseWrapper wrapResp(URL url, String serializeType, Object resp, MethodDescriptor desc,
public static TripleWrapper.TripleResponseWrapper wrapResp(URL url, String serializeType, Object resp,
MethodDescriptor desc,
MultipleSerialization multipleSerialization) {
try {
final TripleWrapper.TripleResponseWrapper.Builder builder = TripleWrapper.TripleResponseWrapper.newBuilder()
@ -196,6 +140,46 @@ public class TripleUtil {
}
}
public static TripleWrapper.TripleRequestWrapper wrapReq(URL url, String serializeType, Object req,
String type,
MultipleSerialization multipleSerialization) {
try {
final TripleWrapper.TripleRequestWrapper.Builder builder = TripleWrapper.TripleRequestWrapper.newBuilder()
.addArgTypes(type)
.setSerializeType(convertHessianToWrapper(serializeType));
ByteArrayOutputStream bos = new ByteArrayOutputStream();
multipleSerialization.serialize(url, serializeType, type, req, bos);
builder.addArgs(ByteString.copyFrom(bos.toByteArray()));
bos.close();
return builder.build();
} catch (IOException e) {
throw new RuntimeException("Failed to pack wrapper req", e);
}
}
public static TripleWrapper.TripleRequestWrapper wrapReq(URL url, RpcInvocation invocation,
MultipleSerialization serialization) {
try {
String serializationName = (String) invocation.getObjectAttachment(Constants.SERIALIZATION_KEY);
final TripleWrapper.TripleRequestWrapper.Builder builder = TripleWrapper.TripleRequestWrapper.newBuilder()
.setSerializeType(convertHessianToWrapper(serializationName));
for (int i = 0; i < invocation.getArguments().length; i++) {
final String clz = invocation.getParameterTypes()[i].getName();
builder.addArgTypes(clz);
ByteArrayOutputStream bos = new ByteArrayOutputStream();
serialization.serialize(url, serializationName, clz, invocation.getArguments()[i], bos);
builder.addArgs(ByteString.copyFrom(bos.toByteArray()));
}
return builder.build();
} catch (IOException e) {
throw new RuntimeException("Failed to pack wrapper req", e);
}
}
public static <T> T unpack(byte[] data, Class<T> clz) {
return unpack(new ByteArrayInputStream(data), clz);
}
public static <T> T unpack(InputStream is, Class<T> clz) {
try {
final T req = (T) pbSerialization.deserialize(is, clz);
@ -218,40 +202,18 @@ public class TripleUtil {
}
}
public static ByteBuf pack(ChannelHandlerContext ctx, Object obj) {
public static byte[] pack(Object obj) {
final ByteArrayOutputStream baos = new ByteArrayOutputStream();
try {
final ByteBuf buf = ctx.alloc().buffer();
buf.writeByte(0);
buf.writeInt(0);
final ByteBufOutputStream bos = new ByteBufOutputStream(buf);
final int size = pbSerialization.serialize(obj, bos);
buf.setInt(1, size);
return buf;
pbSerialization.serialize(obj, baos);
} catch (IOException e) {
throw new RuntimeException("Failed to pack req", e);
throw new RuntimeException("Failed to pack protobuf object", e);
}
return baos.toByteArray();
}
public static TripleWrapper.TripleRequestWrapper wrapReq(URL url, RpcInvocation invocation, MultipleSerialization serialization) {
try {
String serializationName = (String) invocation.getObjectAttachment(Constants.SERIALIZATION_KEY);
final TripleWrapper.TripleRequestWrapper.Builder builder = TripleWrapper.TripleRequestWrapper.newBuilder()
.setSerializeType(convertHessianToWrapper(serializationName));
for (int i = 0; i < invocation.getArguments().length; i++) {
final String clz = invocation.getParameterTypes()[i].getName();
builder.addArgTypes(clz);
ByteArrayOutputStream bos = new ByteArrayOutputStream();
serialization.serialize(url, serializationName, clz, invocation.getArguments()[i], bos);
builder.addArgs(ByteString.copyFrom(bos.toByteArray()));
}
return builder.build();
} catch (IOException e) {
throw new RuntimeException("Failed to pack wrapper req", e);
}
}
public static String encodeWrapper(URL url, Object obj, String serializeType, MultipleSerialization serialization) throws IOException {
public static String encodeWrapper(URL url, Object obj, String serializeType, MultipleSerialization serialization)
throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
serialization.serialize(url, serializeType, obj.getClass().getName(), obj, bos);
final TripleWrapper.TripleRequestWrapper wrap = TripleWrapper.TripleRequestWrapper.newBuilder()
@ -271,7 +233,8 @@ public class TripleUtil {
return BASE64_ENCODER.encode(in);
}
public static Object decodeObjFromHeader(URL url, CharSequence value, MultipleSerialization serialization) throws InvalidProtocolBufferException {
public static Object decodeObjFromHeader(URL url, CharSequence value, MultipleSerialization serialization)
throws InvalidProtocolBufferException {
final byte[] decode = decodeASCIIByte(value);
final TripleWrapper.TripleRequestWrapper wrapper = TripleWrapper.TripleRequestWrapper.parseFrom(decode);
final Object[] objects = TripleUtil.unwrapReq(url, wrapper, serialization);
@ -296,5 +259,4 @@ public class TripleUtil {
return serializeType;
}
}

78
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryClientStream.java Normal file
View File

@ -0,0 +1,78 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.remoting.exchange.Response;
import org.apache.dubbo.remoting.exchange.support.DefaultFuture2;
import org.apache.dubbo.rpc.AppResponse;
import java.util.concurrent.Executor;
public class UnaryClientStream extends AbstractClientStream implements Stream {
protected UnaryClientStream(URL url, Executor executor) {
super(url, executor);
}
@Override
protected StreamObserver<Object> createStreamObserver() {
return new ClientStreamObserver();
}
@Override
protected TransportObserver createTransportObserver() {
return new UnaryClientTransportObserver();
}
private class UnaryClientTransportObserver extends UnaryTransportObserver implements TransportObserver {
@Override
public void doOnComplete(OperationHandler handler) {
execute(() -> {
try {
final Object resp = deserializeResponse(getData());
Response response = new Response(getRequest().getId(), TripleConstant.TRI_VERSION);
final AppResponse result = new AppResponse(resp);
result.setObjectAttachments(parseMetadataToMap(getTrailers()));
response.setResult(result);
DefaultFuture2.received(getConnection(), response);
} catch (Exception e) {
final GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withCause(e)
.withDescription("Failed to deserialize response");
onError(status);
}
});
}
protected void onError(GrpcStatus status) {
Response response = new Response(getRequest().getId(), TripleConstant.TRI_VERSION);
if (status.description != null) {
response.setErrorMessage(status.description);
} else {
response.setErrorMessage(status.cause.getMessage());
}
final byte code = GrpcStatus.toDubboStatus(status.code);
response.setStatus(code);
DefaultFuture2.received(getConnection(), response);
}
}
}

148
dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStream.java Normal file
View File

@ -0,0 +1,148 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.remoting.TimeoutException;
import org.apache.dubbo.rpc.AppResponse;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcInvocation;
import io.netty.handler.codec.http.HttpHeaderNames;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.function.BiConsumer;
import java.util.function.Function;
public class UnaryServerStream extends AbstractServerStream implements Stream {
protected UnaryServerStream(URL url) {
super(url);
}
@Override
protected StreamObserver<Object> createStreamObserver() {
return null;
}
@Override
protected TransportObserver createTransportObserver() {
return new UnaryServerTransportObserver();
}
private class UnaryServerTransportObserver extends UnaryTransportObserver implements TransportObserver {
@Override
protected void onError(GrpcStatus status) {
transportError(status);
}
@Override
public void doOnComplete(OperationHandler handler) {
if (getData() == null) {
onError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription("Missing request data"));
return;
}
execute(this::invoke);
}
public void invoke() {
final RpcInvocation invocation;
try {
final Object[] arguments = deserializeRequest(getData());
if (arguments != null) {
invocation = buildInvocation(getHeaders());
invocation.setArguments(arguments);
} else {
return;
}
} catch (Throwable t) {
LOGGER.warn("Exception processing triple message", t);
transportError(GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription("Decode request failed:" + t.getMessage()));
return;
}
final Result result = getInvoker().invoke(invocation);
CompletionStage<Object> future = result.thenApply(Function.identity());
BiConsumer<Object, Throwable> onComplete = (appResult, t) -> {
try {
if (t != null) {
if (t instanceof TimeoutException) {
transportError(GrpcStatus.fromCode(GrpcStatus.Code.DEADLINE_EXCEEDED).withCause(t));
} else {
transportError(GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN).withCause(t));
}
return;
}
AppResponse response = (AppResponse) appResult;
if (response.hasException()) {
final Throwable exception = response.getException();
if (exception instanceof TripleRpcException) {
transportError(((TripleRpcException) exception).getStatus());
} else {
transportError(GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN)
.withCause(exception));
}
return;
}
Metadata metadata = new DefaultMetadata();
metadata.put(HttpHeaderNames.CONTENT_TYPE, TripleConstant.CONTENT_PROTO);
getTransportSubscriber().tryOnMetadata(metadata, false);
final ClassLoader tccl = Thread.currentThread().getContextClassLoader();
final byte[] data;
try {
ClassLoadUtil.switchContextLoader(
getProviderModel().getServiceInterfaceClass().getClassLoader());
data = encodeResponse(response.getValue());
} finally {
ClassLoadUtil.switchContextLoader(tccl);
}
getTransportSubscriber().tryOnData(data, false);
Metadata trailers = new DefaultMetadata()
.put(TripleConstant.STATUS_KEY, Integer.toString(GrpcStatus.Code.OK.code));
final Map<String, Object> attachments = response.getObjectAttachments();
if (attachments != null) {
convertAttachment(trailers, attachments);
}
getTransportSubscriber().tryOnMetadata(trailers, true);
} catch (Throwable e) {
LOGGER.warn("Exception processing triple message", e);
if (e instanceof TripleRpcException) {
transportError(((TripleRpcException) e).getStatus());
} else {
transportError(GrpcStatus.fromCode(GrpcStatus.Code.UNKNOWN)
.withDescription("Exception occurred in provider's execution:" + e.getMessage())
.withCause(e));
}
}
};
future.whenComplete(onComplete);
RpcContext.removeContext();
}
}
}

42
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/GrpcStatusTest.java Normal file
View File

@ -0,0 +1,42 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
class GrpcStatusTest {
@Test
public void fromMessage() {
String origin = "haha test 😊";
final GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription(origin);
Assertions.assertNotEquals(origin, status.toMessage());
final String decoded = GrpcStatus.fromMessage(status.toMessage());
Assertions.assertEquals(origin, decoded);
}
@Test
public void toMessage() {
String content = "\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n";
final GrpcStatus status = GrpcStatus.fromCode(GrpcStatus.Code.INTERNAL)
.withDescription(content);
Assertions.assertNotEquals(content, status.toMessage());
}
}

79
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/UnaryClientStreamTest.java Normal file
View File

@ -0,0 +1,79 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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 org.apache.dubbo.rpc.protocol.tri;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;
import java.util.concurrent.Executor;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class UnaryClientStreamTest {
@Test
@SuppressWarnings("all")
public void testInit() {
URL url = new URL("test", "1.2.3.4", 8080);
final Executor executor = Mockito.mock(Executor.class);
final UnaryClientStream stream = UnaryClientStream.unary(url, executor);
final StreamObserver<Object> observer = stream.asStreamObserver();
RpcInvocation inv = Mockito.mock(RpcInvocation.class);
// no invoker
Assertions.assertThrows(NullPointerException.class, () -> observer.onNext(inv));
final Invoker mockInvoker = Mockito.mock(Invoker.class);
when(mockInvoker.getUrl()).thenReturn(url);
when(inv.getInvoker()).thenReturn(mockInvoker);
// no subscriber
Assertions.assertThrows(NullPointerException.class, () -> observer.onNext(inv));
verify(mockInvoker, times(2)).getUrl();
TransportObserver transportObserver = Mockito.mock(TransportObserver.class);
stream.subscribe(transportObserver);
// no method descriptor
Assertions.assertThrows(NullPointerException.class, () -> observer.onNext(inv));
Mockito.verify(transportObserver).tryOnMetadata(any(), anyBoolean());
MethodDescriptor md = Mockito.mock(MethodDescriptor.class);
when(md.isNeedWrap()).thenReturn(true);
when(md.isStream()).thenReturn(false);
stream.method(md);
// bad request
Assertions.assertThrows(NullPointerException.class, () -> observer.onNext(inv));
String[] params = new String[]{"rainbow ponies!"};
when(inv.getArguments()).thenReturn(params);
// no serialization
Assertions.assertThrows(NullPointerException.class, () -> observer.onNext(inv));
// Map<String,Object> attachemnts=new HashMap<>();
// when(inv.getObjectAttachments()).thenReturn(attachemnts);
// attachemnts.put(Constants.SERIALIZATION_KEY, "hessian2");
// observer.onNext(inv);
}
}

14
dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/TripleUtilTest.java → dubbo-rpc/dubbo-rpc-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/UnaryServerStreamTest.java
View File

@ -14,18 +14,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.dubbo.rpc.protocol.tri;
import org.junit.jupiter.api.Assertions;
import org.apache.dubbo.common.URL;
import org.junit.jupiter.api.Test;
class TripleUtilTest {
class UnaryServerStreamTest {
@Test
void percentEncoding() {
String content="\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n";
final String encoded = TripleUtil.percentEncode(content);
final String decoded = TripleUtil.percentDecode(encoded);
Assertions.assertEquals(content,decoded);
@SuppressWarnings("all")
public void testInit() {
URL url = new URL("test", "1.2.3.4", 8080);
}
}

28
dubbo-rpc/dubbo-rpc-triple/src/test/resources/log4j.xml
View File

@ -1,19 +1,19 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
contributor license agreements. See the NOTICE file distributed with
this work for additional information regarding copyright ownership.
The ASF licenses this file to You 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.
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You 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.
-->
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">