RSocket: add unreliable channels support (#17)

Author: mostroverkhov

gradle/publishing.gradle

@@ -22,7 +22,7 @@ subprojects {
                         afterEvaluate {
                             description = project.description
                         }
-                        url = "https://jauntsdn.com"
+                        url = "https://jauntsdn.github.io"
                         licenses {
                             license {
                                 name = "The Apache Software License, Version 2.0"

rsocket-futures/src/main/java/com/jauntsdn/rsocket/RSocket.java

@@ -16,9 +16,11 @@
 
 package com.jauntsdn.rsocket;
 
+import io.netty.buffer.ByteBuf;
 import java.util.Optional;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Consumer;
 import java.util.function.Function;
 
 /**
@@ -34,6 +36,14 @@ public interface RSocket extends MessageStreams, Availability {
    */
   CompletionStage<Void> metadataPush(Message message);
 
+  /**
+   * @return unreliable unidirectional or bidirectional communication channel if underlying
+   *     transport supports It (e.g. QUIC-DATAGRAM extension, WebTransport-QUIC)
+   */
+  default Optional<UnreliableChannel> unreliableChannel() {
+    return Optional.empty();
+  }
+
   /**
    * @return lightweight {@link ScheduledExecutorService} intended for non-fine-grained tasks
    *     scheduling (e.g. timeouts).
@@ -52,6 +62,26 @@ default double availability() {
     return availability(0);
   }
 
+  interface UnreliableChannel {
+
+    Consumer<ByteBuf> unreliableChannel(Consumer<ByteBuf> messages);
+
+    Attributes attributes();
+
+    interface Attributes {
+
+      int messageSizeLimit();
+
+      Direction direction();
+    }
+
+    enum Direction {
+      INBOUND,
+      OUTBOUND,
+      BIDIRECTIONAL
+    }
+  }
+
   @FunctionalInterface
   interface Interceptor extends Function<RSocket, RSocket> {}
 

rsocket-grpc/src/main/java/com/jauntsdn/rsocket/RSocket.java

@@ -17,6 +17,7 @@
 package com.jauntsdn.rsocket;
 
 import io.grpc.stub.StreamObserver;
+import io.netty.buffer.ByteBuf;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Function;
@@ -34,6 +35,14 @@ public interface RSocket extends MessageStreams, Availability {
    */
   void metadataPush(Message message, StreamObserver<Message> responseObserver);
 
+  /**
+   * @return unreliable unidirectional or bidirectional communication channel if underlying
+   *     transport supports It (e.g. QUIC-DATAGRAM extension, WebTransport-QUIC)
+   */
+  default Optional<UnreliableChannel> unreliableChannel() {
+    return Optional.empty();
+  }
+
   /**
    * @return lightweight {@link ScheduledExecutorService} intended for non-fine-grained tasks
    *     scheduling (e.g. timeouts).
@@ -52,6 +61,26 @@ default double availability() {
     return availability(0);
   }
 
+  interface UnreliableChannel {
+
+    StreamObserver<ByteBuf> unreliableChannel(StreamObserver<ByteBuf> messages);
+
+    Attributes attributes();
+
+    interface Attributes {
+
+      int messageSizeLimit();
+
+      Direction direction();
+    }
+
+    enum Direction {
+      INBOUND,
+      OUTBOUND,
+      BIDIRECTIONAL
+    }
+  }
+
   @FunctionalInterface
   interface Interceptor extends Function<RSocket, RSocket> {}
 

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

@@ -16,8 +16,11 @@
 
 package com.jauntsdn.rsocket;
 
+import io.netty.buffer.ByteBuf;
+import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
 import java.util.Optional;
+import java.util.concurrent.Flow;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Function;
 
@@ -34,6 +37,14 @@ public interface RSocket extends MessageStreams, Availability {
    */
   Uni<Void> metadataPush(Message message);
 
+  /**
+   * @return unreliable unidirectional or bidirectional communication channel if underlying
+   *     transport supports It (e.g. QUIC-DATAGRAM extension, WebTransport-QUIC)
+   */
+  default Optional<UnreliableChannel> unreliableChannel() {
+    return Optional.empty();
+  }
+
   /**
    * @return lightweight {@link ScheduledExecutorService} intended for non-fine-grained tasks
    *     scheduling (e.g. timeouts).
@@ -52,6 +63,26 @@ default double availability() {
     return availability(0);
   }
 
+  interface UnreliableChannel {
+
+    Multi<ByteBuf> unreliableChannel(Flow.Publisher<ByteBuf> messages);
+
+    Attributes attributes();
+
+    interface Attributes {
+
+      int messageSizeLimit();
+
+      Direction direction();
+    }
+
+    enum Direction {
+      INBOUND,
+      OUTBOUND,
+      BIDIRECTIONAL
+    }
+  }
+
   @FunctionalInterface
   interface Interceptor extends Function<RSocket, RSocket> {}
 

rsocket-reactor/src/main/java/com/jauntsdn/rsocket/RSocket.java

@@ -16,9 +16,12 @@
 
 package com.jauntsdn.rsocket;
 
+import io.netty.buffer.ByteBuf;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Function;
+import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
 
@@ -35,6 +38,14 @@ public interface RSocket extends MessageStreams, Availability {
    */
   Mono<Void> metadataPush(Message message);
 
+  /**
+   * @return unreliable unidirectional or bidirectional communication channel if underlying
+   *     transport supports It (e.g. QUIC-DATAGRAM extension, WebTransport-QUIC)
+   */
+  default Optional<UnreliableChannel> unreliableChannel() {
+    return Optional.empty();
+  }
+
   /**
    * @return lightweight {@link ScheduledExecutorService} intended for non-fine-grained tasks
    *     scheduling (e.g. timeouts).
@@ -53,6 +64,26 @@ default double availability() {
     return availability(0);
   }
 
+  interface UnreliableChannel {
+
+    Flux<ByteBuf> unreliableChannel(Publisher<ByteBuf> messages);
+
+    Attributes attributes();
+
+    interface Attributes {
+
+      int messageSizeLimit();
+
+      Direction direction();
+    }
+
+    enum Direction {
+      INBOUND,
+      OUTBOUND,
+      BIDIRECTIONAL
+    }
+  }
+
   @FunctionalInterface
   interface Interceptor extends Function<RSocket, RSocket> {}
 

rsocket-rxjava/src/main/java/com/jauntsdn/rsocket/RSocket.java

@@ -16,11 +16,14 @@
 
 package com.jauntsdn.rsocket;
 
+import io.netty.buffer.ByteBuf;
 import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Scheduler;
 import java.util.Optional;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Function;
+import org.reactivestreams.Publisher;
 
 /**
  * Models RSocket interactions as described in
@@ -35,6 +38,14 @@ public interface RSocket extends MessageStreams, Availability {
    */
   Completable metadataPush(Message message);
 
+  /**
+   * @return unreliable unidirectional or bidirectional communication channel if underlying
+   *     transport supports It (e.g. QUIC-DATAGRAM extension, WebTransport-QUIC)
+   */
+  default Optional<UnreliableChannel> unreliableChannel() {
+    return Optional.empty();
+  }
+
   /**
    * @return lightweight {@link ScheduledExecutorService} intended for non-fine-grained tasks
    *     scheduling (e.g. timeouts).
@@ -53,6 +64,26 @@ default double availability() {
     return availability(0);
   }
 
+  interface UnreliableChannel {
+
+    Flowable<ByteBuf> unreliableChannel(Publisher<ByteBuf> messages);
+
+    Attributes attributes();
+
+    interface Attributes {
+
+      int messageSizeLimit();
+
+      Direction direction();
+    }
+
+    enum Direction {
+      INBOUND,
+      OUTBOUND,
+      BIDIRECTIONAL
+    }
+  }
+
   @FunctionalInterface
   interface Interceptor extends Function<RSocket, RSocket> {}