unification HTTP server side

dubbo-distribution/dubbo-all/pom.xml

@@ -482,6 +482,13 @@
             <scope>compile</scope>
             <optional>true</optional>
         </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-remoting-http12</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+            <optional>true</optional>
+        </dependency>
         <dependency>
             <groupId>org.apache.dubbo</groupId>
             <artifactId>dubbo-remoting-netty</artifactId>
@@ -720,6 +727,7 @@
                                     <include>org.apache.dubbo:dubbo-registry-zookeeper</include>
                                     <include>org.apache.dubbo:dubbo-remoting-api</include>
                                     <include>org.apache.dubbo:dubbo-remoting-http</include>
+                                    <include>org.apache.dubbo:dubbo-remoting-http12</include>
                                     <include>org.apache.dubbo:dubbo-remoting-netty4</include>
                                     <include>org.apache.dubbo:dubbo-remoting-netty</include>
                                     <include>org.apache.dubbo:dubbo-remoting-zookeeper-api</include>
@@ -1239,6 +1247,18 @@
                                         META-INF/dubbo/internal/org.apache.dubbo.remoting.http.HttpBinder
                                     </resource>
                                 </transformer>
+                                <transformer
+                                    implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+                                    <resource>
+                                        META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec
+                                    </resource>
+                                </transformer>
+                                <transformer
+                                    implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
+                                    <resource>
+                                        META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory
+                                    </resource>
+                                </transformer>
                                 <transformer
                                     implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer">
                                     <resource>

dubbo-remoting/dubbo-remoting-http12/pom.xml

@@ -0,0 +1,61 @@
+<!--
+  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">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.dubbo</groupId>
+        <artifactId>dubbo-remoting</artifactId>
+        <version>${revision}</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+    <artifactId>dubbo-remoting-http12</artifactId>
+    <packaging>jar</packaging>
+    <name>${project.artifactId}</name>
+    <description>The http1/2 remoting module of dubbo project</description>
+    <properties>
+        <skip_maven_deploy>false</skip_maven_deploy>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-rpc-api</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-common</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo-remoting-api</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-transport</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-codec-http2</artifactId>
+        </dependency>
+
+
+    </dependencies>
+</project>

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerCallListener.java

@@ -0,0 +1,89 @@
+/*
+ * 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.http12;
+
+import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
+import org.apache.dubbo.common.logger.LoggerFactory;
+import org.apache.dubbo.common.stream.StreamObserver;
+import org.apache.dubbo.remoting.http12.exception.HttpRequestTimeout;
+import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcContext;
+import org.apache.dubbo.rpc.RpcInvocation;
+
+import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_TIMEOUT_SERVER;
+
+public abstract class AbstractServerCallListener implements ServerCallListener {
+
+    private static final ErrorTypeAwareLogger LOGGER = LoggerFactory.getErrorTypeAwareLogger(AbstractServerCallListener.class);
+
+    protected final RpcInvocation invocation;
+
+    protected final Invoker<?> invoker;
+
+    protected final StreamObserver<Object> responseObserver;
+
+    public AbstractServerCallListener(RpcInvocation invocation,
+                                      Invoker<?> invoker,
+                                      StreamObserver<Object> responseObserver) {
+        this.invocation = invocation;
+        this.invoker = invoker;
+        this.responseObserver = responseObserver;
+    }
+
+    public void invoke() {
+        try {
+            final long stInMillis = System.currentTimeMillis();
+            final Result response = invoker.invoke(invocation);
+            response.whenCompleteWithContext((r, t) -> {
+                if (t != null) {
+                    responseObserver.onError(t);
+                    return;
+                }
+                if (response.hasException()) {
+                    doOnResponseHasException(response.getException());
+                    return;
+                }
+                final long cost = System.currentTimeMillis() - stInMillis;
+                Long timeout = (Long) invocation.get("timeout");
+                if (timeout != null && timeout < cost) {
+                    LOGGER.error(PROTOCOL_TIMEOUT_SERVER, "", "", String.format(
+                        "Invoke timeout at server side, ignored to send response. service=%s method=%s cost=%s",
+                        invocation.getTargetServiceUniqueName(),
+                        invocation.getMethodName(),
+                        cost));
+                    HttpRequestTimeout serverSideTimeout = HttpRequestTimeout.serverSide();
+                    responseObserver.onError(serverSideTimeout);
+                    return;
+                }
+                onReturn(r.getValue());
+            });
+        } catch (Exception e) {
+            responseObserver.onError(e);
+        } finally {
+            RpcContext.removeCancellationContext();
+            RpcContext.removeContext();
+        }
+    }
+
+    protected void doOnResponseHasException(Throwable t) {
+        responseObserver.onError(t);
+    }
+
+    public abstract void onReturn(Object value);
+
+}

dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java

@@ -0,0 +1,151 @@
+/*
+ * 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.http12;
+
+import org.apache.dubbo.remoting.http12.exception.HttpStatusException;
+import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
+
+public abstract class AbstractServerHttpChannelObserver implements CustomizableHttpChannelObserver<Object> {
+
+    private HeadersCustomizer headersCustomizer = HeadersCustomizer.NO_OP;
+
+    private TrailersCustomizer trailersCustomizer = TrailersCustomizer.NO_OP;
+
+    private ErrorResponseCustomizer errorResponseCustomizer = ErrorResponseCustomizer.NO_OP;
+
+    private final HttpChannel httpChannel;
+
+    private boolean headerSent;
+
+    private HttpMessageCodec httpMessageCodec;
+
+    public AbstractServerHttpChannelObserver(HttpChannel httpChannel) {
+        this.httpChannel = httpChannel;
+    }
+
+    public void setHttpMessageCodec(HttpMessageCodec httpMessageCodec) {
+        this.httpMessageCodec = httpMessageCodec;
+    }
+
+    @Override
+    public void setHeadersCustomizer(HeadersCustomizer headersCustomizer) {
+        this.headersCustomizer = headersCustomizer;
+    }
+
+    @Override
+    public void setTrailersCustomizer(TrailersCustomizer trailersCustomizer) {
+        this.trailersCustomizer = trailersCustomizer;
+    }
+
+    @Override
+    public void setErrorResponseCustomizer(ErrorResponseCustomizer errorResponseCustomizer) {
+        this.errorResponseCustomizer = errorResponseCustomizer;
+    }
+
+    protected HeadersCustomizer getHeadersCustomizer() {
+        return headersCustomizer;
+    }
+
+    protected TrailersCustomizer getTrailersCustomizer() {
+        return trailersCustomizer;
+    }
+
+    @Override
+    public void onNext(Object data) {
+        try {
+            if (!headerSent) {
+                doSendHeaders(HttpStatus.OK.getStatusString());
+            }
+            HttpOutputMessage outputMessage = encodeHttpOutputMessage(data);
+            preOutputMessage(outputMessage);
+            this.httpMessageCodec.encode(outputMessage.getBody(), data);
+            getHttpChannel().writeMessage(outputMessage);
+            postOutputMessage(outputMessage);
+        } catch (Throwable e) {
+            onError(e);
+        }
+    }
+
+    protected void preOutputMessage(HttpOutputMessage outputMessage) throws Throwable {
+
+    }
+
+    protected void postOutputMessage(HttpOutputMessage outputMessage) throws Throwable {
+
+    }
+
+    protected abstract HttpMetadata encodeHttpMetadata();
+
+    protected HttpOutputMessage encodeHttpOutputMessage(Object data) {
+        return getHttpChannel().newOutputMessage();
+    }
+
+    protected HttpMetadata encodeTrailers(Throwable throwable) {
+        return null;
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        int httpStatusCode = HttpStatus.INTERNAL_SERVER_ERROR.getCode();
+        if (throwable instanceof HttpStatusException) {
+            httpStatusCode = ((HttpStatusException) throwable).getStatusCode();
+        }
+        if (!headerSent) {
+            doSendHeaders(String.valueOf(httpStatusCode));
+        }
+        try {
+            ErrorResponse errorResponse = new ErrorResponse();
+            errorResponse.setStatus(String.valueOf(httpStatusCode));
+            errorResponse.setMessage(throwable.getMessage());
+            this.errorResponseCustomizer.accept(errorResponse, throwable);
+            HttpOutputMessage httpOutputMessage = encodeHttpOutputMessage(errorResponse);
+            this.httpMessageCodec.encode(httpOutputMessage.getBody(), errorResponse);
+            getHttpChannel().writeMessage(httpOutputMessage);
+        } finally {
+            doOnCompleted(throwable);
+        }
+    }
+
+    @Override
+    public void onCompleted() {
+        doOnCompleted(null);
+    }
+
+    @Override
+    public HttpChannel getHttpChannel() {
+        return httpChannel;
+    }
+
+    private void doSendHeaders(String statusCode) {
+        HttpMetadata httpMetadata = encodeHttpMetadata();
+        httpMetadata.headers().set(HttpHeaderNames.STATUS.getName(), statusCode);
+        httpMetadata.headers().set(HttpHeaderNames.CONTENT_TYPE.getName(), httpMessageCodec.contentType().getName());
+        this.headersCustomizer.accept(httpMetadata.headers());
+        getHttpChannel().writeHeader(httpMetadata);
+        this.headerSent = true;
+    }
+
+    protected void doOnCompleted(Throwable throwable) {
+        HttpMetadata httpMetadata = encodeTrailers(throwable);
+        if (httpMetadata == null) {
+            return;
+        }
+        this.trailersCustomizer.accept(httpMetadata.headers(), throwable);
+        getHttpChannel().writeHeader(httpMetadata);
+    }
+
+}