adds tests for WeightedLoadbalanceStrategy

build.gradle

@@ -42,11 +42,11 @@ subprojects {
     ext['slf4j.version'] = '1.7.30'
     ext['jmh.version'] = '1.33'
     ext['junit.version'] = '5.8.1'
-    ext['hamcrest.version'] = '1.3'
     ext['micrometer.version'] = '1.7.5'
     ext['assertj.version'] = '3.21.0'
     ext['netflix.limits.version'] = '0.3.6'
     ext['bouncycastle-bcpkix.version'] = '1.68'
+    ext['awaitility.version'] = '4.1.1'
 
     group = "io.rsocket"
 
@@ -80,11 +80,11 @@ subprojects {
             dependency "org.assertj:assertj-core:${ext['assertj.version']}"
             dependency "org.hdrhistogram:HdrHistogram:${ext['hdrhistogram.version']}"
             dependency "org.slf4j:slf4j-api:${ext['slf4j.version']}"
+            dependency "org.awaitility:awaitility:${ext['awaitility.version']}"
             dependencySet(group: 'org.mockito', version: ext['mockito.version']) {
                 entry 'mockito-junit-jupiter'
                 entry 'mockito-core'
             }
-            dependency "org.hamcrest:hamcrest-library:${ext['hamcrest.version']}"
             dependencySet(group: 'org.openjdk.jmh', version: ext['jmh.version']) {
                 entry 'jmh-core'
                 entry 'jmh-generator-annprocess'

rsocket-core/build.gradle

@@ -35,12 +35,11 @@ dependencies {
     testImplementation 'org.junit.jupiter:junit-jupiter-api'
     testImplementation 'org.junit.jupiter:junit-jupiter-params'
     testImplementation 'org.mockito:mockito-core'
+    testImplementation 'org.awaitility:awaitility'
 
     testRuntimeOnly 'ch.qos.logback:logback-classic'
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
 
-    testImplementation 'org.hamcrest:hamcrest-library'
-
     jcstressImplementation(project(":rsocket-test"))
     jcstressImplementation "ch.qos.logback:logback-classic"
 }

rsocket-core/src/main/java/io/rsocket/loadbalance/FrugalQuantile.java

@@ -65,7 +65,8 @@ public synchronized void insert(double x) {
       estimate = x;
       sign = 1;
     } else {
-      double v = rnd.nextDouble();
+      final double v = rnd.nextDouble();
+      final double estimate = this.estimate;
 
       if (x > estimate && v > (1 - quantile)) {
         higher(x);
@@ -76,6 +77,8 @@ public synchronized void insert(double x) {
   }
 
   private void higher(double x) {
+    double estimate = this.estimate;
+
     step += sign * increment;
 
     if (step > 0) {
@@ -94,9 +97,13 @@ private void higher(double x) {
     }
 
     sign = 1;
+
+    this.estimate = estimate;
   }
 
   private void lower(double x) {
+    double estimate = this.estimate;
+
     step -= sign * increment;
 
     if (step > 0) {
@@ -115,6 +122,8 @@ private void lower(double x) {
     }
 
     sign = -1;
+
+    this.estimate = estimate;
   }
 
   @Override

rsocket-core/src/main/java/io/rsocket/loadbalance/Median.java

@@ -33,6 +33,7 @@ public synchronized void insert(double x) {
       estimate = x;
       sign = 1;
     } else {
+      final double estimate = this.estimate;
       if (x > estimate) {
         greaterThanZero(x);
       } else if (x < estimate) {
@@ -42,6 +43,8 @@ public synchronized void insert(double x) {
   }
 
   private void greaterThanZero(double x) {
+    double estimate = this.estimate;
+
     step += sign;
 
     if (step > 0) {
@@ -60,9 +63,13 @@ private void greaterThanZero(double x) {
     }
 
     sign = 1;
+
+    this.estimate = estimate;
   }
 
   private void lessThanZero(double x) {
+    double estimate = this.estimate;
+
     step -= sign;
 
     if (step > 0) {
@@ -81,6 +88,8 @@ private void lessThanZero(double x) {
     }
 
     sign = -1;
+
+    this.estimate = estimate;
   }
 
   @Override

rsocket-core/src/main/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategy.java

@@ -124,7 +124,10 @@ public RSocket select(List<RSocket> sockets) {
 
   private static double algorithmicWeight(
       RSocket rSocket, @Nullable final WeightedStats weightedStats) {
-    if (weightedStats == null || rSocket.isDisposed() || rSocket.availability() == 0.0) {
+    if (weightedStats == null) {
+      return 1.0;
+    }
+    if (rSocket.isDisposed() || rSocket.availability() == 0.0) {
       return 0.0;
     }
     final int pending = weightedStats.pending();