Exceptions Documentation and Testing

build.gradle

@@ -44,7 +44,7 @@ subprojects {
             dependency 'com.netflix.spectator:spectator-api:0.63.0'
             dependency 'io.netty:netty-buffer:4.1.21.Final'
             dependency 'io.aeron:aeron-all:1.4.1'
-            dependency 'org.hamcrest:hamcrest-library:1.3'
+            dependency 'org.assertj:assertj-core:3.9.1'
             dependency 'org.hdrhistogram:HdrHistogram:2.1.10'
             dependency 'org.jctools:jctools-core:2.1.2'
             dependency 'org.mockito:mockito-core:2.16.0'
@@ -57,6 +57,7 @@ subprojects {
 
             // TODO: Remove after JUnit5 migration
             dependency 'junit:junit:4.12'
+            dependency 'org.hamcrest:hamcrest-library:1.3'
             dependencySet(group: 'org.junit.vintage', version: '5.1.0') {
                 entry 'junit-vintage-engine'
             }

rsocket-core/build.gradle

@@ -33,14 +33,15 @@ dependencies {
     implementation 'org.slf4j:slf4j-api'
 
     testImplementation 'io.projectreactor:reactor-test'
-    testImplementation 'org.hamcrest:hamcrest-library'
+    testImplementation 'org.assertj:assertj-core'
     testImplementation 'org.junit.jupiter:junit-jupiter-api'
     testImplementation 'org.mockito:mockito-core'
 
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
 
     // TODO: Remove after JUnit5 migration
     testCompileOnly 'junit:junit'
+    testImplementation 'org.hamcrest:hamcrest-library'
     testRuntimeOnly 'org.junit.vintage:junit-vintage-engine'
 }
 

rsocket-core/src/main/java/io/rsocket/RSocketClient.java

@@ -17,7 +17,7 @@
 package io.rsocket;
 
 import io.netty.buffer.Unpooled;
-import io.rsocket.exceptions.ConnectionException;
+import io.rsocket.exceptions.ConnectionErrorException;
 import io.rsocket.exceptions.Exceptions;
 import io.rsocket.internal.LimitableRequestPublisher;
 import io.rsocket.internal.UnboundedProcessor;
@@ -150,7 +150,7 @@ private void sendKeepAlive(long ackTimeoutMs, int missedAcks) {
             String.format(
                 "Missed %d keep-alive acks with a threshold of %d and a ack timeout of %d ms",
                 count, missedAcks, ackTimeoutMs);
-        throw new ConnectionException(message);
+        throw new ConnectionErrorException(message);
       }
     }
 

rsocket-core/src/main/java/io/rsocket/RSocketServer.java

@@ -22,7 +22,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
-import io.rsocket.exceptions.ApplicationException;
+import io.rsocket.exceptions.ApplicationErrorException;
 import io.rsocket.internal.LimitableRequestPublisher;
 import io.rsocket.internal.UnboundedProcessor;
 import java.util.function.Consumer;
@@ -253,7 +253,7 @@ private void handleFrame(Frame frame) {
         case ERROR:
           receiver = channelProcessors.get(streamId);
           if (receiver != null) {
-            receiver.onError(new ApplicationException(Frame.Error.message(frame)));
+            receiver.onError(new ApplicationErrorException(Frame.Error.message(frame)));
           }
           break;
         case NEXT_COMPLETE:

rsocket-core/src/main/java/io/rsocket/exceptions/ApplicationErrorException.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2015-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.exceptions;
+
+import io.rsocket.frame.ErrorFrameFlyweight;
+
+/**
+ * Application layer logic generating a Reactive Streams {@code onError} event.
+ *
+ * @see <a href="https://github.com/rsocket/rsocket/blob/master/Protocol.md#error-codes">Error
+ *     Codes</a>
+ */
+public final class ApplicationErrorException extends RSocketException {
+
+  private static final long serialVersionUID = 7873267740343446585L;
+
+  /**
+   * Constructs a new exception with the specified message.
+   *
+   * @param message the message
+   * @throws NullPointerException if {@code message} is {@code null}
+   */
+  public ApplicationErrorException(String message) {
+    super(message);
+  }
+
+  /**
+   * Constructs a new exception with the specified message and cause.
+   *
+   * @param message the message
+   * @param cause the cause of this exception
+   * @throws NullPointerException if {@code message} or {@code cause} is {@code null}
+   */
+  public ApplicationErrorException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  @Override
+  public int errorCode() {
+    return ErrorFrameFlyweight.APPLICATION_ERROR;
+  }
+}

rsocket-core/src/main/java/io/rsocket/exceptions/CanceledException.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2015-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.exceptions;
+
+import io.rsocket.frame.ErrorFrameFlyweight;
+
+/**
+ * The Responder canceled the request but may have started processing it (similar to REJECTED but
+ * doesn't guarantee lack of side-effects).
+ *
+ * @see <a href="https://github.com/rsocket/rsocket/blob/master/Protocol.md#error-codes">Error
+ *     Codes</a>
+ */
+public final class CanceledException extends RSocketException {
+
+  private static final long serialVersionUID = 5074789326089722770L;
+
+  /**
+   * Constructs a new exception with the specified message.
+   *
+   * @param message the message
+   * @throws NullPointerException if {@code message} is {@code null}
+   */
+  public CanceledException(String message) {
+    super(message);
+  }
+
+  /**
+   * Constructs a new exception with the specified message and cause.
+   *
+   * @param message the message
+   * @param cause the cause of this exception
+   * @throws NullPointerException if {@code message} or {@code cause} is {@code null}
+   */
+  public CanceledException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  @Override
+  public int errorCode() {
+    return ErrorFrameFlyweight.CANCELED;
+  }
+}

rsocket-core/src/main/java/io/rsocket/exceptions/ConnectionCloseException.java

@@ -18,14 +18,34 @@
 
 import io.rsocket.frame.ErrorFrameFlyweight;
 
-public class ConnectionCloseException extends RSocketException {
+/**
+ * The connection is being terminated. Sender or Receiver of this frame MUST wait for outstanding
+ * streams to terminate before closing the connection. New requests MAY not be accepted.
+ *
+ * @see <a href="https://github.com/rsocket/rsocket/blob/master/Protocol.md#error-codes">Error
+ *     Codes</a>
+ */
+public final class ConnectionCloseException extends RSocketException {
 
-  private static final long serialVersionUID = -7659717517940756969L;
+  private static final long serialVersionUID = -2214953527482377471L;
 
+  /**
+   * Constructs a new exception with the specified message.
+   *
+   * @param message the message
+   * @throws NullPointerException if {@code message} is {@code null}
+   */
   public ConnectionCloseException(String message) {
     super(message);
   }
 
+  /**
+   * Constructs a new exception with the specified message and cause.
+   *
+   * @param message the message
+   * @param cause the cause of this exception
+   * @throws NullPointerException if {@code message} or {@code cause} is {@code null}
+   */
   public ConnectionCloseException(String message, Throwable cause) {
     super(message, cause);
   }

rsocket-core/src/main/java/io/rsocket/exceptions/ConnectionErrorException.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2015-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.exceptions;
+
+import io.rsocket.frame.ErrorFrameFlyweight;
+
+/**
+ * The connection is being terminated. Sender or Receiver of this frame MAY close the connection
+ * immediately without waiting for outstanding streams to terminate.
+ *
+ * @see <a href="https://github.com/rsocket/rsocket/blob/master/Protocol.md#error-codes">Error
+ *     Codes</a>
+ */
+public final class ConnectionErrorException extends RSocketException implements Retryable {
+
+  private static final long serialVersionUID = 512325887785119744L;
+
+  /**
+   * Constructs a new exception with the specified message.
+   *
+   * @param message the message
+   * @throws NullPointerException if {@code message} is {@code null}
+   */
+  public ConnectionErrorException(String message) {
+    super(message);
+  }
+
+  /**
+   * Constructs a new exception with the specified message and cause.
+   *
+   * @param message the message
+   * @param cause the cause of this exception
+   * @throws NullPointerException if {@code message} or {@code cause} is {@code null}
+   */
+  public ConnectionErrorException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  @Override
+  public int errorCode() {
+    return ErrorFrameFlyweight.CONNECTION_ERROR;
+  }
+}

rsocket-core/src/main/java/io/rsocket/exceptions/Exceptions.java

@@ -16,29 +16,49 @@
 
 package io.rsocket.exceptions;
 
-import static io.rsocket.frame.ErrorFrameFlyweight.*;
+import static io.rsocket.frame.ErrorFrameFlyweight.APPLICATION_ERROR;
+import static io.rsocket.frame.ErrorFrameFlyweight.CANCELED;
+import static io.rsocket.frame.ErrorFrameFlyweight.CONNECTION_CLOSE;
+import static io.rsocket.frame.ErrorFrameFlyweight.CONNECTION_ERROR;
+import static io.rsocket.frame.ErrorFrameFlyweight.INVALID;
+import static io.rsocket.frame.ErrorFrameFlyweight.INVALID_SETUP;
+import static io.rsocket.frame.ErrorFrameFlyweight.REJECTED;
+import static io.rsocket.frame.ErrorFrameFlyweight.REJECTED_RESUME;
+import static io.rsocket.frame.ErrorFrameFlyweight.REJECTED_SETUP;
+import static io.rsocket.frame.ErrorFrameFlyweight.UNSUPPORTED_SETUP;
 
 import io.rsocket.Frame;
+import java.util.Objects;
 
-public class Exceptions {
+/** Utility class that generates an exception from a frame. */
+public final class Exceptions {
 
   private Exceptions() {}
 
+  /**
+   * Create a {@link RSocketException} from a {@link Frame} that matches the error code it contains.
+   *
+   * @param frame the frame to retrieve the error code and message from
+   * @return a {@link RSocketException} that matches the error code in the {@link Frame}
+   * @throws NullPointerException if {@code frame} is {@code null}
+   */
   public static RuntimeException from(Frame frame) {
-    final int errorCode = Frame.Error.errorCode(frame);
+    Objects.requireNonNull(frame, "frame must not be null");
 
+    int errorCode = Frame.Error.errorCode(frame);
     String message = frame.getDataUtf8();
+
     switch (errorCode) {
       case APPLICATION_ERROR:
-        return new ApplicationException(message);
+        return new ApplicationErrorException(message);
       case CANCELED:
-        return new CancelException(message);
+        return new CanceledException(message);
       case CONNECTION_CLOSE:
         return new ConnectionCloseException(message);
       case CONNECTION_ERROR:
-        return new ConnectionException(message);
+        return new ConnectionErrorException(message);
       case INVALID:
-        return new InvalidRequestException(message);
+        return new InvalidException(message);
       case INVALID_SETUP:
         return new InvalidSetupException(message);
       case REJECTED:
@@ -50,8 +70,8 @@ public static RuntimeException from(Frame frame) {
       case UNSUPPORTED_SETUP:
         return new UnsupportedSetupException(message);
       default:
-        return new InvalidRequestException(
-            "Invalid Error frame: " + errorCode + " '" + message + "'");
+        return new IllegalArgumentException(
+            String.format("Invalid Error frame: %d '%s'", errorCode, message));
     }
   }
 }