Merge branch 'release/1.1.1'

README.md

@@ -24,7 +24,7 @@ RSocket-JVM includes RSocket-RPC: remote procedure call system on top of Protoco
 so usable by each vendor library. Currently transports are comprised of TCP, unix & websocket-over-http2, and 
 are considered part of RSocket-JVM runtime.
 
-**Non-intrusive**. API & runtime are clearly split so from end-user perspective there is 
+**Non-intrusive**. API ([MessageStreams](https://github.com/jauntsdn/rsocket-jvm/blob/1.1.0/rsocket-reactor/src/main/java/com/jauntsdn/rsocket/MessageStreams.java)) & runtime ([RSocket](https://github.com/jauntsdn/rsocket-jvm/blob/1.1.0/rsocket-reactor/src/main/java/com/jauntsdn/rsocket/RSocket.java)) are clearly split so from end-user perspective there is 
 only defined set of basic interactions on buffers/messages:
 ```groovy
   Publisher<Message> requestResponse(Message message);
@@ -75,10 +75,10 @@ repositories {
 }
 
 dependencies {
-    implementation "com.jauntsdn.rsocket:rsocket-messages:1.0.0"
-    implementation "com.jauntsdn.rsocket:rsocket-rpc-idl:1.0.0"
-    implementation "com.jauntsdn.rsocket:rsocket-<VENDOR>:1.0.0"
-    implementation "com.jauntsdn.rsocket:rsocket-rpc-<VENDOR>:1.0.0"
+    implementation "com.jauntsdn.rsocket:rsocket-messages:1.1.0"
+    implementation "com.jauntsdn.rsocket:rsocket-rpc-idl:1.1.0"
+    implementation "com.jauntsdn.rsocket:rsocket-<VENDOR>:1.1.0"
+    implementation "com.jauntsdn.rsocket:rsocket-rpc-<VENDOR>:1.1.0"
 }
 ```
 
@@ -87,7 +87,7 @@ RSocket-RPC compiler binaries are for linux only
 protobuf {
      plugins {
           rsocketRpc {
-              artifact = "com.jauntsdn.rsocket:rsocket-rpc-<VENDOR>-compiler:1.0.0"
+              artifact = "com.jauntsdn.rsocket:rsocket-rpc-<VENDOR>-compiler:1.1.0"
           }
      }
 }

gradle.properties

@@ -1,5 +1,5 @@
 group=com.jauntsdn.rsocket
-version=1.1.0
+version=1.1.1
 
 dependencyManagementPluginVersion=1.0.11.RELEASE
 protobufPluginVersion=0.8.17
@@ -11,6 +11,7 @@ nettyBomVersion=4.1.72.Final
 reactorBomVersion=Dysprosium-SR22
 rxjavaVersion=3.1.3
 helidonCommonReactiveVersion=2.3.4
+mutinyVersion=1.2.0
 jsr305Version=3.0.2
 javaxInjectVersion=1
 javaxAnnotationVersion=1.3.2

gradle/dependency-management.gradle

@@ -14,6 +14,8 @@ subprojects {
         dependencies {
             dependency "io.reactivex.rxjava3:rxjava:${rxjavaVersion}"
             dependency "io.helidon.common:helidon-common-reactive:${helidonCommonReactiveVersion}"
+            dependency "io.smallrye.reactive:mutiny:${mutinyVersion}"
+
             dependency "javax.inject:javax.inject:${javaxInjectVersion}"
             dependency "javax.annotation:javax.annotation-api:${javaxAnnotationVersion}"
             dependency "com.google.code.findbugs:jsr305:${jsr305Version}"

rsocket-bom/build.gradle

@@ -23,13 +23,15 @@ dependencies {
 
         api "io.reactivex.rxjava3:rxjava:${rxjavaVersion}"
         api "io.helidon.common:helidon-common-reactive:${helidonCommonReactiveVersion}"
+        api "io.smallrye.reactive:mutiny:${mutinyVersion}"
         api "javax.inject:javax.inject:${javaxInjectVersion}"
         api "javax.annotation:javax.annotation-api:${javaxAnnotationVersion}"
         api "com.google.protobuf:protobuf-java:${protobufVersion}"
 
         api "com.jauntsdn.rsocket:rsocket-rpc-reactor-compiler:${version}"
         api "com.jauntsdn.rsocket:rsocket-rpc-rxjava-compiler:${version}"
         api "com.jauntsdn.rsocket:rsocket-rpc-helidon-compiler:${version}"
+        api "com.jauntsdn.rsocket:rsocket-rpc-mutiny-compiler:${version}"
         api "com.google.protobuf:protoc:${protobufVersion}"
     }
 }

rsocket-messages/src/main/java/com/jauntsdn/rsocket/Errors.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2020 - present Maksym Ostroverkhov.
+ *
+ * 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 com.jauntsdn.rsocket;
+
+import com.jauntsdn.rsocket.exceptions.ChannelException;
+import javax.annotation.Nullable;
+
+public final class Errors {
+  private Errors() {}
+
+  public static final class Configurer {
+    Connection.SendErrors connectionSendErrors;
+    Connection.ReceiveErrors connectionReceiveErrors;
+    Stream.SendErrors streamSendErrors;
+    Stream.ReceiveErrors streamReceiveErrors;
+
+    Configurer() {}
+
+    public Configurer connectionSendErrors(Connection.SendErrors connectionSendErrors) {
+      this.connectionSendErrors = connectionSendErrors;
+      return this;
+    }
+
+    public Configurer connectionReceiveErrors(Connection.ReceiveErrors connectionReceiveErrors) {
+      this.connectionReceiveErrors = connectionReceiveErrors;
+      return this;
+    }
+
+    public Configurer streamSendErrors(Stream.SendErrors streamSendErrors) {
+      this.streamSendErrors = streamSendErrors;
+      return this;
+    }
+
+    public Configurer streamReceiveErrors(Stream.ReceiveErrors streamReceiveErrors) {
+      this.streamReceiveErrors = streamReceiveErrors;
+      return this;
+    }
+  }
+
+  public static final class Connection {
+
+    private Connection() {}
+
+    public interface SendErrors {
+
+      String translate(int errorCode, @Nullable String errorMessage);
+    }
+
+    public interface ReceiveErrors {
+
+      Exception translate(int errorCode, @Nullable String errorMessage);
+    }
+  }
+
+  public static final class Stream {
+
+    private Stream() {}
+
+    public enum StreamType {
+      REQUEST,
+      RESPONSE
+    }
+
+    /** Converts stream {@link Throwable} error to RSocket error code and message */
+    public interface SendErrors {
+
+      /**
+       * @param streamType type of stream: request or response
+       * @param t {@link Throwable} that should be converted to {@link Error}
+       * @return one of stream errors (code and message): reject, cancel, invalid, application. Null
+       *     if default conversion should be applied.
+       */
+      @Nullable
+      Error translate(StreamType streamType, Throwable t);
+
+      /** Represents one of stream errors: reject, cancel, invalid, application */
+      final class Error {
+        private final int errorCode;
+        private final String message;
+
+        private Error(int errorCode, String message) {
+          this.errorCode = errorCode;
+          this.message = message;
+        }
+
+        public static Error reject(String errorMessage) {
+          return new Error(ChannelException.ErrorCodes.REJECTED, errorMessage);
+        }
+
+        public static Error cancel(String errorMessage) {
+          return new Error(ChannelException.ErrorCodes.CANCELED, errorMessage);
+        }
+
+        public static Error invalid(String errorMessage) {
+          return new Error(ChannelException.ErrorCodes.INVALID, errorMessage);
+        }
+
+        public static Error application(String errorMessage) {
+          return new Error(ChannelException.ErrorCodes.APPLICATION_ERROR, errorMessage);
+        }
+
+        public String message() {
+          return message;
+        }
+
+        public int code() {
+          return errorCode;
+        }
+      }
+    }
+
+    /**
+     * Converts RSocket error code {@link ErrorType} and message to stream {@link Throwable} error
+     */
+    public interface ReceiveErrors {
+
+      /**
+       * @param streamType type of stream: request or response
+       * @param errorType one of stream errors: reject, cancel, invalid, application
+       * @param errorMessage stream error message
+       * @return stream Throwable converted from error code and error message. Null if default
+       *     conversion should be applied
+       */
+      @Nullable
+      Throwable translate(StreamType streamType, ErrorType errorType, String errorMessage);
+
+      enum ErrorType {
+        REJECTED,
+        CANCELED,
+        INVALID,
+        APPLICATION;
+
+        static ErrorType fromCode(int code) {
+          switch (code) {
+            case ChannelException.ErrorCodes.REJECTED:
+              return REJECTED;
+            case ChannelException.ErrorCodes.CANCELED:
+              return CANCELED;
+            case ChannelException.ErrorCodes.INVALID:
+              return INVALID;
+            default:
+              return APPLICATION;
+          }
+        }
+      }
+    }
+  }
+}

rsocket-mutiny/build.gradle

@@ -0,0 +1,18 @@
+/*
+ * Copyright 2021 - present Maksym Ostroverkhov.
+ */
+
+plugins {
+    id "java-library"
+    id "signing"
+    id "maven-publish"
+}
+
+dependencies {
+    api project(":rsocket-messages")
+    api "io.smallrye.reactive:mutiny"
+
+    compileOnly "com.google.code.findbugs:jsr305"
+}
+
+description = "RSocket-smallrye-mutiny api library"

rsocket-mutiny/gradle/dependency-locks/annotationProcessor.lockfile

@@ -0,0 +1,3 @@
+# This is a Gradle generated file for dependency locking.
+# Manual edits can break the build and are not advised.
+# This file is expected to be part of source control.

rsocket-mutiny/gradle/dependency-locks/compileClasspath.lockfile

@@ -0,0 +1,9 @@
+# This is a Gradle generated file for dependency locking.
+# Manual edits can break the build and are not advised.
+# This file is expected to be part of source control.
+com.google.code.findbugs:jsr305:3.0.2
+io.netty:netty-buffer:4.1.72.Final
+io.netty:netty-common:4.1.72.Final
+io.smallrye.common:smallrye-common-annotation:1.8.0
+io.smallrye.reactive:mutiny:1.2.0
+org.reactivestreams:reactive-streams:1.0.3

rsocket-mutiny/gradle/dependency-locks/googleJavaFormat1.6.lockfile

@@ -0,0 +1,10 @@
+# This is a Gradle generated file for dependency locking.
+# Manual edits can break the build and are not advised.
+# This file is expected to be part of source control.
+com.google.code.findbugs:jsr305:3.0.2
+com.google.errorprone:error_prone_annotations:2.0.18
+com.google.errorprone:javac-shaded:9+181-r4173-1
+com.google.googlejavaformat:google-java-format:1.6
+com.google.guava:guava:22.0
+com.google.j2objc:j2objc-annotations:1.1
+org.codehaus.mojo:animal-sniffer-annotations:1.14

rsocket-mutiny/gradle/dependency-locks/runtimeClasspath.lockfile

@@ -0,0 +1,8 @@
+# This is a Gradle generated file for dependency locking.
+# Manual edits can break the build and are not advised.
+# This file is expected to be part of source control.
+io.netty:netty-buffer:4.1.72.Final
+io.netty:netty-common:4.1.72.Final
+io.smallrye.common:smallrye-common-annotation:1.8.0
+io.smallrye.reactive:mutiny:1.2.0
+org.reactivestreams:reactive-streams:1.0.3

rsocket-mutiny/src/main/java/com/jauntsdn/rsocket/AbstractRSocket.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2021 - present Maksym Ostroverkhov.
+ *
+ * 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 com.jauntsdn.rsocket;
+
+import io.smallrye.mutiny.Multi;
+import io.smallrye.mutiny.Uni;
+import org.reactivestreams.Publisher;
+
+public abstract class AbstractRSocket implements RSocketHandler {
+
+  @Override
+  public Uni<Void> fireAndForget(Message message) {
+    message.release();
+    return Uni.createFrom()
+        .failure(new UnsupportedOperationException("fire-and-forget not implemented"));
+  }
+
+  @Override
+  public Uni<Message> requestResponse(Message message) {
+    message.release();
+    return Uni.createFrom()
+        .failure(new UnsupportedOperationException("request-response not implemented"));
+  }
+
+  @Override
+  public Multi<Message> requestStream(Message message) {
+    message.release();
+    return Multi.createFrom()
+        .failure(new UnsupportedOperationException("request-stream not implemented"));
+  }
+
+  @Override
+  public Multi<Message> requestChannel(Publisher<Message> messages) {
+    return Multi.createFrom()
+        .failure(new UnsupportedOperationException("request-channel(messages) not implemented"));
+  }
+
+  @Override
+  public Multi<Message> requestChannel(Message message, Publisher<Message> messages) {
+    message.release();
+    return Multi.createFrom()
+        .failure(
+            new UnsupportedOperationException(
+                "request-channel(message, messages) not implemented"));
+  }
+
+  @Override
+  public Uni<Void> metadataPush(Message message) {
+    message.release();
+    return Uni.createFrom()
+        .failure(new UnsupportedOperationException("metadata-push not implemented"));
+  }
+
+  @Override
+  public void dispose() {}
+
+  @Override
+  public boolean isDisposed() {
+    return false;
+  }
+
+  @Override
+  public Uni<Void> onClose() {
+    return Uni.createFrom().nothing();
+  }
+}

rsocket-mutiny/src/main/java/com/jauntsdn/rsocket/ClientAcceptor.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2021 - present Maksym Ostroverkhov.
+ *
+ * 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 com.jauntsdn.rsocket;
+
+import java.util.function.Function;
+
+public interface ClientAcceptor {
+
+  RSocket accept(SetupMessage setup, RSocket requesterRSocket);
+
+  @FunctionalInterface
+  interface Interceptor extends Function<ClientAcceptor, ClientAcceptor> {}
+}