Merge pull request #745 from zongtanghu/snode

[ISSUE #743]adjust some codes of supporting Http2 for rocketmq.

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

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,

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;

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;