Plugged a few gaps in token handling for error path cases

instrumentation/netty-reactor-0.9.0/src/main/java/reactor/core/publisher/FluxMapFuseable_Instrumentation.java

@@ -41,6 +41,8 @@ public void onComplete() {
 
         public void onNext(T t) {
             if (token != null) {
+                // not ideal to do this in onNext, but we're seeing situations where nothing else is ever called
+                // no onComplete, no onError, no cancel
                 token.linkAndExpire();
                 token = null;
             }
@@ -54,5 +56,13 @@ public void onError(Throwable t) {
             }
             Weaver.callOriginal();
         }
+
+        public void cancel() {
+            if (token != null) {
+                token.linkAndExpire();
+                token = null;
+            }
+            Weaver.callOriginal();
+        }
     }
 }

instrumentation/netty-reactor-0.9.0/src/main/java/reactor/core/publisher/LambdaMonoSubscriber_Instrumentation.java

@@ -50,6 +50,15 @@ public final void onError(Throwable t) {
         Weaver.callOriginal();
     }
 
+    public void dispose() {
+        Token token = this.currentContext().getOrDefault("newrelic-token", null);
+        if (token != null) {
+            token.expire();
+            this.nrContext = null;
+        }
+        Weaver.callOriginal();
+    }
+
     public Context currentContext() {
         if (nrContext != null) {
             return initialContext.putAll(nrContext);

instrumentation/netty-reactor-0.9.0/src/main/java/reactor/core/publisher/LambdaSubscriber_Instrumentation.java

@@ -48,6 +48,15 @@ public final void onError(Throwable t) {
         Weaver.callOriginal();
     }
 
+    public void dispose() {
+        Token token = this.currentContext().getOrDefault("newrelic-token", null);
+        if (token != null) {
+            token.expire();
+            this.nrContext = null;
+        }
+        Weaver.callOriginal();
+    }
+
     public Context currentContext() {
         if (nrContext != null) {
             //return nrContext;

instrumentation/netty-reactor-0.9.0/src/main/java/reactor/core/publisher/MonoSubscribeOn_Instrumentation.java

@@ -19,7 +19,7 @@
 abstract class MonoSubscribeOn_Instrumentation {
 
     @Weave(type = MatchType.ExactClass, originalName = "reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber")
-    static final class SubscribeOnSubscriber_Instrumentation {
+    static final class SubscribeOnSubscriber_Instrumentation<T> {
         @NewField
         private Token token;
 
@@ -32,7 +32,31 @@ static final class SubscribeOnSubscriber_Instrumentation {
 
         public void run () {
             if (token != null) {
-                Boolean result = token.linkAndExpire();
+                token.linkAndExpire();
+                token = null;
+            }
+            Weaver.callOriginal();
+        }
+
+        public void onNext(T t) {
+            if (token != null) {
+                token.linkAndExpire();
+                token = null;
+            }
+            Weaver.callOriginal();
+        }
+
+        public void onComplete() {
+            if (token != null) {
+                token.linkAndExpire();
+                token = null;
+            }
+            Weaver.callOriginal();
+        }
+
+        public void onError(Throwable t) {
+            if (token != null) {
+                token.linkAndExpire();
                 token = null;
             }
             Weaver.callOriginal();