Use CHMV8 fork until we target JDK8, close AsyncHttpClient#886

Author: slandelle

src/main/java/com/ning/http/client/providers/netty/channel/pool/DefaultChannelPool.java

@@ -25,26 +25,33 @@
 
 import com.ning.http.client.AsyncHttpClientConfig;
 import com.ning.http.client.providers.netty.channel.Channels;
+import com.ning.http.client.providers.netty.chmv8.ConcurrentHashMapV8;
 import com.ning.http.client.providers.netty.future.NettyResponseFuture;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
- * A simple implementation of {@link com.ning.http.client.providers.netty.channel.pool.ChannelPool} based on a {@link java.util.concurrent.ConcurrentHashMap}
+ * A simple implementation of {@link com.ning.http.client.providers.netty.channel.pool.ChannelPool} based on a {@link com.ning.http.client.providers.netty.chmv8.ConcurrentHashMapV8}
  */
 public final class DefaultChannelPool implements ChannelPool {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DefaultChannelPool.class);
 
-    private final ConcurrentHashMap<String, ConcurrentLinkedQueue<IdleChannel>> partitions = new ConcurrentHashMap<>();
-    private final ConcurrentHashMap<Integer, ChannelCreation> channelId2Creation = new ConcurrentHashMap<>();
+    private static final ConcurrentHashMapV8.Fun<String, ConcurrentLinkedQueue<IdleChannel>> PARTITION_COMPUTER = new ConcurrentHashMapV8.Fun<String, ConcurrentLinkedQueue<IdleChannel>>() {
+        @Override
+        public ConcurrentLinkedQueue<IdleChannel> apply(String partitionId) {
+            return new ConcurrentLinkedQueue<>();
+        }
+    };
+    
+    private final ConcurrentHashMapV8<String, ConcurrentLinkedQueue<IdleChannel>> partitions = new ConcurrentHashMapV8<>();
+    private final ConcurrentHashMapV8<Integer, ChannelCreation> channelId2Creation = new ConcurrentHashMapV8<>();
     private final AtomicBoolean isClosed = new AtomicBoolean(false);
     private final Timer nettyTimer;
     private final boolean sslConnectionPoolEnabled;
@@ -228,19 +235,7 @@ public void run(Timeout timeout) throws Exception {
         }
     }
 
-    private ConcurrentLinkedQueue<IdleChannel> getPartition(String partitionId) {
-        ConcurrentLinkedQueue<IdleChannel> partition = partitions.get(partitionId);
-        if (partition == null) {
-            // lazy init partition
-            ConcurrentLinkedQueue<IdleChannel> newPartition = new ConcurrentLinkedQueue<>();
-            partition = partitions.putIfAbsent(partitionId, newPartition);
-            if (partition == null)
-                partition = newPartition;
-        }
-        return partition;
-    }
-    
-    public boolean offer(Channel channel, String partition) {
+    public boolean offer(Channel channel, String partitionId) {
         if (isClosed.get() || (!sslConnectionPoolEnabled && channel.getPipeline().get(SslHandler.class) != null))
             return false;
 
@@ -249,9 +244,9 @@ public boolean offer(Channel channel, String partition) {
         if (isTTLExpired(channel, now))
             return false;
 
-        boolean added = getPartition(partition).add(new IdleChannel(channel, now));
+        boolean added = partitions.computeIfAbsent(partitionId, PARTITION_COMPUTER).add(new IdleChannel(channel, now));
         if (added)
-            channelId2Creation.putIfAbsent(channel.getId(), new ChannelCreation(now, partition));
+            channelId2Creation.putIfAbsent(channel.getId(), new ChannelCreation(now, partitionId));
 
         return added;
     }