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

unification HTTP server side (#12815)

This commit is contained in:
icodening 2023-08-09 14:00:25 +08:00 committed by GitHub
parent 059fdf2613
commit 8bfdf1e38c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
87 changed files with 4590 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
dubbo-distribution/dubbo-all/pom.xml
View File

@ -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>

61
dubbo-remoting/dubbo-remoting-http12/pom.xml Normal file
View File

@ -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>

89
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerCallListener.java Normal file
View File

@ -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);
}

151
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/AbstractServerHttpChannelObserver.java Normal file
View File

@ -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);
}
}

63
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/BiStreamServerCallListener.java Normal file
View File

@ -0,0 +1,63 @@
/*
* 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.stream.StreamObserver;
import org.apache.dubbo.remoting.http12.exception.HttpStatusException;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
public class BiStreamServerCallListener extends AbstractServerCallListener {
private StreamObserver<Object> requestObserver;
public BiStreamServerCallListener(RpcInvocation invocation,
Invoker<?> invoker,
FlowControlStreamObserver<Object> responseObserver) {
super(invocation, invoker, responseObserver);
invocation.setArguments(new Object[]{responseObserver});
invoke();
}
@Override
@SuppressWarnings("all")
public void onReturn(Object value) {
this.requestObserver = (StreamObserver<Object>) value;
}
@Override
public void onMessage(Object message) {
if (message instanceof Object[]) {
message = ((Object[]) message)[0];
}
requestObserver.onNext(message);
if (((FlowControlStreamObserver<Object>) responseObserver).isAutoRequestN()) {
((FlowControlStreamObserver<Object>) responseObserver).request(1);
}
}
@Override
public void onCancel(long code) {
requestObserver.onError(new HttpStatusException((int) code));
}
@Override
public void onComplete() {
requestObserver.onCompleted();
}
}

88
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/CompositeInputStream.java Normal file
View File

@ -0,0 +1,88 @@
/*
* 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.DecodeException;
import java.io.IOException;
import java.io.InputStream;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
public class CompositeInputStream extends InputStream {
private final Queue<InputStream> inputStreams = new ConcurrentLinkedQueue<>();
private int totalAvailable = 0;
private int readIndex = 0;
public void addInputStream(InputStream inputStream) {
this.inputStreams.offer(inputStream);
try {
this.totalAvailable += inputStream.available();
} catch (IOException e) {
throw new DecodeException(e);
}
}
@Override
public int read() throws IOException {
InputStream inputStream;
while ((inputStream = inputStreams.peek()) != null) {
int available = inputStream.available();
if (available == 0) {
releaseHeadStream();
continue;
}
int read = inputStream.read();
if (read != -1) {
++readIndex;
releaseIfNecessary(inputStream);
return read;
}
releaseHeadStream();
}
return -1;
}
@Override
public int available() {
return totalAvailable - readIndex;
}
@Override
public void close() throws IOException {
InputStream inputStream;
while ((inputStream = inputStreams.poll()) != null) {
inputStream.close();
}
}
private void releaseHeadStream() throws IOException {
InputStream removeStream = inputStreams.remove();
removeStream.close();
}
private void releaseIfNecessary(InputStream inputStream) throws IOException {
int available = inputStream.available();
if (available == 0) {
inputStream.close();
releaseHeadStream();
}
}
}

27
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/CustomizableHttpChannelObserver.java Normal file
View File

@ -0,0 +1,27 @@
/*
* 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;
public interface CustomizableHttpChannelObserver<T> extends HttpChannelObserver<T> {
void setHeadersCustomizer(HeadersCustomizer headersCustomizer);
void setTrailersCustomizer(TrailersCustomizer trailersCustomizer);
void setErrorResponseCustomizer(ErrorResponseCustomizer errorResponseCustomizer);
}

23
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ErrorCodeHolder.java Normal file
View File

@ -0,0 +1,23 @@
/*
* 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;
public interface ErrorCodeHolder {
long getErrorCode();
}

42
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ErrorResponse.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.remoting.http12;
import java.io.Serializable;
public class ErrorResponse implements Serializable {
private String status;
private String message;
public String getStatus() {
return status;
}
public void setStatus(String status) {
this.status = status;
}
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}

26
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ErrorResponseCustomizer.java Normal file
View File

@ -0,0 +1,26 @@
/*
* 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 java.util.function.BiConsumer;
@FunctionalInterface
public interface ErrorResponseCustomizer extends BiConsumer<ErrorResponse, Throwable> {
ErrorResponseCustomizer NO_OP = (errorResponse, throwable) -> {
};
}

43
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/FlowControlStreamObserver.java Normal file
View File

@ -0,0 +1,43 @@
/*
* 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.stream.StreamObserver;
public interface FlowControlStreamObserver<T> extends StreamObserver<T> {
/**
* Requests the peer to produce {@code count} more messages to be delivered to the 'inbound'
* {@link StreamObserver}.
*
* <p>This method is safe to call from multiple threads without external synchronization.
*
* @param count more messages
*/
void request(int count);
boolean isAutoRequestN();
/**
* Swaps to manual flow control where no message will be delivered to {@link
* StreamObserver#onNext(Object)} unless it is {@link #request request()}ed. Since {@code
* request()} may not be called before the call is started, a number of initial requests may be
* specified.
*/
void disableAutoFlowControl();
}

26
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HeadersCustomizer.java Normal file
View File

@ -0,0 +1,26 @@
/*
* 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 java.util.function.Consumer;
@FunctionalInterface
public interface HeadersCustomizer extends Consumer<HttpHeaders> {
HeadersCustomizer NO_OP = headers -> {
};
}

34
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpChannel.java Normal file
View File

@ -0,0 +1,34 @@
/*
* 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 java.net.SocketAddress;
import java.util.concurrent.CompletableFuture;
public interface HttpChannel {
CompletableFuture<Void> writeHeader(HttpMetadata httpMetadata);
CompletableFuture<Void> writeMessage(HttpOutputMessage httpOutputMessage);
HttpOutputMessage newOutputMessage();
SocketAddress remoteAddress();
void flush();
}

22
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpChannelHolder.java Normal file
View File

@ -0,0 +1,22 @@
/*
* 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;
public interface HttpChannelHolder {
HttpChannel getHttpChannel();
}

25
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpChannelObserver.java Normal file
View File

@ -0,0 +1,25 @@
/*
* 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.stream.StreamObserver;
public interface HttpChannelObserver<T> extends StreamObserver<T> {
HttpChannel getHttpChannel();
}

40
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpHeaderNames.java Normal file
View File

@ -0,0 +1,40 @@
/*
* 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;
public enum HttpHeaderNames {
STATUS(":status"),
CONTENT_TYPE("content-type"),
CONTENT_LENGTH("content-length"),
TRANSFER_ENCODING("transfer-encoding"),
TE("te");
private final String name;
HttpHeaderNames(String name) {
this.name = name;
}
public String getName() {
return name;
}
}

449
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpHeaders.java Normal file
View File

@ -0,0 +1,449 @@
/*
* 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 java.io.Serializable;
import java.util.AbstractCollection;
import java.util.AbstractSet;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.Spliterator;
import java.util.function.Consumer;
import java.util.function.Function;
public class HttpHeaders implements Map<String, List<String>>, Serializable, Cloneable {
private final LinkedHashMap<String, List<String>> targetMap;
private final HashMap<String, String> caseInsensitiveKeys;
private final Locale locale;
private transient volatile Set<String> keySet;
private transient volatile Collection<List<String>> values;
private transient volatile Set<Entry<String, List<String>>> entrySet;
public HttpHeaders() {
this.targetMap = new LinkedHashMap<>();
this.caseInsensitiveKeys = new HashMap<>();
this.locale = Locale.US;
}
// Implementation of java.util.Map
@Override
public int size() {
return this.targetMap.size();
}
@Override
public boolean isEmpty() {
return this.targetMap.isEmpty();
}
@Override
public boolean containsKey(Object key) {
return (key instanceof String && this.caseInsensitiveKeys.containsKey(convertKey((String) key)));
}
@Override
public boolean containsValue(Object value) {
return this.targetMap.containsValue(value);
}
@Override
public List<String> get(Object key) {
if (key instanceof String) {
String caseInsensitiveKey = this.caseInsensitiveKeys.get(convertKey((String) key));
if (caseInsensitiveKey != null) {
return this.targetMap.get(caseInsensitiveKey);
}
}
return null;
}
public String getFirst(String name) {
List<String> values = get(name);
if (values == null || values.isEmpty()) {
return null;
}
return values.get(0);
}
public void set(String name, String value) {
List<String> values = new ArrayList<>();
values.add(value);
put(name, values);
}
@Override
public List<String> getOrDefault(Object key, List<String> defaultValue) {
if (key instanceof String) {
String caseInsensitiveKey = this.caseInsensitiveKeys.get(convertKey((String) key));
if (caseInsensitiveKey != null) {
return this.targetMap.get(caseInsensitiveKey);
}
}
return defaultValue;
}
@Override
public List<String> put(String key, List<String> value) {
String oldKey = this.caseInsensitiveKeys.put(convertKey(key), key);
List<String> oldKeyValue = null;
if (oldKey != null && !oldKey.equals(key)) {
oldKeyValue = this.targetMap.remove(oldKey);
}
List<String> oldValue = this.targetMap.put(key, value);
return (oldKeyValue != null ? oldKeyValue : oldValue);
}
@Override
public void putAll(Map<? extends String, ? extends List<String>> map) {
if (map.isEmpty()) {
return;
}
map.forEach(this::put);
}
@Override
public List<String> putIfAbsent(String key, List<String> value) {
String oldKey = this.caseInsensitiveKeys.putIfAbsent(convertKey(key), key);
if (oldKey != null) {
List<String> oldKeyValue = this.targetMap.get(oldKey);
if (oldKeyValue != null) {
return oldKeyValue;
} else {
key = oldKey;
}
}
return this.targetMap.putIfAbsent(key, value);
}
@Override
public List<String> computeIfAbsent(String key, Function<? super String, ? extends List<String>> mappingFunction) {
String oldKey = this.caseInsensitiveKeys.putIfAbsent(convertKey(key), key);
if (oldKey != null) {
List<String> oldKeyValue = this.targetMap.get(oldKey);
if (oldKeyValue != null) {
return oldKeyValue;
} else {
key = oldKey;
}
}
return this.targetMap.computeIfAbsent(key, mappingFunction);
}
@Override
public List<String> remove(Object key) {
if (key instanceof String) {
String caseInsensitiveKey = removeCaseInsensitiveKey((String) key);
if (caseInsensitiveKey != null) {
return this.targetMap.remove(caseInsensitiveKey);
}
}
return null;
}
@Override
public void clear() {
this.caseInsensitiveKeys.clear();
this.targetMap.clear();
}
@Override
public Set<String> keySet() {
Set<String> keySet = this.keySet;
if (keySet == null) {
keySet = new KeySet(this.targetMap.keySet());
this.keySet = keySet;
}
return keySet;
}
@Override
public Collection<List<String>> values() {
Collection<List<String>> values = this.values;
if (values == null) {
values = new Values(this.targetMap.values());
this.values = values;
}
return values;
}
@Override
public Set<Entry<String, List<String>>> entrySet() {
Set<Entry<String, List<String>>> entrySet = this.entrySet;
if (entrySet == null) {
entrySet = new EntrySet(this.targetMap.entrySet());
this.entrySet = entrySet;
}
return entrySet;
}
public Map<String, String> toSingleValueMap() {
Map<String, String> result = new HashMap<>(this.targetMap.size());
for (String key : keySet()) {
result.put(key, getFirst(key));
}
return result;
}
@Override
public boolean equals(Object other) {
return (this == other || this.targetMap.equals(other));
}
@Override
public int hashCode() {
return this.targetMap.hashCode();
}
@Override
public String toString() {
return this.targetMap.toString();
}
public Locale getLocale() {
return this.locale;
}
protected String convertKey(String key) {
return key.toLowerCase(getLocale());
}
protected boolean removeEldestEntry(Map.Entry<String, List<String>> eldest) {
return false;
}
private String removeCaseInsensitiveKey(String key) {
return this.caseInsensitiveKeys.remove(convertKey(key));
}
private class KeySet extends AbstractSet<String> {
private final Set<String> delegate;
KeySet(Set<String> delegate) {
this.delegate = delegate;
}
@Override
public int size() {
return this.delegate.size();
}
@Override
public boolean contains(Object o) {
return this.delegate.contains(o);
}
@Override
public Iterator<String> iterator() {
return new KeySetIterator();
}
@Override
public boolean remove(Object o) {
return HttpHeaders.this.remove(o) != null;
}
@Override
public void clear() {
HttpHeaders.this.clear();
}
@Override
public Spliterator<String> spliterator() {
return this.delegate.spliterator();
}
@Override
public void forEach(Consumer<? super String> action) {
this.delegate.forEach(action);
}
}
private class Values extends AbstractCollection<List<String>> {
private final Collection<List<String>> delegate;
Values(Collection<List<String>> delegate) {
this.delegate = delegate;
}
@Override
public int size() {
return this.delegate.size();
}
@Override
public boolean contains(Object o) {
return this.delegate.contains(o);
}
@Override
public Iterator<List<String>> iterator() {
return new ValuesIterator();
}
@Override
public void clear() {
HttpHeaders.this.clear();
}
@Override
public Spliterator<List<String>> spliterator() {
return this.delegate.spliterator();
}
@Override
public void forEach(Consumer<? super List<String>> action) {
this.delegate.forEach(action);
}
}
private class EntrySet extends AbstractSet<Entry<String, List<String>>> {
private final Set<Entry<String, List<String>>> delegate;
public EntrySet(Set<Entry<String, List<String>>> delegate) {
this.delegate = delegate;
}
@Override
public int size() {
return this.delegate.size();
}
@Override
public boolean contains(Object o) {
return this.delegate.contains(o);
}
@Override
public Iterator<Entry<String, List<String>>> iterator() {
return new EntrySetIterator();
}
@Override
@SuppressWarnings("unchecked")
public boolean remove(Object o) {
if (this.delegate.remove(o)) {
removeCaseInsensitiveKey(((Map.Entry<String, List<String>>) o).getKey());
return true;
}
return false;
}
@Override
public void clear() {
this.delegate.clear();
caseInsensitiveKeys.clear();
}
@Override
public Spliterator<Entry<String, List<String>>> spliterator() {
return this.delegate.spliterator();
}
@Override
public void forEach(Consumer<? super Entry<String, List<String>>> action) {
this.delegate.forEach(action);
}
}
private abstract class EntryIterator<T> implements Iterator<T> {
private final Iterator<Entry<String, List<String>>> delegate;
private Entry<String, List<String>> last;
public EntryIterator() {
this.delegate = targetMap.entrySet().iterator();
}
protected Entry<String, List<String>> nextEntry() {
Entry<String, List<String>> entry = this.delegate.next();
this.last = entry;
return entry;
}
@Override
public boolean hasNext() {
return this.delegate.hasNext();
}
@Override
public void remove() {
this.delegate.remove();
if (this.last != null) {
removeCaseInsensitiveKey(this.last.getKey());
this.last = null;
}
}
}
private class KeySetIterator extends EntryIterator<String> {
@Override
public String next() {
return nextEntry().getKey();
}
}
private class ValuesIterator extends EntryIterator<List<String>> {
@Override
public List<String> next() {
return nextEntry().getValue();
}
}
private class EntrySetIterator extends EntryIterator<Entry<String, List<String>>> {
@Override
public Entry<String, List<String>> next() {
return nextEntry();
}
}
}

25
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpInputMessage.java Normal file
View File

@ -0,0 +1,25 @@
/*
* 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 java.io.InputStream;
public interface HttpInputMessage {
InputStream getBody();
}

23
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpMetadata.java Normal file
View File

@ -0,0 +1,23 @@
/*
* 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;
public interface HttpMetadata {
HttpHeaders headers();
}

36
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpOutputMessage.java Normal file
View File

@ -0,0 +1,36 @@
/*
* 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 java.io.ByteArrayOutputStream;
import java.io.OutputStream;
public interface HttpOutputMessage {
HttpOutputMessage EMPTY_MESSAGE = new HttpOutputMessage() {
private final OutputStream INPUT_STREAM = new ByteArrayOutputStream(0);
@Override
public OutputStream getBody() {
return INPUT_STREAM;
}
};
OutputStream getBody();
}

42
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpStatus.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.remoting.http12;
public enum HttpStatus {
OK(200),
REQUEST_TIMEOUT(408),
UNSUPPORTED_MEDIA_TYPE(415),
INTERNAL_SERVER_ERROR(500);
private final int code;
private final String statusString;
HttpStatus(int code) {
this.code = code;
this.statusString = String.valueOf(code);
}
public int getCode() {
return code;
}
public String getStatusString() {
return statusString;
}
}

25
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/HttpTransportListener.java Normal file
View File

@ -0,0 +1,25 @@
/*
* 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;
public interface HttpTransportListener<HEADER extends HttpMetadata, MESSAGE extends HttpInputMessage> {
void onMetadata(HEADER metadata);
void onData(MESSAGE message);
}

25
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/RequestMetadata.java Normal file
View File

@ -0,0 +1,25 @@
/*
* 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;
public interface RequestMetadata extends HttpMetadata {
String method();
String path();
}

37
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerCallListener.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.remoting.http12;
public interface ServerCallListener {
/**
* Callback when a request message is received.
*
* @param message message received
*/
void onMessage(Object message);
/**
* @param code when the call is canceled.
*/
void onCancel(long code);
/**
* Request completed.
*/
void onComplete();
}

53
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/ServerStreamServerCallListener.java Normal file
View File

@ -0,0 +1,53 @@
/*
* 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.stream.StreamObserver;
import org.apache.dubbo.remoting.http12.exception.HttpStatusException;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
public class ServerStreamServerCallListener extends AbstractServerCallListener {
public ServerStreamServerCallListener(RpcInvocation invocation, Invoker<?> invoker,
StreamObserver<Object> responseObserver) {
super(invocation, invoker, responseObserver);
}
@Override
public void onReturn(Object value) {
}
@Override
public void onMessage(Object message) {
if (message instanceof Object[]) {
message = ((Object[]) message)[0];
}
invocation.setArguments(new Object[]{message, responseObserver});
}
@Override
public void onCancel(long code) {
responseObserver.onError(new HttpStatusException((int) code));
}
@Override
public void onComplete() {
invoke();
}
}

31
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/TrailersCustomizer.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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 java.util.function.BiConsumer;
@FunctionalInterface
public interface TrailersCustomizer extends BiConsumer<HttpHeaders, Throwable>, HeadersCustomizer {
TrailersCustomizer NO_OP = (headers, throwable) -> {
};
@Override
default void accept(HttpHeaders headers) {
this.accept(headers, null);
}
}

55
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/UnaryServerCallListener.java Normal file
View File

@ -0,0 +1,55 @@
/*
* 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.stream.StreamObserver;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.RpcInvocation;
public class UnaryServerCallListener extends AbstractServerCallListener {
public UnaryServerCallListener(RpcInvocation invocation,
Invoker<?> invoker,
StreamObserver<Object> responseObserver) {
super(invocation, invoker, responseObserver);
}
@Override
public void onReturn(Object value) {
responseObserver.onNext(value);
responseObserver.onCompleted();
}
@Override
public void onMessage(Object message) {
if (message instanceof Object[]) {
invocation.setArguments((Object[]) message);
} else {
invocation.setArguments(new Object[]{message});
}
}
@Override
public void onCancel(long code) {
//ignore
}
@Override
public void onComplete() {
invoke();
}
}

40
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/command/DataQueueCommand.java Normal file
View File

@ -0,0 +1,40 @@
/*
* 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.command;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
public class DataQueueCommand extends HttpChannelQueueCommand {
private final HttpOutputMessage httpOutputMessage;
public DataQueueCommand(HttpOutputMessage httpMessage) {
this.httpOutputMessage = httpMessage;
}
@Override
public void run() {
getHttpChannel().writeMessage(httpOutputMessage)
.whenComplete((unused, throwable) -> {
if (throwable != null) {
completeExceptionally(throwable);
} else {
complete(unused);
}
});
}
}

40
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/command/HeaderQueueCommand.java Normal file
View File

@ -0,0 +1,40 @@
/*
* 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.command;
import org.apache.dubbo.remoting.http12.HttpMetadata;
public class HeaderQueueCommand extends HttpChannelQueueCommand {
private final HttpMetadata httpMetadata;
public HeaderQueueCommand(HttpMetadata httpMetadata) {
this.httpMetadata = httpMetadata;
}
@Override
public void run() {
getHttpChannel().writeHeader(httpMetadata)
.whenComplete((unused, throwable) -> {
if (throwable != null) {
completeExceptionally(throwable);
} else {
complete(unused);
}
});
}
}

36
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/command/HttpChannelQueueCommand.java Normal file
View File

@ -0,0 +1,36 @@
/*
* 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.command;
import org.apache.dubbo.remoting.http12.HttpChannel;
import org.apache.dubbo.remoting.http12.HttpChannelHolder;
import java.util.concurrent.CompletableFuture;
public abstract class HttpChannelQueueCommand extends CompletableFuture<Void> implements QueueCommand, HttpChannelHolder {
private HttpChannelHolder httpChannelHolder;
public void setHttpChannel(HttpChannelHolder httpChannelHolder) {
this.httpChannelHolder = httpChannelHolder;
}
public HttpChannel getHttpChannel() {
return httpChannelHolder.getHttpChannel();
}
}

47
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/command/HttpWriteQueue.java Normal file
View File

@ -0,0 +1,47 @@
/*
* 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.command;
import org.apache.dubbo.common.BatchExecutorQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
public class HttpWriteQueue extends BatchExecutorQueue<HttpChannelQueueCommand> {
private final Executor executor;
public HttpWriteQueue(Executor executor) {
this.executor = executor;
}
public CompletableFuture<Void> enqueue(HttpChannelQueueCommand cmd) {
this.enqueue(cmd, this.executor);
return cmd;
}
@Override
protected void prepare(HttpChannelQueueCommand item) {
item.run();
}
@Override
protected void flush(HttpChannelQueueCommand item) {
item.run();
item.getHttpChannel().flush();
}
}

21
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/command/QueueCommand.java Normal file
View File

@ -0,0 +1,21 @@
/*
* 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.command;
public interface QueueCommand extends Runnable {
}

32
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/exception/DecodeException.java Normal file
View File

@ -0,0 +1,32 @@
/*
* 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.exception;
public class DecodeException extends HttpStatusException {
public DecodeException() {
super(500);
}
public DecodeException(String message) {
super(500, message);
}
public DecodeException(Throwable cause) {
super(500, cause);
}
}

32
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/exception/EncodeException.java Normal file
View File

@ -0,0 +1,32 @@
/*
* 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.exception;
public class EncodeException extends HttpStatusException {
public EncodeException(String message) {
super(500, message);
}
public EncodeException(String message, Throwable cause) {
super(500, message);
}
public EncodeException(Throwable cause) {
super(500, cause);
}
}

42
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/exception/HttpRequestTimeout.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.remoting.http12.exception;
import org.apache.dubbo.remoting.http12.HttpStatus;
public class HttpRequestTimeout extends HttpStatusException {
private final String side;
private HttpRequestTimeout(String side) {
super(HttpStatus.REQUEST_TIMEOUT.getCode());
this.side = side;
}
public String getSide() {
return side;
}
public static HttpRequestTimeout serverSide(){
return new HttpRequestTimeout("server");
}
public static HttpRequestTimeout clientSide(){
return new HttpRequestTimeout("client");
}
}

40
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/exception/HttpStatusException.java Normal file
View File

@ -0,0 +1,40 @@
/*
* 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.exception;
public class HttpStatusException extends RuntimeException {
private final int statusCode;
public HttpStatusException(int statusCode) {
this(statusCode, "Unknown Error");
}
public HttpStatusException(int statusCode, String message) {
super(message);
this.statusCode = statusCode;
}
public HttpStatusException(int statusCode, Throwable cause) {
super(cause);
this.statusCode = statusCode;
}
public int getStatusCode() {
return statusCode;
}
}

31
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/exception/IllegalPathException.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.exception;
public class IllegalPathException extends RuntimeException{
public IllegalPathException() {
}
public IllegalPathException(String message) {
super(message);
}
public String getPath() {
return super.getMessage();
}
}

31
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/exception/UnimplementedException.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.exception;
public class UnimplementedException extends HttpStatusException {
private final String unimplemented;
public UnimplementedException(String unimplemented) {
super(500, "unimplemented " + unimplemented);
this.unimplemented = unimplemented;
}
public String getUnimplemented() {
return unimplemented;
}
}

31
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/exception/UnsupportedMediaTypeException.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.exception;
public class UnsupportedMediaTypeException extends HttpStatusException {
private final String mediaType;
public UnsupportedMediaTypeException(String mediaType) {
super(415, "Unsupported Media Type '" + mediaType + "'");
this.mediaType = mediaType;
}
public String getMediaType() {
return mediaType;
}
}

55
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Request.java Normal file
View File

@ -0,0 +1,55 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.HttpInputMessage;
import org.apache.dubbo.remoting.http12.RequestMetadata;
import java.io.InputStream;
public class DefaultHttp1Request implements Http1Request {
private final RequestMetadata httpMetadata;
private final HttpInputMessage httpInputMessage;
public DefaultHttp1Request(RequestMetadata httpMetadata, HttpInputMessage httpInputMessage) {
this.httpMetadata = httpMetadata;
this.httpInputMessage = httpInputMessage;
}
@Override
public InputStream getBody() {
return httpInputMessage.getBody();
}
@Override
public HttpHeaders headers() {
return httpMetadata.headers();
}
@Override
public String method() {
return httpMetadata.method();
}
@Override
public String path() {
return httpMetadata.path();
}
}

45
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/DefaultHttp1Response.java Normal file
View File

@ -0,0 +1,45 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.HttpInputMessage;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import java.io.InputStream;
public class DefaultHttp1Response implements HttpMetadata, HttpInputMessage {
private final HttpMetadata httpMetadata;
private final HttpInputMessage httpInputMessage;
public DefaultHttp1Response(HttpMetadata httpMetadata, HttpInputMessage httpInputMessage) {
this.httpMetadata = httpMetadata;
this.httpInputMessage = httpInputMessage;
}
@Override
public InputStream getBody() {
return httpInputMessage.getBody();
}
@Override
public HttpHeaders headers() {
return httpMetadata.headers();
}
}

35
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1InputMessage.java Normal file
View File

@ -0,0 +1,35 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpInputMessage;
import java.io.InputStream;
public class Http1InputMessage implements HttpInputMessage {
private final InputStream body;
public Http1InputMessage(InputStream body) {
this.body = body;
}
@Override
public InputStream getBody() {
return body;
}
}

34
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1Metadata.java Normal file
View File

@ -0,0 +1,34 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.HttpMetadata;
public class Http1Metadata implements HttpMetadata {
private final HttpHeaders headers;
public Http1Metadata(HttpHeaders headers) {
this.headers = headers;
}
@Override
public HttpHeaders headers() {
return headers;
}
}

36
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1OutputMessage.java Normal file
View File

@ -0,0 +1,36 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
import java.io.OutputStream;
public class Http1OutputMessage implements HttpOutputMessage {
private final OutputStream outputStream;
public Http1OutputMessage(OutputStream outputStream) {
this.outputStream = outputStream;
}
@Override
public OutputStream getBody() {
return outputStream;
}
}

24
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1Request.java Normal file
View File

@ -0,0 +1,24 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpInputMessage;
import org.apache.dubbo.remoting.http12.RequestMetadata;
public interface Http1Request extends RequestMetadata, HttpInputMessage {
}

56
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1RequestMetadata.java Normal file
View File

@ -0,0 +1,56 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.RequestMetadata;
public class Http1RequestMetadata implements RequestMetadata {
private HttpHeaders headers;
private String method;
private String path;
public void setHeaders(HttpHeaders headers) {
this.headers = headers;
}
public void setMethod(String method) {
this.method = method;
}
public void setPath(String path) {
this.path = path;
}
@Override
public HttpHeaders headers() {
return headers;
}
@Override
public String method() {
return method;
}
@Override
public String path() {
return path;
}
}

45
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerChannelObserver.java Normal file
View File

@ -0,0 +1,45 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.AbstractServerHttpChannelObserver;
import org.apache.dubbo.remoting.http12.HttpChannel;
import org.apache.dubbo.remoting.http12.HttpChannelObserver;
import org.apache.dubbo.remoting.http12.HttpHeaderNames;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
public class Http1ServerChannelObserver extends AbstractServerHttpChannelObserver implements HttpChannelObserver<Object> {
public Http1ServerChannelObserver(HttpChannel httpChannel) {
super(httpChannel);
}
@Override
protected HttpMetadata encodeHttpMetadata() {
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.set(HttpHeaderNames.TRANSFER_ENCODING.getName(), "chunked");
return new Http1Metadata(httpHeaders);
}
@Override
protected void doOnCompleted(Throwable throwable) {
super.doOnCompleted(throwable);
this.getHttpChannel().writeMessage(HttpOutputMessage.EMPTY_MESSAGE);
}
}

49
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerStreamChannelObserver.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpChannel;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
import java.io.IOException;
public class Http1ServerStreamChannelObserver extends Http1ServerChannelObserver {
private static final String SERVER_SENT_EVENT_DATA_PREFIX = "data:";
private static final String SERVER_SENT_EVENT_LF = "\n\n";
private static final byte[] SERVER_SENT_EVENT_DATA_PREFIX_BYTES = SERVER_SENT_EVENT_DATA_PREFIX.getBytes();
private static final byte[] SERVER_SENT_EVENT_LF_BYTES = SERVER_SENT_EVENT_LF.getBytes();
public Http1ServerStreamChannelObserver(HttpChannel httpChannel) {
super(httpChannel);
}
@Override
protected void preOutputMessage(HttpOutputMessage httpMessage) throws IOException {
HttpOutputMessage httpOutputMessage = this.getHttpChannel().newOutputMessage();
httpOutputMessage.getBody().write(SERVER_SENT_EVENT_DATA_PREFIX_BYTES, 0, SERVER_SENT_EVENT_DATA_PREFIX_BYTES.length);
this.getHttpChannel().writeMessage(httpOutputMessage);
}
@Override
protected void postOutputMessage(HttpOutputMessage httpMessage) throws IOException {
HttpOutputMessage httpOutputMessage = this.getHttpChannel().newOutputMessage();
httpOutputMessage.getBody().write(SERVER_SENT_EVENT_LF_BYTES, 0, SERVER_SENT_EVENT_LF_BYTES.length);
this.getHttpChannel().writeMessage(httpOutputMessage);
}
}

26
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerTransportListener.java Normal file
View File

@ -0,0 +1,26 @@
/*
* 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.h1;
import org.apache.dubbo.remoting.http12.HttpChannelHolder;
import org.apache.dubbo.remoting.http12.HttpInputMessage;
import org.apache.dubbo.remoting.http12.HttpTransportListener;
import org.apache.dubbo.remoting.http12.RequestMetadata;
public interface Http1ServerTransportListener extends HttpTransportListener<RequestMetadata, HttpInputMessage>, HttpChannelHolder {
}

30
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h1/Http1ServerTransportListenerFactory.java Normal file
View File

@ -0,0 +1,30 @@
/*
* 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.h1;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionScope;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.remoting.http12.HttpChannel;
import org.apache.dubbo.rpc.model.FrameworkModel;
@SPI(scope = ExtensionScope.FRAMEWORK)
public interface Http1ServerTransportListenerFactory {
Http1ServerTransportListener newInstance(HttpChannel httpChannel, URL url, FrameworkModel frameworkModel);
}

27
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/CancelableTransportListener.java Normal file
View File

@ -0,0 +1,27 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.HttpInputMessage;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.HttpTransportListener;
public interface CancelableTransportListener<HEADER extends HttpMetadata, MESSAGE extends HttpInputMessage> extends HttpTransportListener<HEADER, MESSAGE> {
void cancelByRemote(long errorCode);
}

53
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/DefaultHttp2StreamingDecoder.java Normal file
View File

@ -0,0 +1,53 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.message.DefaultListeningDecoder;
import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
import org.apache.dubbo.remoting.http12.message.StreamingDecoder;
import java.io.InputStream;
public class DefaultHttp2StreamingDecoder implements StreamingDecoder {
private final DefaultListeningDecoder delegate;
public DefaultHttp2StreamingDecoder(HttpMessageCodec httpMessageCodec, Class<?>[] targetTypes) {
this.delegate = new DefaultListeningDecoder(httpMessageCodec, targetTypes);
}
@Override
public void decode(InputStream inputStream) throws DecodeException {
delegate.decode(inputStream);
}
@Override
public void close() {
delegate.close();
}
@Override
public void setListener(Listener listener) {
delegate.setListener(listener);
}
@Override
public void request(int numMessages) {
//no op
}
}

33
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/H2StreamChannel.java Normal file
View File

@ -0,0 +1,33 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.HttpChannel;
import java.util.concurrent.CompletableFuture;
public interface H2StreamChannel extends HttpChannel {
CompletableFuture<Void> writeResetFrame(long errorCode);
@Override
default Http2OutputMessage newOutputMessage() {
return this.newOutputMessage(false);
}
Http2OutputMessage newOutputMessage(boolean endStream);
}

29
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2CancelableStreamObserver.java Normal file
View File

@ -0,0 +1,29 @@
/*
* 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.h2;
import org.apache.dubbo.common.stream.StreamObserver;
import org.apache.dubbo.rpc.CancellationContext;
public interface Http2CancelableStreamObserver<T> extends StreamObserver<T> {
void setCancellationContext(CancellationContext cancellationContext);
CancellationContext getCancellationContext();
void cancel(Throwable throwable);
}

66
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2ChannelDelegate.java Normal file
View File

@ -0,0 +1,66 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
import java.net.SocketAddress;
import java.util.concurrent.CompletableFuture;
public class Http2ChannelDelegate implements H2StreamChannel {
private final H2StreamChannel h2StreamChannel;
public Http2ChannelDelegate(H2StreamChannel h2StreamChannel) {
this.h2StreamChannel = h2StreamChannel;
}
public H2StreamChannel getH2StreamChannel() {
return h2StreamChannel;
}
@Override
public CompletableFuture<Void> writeHeader(HttpMetadata httpMetadata) {
return h2StreamChannel.writeHeader(httpMetadata);
}
@Override
public CompletableFuture<Void> writeMessage(HttpOutputMessage httpOutputMessage) {
return h2StreamChannel.writeMessage(httpOutputMessage);
}
@Override
public SocketAddress remoteAddress() {
return h2StreamChannel.remoteAddress();
}
@Override
public void flush() {
h2StreamChannel.flush();
}
@Override
public CompletableFuture<Void> writeResetFrame(long errorCode) {
return h2StreamChannel.writeResetFrame(errorCode);
}
@Override
public Http2OutputMessage newOutputMessage(boolean endStream) {
return h2StreamChannel.newOutputMessage(endStream);
}
}

41
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2Header.java Normal file
View File

@ -0,0 +1,41 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.RequestMetadata;
public interface Http2Header extends RequestMetadata, Http2StreamFrame {
@Override
default String method() {
return headers().getFirst(Http2Headers.METHOD.getName());
}
@Override
default String path() {
return headers().getFirst(Http2Headers.PATH.getName());
}
default String status() {
return headers().getFirst(Http2Headers.STATUS.getName());
}
@Override
default String name() {
return "HEADER";
}
}

36
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2Headers.java Normal file
View File

@ -0,0 +1,36 @@
/*
* 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.h2;
public enum Http2Headers {
PATH(":path"),
METHOD(":method"),
STATUS(":status");
private final String name;
Http2Headers(String name) {
this.name = name;
}
public String getName() {
return name;
}
}

23
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2InputMessage.java Normal file
View File

@ -0,0 +1,23 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.HttpInputMessage;
public interface Http2InputMessage extends HttpInputMessage, Http2StreamFrame {
}

65
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2InputMessageFrame.java Normal file
View File

@ -0,0 +1,65 @@
/*
* 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.h2;
import java.io.InputStream;
public class Http2InputMessageFrame implements Http2InputMessage {
private int id;
private final InputStream body;
private final boolean endStream;
public Http2InputMessageFrame(InputStream body) {
this(body, false);
}
public Http2InputMessageFrame(boolean endStream) {
this(null, endStream);
}
public Http2InputMessageFrame(InputStream body, boolean endStream) {
this.body = body;
this.endStream = endStream;
}
public void setId(int id) {
this.id = id;
}
@Override
public InputStream getBody() {
return body;
}
@Override
public String name() {
return "DATA";
}
@Override
public int id() {
return id;
}
@Override
public boolean isEndStream() {
return endStream;
}
}

57
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2MetadataFrame.java Normal file
View File

@ -0,0 +1,57 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.HttpHeaders;
public class Http2MetadataFrame implements Http2Header {
private final int streamId;
private final HttpHeaders headers;
private final boolean endStream;
public Http2MetadataFrame(HttpHeaders headers) {
this(headers, false);
}
public Http2MetadataFrame(HttpHeaders headers, boolean endStream) {
this(-1, headers, endStream);
}
public Http2MetadataFrame(int streamId, HttpHeaders headers, boolean endStream) {
this.streamId = streamId;
this.headers = headers;
this.endStream = endStream;
}
@Override
public HttpHeaders headers() {
return headers;
}
@Override
public int id() {
return streamId;
}
@Override
public boolean isEndStream() {
return endStream;
}
}

32
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2OutputMessage.java Normal file
View File

@ -0,0 +1,32 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
public interface Http2OutputMessage extends HttpOutputMessage, Http2StreamFrame {
@Override
default String name() {
return "DATA";
}
@Override
default int id() {
return -1;
}
}

49
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2OutputMessageFrame.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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.h2;
import java.io.OutputStream;
public class Http2OutputMessageFrame implements Http2OutputMessage {
private final OutputStream body;
private final boolean endStream;
public Http2OutputMessageFrame(OutputStream body) {
this(body, false);
}
public Http2OutputMessageFrame(boolean endStream) {
this(null, endStream);
}
public Http2OutputMessageFrame(OutputStream body, boolean endStream) {
this.body = body;
this.endStream = endStream;
}
@Override
public OutputStream getBody() {
return body;
}
@Override
public boolean isEndStream() {
return endStream;
}
}

96
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2ServerChannelObserver.java Normal file
View File

@ -0,0 +1,96 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.AbstractServerHttpChannelObserver;
import org.apache.dubbo.remoting.http12.ErrorCodeHolder;
import org.apache.dubbo.remoting.http12.FlowControlStreamObserver;
import org.apache.dubbo.remoting.http12.HttpChannelObserver;
import org.apache.dubbo.remoting.http12.HttpHeaderNames;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.message.StreamingDecoder;
import org.apache.dubbo.rpc.CancellationContext;
public class Http2ServerChannelObserver extends AbstractServerHttpChannelObserver implements HttpChannelObserver<Object>, FlowControlStreamObserver<Object>, Http2CancelableStreamObserver<Object> {
private CancellationContext cancellationContext;
private StreamingDecoder streamingDecoder;
private boolean autoRequestN = true;
public Http2ServerChannelObserver(H2StreamChannel h2StreamChannel) {
super(h2StreamChannel);
}
public void setStreamingDecoder(StreamingDecoder streamingDecoder) {
this.streamingDecoder = streamingDecoder;
}
@Override
protected HttpMetadata encodeHttpMetadata() {
HttpHeaders httpHeaders = new HttpHeaders();
httpHeaders.set(HttpHeaderNames.TE.getName(), "trailers");
return new Http2MetadataFrame(httpHeaders);
}
@Override
protected HttpMetadata encodeTrailers(Throwable throwable) {
return new Http2MetadataFrame(new HttpHeaders(), true);
}
@Override
public H2StreamChannel getHttpChannel() {
return (H2StreamChannel) super.getHttpChannel();
}
@Override
public void setCancellationContext(CancellationContext cancellationContext) {
this.cancellationContext = cancellationContext;
}
@Override
public CancellationContext getCancellationContext() {
return cancellationContext;
}
@Override
public void cancel(Throwable throwable) {
long errorCode = 0;
if (throwable instanceof ErrorCodeHolder) {
errorCode = ((ErrorCodeHolder) throwable).getErrorCode();
}
getHttpChannel().writeResetFrame(errorCode);
this.cancellationContext.cancel(throwable);
}
@Override
public void request(int count) {
this.streamingDecoder.request(count);
}
@Override
public void disableAutoFlowControl() {
this.autoRequestN = false;
}
@Override
public boolean isAutoRequestN() {
return autoRequestN;
}
}

31
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2ServerTransportListenerFactory.java Normal file
View File

@ -0,0 +1,31 @@
/*
* 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.h2;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionScope;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.rpc.model.FrameworkModel;
@SPI(scope = ExtensionScope.FRAMEWORK)
public interface Http2ServerTransportListenerFactory {
Http2TransportListener newInstance(H2StreamChannel streamChannel, URL url, FrameworkModel frameworkModel);
boolean supportContentType(String contentType);
}

26
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2StreamFrame.java Normal file
View File

@ -0,0 +1,26 @@
/*
* 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.h2;
public interface Http2StreamFrame {
int id();
String name();
boolean isEndStream();
}

23
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/Http2TransportListener.java Normal file
View File

@ -0,0 +1,23 @@
/*
* 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.h2;
import org.apache.dubbo.remoting.http12.HttpChannelHolder;
public interface Http2TransportListener extends CancelableTransportListener<Http2Header, Http2InputMessage>, HttpChannelHolder {
}

58
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/command/Http2WriteQueueChannel.java Normal file
View File

@ -0,0 +1,58 @@
/*
* 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.h2.command;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
import org.apache.dubbo.remoting.http12.command.DataQueueCommand;
import org.apache.dubbo.remoting.http12.command.HeaderQueueCommand;
import org.apache.dubbo.remoting.http12.command.HttpWriteQueue;
import org.apache.dubbo.remoting.http12.h2.H2StreamChannel;
import org.apache.dubbo.remoting.http12.h2.Http2ChannelDelegate;
import java.util.concurrent.CompletableFuture;
public class Http2WriteQueueChannel extends Http2ChannelDelegate {
private final HttpWriteQueue httpWriteQueue;
public Http2WriteQueueChannel(H2StreamChannel h2StreamChannel, HttpWriteQueue httpWriteQueue) {
super(h2StreamChannel);
this.httpWriteQueue = httpWriteQueue;
}
@Override
public CompletableFuture<Void> writeHeader(HttpMetadata httpMetadata) {
HeaderQueueCommand cmd = new HeaderQueueCommand(httpMetadata);
cmd.setHttpChannel(this::getH2StreamChannel);
return httpWriteQueue.enqueue(cmd);
}
@Override
public CompletableFuture<Void> writeMessage(HttpOutputMessage httpOutputMessage) {
DataQueueCommand cmd = new DataQueueCommand(httpOutputMessage);
cmd.setHttpChannel(this::getH2StreamChannel);
return httpWriteQueue.enqueue(cmd);
}
@Override
public CompletableFuture<Void> writeResetFrame(long errorCode) {
ResetQueueCommand cmd = new ResetQueueCommand(errorCode);
cmd.setHttpChannel(this::getH2StreamChannel);
return this.httpWriteQueue.enqueue(cmd);
}
}

42
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/h2/command/ResetQueueCommand.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.remoting.http12.h2.command;
import org.apache.dubbo.remoting.http12.command.HttpChannelQueueCommand;
import org.apache.dubbo.remoting.http12.h2.H2StreamChannel;
public class ResetQueueCommand extends HttpChannelQueueCommand {
private final long errorCode;
public ResetQueueCommand(long errorCode) {
this.errorCode = errorCode;
}
@Override
public void run() {
((H2StreamChannel) getHttpChannel())
.writeResetFrame(errorCode)
.whenComplete((unused, throwable) -> {
if (throwable != null) {
completeExceptionally(throwable);
} else {
complete(unused);
}
});
}
}

49
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/DefaultListeningDecoder.java Normal file
View File

@ -0,0 +1,49 @@
/*
* 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.message;
import java.io.InputStream;
public class DefaultListeningDecoder implements ListeningDecoder {
private final HttpMessageCodec httpMessageCodec;
private final Class<?>[] targetTypes;
private Listener listener;
public DefaultListeningDecoder(HttpMessageCodec httpMessageCodec, Class<?>[] targetTypes) {
this.httpMessageCodec = httpMessageCodec;
this.targetTypes = targetTypes;
}
@Override
public void setListener(Listener listener) {
this.listener = listener;
}
@Override
public void decode(InputStream inputStream) {
Object[] decode = this.httpMessageCodec.decode(inputStream, targetTypes);
this.listener.onMessage(decode);
}
@Override
public void close() {
this.listener.onClose();
}
}

54
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/HttpMessageCodec.java Normal file
View File

@ -0,0 +1,54 @@
/*
* 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.message;
import org.apache.dubbo.common.extension.ExtensionScope;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.exception.EncodeException;
import java.io.InputStream;
import java.io.OutputStream;
/**
* for http body codec
*/
@SPI(scope = ExtensionScope.FRAMEWORK)
public interface HttpMessageCodec {
void encode(OutputStream outputStream, Object data) throws EncodeException;
default void encode(OutputStream outputStream, Object[] data) throws EncodeException {
//default encode first data
this.encode(outputStream, data[0]);
}
Object decode(InputStream inputStream, Class<?> targetType) throws DecodeException;
default Object[] decode(InputStream inputStream, Class<?>[] targetTypes) throws DecodeException {
//default decode first target type
return new Object[]{this.decode(inputStream, targetTypes[0])};
}
MediaType contentType();
default boolean support(String contentType) {
MediaType mediaType = this.contentType();
return mediaType.getName().startsWith(contentType);
}
}

127
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/JsonCodec.java Normal file
View File

@ -0,0 +1,127 @@
/*
* 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.message;
import com.alibaba.fastjson2.JSONObject;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.utils.JsonUtils;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import org.apache.dubbo.remoting.http12.exception.EncodeException;
import java.io.InputStream;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
/**
* body is json
*/
@Activate
public class JsonCodec implements HttpMessageCodec {
public static final HttpMessageCodec INSTANCE = new JsonCodec();
@Override
public MediaType contentType() {
return MediaType.APPLICATION_JSON_VALUE;
}
@Override
public void encode(OutputStream outputStream, Object unSerializedBody) throws EncodeException {
try {
try {
String jsonString = JsonUtils.toJson(unSerializedBody);
outputStream.write(jsonString.getBytes(StandardCharsets.UTF_8));
} finally {
outputStream.flush();
}
} catch (Throwable e) {
throw new EncodeException(e);
}
}
@Override
public void encode(OutputStream outputStream, Object[] data) throws EncodeException {
try {
try {
String jsonString = JsonUtils.toJson(data);
outputStream.write(jsonString.getBytes(StandardCharsets.UTF_8));
} finally {
outputStream.flush();
}
} catch (Throwable e) {
throw new EncodeException(e);
}
}
@Override
public Object decode(InputStream body, Class<?> targetType) throws DecodeException {
try {
try {
int len;
byte[] data = new byte[4096];
StringBuilder builder = new StringBuilder(4096);
while ((len = body.read(data)) != -1) {
builder.append(new String(data, 0, len));
}
return JsonUtils.toJavaObject(builder.toString(), targetType);
} finally {
body.close();
}
} catch (Throwable e) {
throw new DecodeException(e);
}
}
@Override
public Object[] decode(InputStream dataInputStream, Class<?>[] targetTypes) throws DecodeException {
List<Object> result = new ArrayList<>();
try {
try {
if (targetTypes.length == 1) {
return new Object[]{this.decode(dataInputStream, targetTypes[0])};
}
int len;
byte[] data = new byte[4096];
StringBuilder builder = new StringBuilder(4096);
while ((len = dataInputStream.read(data)) != -1) {
builder.append(new String(data, 0, len));
}
String jsonString = builder.toString();
List<Object> jsonObjects = JsonUtils.toJavaList(jsonString, Object.class);
for (int i = 0; i < targetTypes.length; i++) {
Object jsonObject = jsonObjects.get(i);
Class<?> type = targetTypes[i];
if (jsonObject instanceof JSONObject) {
Object o = ((JSONObject) jsonObject).toJavaObject(type);
result.add(o);
} else {
result.add(jsonObject);
}
}
return result.toArray();
} finally {
dataInputStream.close();
}
} catch (Throwable e) {
throw new DecodeException(e);
}
}
}

189
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java Normal file
View File

@ -0,0 +1,189 @@
/*
* 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.message;
import org.apache.dubbo.remoting.http12.CompositeInputStream;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
public class LengthFieldStreamingDecoder implements StreamingDecoder {
private long pendingDeliveries;
private boolean inDelivery = false;
private boolean closing;
private boolean closed;
private DecodeState state = DecodeState.HEADER;
private final CompositeInputStream accumulate = new CompositeInputStream();
private final Class<?>[] targetTypes;
private HttpMessageCodec httpMessageCodec;
private Listener listener;
private final int lengthFieldOffset;
private final int lengthFieldLength;
private int requiredLength;
public LengthFieldStreamingDecoder(Class<?>[] targetTypes) {
this(4, targetTypes);
}
public LengthFieldStreamingDecoder(int lengthFieldLength, Class<?>[] targetTypes) {
this(0, lengthFieldLength, targetTypes);
}
public LengthFieldStreamingDecoder(int lengthFieldOffset, int lengthFieldLength, Class<?>[] targetTypes) {
this.lengthFieldOffset = lengthFieldOffset;
this.lengthFieldLength = lengthFieldLength;
this.targetTypes = targetTypes;
this.requiredLength = lengthFieldOffset + lengthFieldLength;
}
public void setHttpMessageCodec(HttpMessageCodec httpMessageCodec) {
this.httpMessageCodec = httpMessageCodec;
}
@Override
public final void decode(InputStream inputStream) throws DecodeException {
if (closing || closed) {
// ignored
return;
}
accumulate.addInputStream(inputStream);
deliver();
}
@Override
public final void request(int numMessages) {
pendingDeliveries += numMessages;
deliver();
}
@Override
public final void close() {
closing = true;
deliver();
}
@Override
public final void setListener(Listener listener) {
this.listener = listener;
}
private void deliver() {
// We can have reentrancy here when using a direct executor, triggered by calls to
// request more messages. This is safe as we simply loop until pendingDelivers = 0
if (inDelivery) {
return;
}
inDelivery = true;
try {
// Process the uncompressed bytes.
while (pendingDeliveries > 0 && hasEnoughBytes()) {
switch (state) {
case HEADER:
processHeader();
break;
case PAYLOAD:
// Read the body and deliver the message.
processBody();
// Since we've delivered a message, decrement the number of pending
// deliveries remaining.
pendingDeliveries--;
break;
default:
throw new AssertionError("Invalid state: " + state);
}
}
if (closing) {
if (!closed) {
closed = true;
accumulate.close();
listener.onClose();
}
}
} catch (IOException e) {
throw new DecodeException(e);
} finally {
inDelivery = false;
}
}
private void processHeader() throws IOException {
processOffset(accumulate, lengthFieldOffset);
byte[] lengthBytes = new byte[lengthFieldLength];
accumulate.read(lengthBytes);
requiredLength = bytesToInt(lengthBytes);
// Continue reading the frame body.
state = DecodeState.PAYLOAD;
}
protected void processOffset(InputStream inputStream, int lengthFieldOffset) throws IOException {
//default skip offset
skipOffset(inputStream, lengthFieldOffset);
}
private void skipOffset(InputStream inputStream, int lengthFieldOffset) throws IOException {
if (lengthFieldOffset != 0) {
return;
}
inputStream.read(new byte[lengthFieldOffset]);
}
private void processBody() throws IOException {
byte[] rawMessage = readRawMessage(accumulate, requiredLength);
InputStream inputStream = new ByteArrayInputStream(rawMessage);
Object[] decodeParameters = httpMessageCodec.decode(inputStream, targetTypes);
this.listener.onMessage(decodeParameters);
// Done with this frame, begin processing the next header.
state = DecodeState.HEADER;
requiredLength = lengthFieldOffset + lengthFieldLength;
}
protected byte[] readRawMessage(InputStream inputStream, int length) throws IOException {
byte[] data = new byte[length];
inputStream.read(data, 0, length);
return data;
}
private boolean hasEnoughBytes() {
return requiredLength - accumulate.available() <= 0;
}
protected static int bytesToInt(byte[] bytes) {
return (bytes[0] << 24) & 0xFF | (bytes[1] << 16) & 0xFF | (bytes[2] << 8) & 0xFF | (bytes[3]) & 0xFF;
}
private enum DecodeState {
HEADER,
PAYLOAD
}
}

43
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/ListeningDecoder.java Normal file
View File

@ -0,0 +1,43 @@
/*
* 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.message;
import org.apache.dubbo.remoting.http12.exception.DecodeException;
import java.io.InputStream;
public interface ListeningDecoder {
void decode(InputStream inputStream) throws DecodeException;
void close();
void setListener(Listener listener);
interface Listener {
/**
* call on decode finish
* @param message decoded object
*/
void onMessage(Object message);
default void onClose() {
}
}
}

66
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/MediaType.java Normal file
View File

@ -0,0 +1,66 @@
/*
* 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.message;
import java.nio.charset.Charset;
import java.util.Collections;
import java.util.Map;
public class MediaType {
public static final MediaType ALL_VALUE = new MediaType("*", "*");
public static final MediaType APPLICATION_JSON_VALUE = new MediaType("application", "json");
public static final MediaType TEXT_EVENT_STREAM_VALUE = new MediaType("text", "event-stream");
private final String name;
private final String type;
private final String subType;
private final Charset charset;
public MediaType(String type, String subType) {
this(type, subType, Collections.singletonMap("charset", "UTF-8"));
}
public MediaType(String type, String subType, Map<String, String> parameters) {
this.type = type;
this.subType = subType;
this.name = type + "/" + subType;
this.charset = Charset.forName(parameters.getOrDefault("charset", "UTF-8"));
}
public String getName() {
return name;
}
public String getType() {
return type;
}
public String getSubType() {
return subType;
}
public Charset getCharset() {
return charset;
}
}

84
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/MethodMetadata.java Normal file
View File

@ -0,0 +1,84 @@
/*
* 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.message;
import org.apache.dubbo.rpc.model.MethodDescriptor;
import org.apache.dubbo.rpc.model.ReflectionMethodDescriptor;
import org.apache.dubbo.rpc.model.StubMethodDescriptor;
import java.lang.reflect.ParameterizedType;
public class MethodMetadata {
private final Class<?>[] actualRequestTypes;
private final Class<?> actualResponseType;
private MethodMetadata(Class<?>[] actualRequestTypes, Class<?> actualResponseType) {
this.actualRequestTypes = actualRequestTypes;
this.actualResponseType = actualResponseType;
}
public Class<?>[] getActualRequestTypes() {
return actualRequestTypes;
}
public Class<?> getActualResponseType() {
return actualResponseType;
}
public static MethodMetadata fromMethodDescriptor(MethodDescriptor method) {
if (method instanceof ReflectionMethodDescriptor) {
return doResolveReflection((ReflectionMethodDescriptor) method);
}
if (method instanceof StubMethodDescriptor) {
return doResolveStub((StubMethodDescriptor) method);
}
throw new IllegalStateException("Can not reach here");
}
private static MethodMetadata doResolveStub(StubMethodDescriptor method) {
Class<?>[] actualRequestTypes = method.getParameterClasses();
Class<?> actualResponseType = method.getReturnClass();
return new MethodMetadata(actualRequestTypes, actualResponseType);
}
private static MethodMetadata doResolveReflection(ReflectionMethodDescriptor method) {
Class<?>[] actualRequestTypes;
Class<?> actualResponseType;
switch (method.getRpcType()) {
case CLIENT_STREAM:
case BI_STREAM:
actualRequestTypes = new Class<?>[]{
(Class<?>) ((ParameterizedType) method.getMethod()
.getGenericReturnType()).getActualTypeArguments()[0]};
actualResponseType = (Class<?>) ((ParameterizedType) method.getMethod()
.getGenericParameterTypes()[0]).getActualTypeArguments()[0];
return new MethodMetadata(actualRequestTypes, actualResponseType);
case SERVER_STREAM:
actualRequestTypes = new Class[]{method.getMethod().getParameterTypes()[0]};
actualResponseType = (Class<?>) ((ParameterizedType) method.getMethod()
.getGenericParameterTypes()[1]).getActualTypeArguments()[0];
return new MethodMetadata(actualRequestTypes, actualResponseType);
case UNARY:
actualRequestTypes = method.getParameterClasses();
actualResponseType = (Class<?>) method.getReturnTypes()[0];
return new MethodMetadata(actualRequestTypes, actualResponseType);
}
throw new IllegalStateException("Can not reach here");
}
}

23
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/message/StreamingDecoder.java Normal file
View File

@ -0,0 +1,23 @@
/*
* 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.message;
public interface StreamingDecoder extends ListeningDecoder {
void request(int numMessages);
}

42
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/HttpWriteQueueHandler.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.remoting.http12.netty4;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.EventLoop;
import org.apache.dubbo.remoting.http12.command.HttpWriteQueue;
public class HttpWriteQueueHandler extends ChannelInboundHandlerAdapter {
private HttpWriteQueue writeQueue;
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
EventLoop eventLoop = ctx.channel().eventLoop();
this.writeQueue = new HttpWriteQueue(eventLoop);
}
@Override
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
this.writeQueue = null;
}
public HttpWriteQueue getWriteQueue() {
return writeQueue;
}
}

35
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/NettyHttpChannelFutureListener.java Normal file
View File

@ -0,0 +1,35 @@
/*
* 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.netty4;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import java.util.concurrent.CompletableFuture;
public class NettyHttpChannelFutureListener extends CompletableFuture<Void> implements ChannelFutureListener {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
Throwable cause = future.cause();
this.completeExceptionally(cause);
} else {
this.complete(null);
}
}
}

66
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h1/NettyHttp1Channel.java Normal file
View File

@ -0,0 +1,66 @@
/*
* 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.netty4.h1;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.channel.Channel;
import org.apache.dubbo.remoting.http12.HttpChannel;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
import org.apache.dubbo.remoting.http12.h1.Http1OutputMessage;
import org.apache.dubbo.remoting.http12.netty4.NettyHttpChannelFutureListener;
import java.net.SocketAddress;
import java.util.concurrent.CompletableFuture;
public class NettyHttp1Channel implements HttpChannel {
private final Channel channel;
public NettyHttp1Channel(Channel channel) {
this.channel = channel;
}
@Override
public CompletableFuture<Void> writeHeader(HttpMetadata httpMetadata) {
NettyHttpChannelFutureListener nettyHttpChannelFutureListener = new NettyHttpChannelFutureListener();
this.channel.writeAndFlush(httpMetadata).addListener(nettyHttpChannelFutureListener);
return nettyHttpChannelFutureListener;
}
@Override
public CompletableFuture<Void> writeMessage(HttpOutputMessage httpOutputMessage) {
NettyHttpChannelFutureListener nettyHttpChannelFutureListener = new NettyHttpChannelFutureListener();
this.channel.writeAndFlush(httpOutputMessage).addListener(nettyHttpChannelFutureListener);
return nettyHttpChannelFutureListener;
}
@Override
public HttpOutputMessage newOutputMessage() {
return new Http1OutputMessage(new ByteBufOutputStream(channel.alloc().buffer()));
}
@Override
public SocketAddress remoteAddress() {
return channel.remoteAddress();
}
@Override
public void flush() {
}
}

110
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h1/NettyHttp1Codec.java Normal file
View File

@ -0,0 +1,110 @@
/*
* 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.netty4.h1;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.http.DefaultHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.LastHttpContent;
import org.apache.dubbo.remoting.http12.HttpHeaderNames;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
import org.apache.dubbo.remoting.http12.h1.DefaultHttp1Request;
import org.apache.dubbo.remoting.http12.h1.Http1InputMessage;
import org.apache.dubbo.remoting.http12.h1.Http1Request;
import org.apache.dubbo.remoting.http12.h1.Http1RequestMetadata;
import java.io.OutputStream;
import java.util.List;
import java.util.Map;
public class NettyHttp1Codec extends ChannelDuplexHandler {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//decode FullHttpRequest
if (msg instanceof FullHttpRequest) {
FullHttpRequest fullHttpRequest = (FullHttpRequest) msg;
HttpHeaders headers = fullHttpRequest.headers();
Http1RequestMetadata http1RequestMetadata = new Http1RequestMetadata();
http1RequestMetadata.setPath(fullHttpRequest.uri());
http1RequestMetadata.setMethod(fullHttpRequest.method().name());
org.apache.dubbo.remoting.http12.HttpHeaders httpHeaders = new org.apache.dubbo.remoting.http12.HttpHeaders();
for (Map.Entry<String, String> header : headers) {
String key = header.getKey();
httpHeaders.set(key, header.getValue());
}
http1RequestMetadata.setHeaders(httpHeaders);
Http1Request http1Request = new DefaultHttp1Request(http1RequestMetadata, new Http1InputMessage(new ByteBufInputStream(fullHttpRequest.content(), true)));
super.channelRead(ctx, http1Request);
return;
}
super.channelRead(ctx, msg);
}
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
if (msg instanceof HttpMetadata) {
doWriteHeader(ctx, ((HttpMetadata) msg), promise);
return;
}
if (msg instanceof HttpOutputMessage) {
doWriteMessage(ctx, ((HttpOutputMessage) msg), promise);
return;
}
super.write(ctx, msg, promise);
}
private void doWriteHeader(ChannelHandlerContext ctx, HttpMetadata msg, ChannelPromise promise) {
//process status
List<String> statusHeaders = msg.headers().remove(HttpHeaderNames.STATUS.getName());
HttpResponseStatus status = HttpResponseStatus.OK;
if (!(statusHeaders == null || statusHeaders.isEmpty())) {
status = HttpResponseStatus.valueOf(Integer.parseInt(statusHeaders.get(0)));
}
//process normal headers
DefaultHttpResponse defaultHttpResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);
HttpHeaders headers = defaultHttpResponse.headers();
for (Map.Entry<String, List<String>> entry : msg.headers().entrySet()) {
headers.set(entry.getKey(), entry.getValue());
}
ctx.writeAndFlush(defaultHttpResponse, promise);
}
private void doWriteMessage(ChannelHandlerContext ctx, HttpOutputMessage msg, ChannelPromise promise) {
if (HttpOutputMessage.EMPTY_MESSAGE == msg) {
ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT, promise);
return;
}
OutputStream body = msg.getBody();
if (body instanceof ByteBufOutputStream) {
ByteBuf buffer = ((ByteBufOutputStream) body).buffer();
ctx.writeAndFlush(buffer, promise);
return;
}
throw new IllegalArgumentException("HttpOutputMessage body must be 'io.netty.buffer.ByteBufOutputStream'");
}
}

110
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h1/NettyHttp1ConnectionHandler.java Normal file
View File

@ -0,0 +1,110 @@
/*
* 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.netty4.h1;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.threadpool.ThreadPool;
import org.apache.dubbo.common.utils.Assert;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.http12.HttpHeaderNames;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.exception.UnsupportedMediaTypeException;
import org.apache.dubbo.remoting.http12.h1.Http1Request;
import org.apache.dubbo.remoting.http12.h1.Http1ServerChannelObserver;
import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListener;
import org.apache.dubbo.remoting.http12.h1.Http1ServerTransportListenerFactory;
import org.apache.dubbo.remoting.http12.message.HttpMessageCodec;
import org.apache.dubbo.rpc.model.FrameworkModel;
import java.util.List;
import java.util.concurrent.Executor;
public class NettyHttp1ConnectionHandler extends SimpleChannelInboundHandler<Http1Request> {
private Http1ServerTransportListenerFactory http1ServerTransportListenerFactory;
private final FrameworkModel frameworkModel;
private final URL url;
private final Executor executor;
private Http1ServerChannelObserver errorResponseObserver;
public NettyHttp1ConnectionHandler(URL url, FrameworkModel frameworkModel) {
this.url = url;
this.frameworkModel = frameworkModel;
this.executor = url.getOrDefaultFrameworkModel().getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
}
public NettyHttp1ConnectionHandler(URL url,
FrameworkModel frameworkModel,
Http1ServerTransportListenerFactory http1ServerTransportListenerFactory) {
this.url = url;
this.frameworkModel = frameworkModel;
this.executor = url.getOrDefaultFrameworkModel().getExtensionLoader(ThreadPool.class).getAdaptiveExtension().getExecutor(url);
this.http1ServerTransportListenerFactory = http1ServerTransportListenerFactory;
}
public void setHttp1ServerTransportListenerFactory(Http1ServerTransportListenerFactory http1ServerTransportListenerFactory) {
this.http1ServerTransportListenerFactory = http1ServerTransportListenerFactory;
}
protected void channelRead0(ChannelHandlerContext ctx, Http1Request http1Request) {
//process h1 request
Http1ServerTransportListener http1TransportListener = initTransportListenerIfNecessary(ctx, http1Request);
executor.execute(() -> {
try {
http1TransportListener.onMetadata(http1Request);
http1TransportListener.onData(http1Request);
} catch (Exception e) {
errorResponseObserver.onError(e);
}
});
}
private Http1ServerTransportListener initTransportListenerIfNecessary(ChannelHandlerContext ctx, Http1Request http1Request) {
//each h1 request create http1TransportListener instance
Http1ServerTransportListenerFactory http1ServerTransportListenerFactory = this.http1ServerTransportListenerFactory;
Assert.notNull(http1ServerTransportListenerFactory, "http1ServerTransportListenerFactory must be not null.");
Http1ServerTransportListener http1TransportListener = http1ServerTransportListenerFactory.newInstance(new NettyHttp1Channel(ctx.channel()), url, frameworkModel);
HttpHeaders headers = http1Request.headers();
String contentType = headers.getFirst(HttpHeaderNames.CONTENT_TYPE.getName());
if (!StringUtils.hasText(contentType)) {
throw new UnsupportedMediaTypeException(contentType);
}
HttpMessageCodec codec = findSuitableCodec(contentType, frameworkModel.getExtensionLoader(HttpMessageCodec.class).getActivateExtensions());
if (codec == null) {
throw new UnsupportedMediaTypeException(contentType);
}
this.errorResponseObserver = new Http1ServerChannelObserver(new NettyHttp1Channel(ctx.channel()));
this.errorResponseObserver.setHttpMessageCodec(codec);
return http1TransportListener;
}
private static HttpMessageCodec findSuitableCodec(String contentType, List<HttpMessageCodec> candidates) {
for (HttpMessageCodec codec : candidates) {
if (codec.support(contentType)) {
return codec;
}
}
return null;
}
}

80
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h2/NettyH2StreamChannel.java Normal file
View File

@ -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.remoting.http12.netty4.h2;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.handler.codec.http2.DefaultHttp2ResetFrame;
import io.netty.handler.codec.http2.Http2StreamChannel;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.HttpOutputMessage;
import org.apache.dubbo.remoting.http12.h2.H2StreamChannel;
import org.apache.dubbo.remoting.http12.h2.Http2OutputMessage;
import org.apache.dubbo.remoting.http12.h2.Http2OutputMessageFrame;
import org.apache.dubbo.remoting.http12.netty4.NettyHttpChannelFutureListener;
import java.net.SocketAddress;
import java.util.concurrent.CompletableFuture;
public class NettyH2StreamChannel implements H2StreamChannel {
private final Http2StreamChannel http2StreamChannel;
public NettyH2StreamChannel(Http2StreamChannel http2StreamChannel) {
this.http2StreamChannel = http2StreamChannel;
}
@Override
public CompletableFuture<Void> writeHeader(HttpMetadata httpMetadata) {
//WriteQueue.enqueue header frame
NettyHttpChannelFutureListener nettyHttpChannelFutureListener = new NettyHttpChannelFutureListener();
http2StreamChannel.write(httpMetadata).addListener(nettyHttpChannelFutureListener);
return nettyHttpChannelFutureListener;
}
@Override
public CompletableFuture<Void> writeMessage(HttpOutputMessage httpOutputMessage) {
NettyHttpChannelFutureListener nettyHttpChannelFutureListener = new NettyHttpChannelFutureListener();
http2StreamChannel.write(httpOutputMessage).addListener(nettyHttpChannelFutureListener);
return nettyHttpChannelFutureListener;
}
@Override
public Http2OutputMessage newOutputMessage(boolean endStream) {
ByteBuf buffer = http2StreamChannel.alloc().buffer();
ByteBufOutputStream outputStream = new ByteBufOutputStream(buffer);
return new Http2OutputMessageFrame(outputStream, endStream);
}
@Override
public SocketAddress remoteAddress() {
return this.http2StreamChannel.remoteAddress();
}
@Override
public void flush() {
this.http2StreamChannel.flush();
}
@Override
public CompletableFuture<Void> writeResetFrame(long errorCode) {
DefaultHttp2ResetFrame resetFrame = new DefaultHttp2ResetFrame(errorCode);
NettyHttpChannelFutureListener nettyHttpChannelFutureListener = new NettyHttpChannelFutureListener();
http2StreamChannel.write(resetFrame).addListener(nettyHttpChannelFutureListener);
return nettyHttpChannelFutureListener;
}
}

113
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h2/NettyHttp2FrameCodec.java Normal file
View File

@ -0,0 +1,113 @@
/*
* 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.netty4.h2;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufInputStream;
import io.netty.buffer.ByteBufOutputStream;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
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.Http2DataFrame;
import io.netty.handler.codec.http2.Http2Headers;
import io.netty.handler.codec.http2.Http2HeadersFrame;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.h2.Http2Header;
import org.apache.dubbo.remoting.http12.h2.Http2InputMessage;
import org.apache.dubbo.remoting.http12.h2.Http2InputMessageFrame;
import org.apache.dubbo.remoting.http12.h2.Http2MetadataFrame;
import org.apache.dubbo.remoting.http12.h2.Http2OutputMessage;
import java.io.IOException;
import java.io.OutputStream;
import java.util.List;
import java.util.Map;
public class NettyHttp2FrameCodec extends ChannelDuplexHandler {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof Http2HeadersFrame) {
Http2Header http2Header = onHttp2HeadersFrame(((Http2HeadersFrame) msg));
super.channelRead(ctx, http2Header);
} else if (msg instanceof Http2DataFrame) {
Http2InputMessage http2Message = onHttp2DataFrame(((Http2DataFrame) msg));
super.channelRead(ctx, http2Message);
} else {
super.channelRead(ctx, msg);
}
}
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
if (msg instanceof Http2Header) {
Http2Header http2Header = (Http2Header) msg;
Http2HeadersFrame http2HeadersFrame = encodeHttp2HeadersFrame(http2Header);
super.write(ctx, http2HeadersFrame, promise);
} else if (msg instanceof Http2OutputMessage) {
Http2OutputMessage http2OutputMessage = (Http2OutputMessage) msg;
Http2DataFrame http2DataFrame = encodeHttp2DataFrame(ctx, http2OutputMessage);
super.write(ctx, http2DataFrame, promise);
} else {
super.write(ctx, msg, promise);
}
}
private Http2Header onHttp2HeadersFrame(Http2HeadersFrame headersFrame) {
Http2Headers headers = headersFrame.headers();
boolean endStream = headersFrame.isEndStream();
HttpHeaders head = new HttpHeaders();
for (Map.Entry<CharSequence, CharSequence> header : headers) {
head.set(header.getKey().toString(), header.getValue().toString());
}
return new Http2MetadataFrame(headersFrame.stream().id(), head, endStream);
}
private Http2InputMessage onHttp2DataFrame(Http2DataFrame dataFrame) {
ByteBuf content = dataFrame.content();
Http2InputMessageFrame message = new Http2InputMessageFrame(new ByteBufInputStream(content, true), dataFrame.isEndStream());
message.setId(dataFrame.stream().id());
return message;
}
private Http2HeadersFrame encodeHttp2HeadersFrame(Http2Header http2Header) {
HttpHeaders headers = http2Header.headers();
DefaultHttp2Headers http2Headers = new DefaultHttp2Headers();
for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
String name = entry.getKey();
List<String> value = entry.getValue();
http2Headers.set(name, value);
}
return new DefaultHttp2HeadersFrame(http2Headers, http2Header.isEndStream());
}
private Http2DataFrame encodeHttp2DataFrame(ChannelHandlerContext ctx, Http2OutputMessage outputMessage) throws IOException {
OutputStream body = outputMessage.getBody();
if (body == null) {
return new DefaultHttp2DataFrame(outputMessage.isEndStream());
}
if (body instanceof ByteBufOutputStream) {
ByteBuf buffer = ((ByteBufOutputStream) body).buffer();
return new DefaultHttp2DataFrame(buffer, outputMessage.isEndStream());
}
throw new IllegalArgumentException("Http2OutputMessage body must be ByteBufOutputStream");
}
}

81
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h2/NettyHttp2FrameHandler.java Normal file
View File

@ -0,0 +1,81 @@
/*
* 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.netty4.h2;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http2.Http2ResetFrame;
import org.apache.dubbo.common.logger.ErrorTypeAwareLogger;
import org.apache.dubbo.common.logger.LoggerFactory;
import org.apache.dubbo.remoting.http12.HttpStatus;
import org.apache.dubbo.remoting.http12.exception.HttpStatusException;
import org.apache.dubbo.remoting.http12.h2.H2StreamChannel;
import org.apache.dubbo.remoting.http12.h2.Http2Header;
import org.apache.dubbo.remoting.http12.h2.Http2InputMessage;
import org.apache.dubbo.remoting.http12.h2.Http2TransportListener;
import static org.apache.dubbo.common.constants.LoggerCodeConstants.PROTOCOL_FAILED_RESPONSE;
public class NettyHttp2FrameHandler extends ChannelDuplexHandler {
private static final ErrorTypeAwareLogger LOGGER = LoggerFactory.getErrorTypeAwareLogger(
NettyHttp2FrameHandler.class);
private final H2StreamChannel h2StreamChannel;
private final Http2TransportListener transportListener;
public NettyHttp2FrameHandler(H2StreamChannel h2StreamChannel, Http2TransportListener transportListener) {
this.h2StreamChannel = h2StreamChannel;
this.transportListener = transportListener;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof Http2Header) {
transportListener.onMetadata((Http2Header) msg);
} else if (msg instanceof Http2InputMessage) {
transportListener.onData((Http2InputMessage) msg);
} else {
super.channelRead(ctx, msg);
}
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
//reset frame
if (evt instanceof Http2ResetFrame) {
long errorCode = ((Http2ResetFrame) evt).errorCode();
transportListener.cancelByRemote(errorCode);
} else {
super.userEventTriggered(ctx, evt);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
if (LOGGER.isWarnEnabled()) {
LOGGER.warn(PROTOCOL_FAILED_RESPONSE, "", "", "Exception in processing triple message", cause);
}
int statusCode = HttpStatus.INTERNAL_SERVER_ERROR.getCode();
if (cause instanceof HttpStatusException) {
statusCode = ((HttpStatusException) cause).getStatusCode();
}
h2StreamChannel.writeResetFrame(statusCode);
}
}

90
dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/dubbo/remoting/http12/netty4/h2/NettyHttp2ProtocolSelectorHandler.java Normal file
View File

@ -0,0 +1,90 @@
/*
* 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.netty4.h2;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http2.Http2StreamChannel;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.utils.StringUtils;
import org.apache.dubbo.remoting.http12.HttpHeaderNames;
import org.apache.dubbo.remoting.http12.HttpHeaders;
import org.apache.dubbo.remoting.http12.HttpMetadata;
import org.apache.dubbo.remoting.http12.command.HttpWriteQueue;
import org.apache.dubbo.remoting.http12.exception.UnsupportedMediaTypeException;
import org.apache.dubbo.remoting.http12.h2.H2StreamChannel;
import org.apache.dubbo.remoting.http12.h2.Http2ServerTransportListenerFactory;
import org.apache.dubbo.remoting.http12.h2.command.Http2WriteQueueChannel;
import org.apache.dubbo.remoting.http12.netty4.HttpWriteQueueHandler;
import org.apache.dubbo.rpc.model.FrameworkModel;
import java.util.Set;
public class NettyHttp2ProtocolSelectorHandler extends SimpleChannelInboundHandler<HttpMetadata> {
private Http2ServerTransportListenerFactory defaultHttp2ServerTransportListenerFactory;
private final URL url;
private final FrameworkModel frameworkModel;
public NettyHttp2ProtocolSelectorHandler(URL url, FrameworkModel frameworkModel) {
this.url = url;
this.frameworkModel = frameworkModel;
}
public NettyHttp2ProtocolSelectorHandler(URL url, FrameworkModel frameworkModel, Http2ServerTransportListenerFactory defaultHttp2ServerTransportListenerFactory) {
this.url = url;
this.frameworkModel = frameworkModel;
this.defaultHttp2ServerTransportListenerFactory = defaultHttp2ServerTransportListenerFactory;
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, HttpMetadata metadata) {
HttpHeaders headers = metadata.headers();
String contentType = headers.getFirst(HttpHeaderNames.CONTENT_TYPE.getName());
//415
if (!StringUtils.hasText(contentType)) {
throw new UnsupportedMediaTypeException(contentType);
}
Http2ServerTransportListenerFactory factory = adaptHttp2ServerTransportListenerFactory(contentType);
if (factory == null) {
throw new UnsupportedMediaTypeException(contentType);
}
H2StreamChannel h2StreamChannel = new NettyH2StreamChannel((Http2StreamChannel) ctx.channel());
HttpWriteQueueHandler writeQueueHandler = ctx.channel().parent().pipeline().get(HttpWriteQueueHandler.class);
if (writeQueueHandler != null) {
HttpWriteQueue writeQueue = writeQueueHandler.getWriteQueue();
h2StreamChannel = new Http2WriteQueueChannel(h2StreamChannel, writeQueue);
}
ChannelPipeline pipeline = ctx.pipeline();
pipeline.addLast(new NettyHttp2FrameHandler(h2StreamChannel, factory.newInstance(h2StreamChannel, url, frameworkModel)));
pipeline.remove(this);
ctx.fireChannelRead(metadata);
}
private Http2ServerTransportListenerFactory adaptHttp2ServerTransportListenerFactory(String contentType) {
Set<Http2ServerTransportListenerFactory> http2ServerTransportListenerFactories = frameworkModel.getExtensionLoader(Http2ServerTransportListenerFactory.class).getSupportedExtensionInstances();
for (Http2ServerTransportListenerFactory factory : http2ServerTransportListenerFactories) {
if (factory.supportContentType(contentType)) {
return factory;
}
}
return defaultHttp2ServerTransportListenerFactory;
}
}

1
dubbo-remoting/dubbo-remoting-http12/src/main/resources/META-INF/dubbo/internal/org.apache.dubbo.remoting.http12.message.HttpMessageCodec Normal file
View File

@ -0,0 +1 @@
json=org.apache.dubbo.remoting.http12.message.JsonCodec

1
dubbo-remoting/pom.xml
View File

@ -38,6 +38,7 @@
<module>dubbo-remoting-zookeeper</module>
<module>dubbo-remoting-zookeeper-curator5</module>
<module>dubbo-remoting-netty4</module>
<module>dubbo-remoting-http12</module>
</modules>
<dependencies>