Merge branch 'master' into rxsocks

rxlib-x/src/main/java/org/rx/jdbc/JdbcUtil.java

@@ -205,6 +205,16 @@ public static void print(ResultSet resultSet) {
         }
     }
 
+    @SneakyThrows
+    public static <T> T executeScalar(ResultSet resultSet) {
+        try (ResultSet rs = resultSet) {
+            if (rs.next()) {
+                return (T) rs.getObject(1);
+            }
+            return null;
+        }
+    }
+
     public static final BiFunc<String, String> TO_CAMEL_COLUMN_MAPPING = p -> CaseFormat.LOWER_UNDERSCORE.to(CaseFormat.LOWER_CAMEL, p);
 
     public static <T> List<T> readAs(ResultSet resultSet, Type type) {

rxlib-x/src/main/java/org/rx/spring/BeanRegister.java

@@ -45,7 +45,7 @@ public RateLimiterAdapter httpRateLimiterAdapter(RedisCache<?, ?> rCache, Middle
             }
 
             String rk = "RateLimiter:" + clientIp;
-            RateLimiterAdapter adapter = IOC.<String, RateLimiterAdapter>weakMap(false)
+            RateLimiterAdapter adapter = IOC.<String, RateLimiterAdapter>weakMap(true)
                     .computeIfAbsent(rk, k -> RedisUtil.wrapRateLimiter(new RedisRateLimiter(rCache, k, conf.getLimiterPermits())));
             return adapter.tryAcquire();
         };

rxlib/src/main/java/org/rx/Main.java

@@ -12,6 +12,7 @@
 import org.rx.net.TransportFlags;
 import org.rx.net.dns.DnsClient;
 import org.rx.net.dns.DnsServer;
+import org.rx.net.http.*;
 import org.rx.net.rpc.Remoting;
 import org.rx.net.rpc.RpcClientConfig;
 import org.rx.net.rpc.RpcServerConfig;
@@ -31,6 +32,7 @@
 import java.math.BigInteger;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.Proxy;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -88,8 +90,8 @@ public static class RSSConf {
         public String udp2rawEndpoint;
     }
 
-    static RSSConf conf;
     static boolean udp2raw = false;
+    static RSSConf conf;
 
     @SneakyThrows
     static void launchClient(Map<String, String> options, Integer port, Integer connectTimeout) {
@@ -302,11 +304,36 @@ public void addWhiteList(InetAddress endpoint) {
             });
         }
 
-        app.ddns();
+        clientInit();
         log.info("Server started..");
         app.await();
     }
 
+    static void clientInit() {
+        Tasks.schedulePeriod(() -> {
+            if (conf == null) {
+                log.warn("conf is null");
+            }
+
+            InetAddress wanIp = InetAddress.getByName(IPSearcher.DEFAULT.currentIp());
+            for (String ddns : conf.ddnsDomains) {
+                List<InetAddress> currentIps = DnsClient.inlandClient().resolveAll(ddns);
+                if (currentIps.contains(wanIp)) {
+                    continue;
+                }
+                int i = ddns.indexOf(".");
+                String domain = ddns.substring(i + 1), name = ddns.substring(0, i);
+                log.info("ddns-{}.{}: {}->{}", name, domain, currentIps, wanIp);
+                AuthenticProxy p = conf.godaddyProxy != null
+                        ? new AuthenticProxy(Proxy.Type.SOCKS, Sockets.parseEndpoint(conf.godaddyProxy))
+                        : null;
+                IPSearcher.godaddyDns(conf.getGodaddyKey(), domain, name, wanIp.getHostAddress(), p);
+            }
+        }, conf.ddnsSeconds * 1000L);
+    }
+
+    static HttpServer httpServer;
+
     static void launchServer(Map<String, String> options, Integer port, Integer connectTimeout) {
         AuthenticEndpoint shadowUser = Reflects.convertQuietly(options.get("shadowUser"), AuthenticEndpoint.class);
         if (shadowUser == null) {
@@ -330,34 +357,24 @@ static void launchServer(Map<String, String> options, Integer port, Integer conn
         rpcConf.getTcpConfig().setTransportFlags(TransportFlags.FRONTEND_AES_COMBO.flags());
         Main app = new Main(backSvr);
         Remoting.register(app, rpcConf);
+        serverInit();
         app.await();
     }
 
-    final SocksProxyServer proxyServer;
-
-    void ddns() {
-//        Tasks.schedulePeriod(() -> {
-//            if (conf == null) {
-//                log.warn("conf is null");
-//            }
-//
-//            InetAddress wanIp = InetAddress.getByName(IPSearcher.DEFAULT.currentIp());
-//            for (String ddns : conf.ddnsDomains) {
-//                List<InetAddress> currentIps = DnsClient.inlandClient().resolveAll(ddns);
-//                if (currentIps.contains(wanIp)) {
-//                    continue;
-//                }
-//                int i = ddns.indexOf(".");
-//                String domain = ddns.substring(i + 1), name = ddns.substring(0, i);
-//                log.info("ddns-{}.{}: {}->{}", name, domain, currentIps, wanIp);
-//                AuthenticProxy p = conf.godaddyProxy != null
-//                        ? new AuthenticProxy(Proxy.Type.SOCKS, Sockets.parseEndpoint(conf.godaddyProxy))
-//                        : null;
-//                IPSearcher.godaddyDns(conf.getGodaddyKey(), domain, name, wanIp.getHostAddress(), p);
-//            }
-//        }, conf.ddnsSeconds * 1000L);
+    static void serverInit() {
+        httpServer = new HttpServer(8082, true).requestMapping("/hf", (request, response) -> {
+            String url = request.getQueryString().getFirst("fu");
+            Integer tm = Reflects.convertQuietly(request.getQueryString().getFirst("tm"), Integer.class);
+            HttpClient client = new HttpClient();
+            if (tm != null) {
+                client.withTimeoutMillis(tm);
+            }
+            response.jsonBody(client.get(url).toJson());
+        });
     }
 
+    final SocksProxyServer proxyServer;
+
     @Override
     public void fakeEndpoint(BigInteger hash, String endpoint) {
         SocksSupport.fakeDict().putIfAbsent(hash, UnresolvedEndpoint.valueOf(endpoint));

rxlib/src/main/java/org/rx/bean/WeakIdentityMap.java → rxlib/src/main/java/org/rx/bean/ConcurrentWeakMap.java

@@ -1,5 +1,7 @@
 package org.rx.bean;
 
+import lombok.NonNull;
+
 import java.lang.ref.Reference;
 import java.lang.ref.ReferenceQueue;
 import java.lang.ref.WeakReference;
@@ -9,18 +11,25 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
 
-//ReferenceIdentityMap
-public class WeakIdentityMap<K, V> implements AbstractMap<K, V> {
-    final Map<WeakReference<K>, V> map;
+//不要放值类型
+//ReferenceMap, ReferenceIdentityMap
+public class ConcurrentWeakMap<K, V> implements AbstractMap<K, V> {
     final ReferenceQueue<K> refQueue = new ReferenceQueue<>();
-    transient MapView.EntrySetView<WeakReference<K>, K, V> entrySet;
+    final Map<Reference<K>, V> map;
+    final boolean identityReference;
+    transient MapView.EntrySetView<Reference<K>, K, V> entrySet;
 
-    public WeakIdentityMap() {
-        this(16);
+    public ConcurrentWeakMap(boolean identityReference) {
+        this(identityReference, 16);
     }
 
-    public WeakIdentityMap(int initialCapacity) {
+    public ConcurrentWeakMap(boolean identityReference, int initialCapacity) {
         map = new ConcurrentHashMap<>(initialCapacity);
+        this.identityReference = identityReference;
+    }
+
+    Reference<K> toKeyReference(K key) {
+        return identityReference ? new IdentityWeakReference<>(key, refQueue) : new WeakReference<>(key, refQueue);
     }
 
     @Override
@@ -30,43 +39,35 @@ public int size() {
     }
 
     @Override
-    public V get(Object key) {
+    public V get(@NonNull Object key) {
         expunge();
-        Objects.requireNonNull(key, "key");
-        WeakReference<K> keyref = new IdentityWeakReference<>((K) key);
-        return map.get(keyref);
+        return map.get(toKeyReference((K) key));
     }
 
     @Override
-    public V put(K key, V value) {
+    public V put(@NonNull K key, V value) {
         expunge();
-        Objects.requireNonNull(key, "key");
-        WeakReference<K> keyref = new IdentityWeakReference<>(key, refQueue);
-        return map.put(keyref, value);
+        return map.put(toKeyReference(key), value);
     }
 
     @Override
-    public V remove(Object key) {
+    public V remove(@NonNull Object key) {
         expunge();
-        Objects.requireNonNull(key, "key");
-        WeakReference<K> keyref = new IdentityWeakReference<>((K) key);
-        return map.remove(keyref);
+        return map.remove(toKeyReference((K) key));
     }
 
     @Override
     public Set<Entry<K, V>> entrySet() {
         expunge();
-        MapView.EntrySetView<WeakReference<K>, K, V> es;
+        MapView.EntrySetView<Reference<K>, K, V> es;
         return (es = entrySet) != null ? es : (entrySet = new MapView.EntrySetView<>(map, Reference::get));
     }
 
     @Override
-    public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
+    public V computeIfAbsent(@NonNull K key, Function<? super K, ? extends V> mappingFunction) {
         expunge();
-        Objects.requireNonNull(key, "key");
         Objects.requireNonNull(mappingFunction, "mappingFunction");
-        WeakReference<K> keyref = new IdentityWeakReference<>(key, refQueue);
-        return map.computeIfAbsent(keyref, p -> mappingFunction.apply(key));
+        return map.computeIfAbsent(toKeyReference(key), p -> mappingFunction.apply(key));
     }
 
 //    public Stream<K> keysForValue(V value) {
@@ -115,11 +116,10 @@ public boolean equals(Object o) {
             if (this == o) {
                 return true;
             }
-            if (!(o instanceof WeakIdentityMap.IdentityWeakReference<?>)) {
+            if (!(o instanceof IdentityWeakReference)) {
                 return false;
             }
-            Object got = get();
-            return got != null && got == ((IdentityWeakReference<?>) o).get();
+            return get() == ((IdentityWeakReference<?>) o).get();
         }
 
         @Override

rxlib/src/main/java/org/rx/bean/DataTable.java

@@ -1,6 +1,5 @@
 package org.rx.bean;
 
-import com.alibaba.fastjson2.JSONObject;
 import lombok.*;
 import lombok.extern.slf4j.Slf4j;
 import org.h2.expression.Alias;

rxlib/src/main/java/org/rx/bean/Tuple.java

@@ -23,10 +23,10 @@ public static <T1, T2> Tuple<T1, T2> of(T1 t1, T2 t2) {
         return new Tuple<>(t1, t2);
     }
 
+    public static <T1, T2> Map.Entry<T1, T2> toMapEntry(T1 t1, T2 t2) {
+        return new AbstractMap.SimpleImmutableEntry<>(t1, t2);
+    }
+
     public T1 left;
     public T2 right;
-
-    public Map.Entry<T1, T2> toMapEntry() {
-        return new AbstractMap.SimpleImmutableEntry<>(left, right);
-    }
 }

rxlib/src/main/java/org/rx/core/CpuWatchman.java

@@ -159,7 +159,7 @@ static int decrSize(ThreadPoolExecutor pool) {
         return poolSize;
     }
 
-    final Map<ThreadPoolExecutor, Tuple<IntWaterMark, int[]>> holder = new WeakIdentityMap<>(8);
+    final Map<ThreadPoolExecutor, Tuple<IntWaterMark, int[]>> holder = new ConcurrentWeakMap<>(true, 8);
 
     private CpuWatchman() {
         timer.newTimeout(this, RxConfig.INSTANCE.threadPool.samplingPeriod, TimeUnit.MILLISECONDS);

rxlib/src/main/java/org/rx/core/IOC.java

@@ -2,7 +2,10 @@
 
 import lombok.NonNull;
 import lombok.SneakyThrows;
-import org.rx.bean.WeakIdentityMap;
+import org.apache.commons.collections4.map.AbstractReferenceMap;
+import org.apache.commons.collections4.map.ReferenceIdentityMap;
+import org.apache.commons.collections4.map.ReferenceMap;
+import org.rx.bean.ConcurrentWeakMap;
 import org.rx.exception.InvalidException;
 
 import java.util.*;
@@ -13,9 +16,9 @@
 @SuppressWarnings(NON_UNCHECKED)
 public final class IOC {
     static final Map<Class<?>, Object> container = new ConcurrentHashMap<>(8);
-    static final Map WEAK_MAP = Collections.synchronizedMap(new WeakHashMap<>());
-    //不要放值类型
-    static final Map WEAK_IDENTITY_MAP = new WeakIdentityMap<>();
+    //    static final Map WEAK_KEY_MAP = Collections.synchronizedMap(new WeakHashMap<>());
+    static final Map WEAK_KEY_MAP = new ConcurrentWeakMap<>(false);
+    static Map weakValMap, wKeyIdentityMap, wValIdentityMap;
 
     public static <T> boolean isInit(Class<T> type) {
         return container.containsKey(type);
@@ -67,11 +70,32 @@ public static <T> void unregister(Class<T> type) {
         container.remove(type);
     }
 
-    public static <K, V> Map<K, V> weakMap(boolean identity) {
-        return identity ? WEAK_IDENTITY_MAP : WEAK_MAP;
+    public static <K, V> Map<K, V> weakMap(boolean weakValue) {
+        if (!weakValue) {
+            return WEAK_KEY_MAP;
+        }
+        synchronized (WEAK_KEY_MAP) {
+            if (weakValMap == null) {
+                weakValMap = Collections.synchronizedMap(new ReferenceMap<>(AbstractReferenceMap.ReferenceStrength.HARD, AbstractReferenceMap.ReferenceStrength.WEAK));
+            }
+            return weakValMap;
+        }
+    }
+
+    static <K, V> Map<K, V> weakMap(Object ref, boolean weakValue) {
+        return (Map<K, V>) weakMap(weakValue).computeIfAbsent(ref, k -> new ConcurrentHashMap<>(4));
     }
 
-    static <K, V> Map<K, V> weakMap(Object ref, boolean identity) {
-        return (Map<K, V>) (identity ? WEAK_IDENTITY_MAP : WEAK_MAP).computeIfAbsent(ref, k -> new ConcurrentHashMap<>(4));
+    public synchronized static <K, V> Map<K, V> weakIdentityMap(boolean weakValue) {
+        if (weakValue) {
+            if (wValIdentityMap == null) {
+                wValIdentityMap = Collections.synchronizedMap(new ReferenceIdentityMap<>(AbstractReferenceMap.ReferenceStrength.HARD, AbstractReferenceMap.ReferenceStrength.WEAK));
+            }
+            return wValIdentityMap;
+        }
+        if (wKeyIdentityMap == null) {
+            wKeyIdentityMap = new ConcurrentWeakMap<>(true);
+        }
+        return wKeyIdentityMap;
     }
 }

rxlib/src/main/java/org/rx/core/Locker.java

@@ -1,21 +1,21 @@
-package org.rx.core;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import org.rx.bean.WeakIdentityMap;
-
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.locks.ReentrantLock;
-
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public class Locker {
-    public static final Locker INSTANCE = new Locker();
-    //key1: ref, key2: key
-    final Map<Object, Map<Object, ReentrantLock>> holder = new WeakIdentityMap<>();
-
-    public ReentrantLock getLock(Object ref, Object key) {
-        return holder.computeIfAbsent(ref, k -> new ConcurrentHashMap<>())
-                .computeIfAbsent(key, k -> new ReentrantLock());
-    }
-}
+//package org.rx.core;
+//
+//import lombok.AccessLevel;
+//import lombok.NoArgsConstructor;
+//import org.rx.bean.WeakIdentityMap;
+//
+//import java.util.Map;
+//import java.util.concurrent.ConcurrentHashMap;
+//import java.util.concurrent.locks.ReentrantLock;
+//
+//@NoArgsConstructor(access = AccessLevel.PRIVATE)
+//public class Locker {
+//    public static final Locker INSTANCE = new Locker();
+//    //key1: ref, key2: key
+//    final Map<Object, Map<Object, ReentrantLock>> holder = new WeakIdentityMap<>();
+//
+//    public ReentrantLock getLock(Object ref, Object key) {
+//        return holder.computeIfAbsent(ref, k -> new ConcurrentHashMap<>())
+//                .computeIfAbsent(key, k -> new ReentrantLock());
+//    }
+//}

rxlib/src/main/java/org/rx/core/Numbers.java

@@ -1,7 +1,6 @@
 package org.rx.core;
 
 import org.apache.commons.lang3.math.NumberUtils;
-import org.rx.exception.InvalidException;
 
 import java.math.BigDecimal;