You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by du...@apache.org on 2019/02/01 03:39:14 UTC
[rocketmq] branch snode updated: [ISSUE #743]adjust some codes of
supporting Http2 for rocketmq.
This is an automated email from the ASF dual-hosted git repository.
duhengforever pushed a commit to branch snode
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/snode by this push:
new 7268076 [ISSUE #743]adjust some codes of supporting Http2 for rocketmq.
new 81b4293 Merge pull request #745 from zongtanghu/snode
7268076 is described below
commit 72680765294608c586270e162d3b33accfca3e67
Author: huzongtang <hu...@cmss.chinamobile.com>
AuthorDate: Fri Feb 1 10:50:59 2019 +0800
[ISSUE #743]adjust some codes of supporting Http2 for rocketmq.
---
.../remoting/transport/http2/Http2ClientImpl.java | 16 ++++-------
.../remoting/transport/http2/Http2Handler.java | 15 +++++-----
.../remoting/transport/http2/Http2ServerImpl.java | 4 +--
.../remoting/RemotingHttp2InvokingTest.java | 32 ++++++++++++++++++++++
4 files changed, 48 insertions(+), 19 deletions(-)
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java
index 761d2bf..2946d85 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ClientImpl.java
@@ -89,11 +89,11 @@ public class Http2ClientImpl extends NettyRemotingClientAbstract implements Remo
10000, "Remoting-PublicExecutor", true);
this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(clientConfig.getClientWorkerThreads(),
ThreadUtils.newGenericThreadFactory("NettyClientWorkerThreads", clientConfig.getClientWorkerThreads()));
- buildSslContext();
+ buildHttp2SslClientContext();
return this;
}
- private void buildSslContext() {
+ private void buildHttp2SslClientContext() {
SslProvider provider = OpenSsl.isAvailable() ? SslProvider.OPENSSL : SslProvider.JDK;
try {
sslContext = SslContextBuilder.forClient()
@@ -104,7 +104,7 @@ public class Http2ClientImpl extends NettyRemotingClientAbstract implements Remo
.trustManager(InsecureTrustManagerFactory.INSTANCE)
.build();
} catch (SSLException e) {
- e.printStackTrace();
+ log.error("Can not build Http2 SSL Client context !", e);
}
}
@@ -147,16 +147,12 @@ public class Http2ClientImpl extends NettyRemotingClientAbstract implements Remo
if (this.defaultEventExecutorGroup != null) {
this.defaultEventExecutorGroup.shutdownGracefully();
}
- } catch (Exception e) {
- log.error("Http2ClientImpl shutdown exception, ", e);
- }
- if (this.publicExecutor != null) {
- try {
+ if (this.publicExecutor != null) {
this.publicExecutor.shutdown();
- } catch (Exception e) {
- log.error("NettyRemotingServer shutdown exception, ", e);
}
+ } catch (Exception e) {
+ log.error("Http2ClientImpl shutdown exception, ", e);
}
}
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2Handler.java b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2Handler.java
index 2f4c8a1..9197be1 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2Handler.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2Handler.java
@@ -78,12 +78,11 @@ public class Http2Handler extends Http2ConnectionHandler {
frameReader);
Http2Settings settings = new Http2Settings();
-
if (!isServer) {
settings.pushEnabled(true);
}
-
- settings.initialWindowSize(1048576 * 10); //10MiB
+ //10MiB
+ settings.initialWindowSize(10485760);
settings.maxConcurrentStreams(Integer.MAX_VALUE);
return newHandler(decoder, encoder, settings, isServer);
@@ -98,8 +97,10 @@ public class Http2Handler extends Http2ConnectionHandler {
public void write(final ChannelHandlerContext ctx, final Object msg,
final ChannelPromise promise) throws Exception {
if (isServer) {
- assert msg instanceof ByteBuf;
- sendAPushPromise(ctx, lastStreamId, lastStreamId + 1, (ByteBuf) msg);
+ if (!(msg instanceof ByteBuf)) {
+ return;
+ }
+ sendAndPushPromise(ctx, lastStreamId, lastStreamId + 1, (ByteBuf) msg);
} else {
final Http2Headers headers = new DefaultHttp2Headers();
@@ -111,13 +112,13 @@ public class Http2Handler extends Http2ConnectionHandler {
encoder().writeData(ctx, (int) streamId, (ByteBuf) msg, 0, false, ctx.newPromise());
ctx.flush();
} catch (Exception e) {
- e.printStackTrace();
+ log.error("Http2Handler write data exception,", e);
}
}
}
- private void sendAPushPromise(ChannelHandlerContext ctx, int streamId, int pushPromiseStreamId,
+ private void sendAndPushPromise(ChannelHandlerContext ctx, int streamId, int pushPromiseStreamId,
ByteBuf payload) throws Http2Exception {
encoder().writePushPromise(ctx, streamId, pushPromiseStreamId,
diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java
index 5f95edf..fbf60d5 100644
--- a/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java
+++ b/remoting/src/main/java/org/apache/rocketmq/remoting/transport/http2/Http2ServerImpl.java
@@ -115,11 +115,11 @@ public class Http2ServerImpl extends NettyRemotingServerAbstract implements Remo
this.workerGroup = new DefaultEventExecutorGroup(serverConfig.getServerWorkerThreads(),
ThreadUtils.newGenericThreadFactory("NettyWorkerThreads", serverConfig.getServerWorkerThreads()));
this.port = nettyServerConfig.getListenPort();
- buildHttp2SslContext();
+ buildHttp2SslServerContext();
return this;
}
- private void buildHttp2SslContext() {
+ private void buildHttp2SslServerContext() {
try {
SslProvider provider = OpenSsl.isAvailable() ? SslProvider.OPENSSL : SslProvider.JDK;
SelfSignedCertificate ssc;
diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/RemotingHttp2InvokingTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/RemotingHttp2InvokingTest.java
index 3e1c04f..881c5f3 100644
--- a/remoting/src/test/java/org/apache/rocketmq/remoting/RemotingHttp2InvokingTest.java
+++ b/remoting/src/test/java/org/apache/rocketmq/remoting/RemotingHttp2InvokingTest.java
@@ -17,13 +17,17 @@
package org.apache.rocketmq.remoting;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.remoting.annotation.CFNullable;
import org.apache.rocketmq.remoting.common.RemotingUtil;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.serialize.LanguageCode;
import org.junit.After;
@@ -90,6 +94,34 @@ public class RemotingHttp2InvokingTest {
}
+ @Test
+ public void testInvokeOneway() throws InterruptedException, RemotingConnectException,
+ RemotingTimeoutException, RemotingTooMuchRequestException, RemotingSendRequestException {
+
+ RemotingCommand request = RemotingCommand.createRequestCommand(0, null);
+ request.setRemark("messi");
+ remotingHttp2Client.invokeOneway("localhost:8888", request, 1000 * 3);
+ }
+
+ @Test
+ public void testInvokeAsync() throws InterruptedException, RemotingConnectException,
+ RemotingTimeoutException, RemotingTooMuchRequestException, RemotingSendRequestException {
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ RemotingCommand request = RemotingCommand.createRequestCommand(0, null);
+ request.setRemark("messi");
+ remotingHttp2Client.invokeAsync("localhost:8888", request, 1000 * 3, new InvokeCallback() {
+ @Override
+ public void operationComplete(ResponseFuture responseFuture) {
+ latch.countDown();
+ assertTrue(responseFuture != null);
+ assertThat(responseFuture.getResponseCommand().getLanguage()).isEqualTo(LanguageCode.JAVA);
+ assertThat(responseFuture.getResponseCommand().getExtFields()).hasSize(2);
+ }
+ });
+ latch.await(3000, TimeUnit.SECONDS);
+ }
+
class Http2RequestHeader implements CommandCustomHeader {
@CFNullable
private Integer count;