Merge pull request #18 from sunquakes/development

Replace httpclient with nettyhttpclient.
sunquakes · Oct 16, 2024 · 8f11d35 · 8f11d35
2 parents 952b2d0 + 081143c
commit 8f11d35
Show file tree

Hide file tree

Showing 13 changed files with 479 additions and 163 deletions.
diff --git a/pom.xml b/pom.xml
@@ -17,13 +17,11 @@
         <maven.compiler.target>17</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <spring.version>6.1.11</spring.version>
-        <httpclient.version>4.5.14</httpclient.version>
         <fastjson.version>2.0.52</fastjson.version>
         <lombok.version>1.18.34</lombok.version>
         <slf4j.version>2.0.13</slf4j.version>
         <spring-boot.version>3.3.2</spring-boot.version>
         <netty.version>4.1.112.Final</netty.version>
-        <consul.version>1.4.5</consul.version>
         <jupiter.version>5.10.3</jupiter.version>
         <mockito.version>5.2.0</mockito.version>
         <javax.version>1.3.2</javax.version>
@@ -56,11 +54,6 @@
             <artifactId>spring-web</artifactId>
             <version>${spring.version}</version>
         </dependency>
-        <dependency>
-            <groupId>org.apache.httpcomponents</groupId>
-            <artifactId>httpclient</artifactId>
-            <version>${httpclient.version}</version>
-        </dependency>
         <dependency>
             <groupId>com.alibaba.fastjson2</groupId>
             <artifactId>fastjson2</artifactId>
@@ -87,11 +80,6 @@
             <artifactId>netty-all</artifactId>
             <version>${netty.version}</version>
         </dependency>
-        <dependency>
-            <groupId>com.ecwid.consul</groupId>
-            <artifactId>consul-api</artifactId>
-            <version>${consul.version}</version>
-        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>

diff --git a/src/main/java/com/sunquakes/jsonrpc4j/client/JsonRpcClientHandler.java b/src/main/java/com/sunquakes/jsonrpc4j/client/JsonRpcClientHandler.java
@@ -24,7 +24,7 @@
 @Sharable
 public abstract class JsonRpcClientHandler extends ChannelInboundHandlerAdapter {
 
-    protected Map<Channel, Promise> promiseMap = new ConcurrentHashMap<>();
+    protected Map<Channel, Promise<String>> promiseMap = new ConcurrentHashMap<>();
 
     @Override
     public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
@@ -39,7 +39,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
     protected void handleInternalError(ChannelHandlerContext ctx) {
         Channel channel = ctx.channel();
         ErrorResponseDto errorResponseDto = new ErrorResponseDto(null, RequestUtils.JSONRPC, new ErrorDto(ErrorEnum.INTERNAL_ERROR.getCode(), ErrorEnum.INTERNAL_ERROR.getText(), null));
-        Promise promise = promiseMap.get(channel);
+        Promise<String> promise = promiseMap.get(channel);
         if (promise != null) {
             promise.setSuccess(JSON.toJSONString(errorResponseDto));
             promiseMap.remove(channel);

diff --git a/src/main/java/com/sunquakes/jsonrpc4j/client/JsonRpcHttpClient.java b/src/main/java/com/sunquakes/jsonrpc4j/client/JsonRpcHttpClient.java
@@ -16,10 +16,10 @@
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.pool.FixedChannelPool;
+import io.netty.channel.socket.SocketChannel;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.HttpClientCodec;
 import io.netty.handler.codec.http.HttpObjectAggregator;
-import io.netty.handler.ssl.OptionalSslHandler;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
@@ -54,7 +54,7 @@ public void initLoadBalancer() {
         bootstrap.group(eventLoopGroup)
                 .channel(NioSocketChannel.class)
                 .option(ChannelOption.SO_KEEPALIVE, true);
-        int defaultPort = protocol.equals(JsonRpcProtocol.HTTPS.name()) ? DEFAULT_HTTPS_PORT : DEFAULT_HTTP_PORT;
+        int defaultPort = isHttps(protocol) ? DEFAULT_HTTPS_PORT : DEFAULT_HTTP_PORT;
         if (discovery != null) {
             loadBalancer = new JsonRpcLoadBalancer(() -> discovery.value().get(name), defaultPort, bootstrap, poolHandler);
         } else {
@@ -97,15 +97,19 @@ public Object handle(String method, Object[] args) throws JsonRpcException {
     class Handler implements JsonRpcChannelHandler {
         @Override
         public void channelUpdated(Channel ch) throws SSLException {
-            ch.pipeline()
-                    .addLast("codec", new HttpClientCodec())
-                    .addLast("http-aggregator", new HttpObjectAggregator(1024 * 1024))
-                    .addLast(jsonRpcHttpClientHandler);
-            protocol = protocol.toUpperCase();
-            if (protocol.equals(JsonRpcProtocol.HTTPS.name())) {
+            SocketChannel sc = (SocketChannel) ch;
+            if (isHttps(protocol)) {
                 SslContext sslContext = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
-                ch.pipeline().addLast(new OptionalSslHandler(sslContext));
+                ch.pipeline().addLast(sslContext.newHandler(ch.alloc()));
             }
+            sc.pipeline()
+                    .addLast("codec", new HttpClientCodec())
+                    .addLast("http-aggregator", new HttpObjectAggregator(1024 * 1024));
+            sc.pipeline().addLast(jsonRpcHttpClientHandler);
         }
     }
+
+    private boolean isHttps(String scheme) {
+        return scheme.toUpperCase().equals(JsonRpcProtocol.HTTPS.name());
+    }
 }
diff --git a/src/main/java/com/sunquakes/jsonrpc4j/client/JsonRpcLoadBalancer.java b/src/main/java/com/sunquakes/jsonrpc4j/client/JsonRpcLoadBalancer.java
@@ -7,10 +7,10 @@
 import org.springframework.util.StringUtils;
 
 import java.net.InetSocketAddress;
+import java.security.SecureRandom;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.ThreadLocalRandom;
 import java.util.function.Supplier;
 
 /**
@@ -34,6 +34,8 @@ public class JsonRpcLoadBalancer {
 
     private static final int MAX_RETRY_TIMES = 3;
 
+    private static final SecureRandom secureRandom = new SecureRandom();
+
     public JsonRpcLoadBalancer(Supplier<String> url, int defaultPort, Bootstrap bootstrap, JsonRpcChannelPoolHandler poolHandler) {
         this.bootstrap = bootstrap;
         this.poolHandler = poolHandler;
@@ -68,7 +70,8 @@ public FixedChannelPool getPool() {
             initPools();
             return getPool();
         } else {
-            int index = ThreadLocalRandom.current().nextInt(pools.size());
+            secureRandom.nextBytes(new byte[8]);
+            int index = secureRandom.nextInt(pools.size());
             pool = pools.get(index);
         }
         return pool;

diff --git a/src/main/java/com/sunquakes/jsonrpc4j/client/JsonRpcTcpClientHandler.java b/src/main/java/com/sunquakes/jsonrpc4j/client/JsonRpcTcpClientHandler.java
@@ -84,7 +84,7 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
             if (bytes.length > 0) {
                 String body = new String(bytes);
                 synchronized (channel) {
-                    Promise promise = promiseMap.get(channel);
+                    Promise<String> promise = promiseMap.get(channel);
                     if (promise != null) {
                         promise.setSuccess(body);
                     }