http protocol detector (#12452)
* http protocol detector * make differnece between rest and qos http * set rest http wire protocol spi order after qos http detector * make differnce between rest and qos http detector * add differnt service to server url * adjust code position * complete some change request * passing ServiceDeployer to PortUnificationServer through URL * merge code and add qos detect ut
This commit is contained in:
suncairong163
GitHub
parent
391d3cdb35
commit
85bc6ea472
|
|
@ -609,9 +609,7 @@ public interface CommonConstants {
|
|||
|
||||
|
||||
/**
|
||||
*
|
||||
* used in JVMUtil.java ,Control stack print lines, default is 32 lines
|
||||
*
|
||||
*/
|
||||
String DUBBO_JSTACK_MAXLINE = "dubbo.jstack-dump.max-line";
|
||||
|
||||
|
|
@ -634,4 +632,6 @@ public interface CommonConstants {
|
|||
|
||||
String DUBBO_PACKABLE_METHOD_FACTORY = "dubbo.application.parameters." + PACKABLE_METHOD_FACTORY_KEY;
|
||||
|
||||
String REST_SERVICE_DEPLOYER_URL_ATTRIBUTE_KEY = "restServiceDeployerAttributeKey";
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,7 +22,7 @@ import org.apache.dubbo.rpc.model.FrameworkModel;
|
|||
|
||||
public class QosDetector implements ProtocolDetector {
|
||||
|
||||
private final QosHTTP1Detector qosHTTP1Detector = new QosHTTP1Detector();
|
||||
private final QosHTTP1Detector qosHTTP1Detector;
|
||||
private final TelnetDetector telnetDetector;
|
||||
private boolean QosEnableFlag = true;
|
||||
|
||||
|
|
@ -32,6 +32,7 @@ public class QosDetector implements ProtocolDetector {
|
|||
|
||||
public QosDetector(FrameworkModel frameworkModel) {
|
||||
this.telnetDetector = new TelnetDetector(frameworkModel);
|
||||
qosHTTP1Detector = new QosHTTP1Detector(frameworkModel);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
|
|||
|
|
@ -16,24 +16,61 @@
|
|||
*/
|
||||
package org.apache.dubbo.qos.pu;
|
||||
|
||||
import org.apache.dubbo.remoting.api.ProtocolDetector;
|
||||
import org.apache.dubbo.qos.api.BaseCommand;
|
||||
import org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector;
|
||||
import org.apache.dubbo.remoting.buffer.ChannelBuffer;
|
||||
import org.apache.dubbo.rpc.model.FrameworkModel;
|
||||
|
||||
public class QosHTTP1Detector implements ProtocolDetector {
|
||||
private static boolean isHttp(int magic) {
|
||||
return magic == 'G' || magic == 'P';
|
||||
|
||||
public class QosHTTP1Detector extends AbstractHttpProtocolDetector {
|
||||
private static final char[][] QOS_METHODS_PREFIX = getQOSHttpMethodsPrefix();
|
||||
|
||||
FrameworkModel frameworkModel;
|
||||
|
||||
public QosHTTP1Detector(FrameworkModel frameworkModel) {
|
||||
this.frameworkModel = frameworkModel;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public Result detect(ChannelBuffer in) {
|
||||
if (in.readableBytes() < 2) {
|
||||
return Result.needMoreData();
|
||||
}
|
||||
final int magic = in.getByte(in.readerIndex());
|
||||
// h2 starts with "PR"
|
||||
if (isHttp(magic) && in.getByte(in.readerIndex()+1) != 'R' ){
|
||||
|
||||
if (prefixMatch(QOS_METHODS_PREFIX, in, 3)) {
|
||||
|
||||
// make distinguish from rest ,read request url
|
||||
String requestURL = readRequestLine(in);
|
||||
|
||||
// url split by / length judge
|
||||
if (!isQosRequestURL(requestURL)) {
|
||||
return Result.unrecognized();
|
||||
}
|
||||
|
||||
// command exist judge, when /cmd or /cmd/appName we prefer response by rest http
|
||||
// decrease the affect to user
|
||||
BaseCommand command = commandExist(requestURL);
|
||||
|
||||
if (command == null) {
|
||||
return Result.unrecognized();
|
||||
}
|
||||
|
||||
|
||||
return Result.recognized();
|
||||
}
|
||||
|
||||
return Result.unrecognized();
|
||||
}
|
||||
|
||||
private BaseCommand commandExist(String requestURL) {
|
||||
BaseCommand command = null;
|
||||
try {
|
||||
String cmd = splitAndGetFirst(requestURL);
|
||||
command = frameworkModel.getExtensionLoader(BaseCommand.class).getExtension(cmd);
|
||||
} catch (Throwable throwable) {
|
||||
//can't find command
|
||||
}
|
||||
return command;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,9 +18,12 @@
|
|||
package org.apache.dubbo.qos.pu;
|
||||
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.remoting.api.ProtocolDetector;
|
||||
import org.apache.dubbo.remoting.api.pu.ChannelOperator;
|
||||
import org.apache.dubbo.remoting.buffer.HeapChannelBuffer;
|
||||
import org.apache.dubbo.rpc.model.FrameworkModel;
|
||||
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.mockito.ArgumentMatchers.anyList;
|
||||
|
|
@ -38,4 +41,25 @@ class QosWireProtocolTest {
|
|||
verify(channelOperator).configChannelHandler(anyList());
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
void testQosHttp1Detector() {
|
||||
QosHTTP1Detector qosHTTP1Detector = new QosHTTP1Detector(FrameworkModel.defaultModel());
|
||||
HeapChannelBuffer heapChannelBuffer = new HeapChannelBuffer(30);
|
||||
heapChannelBuffer.writeBytes("GET /ls HTTP/1.1".getBytes());
|
||||
ProtocolDetector.Result detect = qosHTTP1Detector.detect(heapChannelBuffer);
|
||||
Assertions.assertEquals(ProtocolDetector.Result.recognized().flag(), detect.flag());
|
||||
|
||||
heapChannelBuffer = new HeapChannelBuffer(30);
|
||||
heapChannelBuffer.writeBytes("GET /ls/appName HTTP/1.1".getBytes());
|
||||
detect = qosHTTP1Detector.detect(heapChannelBuffer);
|
||||
Assertions.assertEquals(ProtocolDetector.Result.recognized().flag(), detect.flag());
|
||||
|
||||
heapChannelBuffer = new HeapChannelBuffer(30);
|
||||
heapChannelBuffer.writeBytes("GET /rest/demo HTTP/1.1".getBytes());
|
||||
detect = qosHTTP1Detector.detect(heapChannelBuffer);
|
||||
Assertions.assertEquals(ProtocolDetector.Result.unrecognized().flag(), detect.flag());
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -18,6 +18,9 @@
|
|||
package org.apache.dubbo.remoting;
|
||||
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
public interface Constants {
|
||||
|
||||
String BUFFER_KEY = "buffer";
|
||||
|
|
@ -161,4 +164,8 @@ public interface Constants {
|
|||
String OK_HTTP = "ok-http";
|
||||
String URL_CONNECTION = "url-connection";
|
||||
String APACHE_HTTP_CLIENT = "apache-http-client";
|
||||
String PORT_UNIFICATION_NETTY4_SERVER = "netty4";
|
||||
|
||||
List<String> REST_SERVER = Arrays.asList("jetty", "tomcat", "netty");
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,176 @@
|
|||
/*
|
||||
* 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.remoting.api;
|
||||
|
||||
|
||||
import org.apache.dubbo.remoting.buffer.ChannelBuffer;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import static org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector.HttpMethod.DELETE;
|
||||
import static org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector.HttpMethod.GET;
|
||||
import static org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector.HttpMethod.HEAD;
|
||||
import static org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector.HttpMethod.OPTIONS;
|
||||
import static org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector.HttpMethod.PATCH;
|
||||
import static org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector.HttpMethod.POST;
|
||||
import static org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector.HttpMethod.PUT;
|
||||
import static org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector.HttpMethod.TRACE;
|
||||
|
||||
|
||||
/**
|
||||
* http protocol detector
|
||||
*/
|
||||
public abstract class AbstractHttpProtocolDetector implements ProtocolDetector {
|
||||
|
||||
protected int empty = ' ';
|
||||
protected static String SIMPLE_HTTP = "XXX HTTP/1";
|
||||
|
||||
|
||||
|
||||
protected static final List<HttpMethod> QOS_HTTP_METHOD = Arrays.asList(GET, POST);
|
||||
|
||||
/**
|
||||
* rank by frequency
|
||||
* first GET ,POST,DELETE,PUT
|
||||
* second HEAD,PATCH,OPTIONS,TRACE
|
||||
*/
|
||||
protected static final List<HttpMethod> HTTP_METHODS = Arrays.asList(GET, POST, DELETE, PUT, HEAD, PATCH, OPTIONS, TRACE);
|
||||
|
||||
protected static char[][] getHttpMethodsPrefix(int length, List<HttpMethod> httpMethods) {
|
||||
if (0 >= length || length > 3) {
|
||||
throw new IllegalArgumentException("Current substring length is beyond Http methods length");
|
||||
}
|
||||
|
||||
List<char[]> prefix = new ArrayList<>();
|
||||
for (HttpMethod httpMethod : httpMethods) {
|
||||
prefix.add(httpMethod.getValue().substring(0, length).toCharArray());
|
||||
}
|
||||
|
||||
return prefix.toArray(new char[0][]);
|
||||
|
||||
}
|
||||
|
||||
|
||||
protected static char[][] getHttpMethodsPrefix() {
|
||||
return getHttpMethodsPrefix(3, HTTP_METHODS);
|
||||
}
|
||||
|
||||
protected static char[][] getQOSHttpMethodsPrefix() {
|
||||
return getHttpMethodsPrefix(3, QOS_HTTP_METHOD);
|
||||
}
|
||||
|
||||
/**
|
||||
* qos /name/appName
|
||||
*
|
||||
* @param requestUrl
|
||||
* @return
|
||||
*/
|
||||
protected boolean isQosRequestURL(String requestUrl) {
|
||||
|
||||
if (requestUrl == null) {
|
||||
return false;
|
||||
}
|
||||
|
||||
String[] split = requestUrl.split("/");
|
||||
|
||||
if (split.length <= 3) {
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
protected String splitAndGetFirst(String str) {
|
||||
|
||||
return splitAndGet(str, 1);
|
||||
}
|
||||
|
||||
protected String splitAndGet(String str, int index) {
|
||||
if (str == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
String[] split = str.split("/");
|
||||
|
||||
if (split.length - 1 < index) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return split[index];
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* between first and second empty char
|
||||
*
|
||||
* @param buffer
|
||||
* @return
|
||||
*/
|
||||
protected String readRequestLine(ChannelBuffer buffer) {
|
||||
|
||||
// GET /test/demo HTTP/1.1
|
||||
int firstEmptyIndex = 0;
|
||||
// read first empty
|
||||
for (int i = 0; i < Integer.MAX_VALUE; i++) {
|
||||
|
||||
int read = getByteByIndex(buffer, i);
|
||||
if (read == empty) {
|
||||
firstEmptyIndex = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
StringBuilder stringBuilder = new StringBuilder();
|
||||
|
||||
for (int i = firstEmptyIndex + 1; i < Integer.MAX_VALUE; i++) {
|
||||
int read = getByteByIndex(buffer, i);
|
||||
// second empty break
|
||||
if (read == empty) {
|
||||
break;
|
||||
}
|
||||
stringBuilder.append((char) read);
|
||||
}
|
||||
|
||||
return stringBuilder.toString();
|
||||
|
||||
}
|
||||
|
||||
public enum HttpMethod {
|
||||
GET("GET"),
|
||||
HEAD("HEAD"),
|
||||
POST("POST"),
|
||||
PUT("PUT"),
|
||||
|
||||
PATCH("PATCH"),
|
||||
DELETE("DELETE"),
|
||||
OPTIONS("OPTIONS"),
|
||||
TRACE("TRACE");
|
||||
|
||||
|
||||
HttpMethod(String value) {
|
||||
this.value = value;
|
||||
}
|
||||
|
||||
private String value;
|
||||
|
||||
public String getValue() {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -74,4 +74,33 @@ public interface ProtocolDetector {
|
|||
enum Flag {
|
||||
RECOGNIZED, UNRECOGNIZED, NEED_MORE_DATA
|
||||
}
|
||||
|
||||
default int getByteByIndex(ChannelBuffer buffer, int index) {
|
||||
return buffer.getByte(buffer.readerIndex() + index);
|
||||
}
|
||||
|
||||
default boolean prefixMatch(char[][] prefixes, ChannelBuffer buffer, int length) {
|
||||
|
||||
int[] ints = new int[length];
|
||||
for (int i = 0; i < length; i++) {
|
||||
ints[i] = getByteByIndex(buffer, i);
|
||||
}
|
||||
|
||||
// prefix match
|
||||
for (char[] prefix : prefixes) {
|
||||
|
||||
boolean matched = true;
|
||||
for (int j = 0; j < length; j++) {
|
||||
if (prefix[j] != ints[j]) {
|
||||
matched = false;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (matched) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ import org.apache.dubbo.remoting.api.pu.ChannelOperator;
|
|||
import org.apache.dubbo.remoting.api.pu.DefaultCodec;
|
||||
import org.apache.dubbo.remoting.transport.codec.CodecAdapter;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
|
||||
public class NettyConfigOperator implements ChannelOperator {
|
||||
|
|
@ -55,14 +56,14 @@ public class NettyConfigOperator implements ChannelOperator {
|
|||
}
|
||||
if (url.getOrDefaultFrameworkModel().getExtensionLoader(Codec2.class).hasExtension(codecName)) {
|
||||
codec2 = url.getOrDefaultFrameworkModel().getExtensionLoader(Codec2.class).getExtension(codecName);
|
||||
} else if(url.getOrDefaultFrameworkModel().getExtensionLoader(Codec.class).hasExtension(codecName)){
|
||||
} else if (url.getOrDefaultFrameworkModel().getExtensionLoader(Codec.class).hasExtension(codecName)) {
|
||||
codec2 = new CodecAdapter(url.getOrDefaultFrameworkModel().getExtensionLoader(Codec.class)
|
||||
.getExtension(codecName));
|
||||
}else {
|
||||
} else {
|
||||
codec2 = url.getOrDefaultFrameworkModel().getExtensionLoader(Codec2.class).getExtension("default");
|
||||
}
|
||||
|
||||
if (!(codec2 instanceof DefaultCodec)){
|
||||
if (!(codec2 instanceof DefaultCodec)) {
|
||||
((NettyChannel) channel).setCodec(codec2);
|
||||
NettyCodecAdapter codec = new NettyCodecAdapter(codec2, channel.getUrl(), handler);
|
||||
((NettyChannel) channel).getNioChannel().pipeline().addLast(
|
||||
|
|
@ -72,21 +73,17 @@ public class NettyConfigOperator implements ChannelOperator {
|
|||
);
|
||||
}
|
||||
|
||||
for (ChannelHandler handler: handlerList) {
|
||||
for (ChannelHandler handler : handlerList) {
|
||||
if (handler instanceof ChannelHandlerPretender) {
|
||||
Object realHandler = ((ChannelHandlerPretender) handler).getRealHandler();
|
||||
if(realHandler instanceof io.netty.channel.ChannelHandler) {
|
||||
((NettyChannel) channel).getNioChannel().pipeline().addLast(
|
||||
(io.netty.channel.ChannelHandler) realHandler
|
||||
);
|
||||
}
|
||||
addRealHandler(realHandler);
|
||||
}
|
||||
}
|
||||
|
||||
// todo distinguish between client and server channel
|
||||
if( isClientSide(channel)){
|
||||
if (isClientSide(channel)) {
|
||||
//todo config client channel handler
|
||||
}else {
|
||||
} else {
|
||||
NettyServerHandler sh = new NettyServerHandler(channel.getUrl(), handler);
|
||||
((NettyChannel) channel).getNioChannel().pipeline().addLast(
|
||||
sh
|
||||
|
|
@ -94,6 +91,28 @@ public class NettyConfigOperator implements ChannelOperator {
|
|||
}
|
||||
}
|
||||
|
||||
private void addRealHandler(Object realHandler) {
|
||||
if (realHandler instanceof Collection) {
|
||||
Collection realHandlers = (Collection) realHandler;
|
||||
|
||||
for (Object handler : realHandlers) {
|
||||
addChannelHandler(handler);
|
||||
}
|
||||
} else {
|
||||
addChannelHandler(realHandler);
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
private void addChannelHandler(Object channelHandler) {
|
||||
if (!(channelHandler instanceof io.netty.channel.ChannelHandler)) {
|
||||
return;
|
||||
}
|
||||
((NettyChannel) channel).getNioChannel().pipeline().addLast((io.netty.channel.ChannelHandler) channelHandler);
|
||||
}
|
||||
|
||||
public void setDetectResult(ProtocolDetector.Result detectResult) {
|
||||
this.detectResult = detectResult;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -103,6 +103,7 @@ public class NettyPortUnificationServerHandler extends ByteToMessageDecoder {
|
|||
enableSsl(ctx, providerConnectionConfig);
|
||||
} else {
|
||||
for (final WireProtocol protocol : protocols) {
|
||||
|
||||
in.markReaderIndex();
|
||||
ChannelBuffer buf = new NettyBackedChannelBuffer(in);
|
||||
final ProtocolDetector.Result result = protocol.detector().detect(buf);
|
||||
|
|
|
|||
|
|
@ -25,7 +25,7 @@
|
|||
<artifactId>dubbo-rpc-rest</artifactId>
|
||||
<packaging>jar</packaging>
|
||||
<name>${project.artifactId}</name>
|
||||
<description>The JAX-RS rpc module of dubbo project</description>
|
||||
<description>The rest rpc module of dubbo project</description>
|
||||
<properties>
|
||||
<skip_maven_deploy>false</skip_maven_deploy>
|
||||
</properties>
|
||||
|
|
@ -40,6 +40,12 @@
|
|||
<artifactId>dubbo-remoting-http</artifactId>
|
||||
<version>${project.parent.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.dubbo</groupId>
|
||||
<artifactId>dubbo-remoting-netty4</artifactId>
|
||||
<version>${project.parent.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.dubbo</groupId>
|
||||
<artifactId>dubbo-cluster</artifactId>
|
||||
|
|
@ -119,12 +125,6 @@
|
|||
<version>${project.parent.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>javax.servlet</groupId>
|
||||
<artifactId>javax.servlet-api</artifactId>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.netty</groupId>
|
||||
<artifactId>netty-codec-http</artifactId>
|
||||
|
|
|
|||
|
|
@ -1,214 +0,0 @@
|
|||
/*
|
||||
* 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.rest;
|
||||
|
||||
import io.netty.buffer.PooledByteBufAllocator;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelOption;
|
||||
import io.netty.handler.codec.http.HttpObjectAggregator;
|
||||
import io.netty.handler.codec.http.HttpRequestDecoder;
|
||||
import io.netty.handler.codec.http.HttpResponseEncoder;
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.common.utils.NetUtils;
|
||||
import org.apache.dubbo.common.utils.StringUtils;
|
||||
import org.apache.dubbo.metadata.rest.PathMatcher;
|
||||
import org.apache.dubbo.metadata.rest.RestMethodMetadata;
|
||||
import org.apache.dubbo.metadata.rest.ServiceRestMetadata;
|
||||
import org.apache.dubbo.rpc.Invoker;
|
||||
import org.apache.dubbo.rpc.protocol.rest.constans.RestConstant;
|
||||
import org.apache.dubbo.rpc.protocol.rest.exception.mapper.ExceptionMapper;
|
||||
import org.apache.dubbo.rpc.protocol.rest.handler.NettyHttpHandler;
|
||||
import org.apache.dubbo.rpc.protocol.rest.netty.NettyServer;
|
||||
import org.apache.dubbo.rpc.protocol.rest.netty.RestHttpRequestDecoder;
|
||||
import org.apache.dubbo.rpc.protocol.rest.netty.UnSharedHandlerCreator;
|
||||
import org.apache.dubbo.rpc.protocol.rest.netty.ssl.SslServerTlsHandler;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.BACKLOG_KEY;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.IO_THREADS_KEY;
|
||||
import static org.apache.dubbo.remoting.Constants.BIND_IP_KEY;
|
||||
import static org.apache.dubbo.remoting.Constants.BIND_PORT_KEY;
|
||||
import static org.apache.dubbo.remoting.Constants.DEFAULT_IO_THREADS;
|
||||
|
||||
/**
|
||||
* netty http server
|
||||
*/
|
||||
public class NettyHttpRestServer implements RestProtocolServer {
|
||||
|
||||
private final PathAndInvokerMapper pathAndInvokerMapper = new PathAndInvokerMapper();
|
||||
private final ExceptionMapper exceptionMapper = new ExceptionMapper();
|
||||
private NettyServer server = getNettyServer();
|
||||
|
||||
/**
|
||||
* for triple override
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
protected NettyServer getNettyServer() {
|
||||
return new NettyServer();
|
||||
}
|
||||
|
||||
private String address;
|
||||
|
||||
private final Map<String, Object> attributes = new ConcurrentHashMap<>();
|
||||
|
||||
|
||||
@Override
|
||||
public String getAddress() {
|
||||
return address;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setAddress(String address) {
|
||||
|
||||
this.address = address;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
server.stop();
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, Object> getAttributes() {
|
||||
return attributes;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void start(URL url) {
|
||||
|
||||
registerExceptionMapper(url);
|
||||
|
||||
String bindIp = url.getParameter(BIND_IP_KEY, url.getHost());
|
||||
if (!url.isAnyHost() && NetUtils.isValidLocalHost(bindIp)) {
|
||||
server.setHostname(bindIp);
|
||||
}
|
||||
server.setPort(url.getParameter(BIND_PORT_KEY, url.getPort()));
|
||||
|
||||
// child options
|
||||
server.setChildChannelOptions(getChildChannelOptionMap(url));
|
||||
|
||||
// set options
|
||||
server.setChannelOptions(getChannelOptionMap(url));
|
||||
// set unshared callback
|
||||
server.setUnSharedHandlerCallBack(getUnSharedHttpChannelHandlers());
|
||||
// set channel handler and @Shared
|
||||
server.setChannelHandlers(getChannelHandlers(url));
|
||||
server.setIoWorkerCount(url.getParameter(IO_THREADS_KEY, DEFAULT_IO_THREADS));
|
||||
server.start(url);
|
||||
}
|
||||
|
||||
private UnSharedHandlerCreator getUnSharedHttpChannelHandlers() {
|
||||
return new UnSharedHandlerCreator() {
|
||||
@Override
|
||||
public List<ChannelHandler> getUnSharedHandlers(URL url) {
|
||||
return Arrays.asList(
|
||||
// add SslServerTlsHandler
|
||||
new SslServerTlsHandler(url),
|
||||
new HttpRequestDecoder(
|
||||
url.getParameter(RestConstant.MAX_INITIAL_LINE_LENGTH_PARAM, RestConstant.MAX_INITIAL_LINE_LENGTH),
|
||||
url.getParameter(RestConstant.MAX_HEADER_SIZE_PARAM, RestConstant.MAX_HEADER_SIZE),
|
||||
url.getParameter(RestConstant.MAX_CHUNK_SIZE_PARAM, RestConstant.MAX_CHUNK_SIZE)),
|
||||
new HttpObjectAggregator(url.getParameter(RestConstant.MAX_REQUEST_SIZE_PARAM, RestConstant.MAX_REQUEST_SIZE)),
|
||||
new HttpResponseEncoder(), new RestHttpRequestDecoder(new NettyHttpHandler(pathAndInvokerMapper, exceptionMapper), url))
|
||||
;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* create child channel options map
|
||||
*
|
||||
* @param url
|
||||
* @return
|
||||
*/
|
||||
protected Map<ChannelOption, Object> getChildChannelOptionMap(URL url) {
|
||||
Map<ChannelOption, Object> channelOption = new HashMap<>();
|
||||
channelOption.put(ChannelOption.SO_KEEPALIVE, url.getParameter(Constants.KEEP_ALIVE_KEY, Constants.DEFAULT_KEEP_ALIVE));
|
||||
return channelOption;
|
||||
}
|
||||
|
||||
/**
|
||||
* create channel options map
|
||||
*
|
||||
* @param url
|
||||
* @return
|
||||
*/
|
||||
protected Map<ChannelOption, Object> getChannelOptionMap(URL url) {
|
||||
Map<ChannelOption, Object> options = new HashMap<>();
|
||||
|
||||
options.put(ChannelOption.SO_REUSEADDR, Boolean.TRUE);
|
||||
options.put(ChannelOption.TCP_NODELAY, Boolean.TRUE);
|
||||
options.put(ChannelOption.SO_BACKLOG, url.getPositiveParameter(BACKLOG_KEY, org.apache.dubbo.remoting.Constants.DEFAULT_BACKLOG));
|
||||
options.put(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
|
||||
return options;
|
||||
}
|
||||
|
||||
/**
|
||||
* create channel handler
|
||||
*
|
||||
* @param url
|
||||
* @return
|
||||
*/
|
||||
protected List<ChannelHandler> getChannelHandlers(URL url) {
|
||||
List<ChannelHandler> channelHandlers = new ArrayList<>();
|
||||
|
||||
return channelHandlers;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void deploy(ServiceRestMetadata serviceRestMetadata, Invoker invoker) {
|
||||
Map<PathMatcher, RestMethodMetadata> pathToServiceMapContainPathVariable =
|
||||
serviceRestMetadata.getPathContainPathVariableToServiceMap();
|
||||
pathAndInvokerMapper.addPathAndInvoker(pathToServiceMapContainPathVariable, invoker);
|
||||
|
||||
Map<PathMatcher, RestMethodMetadata> pathToServiceMapUnContainPathVariable =
|
||||
serviceRestMetadata.getPathUnContainPathVariableToServiceMap();
|
||||
pathAndInvokerMapper.addPathAndInvoker(pathToServiceMapUnContainPathVariable, invoker);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void undeploy(ServiceRestMetadata serviceRestMetadata) {
|
||||
Map<PathMatcher, RestMethodMetadata> pathToServiceMapContainPathVariable =
|
||||
serviceRestMetadata.getPathContainPathVariableToServiceMap();
|
||||
pathToServiceMapContainPathVariable.keySet().stream().forEach(pathAndInvokerMapper::removePath);
|
||||
|
||||
Map<PathMatcher, RestMethodMetadata> pathToServiceMapUnContainPathVariable =
|
||||
serviceRestMetadata.getPathUnContainPathVariableToServiceMap();
|
||||
pathToServiceMapUnContainPathVariable.keySet().stream().forEach(pathAndInvokerMapper::removePath);
|
||||
|
||||
}
|
||||
|
||||
private void registerExceptionMapper(URL url) {
|
||||
|
||||
for (String clazz : COMMA_SPLIT_PATTERN.split(url.getParameter(Constants.EXTENSION_KEY, RpcExceptionMapper.class.getName()))) {
|
||||
if (!StringUtils.isEmpty(clazz)) {
|
||||
exceptionMapper.registerMapper(clazz);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
@ -19,6 +19,8 @@ package org.apache.dubbo.rpc.protocol.rest;
|
|||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
|
||||
import org.apache.dubbo.metadata.rest.ServiceRestMetadata;
|
||||
import org.apache.dubbo.remoting.api.pu.DefaultPuHandler;
|
||||
import org.apache.dubbo.remoting.exchange.PortUnificationExchanger;
|
||||
import org.apache.dubbo.remoting.http.RestClient;
|
||||
import org.apache.dubbo.remoting.http.factory.RestClientFactory;
|
||||
import org.apache.dubbo.rpc.Exporter;
|
||||
|
|
@ -30,6 +32,8 @@ import org.apache.dubbo.rpc.protocol.AbstractExporter;
|
|||
import org.apache.dubbo.rpc.protocol.AbstractProtocol;
|
||||
import org.apache.dubbo.rpc.protocol.rest.annotation.consumer.HttpConnectionPreBuildIntercept;
|
||||
import org.apache.dubbo.rpc.protocol.rest.annotation.metadata.MetadataResolver;
|
||||
import org.apache.dubbo.rpc.protocol.rest.deploy.ServiceDeployer;
|
||||
import org.apache.dubbo.rpc.protocol.rest.deploy.ServiceDeployerManager;
|
||||
|
||||
|
||||
import java.util.Map;
|
||||
|
|
@ -39,17 +43,14 @@ import java.util.concurrent.ConcurrentHashMap;
|
|||
import java.util.concurrent.ConcurrentMap;
|
||||
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.INTERFACE_KEY;
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.REST_SERVICE_DEPLOYER_URL_ATTRIBUTE_KEY;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_ERROR_CLOSE_CLIENT;
|
||||
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_ERROR_CLOSE_SERVER;
|
||||
import static org.apache.dubbo.remoting.Constants.SERVER_KEY;
|
||||
import static org.apache.dubbo.rpc.protocol.rest.constans.RestConstant.PATH_SEPARATOR;
|
||||
|
||||
public class RestProtocol extends AbstractProtocol {
|
||||
|
||||
private static final int DEFAULT_PORT = 80;
|
||||
private static final String DEFAULT_SERVER = Constants.NETTY_HTTP;
|
||||
|
||||
private final RestServerFactory serverFactory = new RestServerFactory();
|
||||
|
||||
private final ConcurrentMap<String, ReferenceCountedClient<? extends RestClient>> clients = new ConcurrentHashMap<>();
|
||||
|
||||
|
|
@ -89,24 +90,22 @@ public class RestProtocol extends AbstractProtocol {
|
|||
url, getContextPath(url));
|
||||
|
||||
|
||||
// TODO add Extension filter
|
||||
// create rest server
|
||||
RestProtocolServer server = (RestProtocolServer) ConcurrentHashMapUtils.computeIfAbsent(serverMap, getAddr(url), restServer -> {
|
||||
RestProtocolServer s = serverFactory.createServer(url.getParameter(SERVER_KEY, DEFAULT_SERVER));
|
||||
s.setAddress(url.getAddress());
|
||||
s.start(url);
|
||||
return s;
|
||||
});
|
||||
// deploy service
|
||||
URL newURL = ServiceDeployerManager.deploy(url, serviceRestMetadata, invoker);
|
||||
|
||||
// create server
|
||||
PortUnificationExchanger.bind(newURL, new DefaultPuHandler());
|
||||
|
||||
ServiceDeployer serviceDeployer = (ServiceDeployer) newURL.getAttribute(REST_SERVICE_DEPLOYER_URL_ATTRIBUTE_KEY);
|
||||
|
||||
|
||||
server.deploy(serviceRestMetadata, invoker);
|
||||
|
||||
URL finalUrl = newURL;
|
||||
exporter = new AbstractExporter<T>(invoker) {
|
||||
@Override
|
||||
public void afterUnExport() {
|
||||
destroyInternal(url);
|
||||
destroyInternal(finalUrl);
|
||||
exporterMap.remove(uri);
|
||||
server.undeploy(serviceRestMetadata);
|
||||
serviceDeployer.undeploy(serviceRestMetadata);
|
||||
}
|
||||
};
|
||||
exporterMap.put(uri, exporter);
|
||||
|
|
@ -164,6 +163,10 @@ public class RestProtocol extends AbstractProtocol {
|
|||
if (logger.isInfoEnabled()) {
|
||||
logger.info("Destroying protocol [" + this.getClass().getSimpleName() + "] ...");
|
||||
}
|
||||
|
||||
PortUnificationExchanger.close();
|
||||
ServiceDeployerManager.close();
|
||||
|
||||
super.destroy();
|
||||
|
||||
for (Map.Entry<String, ProtocolServer> entry : serverMap.entrySet()) {
|
||||
|
|
|
|||
|
|
@ -1,34 +0,0 @@
|
|||
/*
|
||||
* 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.rest;
|
||||
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.metadata.rest.ServiceRestMetadata;
|
||||
import org.apache.dubbo.rpc.Invoker;
|
||||
import org.apache.dubbo.rpc.ProtocolServer;
|
||||
|
||||
|
||||
public interface RestProtocolServer extends ProtocolServer {
|
||||
|
||||
void start(URL url);
|
||||
|
||||
|
||||
void deploy(ServiceRestMetadata serviceRestMetadata, Invoker invoker);
|
||||
|
||||
void undeploy(ServiceRestMetadata serviceRestMetadata);
|
||||
|
||||
}
|
||||
|
|
@ -1,30 +0,0 @@
|
|||
/*
|
||||
* 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.rest;
|
||||
|
||||
|
||||
/**
|
||||
* Only the server that implements servlet container
|
||||
* could support something like @Context injection of servlet objects.
|
||||
*/
|
||||
public class RestServerFactory {
|
||||
|
||||
|
||||
public RestProtocolServer createServer(String name) {
|
||||
return new NettyHttpRestServer();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,77 @@
|
|||
/*
|
||||
* 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.rest.deploy;
|
||||
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.common.utils.StringUtils;
|
||||
import org.apache.dubbo.metadata.rest.PathMatcher;
|
||||
import org.apache.dubbo.metadata.rest.RestMethodMetadata;
|
||||
import org.apache.dubbo.metadata.rest.ServiceRestMetadata;
|
||||
import org.apache.dubbo.rpc.Invoker;
|
||||
import org.apache.dubbo.rpc.protocol.rest.Constants;
|
||||
import org.apache.dubbo.rpc.protocol.rest.PathAndInvokerMapper;
|
||||
import org.apache.dubbo.rpc.protocol.rest.RpcExceptionMapper;
|
||||
import org.apache.dubbo.rpc.protocol.rest.exception.mapper.ExceptionMapper;
|
||||
|
||||
import java.util.Map;
|
||||
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.COMMA_SPLIT_PATTERN;
|
||||
|
||||
public class ServiceDeployer {
|
||||
|
||||
private final PathAndInvokerMapper pathAndInvokerMapper = new PathAndInvokerMapper();
|
||||
private final ExceptionMapper exceptionMapper = new ExceptionMapper();
|
||||
|
||||
|
||||
public void deploy(ServiceRestMetadata serviceRestMetadata, Invoker invoker) {
|
||||
Map<PathMatcher, RestMethodMetadata> pathToServiceMapContainPathVariable =
|
||||
serviceRestMetadata.getPathContainPathVariableToServiceMap();
|
||||
pathAndInvokerMapper.addPathAndInvoker(pathToServiceMapContainPathVariable, invoker);
|
||||
|
||||
Map<PathMatcher, RestMethodMetadata> pathToServiceMapUnContainPathVariable =
|
||||
serviceRestMetadata.getPathUnContainPathVariableToServiceMap();
|
||||
pathAndInvokerMapper.addPathAndInvoker(pathToServiceMapUnContainPathVariable, invoker);
|
||||
}
|
||||
|
||||
public void undeploy(ServiceRestMetadata serviceRestMetadata) {
|
||||
Map<PathMatcher, RestMethodMetadata> pathToServiceMapContainPathVariable =
|
||||
serviceRestMetadata.getPathContainPathVariableToServiceMap();
|
||||
pathToServiceMapContainPathVariable.keySet().stream().forEach(pathAndInvokerMapper::removePath);
|
||||
|
||||
Map<PathMatcher, RestMethodMetadata> pathToServiceMapUnContainPathVariable =
|
||||
serviceRestMetadata.getPathUnContainPathVariableToServiceMap();
|
||||
pathToServiceMapUnContainPathVariable.keySet().stream().forEach(pathAndInvokerMapper::removePath);
|
||||
|
||||
}
|
||||
|
||||
public void registerExceptionMapper(URL url) {
|
||||
|
||||
for (String clazz : COMMA_SPLIT_PATTERN.split(url.getParameter(Constants.EXTENSION_KEY, RpcExceptionMapper.class.getName()))) {
|
||||
if (!StringUtils.isEmpty(clazz)) {
|
||||
exceptionMapper.registerMapper(clazz);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public PathAndInvokerMapper getPathAndInvokerMapper() {
|
||||
return pathAndInvokerMapper;
|
||||
}
|
||||
|
||||
public ExceptionMapper getExceptionMapper() {
|
||||
return exceptionMapper;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,80 @@
|
|||
/*
|
||||
* 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.rest.deploy;
|
||||
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.common.utils.ConcurrentHashMapUtils;
|
||||
import org.apache.dubbo.metadata.rest.ServiceRestMetadata;
|
||||
import org.apache.dubbo.rpc.Invoker;
|
||||
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.ConcurrentMap;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.REST_SERVICE_DEPLOYER_URL_ATTRIBUTE_KEY;
|
||||
import static org.apache.dubbo.remoting.Constants.PORT_UNIFICATION_NETTY4_SERVER;
|
||||
import static org.apache.dubbo.remoting.Constants.REST_SERVER;
|
||||
import static org.apache.dubbo.remoting.Constants.SERVER_KEY;
|
||||
|
||||
public class ServiceDeployerManager {
|
||||
private static final ConcurrentMap<String, ServiceDeployer> serviceDeployers = new ConcurrentHashMap<>();
|
||||
|
||||
|
||||
public static URL deploy(final URL currentURL, ServiceRestMetadata serviceRestMetadata, Invoker invoker) {
|
||||
|
||||
AtomicBoolean isNewCreate = new AtomicBoolean();
|
||||
|
||||
ServiceDeployer newServiceDeployer = ConcurrentHashMapUtils.computeIfAbsent(serviceDeployers, currentURL.getAddress(), address -> {
|
||||
ServiceDeployer serviceDeployer = new ServiceDeployer();
|
||||
isNewCreate.set(true);
|
||||
return serviceDeployer;
|
||||
|
||||
});
|
||||
|
||||
|
||||
// register service
|
||||
newServiceDeployer.deploy(serviceRestMetadata, invoker);
|
||||
|
||||
// register exception mapper
|
||||
newServiceDeployer.registerExceptionMapper(currentURL);
|
||||
|
||||
// not new URL
|
||||
if (!isNewCreate.get()) {
|
||||
return currentURL;
|
||||
}
|
||||
|
||||
|
||||
URL tmp = currentURL;
|
||||
// adapt to older rest versions
|
||||
if (REST_SERVER.contains(tmp.getParameter(SERVER_KEY))) {
|
||||
tmp = tmp.addParameter(SERVER_KEY, PORT_UNIFICATION_NETTY4_SERVER);
|
||||
}
|
||||
|
||||
// passing ServiceDeployer to PortUnificationServer through URL
|
||||
// add attribute for server build
|
||||
|
||||
tmp = tmp.putAttribute(REST_SERVICE_DEPLOYER_URL_ATTRIBUTE_KEY, newServiceDeployer);
|
||||
|
||||
|
||||
return tmp;
|
||||
}
|
||||
|
||||
|
||||
public static void close() {
|
||||
serviceDeployers.clear();
|
||||
}
|
||||
}
|
||||
|
|
@ -29,6 +29,7 @@ import org.apache.dubbo.rpc.RpcInvocation;
|
|||
import org.apache.dubbo.rpc.protocol.rest.PathAndInvokerMapper;
|
||||
import org.apache.dubbo.rpc.protocol.rest.RestRPCInvocationUtil;
|
||||
import org.apache.dubbo.rpc.protocol.rest.RestHeaderEnum;
|
||||
import org.apache.dubbo.rpc.protocol.rest.deploy.ServiceDeployer;
|
||||
import org.apache.dubbo.rpc.protocol.rest.exception.MediaTypeUnSupportException;
|
||||
import org.apache.dubbo.rpc.protocol.rest.exception.ParamParseException;
|
||||
import org.apache.dubbo.rpc.protocol.rest.exception.PathNoFoundException;
|
||||
|
|
@ -58,6 +59,11 @@ public class NettyHttpHandler implements HttpHandler<NettyRequestFacade, NettyHt
|
|||
this.exceptionMapper = exceptionMapper;
|
||||
}
|
||||
|
||||
public NettyHttpHandler(ServiceDeployer serviceDeployer) {
|
||||
this.pathAndInvokerMapper = serviceDeployer.getPathAndInvokerMapper();
|
||||
this.exceptionMapper = serviceDeployer.getExceptionMapper();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handle(NettyRequestFacade requestFacade, NettyHttpResponse nettyHttpResponse) throws IOException {
|
||||
|
||||
|
|
|
|||
|
|
@ -1,184 +0,0 @@
|
|||
/*
|
||||
* 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.rest.netty;
|
||||
|
||||
import io.netty.bootstrap.ServerBootstrap;
|
||||
import io.netty.channel.Channel;
|
||||
import io.netty.channel.ChannelHandler;
|
||||
import io.netty.channel.ChannelInitializer;
|
||||
import io.netty.channel.ChannelOption;
|
||||
import io.netty.channel.ChannelPipeline;
|
||||
import io.netty.channel.EventLoopGroup;
|
||||
import io.netty.channel.nio.NioEventLoopGroup;
|
||||
import io.netty.channel.socket.SocketChannel;
|
||||
import io.netty.channel.socket.nio.NioServerSocketChannel;
|
||||
import io.netty.handler.timeout.IdleStateHandler;
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.common.utils.NamedThreadFactory;
|
||||
import org.apache.dubbo.rpc.protocol.rest.constans.RestConstant;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.apache.dubbo.remoting.Constants.EVENT_LOOP_BOSS_POOL_NAME;
|
||||
import static org.apache.dubbo.remoting.Constants.EVENT_LOOP_WORKER_POOL_NAME;
|
||||
|
||||
|
||||
public class NettyServer {
|
||||
|
||||
protected ServerBootstrap bootstrap = new ServerBootstrap();
|
||||
protected String hostname = null;
|
||||
protected int configuredPort = 8080;
|
||||
protected int runtimePort = -1;
|
||||
|
||||
private EventLoopGroup eventLoopGroup;
|
||||
private EventLoopGroup workerLoopGroup;
|
||||
private int ioWorkerCount = Runtime.getRuntime().availableProcessors() * 2;
|
||||
|
||||
private List<ChannelHandler> channelHandlers = Collections.emptyList();
|
||||
private Map<ChannelOption, Object> channelOptions = Collections.emptyMap();
|
||||
private Map<ChannelOption, Object> childChannelOptions = Collections.emptyMap();
|
||||
private UnSharedHandlerCreator unSharedHandlerCallBack;
|
||||
|
||||
public NettyServer() {
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Specify the worker count to use. For more information about this please see the javadocs of {@link EventLoopGroup}
|
||||
*
|
||||
* @param ioWorkerCount worker count
|
||||
*/
|
||||
public void setIoWorkerCount(int ioWorkerCount) {
|
||||
this.ioWorkerCount = ioWorkerCount;
|
||||
}
|
||||
|
||||
public String getHostname() {
|
||||
return hostname;
|
||||
}
|
||||
|
||||
public void setHostname(String hostname) {
|
||||
this.hostname = hostname;
|
||||
}
|
||||
|
||||
public int getPort() {
|
||||
return runtimePort > 0 ? runtimePort : configuredPort;
|
||||
}
|
||||
|
||||
public void setPort(int port) {
|
||||
this.configuredPort = port;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Add additional {@link io.netty.channel.ChannelHandler}s to the {@link io.netty.bootstrap.ServerBootstrap}.
|
||||
* <p>The additional channel handlers are being added <em>before</em> the HTTP handling.</p>
|
||||
*
|
||||
* @param channelHandlers the additional {@link io.netty.channel.ChannelHandler}s.
|
||||
*/
|
||||
public void setChannelHandlers(final List<ChannelHandler> channelHandlers) {
|
||||
this.channelHandlers = channelHandlers == null ? Collections.<ChannelHandler>emptyList() : channelHandlers;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add Netty {@link io.netty.channel.ChannelOption}s to the {@link io.netty.bootstrap.ServerBootstrap}.
|
||||
*
|
||||
* @param channelOptions the additional {@link io.netty.channel.ChannelOption}s.
|
||||
* @see io.netty.bootstrap.ServerBootstrap#option(io.netty.channel.ChannelOption, Object)
|
||||
*/
|
||||
public void setChannelOptions(final Map<ChannelOption, Object> channelOptions) {
|
||||
this.channelOptions = channelOptions == null ? Collections.<ChannelOption, Object>emptyMap() : channelOptions;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add child options to the {@link io.netty.bootstrap.ServerBootstrap}.
|
||||
*
|
||||
* @param channelOptions the additional child {@link io.netty.channel.ChannelOption}s.
|
||||
* @see io.netty.bootstrap.ServerBootstrap#childOption(io.netty.channel.ChannelOption, Object)
|
||||
*/
|
||||
public void setChildChannelOptions(final Map<ChannelOption, Object> channelOptions) {
|
||||
this.childChannelOptions = channelOptions == null ? Collections.<ChannelOption, Object>emptyMap() : channelOptions;
|
||||
}
|
||||
|
||||
public void setUnSharedHandlerCallBack(UnSharedHandlerCreator unSharedHandlerCallBack) {
|
||||
this.unSharedHandlerCallBack = unSharedHandlerCallBack;
|
||||
}
|
||||
|
||||
public void start(URL url) {
|
||||
eventLoopGroup = new NioEventLoopGroup(1, new NamedThreadFactory(EVENT_LOOP_BOSS_POOL_NAME));
|
||||
workerLoopGroup = new NioEventLoopGroup(ioWorkerCount, new NamedThreadFactory(EVENT_LOOP_WORKER_POOL_NAME));
|
||||
|
||||
// Configure the server.
|
||||
bootstrap.group(eventLoopGroup, workerLoopGroup)
|
||||
.channel(NioServerSocketChannel.class)
|
||||
.childHandler(setupHandlers(url));
|
||||
|
||||
|
||||
for (Map.Entry<ChannelOption, Object> entry : channelOptions.entrySet()) {
|
||||
bootstrap.option(entry.getKey(), entry.getValue());
|
||||
}
|
||||
|
||||
for (Map.Entry<ChannelOption, Object> entry : childChannelOptions.entrySet()) {
|
||||
bootstrap.childOption(entry.getKey(), entry.getValue());
|
||||
}
|
||||
|
||||
final InetSocketAddress socketAddress;
|
||||
if (null == getHostname() || getHostname().isEmpty()) {
|
||||
socketAddress = new InetSocketAddress(configuredPort);
|
||||
} else {
|
||||
socketAddress = new InetSocketAddress(hostname, configuredPort);
|
||||
}
|
||||
|
||||
Channel channel = bootstrap.bind(socketAddress).syncUninterruptibly().channel();
|
||||
runtimePort = ((InetSocketAddress) channel.localAddress()).getPort();
|
||||
}
|
||||
|
||||
|
||||
protected ChannelHandler setupHandlers(URL url) {
|
||||
|
||||
return new ChannelInitializer<SocketChannel>() {
|
||||
@Override
|
||||
public void initChannel(SocketChannel ch) throws Exception {
|
||||
ChannelPipeline channelPipeline = ch.pipeline();
|
||||
|
||||
int idleTimeout = url.getParameter(RestConstant.IDLE_TIMEOUT_PARAM, RestConstant.IDLE_TIMEOUT);
|
||||
if (idleTimeout > 0) {
|
||||
channelPipeline.addLast(new IdleStateHandler(0, 0, idleTimeout));
|
||||
}
|
||||
|
||||
channelPipeline.addLast(channelHandlers.toArray(new ChannelHandler[channelHandlers.size()]));
|
||||
|
||||
List<ChannelHandler> unSharedHandlers = unSharedHandlerCallBack.getUnSharedHandlers(url);
|
||||
|
||||
for (ChannelHandler unSharedHandler : unSharedHandlers) {
|
||||
channelPipeline.addLast(unSharedHandler);
|
||||
}
|
||||
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
|
||||
public void stop() {
|
||||
runtimePort = -1;
|
||||
eventLoopGroup.shutdownGracefully();
|
||||
}
|
||||
}
|
||||
|
|
@ -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.rest.pu;
|
||||
|
||||
import org.apache.dubbo.remoting.api.AbstractHttpProtocolDetector;
|
||||
import org.apache.dubbo.remoting.buffer.ChannelBuffer;
|
||||
import org.apache.dubbo.rpc.model.FrameworkModel;
|
||||
|
||||
/**
|
||||
* rest http protocol detector
|
||||
*/
|
||||
|
||||
public class RestHttp1Detector extends AbstractHttpProtocolDetector {
|
||||
private static final char[][] HTTP_METHODS_PREFIX = getHttpMethodsPrefix();
|
||||
|
||||
private FrameworkModel frameworkModel;
|
||||
|
||||
public RestHttp1Detector(FrameworkModel frameworkModel) {
|
||||
this.frameworkModel = frameworkModel;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Result detect(ChannelBuffer in) {
|
||||
|
||||
int i = in.readableBytes();
|
||||
|
||||
// length judge
|
||||
if (i < SIMPLE_HTTP.length()) {
|
||||
return Result.unrecognized();
|
||||
}
|
||||
|
||||
if (prefixMatch(HTTP_METHODS_PREFIX, in, 3)) {
|
||||
return Result.recognized();
|
||||
}
|
||||
|
||||
return Result.unrecognized();
|
||||
}
|
||||
}
|
||||
|
|
@ -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.rest.pu;
|
||||
|
||||
import io.netty.handler.codec.http.HttpObjectAggregator;
|
||||
import io.netty.handler.codec.http.HttpRequestDecoder;
|
||||
import io.netty.handler.codec.http.HttpResponseEncoder;
|
||||
import org.apache.dubbo.common.URL;
|
||||
import org.apache.dubbo.common.extension.Activate;
|
||||
import org.apache.dubbo.remoting.ChannelHandler;
|
||||
import org.apache.dubbo.remoting.api.AbstractWireProtocol;
|
||||
import org.apache.dubbo.remoting.api.pu.ChannelHandlerPretender;
|
||||
import org.apache.dubbo.remoting.api.pu.ChannelOperator;
|
||||
import org.apache.dubbo.rpc.model.FrameworkModel;
|
||||
import org.apache.dubbo.rpc.model.ScopeModelAware;
|
||||
import org.apache.dubbo.rpc.protocol.rest.constans.RestConstant;
|
||||
import org.apache.dubbo.rpc.protocol.rest.deploy.ServiceDeployer;
|
||||
import org.apache.dubbo.rpc.protocol.rest.handler.NettyHttpHandler;
|
||||
import org.apache.dubbo.rpc.protocol.rest.netty.RestHttpRequestDecoder;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import static org.apache.dubbo.common.constants.CommonConstants.REST_SERVICE_DEPLOYER_URL_ATTRIBUTE_KEY;
|
||||
|
||||
@Activate(order = Integer.MAX_VALUE)
|
||||
public class RestHttp1WireProtocol extends AbstractWireProtocol implements ScopeModelAware {
|
||||
|
||||
private static final ServiceDeployer emptyServiceDeployer = new ServiceDeployer();
|
||||
|
||||
public RestHttp1WireProtocol(FrameworkModel frameworkModel) {
|
||||
super(new RestHttp1Detector(frameworkModel));
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void configServerProtocolHandler(URL url, ChannelOperator operator) {
|
||||
|
||||
|
||||
// h1 upgrade to don`t response 101 ,cause there is no h1 stream handler
|
||||
// TODO add h1 stream handler
|
||||
|
||||
// pathAndInvokerMapper, exceptionMapper getFrom url
|
||||
ServiceDeployer serviceDeployer = (ServiceDeployer) url.getAttribute(REST_SERVICE_DEPLOYER_URL_ATTRIBUTE_KEY);
|
||||
|
||||
// maybe current request is qos http or no rest service export
|
||||
if (serviceDeployer == null) {
|
||||
serviceDeployer = emptyServiceDeployer;
|
||||
}
|
||||
|
||||
List<Object> channelHandlers = Arrays.asList(new HttpRequestDecoder(
|
||||
url.getParameter(RestConstant.MAX_INITIAL_LINE_LENGTH_PARAM, RestConstant.MAX_INITIAL_LINE_LENGTH),
|
||||
url.getParameter(RestConstant.MAX_HEADER_SIZE_PARAM, RestConstant.MAX_HEADER_SIZE),
|
||||
url.getParameter(RestConstant.MAX_CHUNK_SIZE_PARAM, RestConstant.MAX_CHUNK_SIZE)),
|
||||
new HttpObjectAggregator(url.getParameter(RestConstant.MAX_REQUEST_SIZE_PARAM, RestConstant.MAX_REQUEST_SIZE)),
|
||||
new HttpResponseEncoder(), new RestHttpRequestDecoder(new NettyHttpHandler(serviceDeployer), url));
|
||||
|
||||
|
||||
List<ChannelHandler> handlers = new ArrayList<>();
|
||||
handlers.add(new ChannelHandlerPretender(channelHandlers));
|
||||
operator.configChannelHandler(handlers);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1 @@
|
|||
rest=org.apache.dubbo.rpc.protocol.rest.pu.RestHttp1WireProtocol
|
||||
|
|
@ -74,6 +74,7 @@ class JaxrsRestProtocolTest {
|
|||
private final URL exportUrl = URL.valueOf("rest://127.0.0.1:" + availablePort + "/rest?interface=org.apache.dubbo.rpc.protocol.rest.DemoService");
|
||||
private final ModuleServiceRepository repository = ApplicationModel.defaultModel().getDefaultModule().getServiceRepository();
|
||||
private final ExceptionMapper exceptionMapper = new ExceptionMapper();
|
||||
private static final String SERVER = "netty4";
|
||||
|
||||
@AfterEach
|
||||
public void tearDown() {
|
||||
|
|
@ -197,7 +198,7 @@ class JaxrsRestProtocolTest {
|
|||
|
||||
URL url = this.registerProvider(exportUrl, server, DemoService.class);
|
||||
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, "netty");
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, SERVER);
|
||||
Exporter<DemoService> exporter = protocol.export(proxy.getInvoker(new DemoServiceImpl(), DemoService.class, nettyUrl));
|
||||
|
||||
DemoService demoService = this.proxy.getProxy(protocol.refer(DemoService.class, nettyUrl));
|
||||
|
|
@ -230,7 +231,7 @@ class JaxrsRestProtocolTest {
|
|||
|
||||
URL url = this.registerProvider(exportUrl, server, DemoService.class);
|
||||
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, "netty");
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, SERVER);
|
||||
Exporter<DemoService> exporter = protocol.export(proxy.getInvoker(server, DemoService.class, nettyUrl));
|
||||
|
||||
DemoService demoService = this.proxy.getProxy(protocol.refer(DemoService.class, nettyUrl));
|
||||
|
|
@ -259,7 +260,7 @@ class JaxrsRestProtocolTest {
|
|||
|
||||
URL url = this.registerProvider(exportUrl, server, DemoService.class);
|
||||
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, "netty")
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, SERVER)
|
||||
.addParameter(EXTENSION_KEY, "org.apache.dubbo.rpc.protocol.rest.support.LoggingFilter");
|
||||
Exporter<DemoService> exporter = protocol.export(proxy.getInvoker(server, DemoService.class, nettyUrl));
|
||||
|
||||
|
|
@ -279,7 +280,7 @@ class JaxrsRestProtocolTest {
|
|||
URL url = this.registerProvider(exportUrl, server, DemoService.class);
|
||||
|
||||
// use RpcContextFilter
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, "netty")
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, SERVER)
|
||||
.addParameter(EXTENSION_KEY, "org.apache.dubbo.rpc.protocol.rest.RpcContextFilter");
|
||||
Exporter<DemoService> exporter = protocol.export(proxy.getInvoker(server, DemoService.class, nettyUrl));
|
||||
|
||||
|
|
@ -402,7 +403,7 @@ class JaxrsRestProtocolTest {
|
|||
|
||||
URL url = this.registerProvider(exportUrl, server, DemoService.class);
|
||||
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, "netty")
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, SERVER)
|
||||
.addParameter(EXTENSION_KEY, "org.apache.dubbo.rpc.protocol.rest.support.LoggingFilter");
|
||||
Exporter<DemoService> exporter = protocol.export(proxy.getInvoker(server, DemoService.class, nettyUrl));
|
||||
|
||||
|
|
@ -453,7 +454,7 @@ class JaxrsRestProtocolTest {
|
|||
|
||||
URL url = this.registerProvider(exportUrl, server, DemoService.class);
|
||||
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, "netty")
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, SERVER)
|
||||
.addParameter(EXTENSION_KEY, "org.apache.dubbo.rpc.protocol.rest.support.LoggingFilter");
|
||||
Exporter<DemoService> exporter = protocol.export(proxy.getInvoker(server, DemoService.class, nettyUrl));
|
||||
|
||||
|
|
@ -485,7 +486,7 @@ class JaxrsRestProtocolTest {
|
|||
|
||||
URL url = this.registerProvider(exportUrl, server, DemoService.class);
|
||||
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, "netty")
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, SERVER)
|
||||
.addParameter(EXTENSION_KEY, "org.apache.dubbo.rpc.protocol.rest.support.LoggingFilter");
|
||||
Exporter<DemoService> exporter = protocol.export(proxy.getInvoker(server, DemoService.class, nettyUrl));
|
||||
|
||||
|
|
|
|||
|
|
@ -39,6 +39,7 @@ public class ResteasyResponseTest {
|
|||
private final int availablePort = NetUtils.getAvailablePort();
|
||||
private final URL exportUrl = URL.valueOf("rest://127.0.0.1:" + availablePort + "/rest?interface=org.apache.dubbo.rpc.protocol.rest.rest.RestDemoService");
|
||||
private final ModuleServiceRepository repository = ApplicationModel.defaultModel().getDefaultModule().getServiceRepository();
|
||||
private static final String SERVER = "netty4";
|
||||
|
||||
@AfterEach
|
||||
public void tearDown() {
|
||||
|
|
@ -50,7 +51,7 @@ public class ResteasyResponseTest {
|
|||
RestDemoService server = new RestDemoServiceImpl();
|
||||
URL url = this.registerProvider(exportUrl, server, RestDemoService.class);
|
||||
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, "netty").addParameter("timeout", 3000000);
|
||||
URL nettyUrl = url.addParameter(SERVER_KEY, SERVER).addParameter("timeout", 3000000);
|
||||
|
||||
protocol.export(proxy.getInvoker(new RestDemoServiceImpl(), RestDemoService.class, nettyUrl));
|
||||
|
||||
|
|
|
|||
|
|
@ -59,9 +59,12 @@ import static org.hamcrest.MatcherAssert.assertThat;
|
|||
public class SpringMvcRestProtocolTest {
|
||||
private Protocol protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getExtension("rest");
|
||||
private ProxyFactory proxy = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
|
||||
private final static int availablePort = NetUtils.getAvailablePort();
|
||||
private final static URL exportUrl = URL.valueOf("rest://127.0.0.1:" + availablePort + "/rest?interface=org.apache.dubbo.rpc.protocol.rest.mvc.SpringRestDemoService");
|
||||
|
||||
private static URL getUrl() {
|
||||
return URL.valueOf("rest://127.0.0.1:" + NetUtils.getAvailablePort() + "/rest?interface=org.apache.dubbo.rpc.protocol.rest.mvc.SpringRestDemoService");
|
||||
}
|
||||
|
||||
private static final String SERVER = "netty4";
|
||||
|
||||
private final ModuleServiceRepository repository = ApplicationModel.defaultModel().getDefaultModule().getServiceRepository();
|
||||
|
||||
|
|
@ -83,12 +86,12 @@ public class SpringMvcRestProtocolTest {
|
|||
}
|
||||
|
||||
public Exporter<SpringRestDemoService> getExport(URL url, SpringRestDemoService server) {
|
||||
url = url.addParameter(SERVER_KEY, Constants.NETTY_HTTP);
|
||||
url = url.addParameter(SERVER_KEY, SERVER);
|
||||
return protocol.export(proxy.getInvoker(server, getServerClass(), url));
|
||||
}
|
||||
|
||||
public Exporter<SpringRestDemoService> getExceptionHandlerExport(URL url, SpringRestDemoService server) {
|
||||
url = url.addParameter(SERVER_KEY, Constants.NETTY_HTTP);
|
||||
url = url.addParameter(SERVER_KEY, SERVER);
|
||||
url = url.addParameter(EXTENSION_KEY, TestExceptionMapper.class.getName());
|
||||
return protocol.export(proxy.getInvoker(server, getServerClass(), url));
|
||||
}
|
||||
|
|
@ -177,7 +180,7 @@ public class SpringMvcRestProtocolTest {
|
|||
void testExport() {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL url = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL url = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
RpcContext.getClientAttachment().setAttachment("timeout", "200");
|
||||
Exporter<SpringRestDemoService> exporter = getExport(url, server);
|
||||
|
|
@ -194,7 +197,7 @@ public class SpringMvcRestProtocolTest {
|
|||
void testNettyServer() {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL nettyUrl = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL nettyUrl = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
|
||||
Exporter<SpringRestDemoService> exporter = getExport(nettyUrl, server);
|
||||
|
|
@ -213,7 +216,7 @@ public class SpringMvcRestProtocolTest {
|
|||
Assertions.assertThrows(RpcException.class, () -> {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL url = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL url = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
URL servletUrl = url.addParameter(SERVER_KEY, "servlet");
|
||||
|
||||
|
|
@ -227,7 +230,7 @@ public class SpringMvcRestProtocolTest {
|
|||
exceptionMapper.unRegisterMapper(RuntimeException.class);
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL nettyUrl = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL nettyUrl = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
Exporter<SpringRestDemoService> exporter = getExport(nettyUrl, server);
|
||||
|
||||
|
|
@ -241,7 +244,7 @@ public class SpringMvcRestProtocolTest {
|
|||
void testInvoke() {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL url = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL url = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
Exporter<SpringRestDemoService> exporter = getExport(url, server);
|
||||
|
||||
|
|
@ -255,7 +258,7 @@ public class SpringMvcRestProtocolTest {
|
|||
void testFilter() {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL url = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL url = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
Exporter<SpringRestDemoService> exporter = getExport(url, server);
|
||||
|
||||
|
|
@ -272,7 +275,7 @@ public class SpringMvcRestProtocolTest {
|
|||
void testRpcContextFilter() {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL nettyUrl = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL nettyUrl = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
// use RpcContextFilter
|
||||
// URL nettyUrl = url.addParameter(SERVER_KEY, "netty")
|
||||
|
|
@ -307,7 +310,7 @@ public class SpringMvcRestProtocolTest {
|
|||
Assertions.assertThrows(RuntimeException.class, () -> {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL url = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL url = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
URL nettyUrl = url.addParameter(EXTENSION_KEY, "com.not.existing.Filter");
|
||||
Exporter<SpringRestDemoService> exporter = getExport(nettyUrl, server);
|
||||
|
|
@ -325,7 +328,7 @@ public class SpringMvcRestProtocolTest {
|
|||
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL exceptionUrl = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL exceptionUrl = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
Exporter<SpringRestDemoService> exporter = getExceptionHandlerExport(exceptionUrl, server);
|
||||
|
||||
|
|
@ -342,7 +345,7 @@ public class SpringMvcRestProtocolTest {
|
|||
void testFormConsumerParser() {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL nettyUrl = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL nettyUrl = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
|
||||
Exporter<SpringRestDemoService> exporter = getExport(nettyUrl, server);
|
||||
|
|
@ -368,7 +371,7 @@ public class SpringMvcRestProtocolTest {
|
|||
void testPrimitive() {
|
||||
SpringRestDemoService server = getServerImpl();
|
||||
|
||||
URL nettyUrl = this.registerProvider(exportUrl, server, SpringRestDemoService.class);
|
||||
URL nettyUrl = this.registerProvider(getUrl(), server, SpringRestDemoService.class);
|
||||
|
||||
|
||||
Exporter<SpringRestDemoService> exporter = getExport(nettyUrl, server);
|
||||
|
|
|
|||
Loading…
Reference in New Issue