fix issue with fuseable metrics subscriber; update reactor/rsocket version to latest

Author: OlegDokuka

gradle/java.gradle

@@ -80,14 +80,14 @@ dependencies {
     testCompile 'junit:junit:4.12'
 
     testCompile 'javax.inject:javax.inject:1'
-    testCompile 'io.projectreactor:reactor-test:3.1.7.RELEASE'
+    testCompile 'io.projectreactor:reactor-test:3.1.9.RELEASE'
     testCompile 'com.google.protobuf:protobuf-java:3.6.0'
     testCompile 'org.hdrhistogram:HdrHistogram:2.1.10'
     testCompile 'org.apache.logging.log4j:log4j-api:2.8.2'
     testCompile 'org.apache.logging.log4j:log4j-core:2.8.2'
     testCompile 'org.apache.logging.log4j:log4j-slf4j-impl:2.8.2'
-    testCompile 'io.rsocket:rsocket-transport-netty:0.11.5'
-    testCompile 'io.rsocket:rsocket-transport-local:0.11.5'
+    testCompile 'io.rsocket:rsocket-transport-netty:0.11.6'
+    testCompile 'io.rsocket:rsocket-transport-local:0.11.6'
     testCompile 'org.mockito:mockito-all:1.10.19'
 }
 

rsocket-rpc-core/build.gradle

@@ -10,9 +10,9 @@ dependencies {
     compile 'io.opentracing:opentracing-api:0.31.0'
     compile 'javax.inject:javax.inject:1'
     compile 'com.google.protobuf:protobuf-java:3.6.0'
-    compile 'io.rsocket:rsocket-core:0.11.5'
-    compile 'io.rsocket:rsocket-transport-netty:0.11.5'
-    compile 'io.micrometer:micrometer-core:1.0.3'
+    compile 'io.rsocket:rsocket-core:0.11.6'
+    compile 'io.rsocket:rsocket-transport-netty:0.11.6'
+    compile 'io.micrometer:micrometer-core:1.0.6'
 
     protobuf project(':rsocket-rpc-metrics-idl')
 

rsocket-rpc-core/src/main/java/io/rsocket/rpc/metrics/Metrics.java

@@ -3,6 +3,7 @@
 import io.micrometer.core.instrument.*;
 import java.util.function.Function;
 import org.reactivestreams.Publisher;
+import reactor.core.Fuseable;
 import reactor.core.publisher.Operators;
 
 public class Metrics {
@@ -32,7 +33,13 @@ public class Metrics {
             .tags(tags)
             .register(registry);
     return Operators.lift(
-        (scannable, subscriber) ->
-            new MetricsSubscriber<>(subscriber, next, complete, error, cancelled, timer));
+        (scannable, subscriber) -> {
+            if (scannable instanceof Fuseable) {
+              return new MetricsFuseableSubscriber<>(subscriber, next, complete, error, cancelled, timer);
+            }
+            else {
+              return new MetricsSubscriber<>(subscriber, next, complete, error, cancelled, timer);
+            }
+        });
   }
 }

rsocket-rpc-core/src/main/java/io/rsocket/rpc/metrics/MetricsFuseableSubscriber.java

@@ -0,0 +1,141 @@
+package io.rsocket.rpc.metrics;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.Timer;
+import org.reactivestreams.Subscription;
+import reactor.core.CoreSubscriber;
+import reactor.core.Fuseable.QueueSubscription;
+import reactor.core.Fuseable;
+import reactor.core.publisher.Operators;
+import reactor.util.annotation.Nullable;
+import reactor.util.context.Context;
+
+import static reactor.core.Fuseable.ASYNC;
+
+public class MetricsFuseableSubscriber<T> extends AtomicBoolean implements QueueSubscription<T>,
+                                                                           CoreSubscriber<T> {
+  private final CoreSubscriber<? super T> actual;
+  private final Counter next, complete, error, cancelled;
+  private final Timer timer;
+
+  private QueueSubscription<T> s;
+  private int sourceMode;
+
+  private long start;
+
+  MetricsFuseableSubscriber(
+      CoreSubscriber<? super T> actual,
+      Counter next,
+      Counter complete,
+      Counter error,
+      Counter cancelled,
+      Timer timer) {
+    this.actual = actual;
+    this.next = next;
+    this.complete = complete;
+    this.error = error;
+    this.cancelled = cancelled;
+    this.timer = timer;
+  }
+
+  @Override
+  @SuppressWarnings("unchecked")
+  public void onSubscribe(Subscription s) {
+    if (Operators.validate(this.s, s)) {
+      this.s = (QueueSubscription<T>) s;
+      this.start = System.nanoTime();
+
+      actual.onSubscribe(this);
+    }
+  }
+
+  @Override
+  public void onNext(T t) {
+    if (sourceMode == ASYNC) {
+      actual.onNext(null);
+    }
+    else {
+      next.increment();
+      actual.onNext(t);
+    }
+  }
+
+  @Override
+  public void onError(Throwable t) {
+    if (compareAndSet(false, true)) {
+      error.increment();
+      timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS);
+    }
+    actual.onError(t);
+  }
+
+  @Override
+  public void onComplete() {
+    if (compareAndSet(false, true)) {
+      complete.increment();
+      timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS);
+    }
+    actual.onComplete();
+  }
+
+  @Override
+  public void request(long n) {
+    s.request(n);
+  }
+
+  @Override
+  public void cancel() {
+    if (compareAndSet(false, true)) {
+      cancelled.increment();
+      timer.record(System.nanoTime() - start, TimeUnit.NANOSECONDS);
+    }
+    s.cancel();
+  }
+
+  @Override
+  public Context currentContext() {
+    return actual.currentContext();
+  }
+
+  @Override
+  public int requestFusion(int requestedMode) {
+    int m;
+    if ((requestedMode & Fuseable.THREAD_BARRIER) != 0) {
+      return Fuseable.NONE;
+    }
+    else {
+      m = s.requestFusion(requestedMode);
+    }
+    sourceMode = m;
+    return m;
+  }
+
+  @Override
+  @Nullable
+  public T poll() {
+    T v = s.poll();
+    if (v != null) {
+      next.increment();
+      return v;
+    }
+    return null;
+  }
+
+  @Override
+  public boolean isEmpty() {
+    return s.isEmpty();
+  }
+
+  @Override
+  public void clear() {
+    s.clear();
+  }
+
+  @Override
+  public int size() {
+    return s.size();
+  }
+}

rsocket-rpc-protobuf/build.gradle

@@ -54,7 +54,7 @@ sourceCompatibility = 1.8
 targetCompatibility = 1.8
 
 dependencies {
-    compileOnly 'io.projectreactor:reactor-core:3.1.7.RELEASE'
+    compileOnly 'io.projectreactor:reactor-core:3.1.9.RELEASE'
     compileOnly 'com.google.protobuf:protobuf-java:3.6.0'
     compileOnly 'javax.inject:javax.inject:1'
 }