provides tests, fixes bugs and polish API

Signed-off-by: Oleh Dokuka <[email protected]>
rsocket · Oct 16, 2020 · cc4595b · cc4595b
1 parent 8c7dfa7
commit cc4595b
Show file tree

Hide file tree

Showing 22 changed files with 1,605 additions and 338 deletions.
diff --git a/rsocket-core/build.gradle b/rsocket-core/build.gradle
@@ -29,6 +29,7 @@ dependencies {
 
     implementation 'org.slf4j:slf4j-api'
 
+    testImplementation (project(":rsocket-transport-local"))
     testImplementation 'io.projectreactor:reactor-test'
     testImplementation 'org.assertj:assertj-core'
     testImplementation 'org.junit.jupiter:junit-jupiter-api'

diff --git a/rsocket-core/src/main/java/io/rsocket/core/FireAndForgetRequesterMono.java b/rsocket-core/src/main/java/io/rsocket/core/FireAndForgetRequesterMono.java
@@ -69,8 +69,15 @@ final class FireAndForgetRequesterMono extends Mono<Void> implements Subscriptio
   public void subscribe(CoreSubscriber<? super Void> actual) {
     long previousState = markSubscribed(STATE, this);
     if (isSubscribedOrTerminated(previousState)) {
-      Operators.error(
-          actual, new IllegalStateException("FireAndForgetMono allows only a single Subscriber"));
+      final IllegalStateException e =
+          new IllegalStateException("FireAndForgetMono allows only a single Subscriber");
+
+      final RequestInterceptor requestInterceptor = this.requestInterceptor;
+      if (requestInterceptor != null) {
+        requestInterceptor.onReject(e, FrameType.REQUEST_FNF);
+      }
+
+      Operators.error(actual, e);
       return;
     }
 
@@ -81,14 +88,28 @@ public void subscribe(CoreSubscriber<? super Void> actual) {
     try {
       if (!isValid(mtu, this.maxFrameLength, p, false)) {
         lazyTerminate(STATE, this);
+
         p.release();
-        actual.onError(
+
+        final IllegalArgumentException e =
             new IllegalArgumentException(
-                String.format(INVALID_PAYLOAD_ERROR_MESSAGE, this.maxFrameLength)));
+                String.format(INVALID_PAYLOAD_ERROR_MESSAGE, this.maxFrameLength));
+        final RequestInterceptor requestInterceptor = this.requestInterceptor;
+        if (requestInterceptor != null) {
+          requestInterceptor.onReject(e, FrameType.REQUEST_FNF);
+        }
+
+        actual.onError(e);
         return;
       }
     } catch (IllegalReferenceCountException e) {
       lazyTerminate(STATE, this);
+
+      final RequestInterceptor requestInterceptor = this.requestInterceptor;
+      if (requestInterceptor != null) {
+        requestInterceptor.onReject(e, FrameType.REQUEST_FNF);
+      }
+
       actual.onError(e);
       return;
     }
@@ -98,14 +119,22 @@ public void subscribe(CoreSubscriber<? super Void> actual) {
       streamId = this.requesterResponderSupport.getNextStreamId();
     } catch (Throwable t) {
       lazyTerminate(STATE, this);
+
       p.release();
-      actual.onError(Exceptions.unwrap(t));
+
+      final Throwable ut = Exceptions.unwrap(t);
+      final RequestInterceptor requestInterceptor = this.requestInterceptor;
+      if (requestInterceptor != null) {
+        requestInterceptor.onReject(ut, FrameType.REQUEST_FNF);
+      }
+
+      actual.onError(ut);
       return;
     }
 
     final RequestInterceptor interceptor = this.requestInterceptor;
     if (interceptor != null) {
-      interceptor.onStart(streamId, FrameType.REQUEST_FNF, p.sliceMetadata());
+      interceptor.onStart(streamId, FrameType.REQUEST_FNF);
     }
 
     try {
@@ -162,19 +191,40 @@ public Void block(Duration m) {
   public Void block() {
     long previousState = markSubscribed(STATE, this);
     if (isSubscribedOrTerminated(previousState)) {
-      throw new IllegalStateException("FireAndForgetMono allows only a single Subscriber");
+      final IllegalStateException e =
+          new IllegalStateException("FireAndForgetMono allows only a single Subscriber");
+      final RequestInterceptor requestInterceptor = this.requestInterceptor;
+      if (requestInterceptor != null) {
+        requestInterceptor.onReject(e, FrameType.REQUEST_FNF);
+      }
+      throw e;
     }
 
     final Payload p = this.payload;
     try {
       if (!isValid(this.mtu, this.maxFrameLength, p, false)) {
         lazyTerminate(STATE, this);
+
         p.release();
-        throw new IllegalArgumentException(
-            String.format(INVALID_PAYLOAD_ERROR_MESSAGE, this.maxFrameLength));
+
+        final IllegalArgumentException e =
+            new IllegalArgumentException(
+                String.format(INVALID_PAYLOAD_ERROR_MESSAGE, this.maxFrameLength));
+        final RequestInterceptor requestInterceptor = this.requestInterceptor;
+        if (requestInterceptor != null) {
+          requestInterceptor.onReject(e, FrameType.REQUEST_FNF);
+        }
+
+        throw e;
       }
     } catch (IllegalReferenceCountException e) {
       lazyTerminate(STATE, this);
+
+      final RequestInterceptor requestInterceptor = this.requestInterceptor;
+      if (requestInterceptor != null) {
+        requestInterceptor.onReject(e, FrameType.REQUEST_FNF);
+      }
+
       throw Exceptions.propagate(e);
     }
 
@@ -183,13 +233,20 @@ public Void block() {
       streamId = this.requesterResponderSupport.getNextStreamId();
     } catch (Throwable t) {
       lazyTerminate(STATE, this);
+
       p.release();
+
+      final RequestInterceptor requestInterceptor = this.requestInterceptor;
+      if (requestInterceptor != null) {
+        requestInterceptor.onReject(t, FrameType.REQUEST_FNF);
+      }
+
       throw Exceptions.propagate(t);
     }
 
     final RequestInterceptor interceptor = this.requestInterceptor;
     if (interceptor != null) {
-      interceptor.onStart(streamId, FrameType.REQUEST_FNF, p.sliceMetadata());
+      interceptor.onStart(streamId, FrameType.REQUEST_FNF);
     }
 
     try {