package com.xxl.rpc.remoting.net.common;

import com.xxl.rpc.remoting.invoker.XxlRpcInvokerFactory;
import com.xxl.rpc.remoting.invoker.reference.XxlRpcReferenceBean;
import com.xxl.rpc.remoting.net.params.BaseCallback;
import com.xxl.rpc.remoting.net.params.XxlRpcRequest;
import com.xxl.rpc.serialize.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author xuxueli 2018-10-19
 */
public abstract class ConnectClient {
    protected static transient Logger logger = LoggerFactory.getLogger(ConnectClient.class);


    // ---------------------- iface ----------------------

    public abstract void init(String address, final Serializer serializer, final XxlRpcInvokerFactory xxlRpcInvokerFactory) throws Exception;

    public abstract void close();

    public abstract boolean isValidate();

    public abstract void send(XxlRpcRequest xxlRpcRequest) throws Exception ;


    // ---------------------- client pool map ----------------------

    /**
     * async send
     */
    public static void asyncSend(XxlRpcRequest xxlRpcRequest, String address,
                                 Class<? extends ConnectClient> connectClientImpl,
                                 final XxlRpcReferenceBean xxlRpcReferenceBean) throws Exception {

        // client pool	[tips03 : may save 35ms/100invoke if move it to constructor, but it is necessary. cause by ConcurrentHashMap.get]
        ConnectClient clientPool = ConnectClient.getPool(address, connectClientImpl, xxlRpcReferenceBean);

        try {
            // do invoke
            clientPool.send(xxlRpcRequest);
        } catch (Exception e) {
            throw e;
        }

    }

    private static ConcurrentHashMap<String, ConnectClient> connectClientMap;        // (static) alread addStopCallBack
    private static ConnectClient getPool(String address, Class<? extends ConnectClient> connectClientImpl,
                                         final XxlRpcReferenceBean xxlRpcReferenceBean) throws Exception {

        // init client-pool-map, avoid repeat init
        if (connectClientMap == null) {
            synchronized (ConnectClient.class) {
                if (connectClientMap == null) {
                    // init
                    connectClientMap = new ConcurrentHashMap<String, ConnectClient>();
                    // stop callback
                    xxlRpcReferenceBean.getInvokerFactory().addStopCallBack(new BaseCallback() {
                        @Override
                        public void run() throws Exception {
                            if (connectClientMap.size() > 0) {
                                for (String key: connectClientMap.keySet()) {
                                    ConnectClient clientPool = connectClientMap.get(key);
                                    clientPool.close();
                                }
                                connectClientMap.clear();
                            }
                        }
                    });
                }
            }
        }

        // get pool
        ConnectClient connectClient = connectClientMap.get(address);

        if (connectClient!=null && !connectClient.isValidate()) {
            connectClient.close();
            connectClient = null;
            connectClientMap.remove(address);
        }

        if (connectClient != null) {
            return connectClient;
        }

        // make pool, avoid repeat init
        synchronized (connectClientMap) {

            // re-get pool
            connectClient = connectClientMap.get(address);
            if (connectClient != null) {
                return connectClient;
            }

            // set pool
            ConnectClient connectClient_new = connectClientImpl.newInstance();
            connectClient_new.init(address, xxlRpcReferenceBean.getSerializer(), xxlRpcReferenceBean.getInvokerFactory());
            connectClientMap.put(address, connectClient_new);

            return connectClient_new;
        }

    }

}

@@ -1729,16 +1729,25 @@
  static 
+volatile 
 Concurre
@@ -1835,24 +1835,136 @@
 topCallBack%0A
+    private static volatile ConcurrentHashMap%3CString, Object%3E connectClientLockMap = new ConcurrentHashMap%3C%3E();%0A
     private 
@@ -2186,23 +2186,20 @@
 nit 
-client-pool-map
+base compont
 , av
@@ -3211,21 +3211,29 @@
   // get
- pool
+-valid client
 %0A       
@@ -3294,17 +3294,16 @@
 dress);%0A
-%0A
         
@@ -3329,17 +3329,16 @@
 null && 
-!
 connectC
@@ -3363,32 +3363,39 @@
 ) %7B%0A            
+return 
 connectClient.cl
@@ -3395,63 +3395,64 @@
 ient
-.close()
 ;%0A     
+   %7D%0A%0A
        
-connectClient = null;%0A           
+ // lock%0A        Object clientLock =
  con
@@ -3465,18 +3465,19 @@
 ient
-Map.remove
+LockMap.get
 (add
@@ -3491,27 +3491,51 @@
         
-%7D%0A%0A
+if (clientLock == null) %7B%0A 
         
 if (conn
@@ -3518,36 +3518,35 @@
 ull) %7B%0A         
-if (
+   
 connectClient !=
@@ -3538,35 +3538,67 @@
 onnectClient
- != null) %7B
+LockMap.putIfAbsent(address, new Object());
 %0A           
@@ -3590,38 +3590,44 @@
 ));%0A            
-return
+clientLock =
  connectClient;%0A
@@ -3616,32 +3616,52 @@
  = connectClient
+LockMap.get(address)
 ;%0A        %7D%0A%0A   
@@ -3672,35 +3672,31 @@
  // 
-make pool, avoid repeat ini
+remove-create new clien
 t%0A  
@@ -3716,31 +3716,25 @@
 nized (c
-onnectC
 lient
-Map
+Lock
 ) %7B%0A%0A   
@@ -3749,19 +3749,38 @@
  // 
-re-get pool
+get-valid client, avlid repeat
 %0A   
@@ -3868,16 +3868,44 @@
 ient
- 
 !=
- 
 null
+ && connectClient.isValidate()
 ) %7B%0A
@@ -3949,32 +3949,203 @@
 %0A            %7D%0A%0A
+            // remove old%0A            if (connectClient != null) %7B%0A                connectClient.close();%0A                connectClientMap.remove(address);%0A            %7D%0A%0A
             // s

connect client optimize