Integrate BeanColloctor into Partitioner

app/src/main/java/org/astraea/partitioner/nodeLoadMetric/LoadPoisson.java

@@ -3,40 +3,41 @@
 import java.util.HashMap;
 import java.util.Map;
 
+/** do poisson for node's load situation */
 public class LoadPoisson {
-  private NodeLoadClient nodeLoadClient;
 
-  public LoadPoisson(NodeLoadClient nodeLoadClient) {
-    this.nodeLoadClient = nodeLoadClient;
-  }
-
-  public synchronized HashMap<String, Double> setAllPoisson() {
-    HashMap<String, Double> poissonMap = new HashMap<>();
-    int lambda = nodeLoadClient.getAvgLoadCount();
-    for (Map.Entry<String, Integer> entry : nodeLoadClient.getAllOverLoadCount().entrySet()) {
-      int x = nodeLoadClient.getBinOneCount(entry.getValue());
-      poissonMap.put(entry.getKey(), doPoisson(lambda, x));
-    }
+  public HashMap<Integer, Double> allPoisson(Map<Integer, Integer> overLoadCount) {
+    var poissonMap = new HashMap<Integer, Double>();
+    var lambda = avgLoadCount(overLoadCount);
+    overLoadCount.forEach((nodeID, count) -> poissonMap.put(nodeID, doPoisson(lambda, count)));
     return poissonMap;
   }
 
-  public double doPoisson(int lambda, int x) {
-    double Probability = 0;
-    double ans = 0;
-
-    for (int i = 0; i <= x; i++) {
-      double j = Math.pow(lambda, i);
-      double e = Math.exp(-lambda);
-      long h = factorial(i);
+  // visible for test
+  double doPoisson(int lambda, int x) {
+    var Probability = 0.0;
+    var ans = 0.0;
+    var i = 0;
+    while (i <= x) {
+      var j = Math.pow(lambda, i);
+      var e = Math.exp(-lambda);
+      var h = factorial(i);
       Probability = (j * e) / h;
       ans += Probability;
+      i++;
     }
-
     return ans;
   }
 
-  public long factorial(long number) {
+  // visible for test
+  long factorial(long number) {
     if (number <= 1) return 1;
     else return number * factorial(number - 1);
   }
+
+  private int avgLoadCount(Map<Integer, Integer> overLoadCount) {
+    var avgLoadCount =
+        overLoadCount.values().stream().mapToDouble(Integer::doubleValue).average().orElse(0);
+    return (int) avgLoadCount;
+  }
 }