Merge branch 'release/1.0.0-RC3'

Author: robertroeser

.travis.yml

@@ -20,7 +20,7 @@ dist: trusty
 
 matrix:
   include:
-    - jdk: oraclejdk8
+    - jdk: openjdk8
     - jdk: openjdk11
       env: SKIP_RELEASE=true
     - jdk: openjdk12

AUTHORS

@@ -18,3 +18,4 @@ somasun         = somasun <[email protected]>
 stevegury       = Steve Gury <[email protected]>
 tmontgomery     = Todd L. Montgomery <[email protected]>
 yschimke        = Yuri Schimke <[email protected]>
+OlegDokuka      = Oleh Dokuka <[email protected]>

README.md

@@ -23,10 +23,10 @@ Example:
 
 ```groovy
 dependencies {
-    implementation 'io.rsocket:rsocket-core:0.12.2-RC4'
-    implementation 'io.rsocket:rsocket-transport-netty:0.12.2-RC4'
-//    implementation 'io.rsocket:rsocket-core:1.0.0-RC1-SNAPSHOT'
-//    implementation 'io.rsocket:rsocket-transport-netty:1.0.0-RC1-SNAPSHOT'
+    implementation 'io.rsocket:rsocket-core:1.0.0-RC3'
+    implementation 'io.rsocket:rsocket-transport-netty:1.0.0-RC3'
+//    implementation 'io.rsocket:rsocket-core:1.0.0-RC4-SNAPSHOT'
+//    implementation 'io.rsocket:rsocket-transport-netty:1.0.0-RC4-SNAPSHOT'
 }
 ```
 

build.gradle

@@ -46,7 +46,13 @@ subprojects {
     googleJavaFormat {
         toolVersion = '1.6'
     }
-    
+
+    ext {
+      if (project.hasProperty('versionSuffix')) {
+          project.version += project.getProperty('versionSuffix')
+      }
+    }
+
     dependencyManagement {
         imports {
             mavenBom "io.projectreactor:reactor-bom:${ext['reactor-bom.version']}"
@@ -177,6 +183,16 @@ subprojects {
                                     name 'Yuri Schimke'
                                     email '[email protected]'
                                 }
+                                developer {
+                                    id 'OlegDokuka'
+                                    name 'Oleh Dokuka'
+                                    email '[email protected]'
+                                }
+                                developer {
+                                    id 'mostroverkhov'
+                                    name 'Maksym Ostroverkhov'
+                                    email '[email protected]'
+                                }
                             }
 
                             scm {

ci/travis.sh

@@ -11,6 +11,7 @@ elif [ "$TRAVIS_PULL_REQUEST" == "false" ] && [ "$TRAVIS_TAG" == "" ] && [ "$bin
     ./gradlew \
         -PbintrayUser="${bintrayUser}" -PbintrayKey="${bintrayKey}" \
         -PsonatypeUsername="${sonatypeUsername}" -PsonatypePassword="${sonatypePassword}" \
+        -PversionSuffix="-SNAPSHOT" \
         build artifactoryPublish --stacktrace
 
 elif [ "$TRAVIS_PULL_REQUEST" == "false" ] && [ "$TRAVIS_TAG" != "" ] && [ "$bintrayUser" != "" ] ; then

gradle.properties

@@ -11,4 +11,4 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-version=1.0.0-RC2
+version=1.0.0-RC3

rsocket-bom/build.gradle

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2015-2019 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.
+ */
+plugins {
+    id 'java-library'
+    id 'maven'
+    id 'maven-publish'
+    id 'com.jfrog.artifactory'
+    id 'com.jfrog.bintray'
+    id 'io.morethan.jmhreport'
+    id 'me.champeau.gradle.jmh'
+}
+
+description = 'Bill of materials to make sure a consistent set of versions is used for RSocket-Java.'
+configurations.archives.artifacts.clear()
+
+build.doLast {
+    pom {
+        customizePom(it, project)
+    }
+}
+
+dependencies {
+    compile project(':rsocket-core')
+    compile project(':rsocket-test')
+    compile project(':rsocket-load-balancer')
+    compile project(':rsocket-micrometer')
+    compile project(':rsocket-transport-local')
+    compile project(':rsocket-transport-netty')
+}
+
+def customizePom(generatedPom, gradleProject) {
+    //make sure that dependencies are under <dependencyManagement>
+    generatedPom.withXml {
+        if (generatedPom.generatedDependencies.size > 0) {
+            asNode().appendNode('dependencyManagement', asNode().dependencies)
+            asNode().dependencies.replaceNode {}
+        }
+    }
+
+    generatedPom.project {
+        name = 'RSocket-Java Release Train - BOM'
+        description = gradleProject.description
+        url = 'http://rsocket.io'
+        groupId = group
+        packaging = "pom"
+        licenses {
+            license {
+                name = 'The Apache License, Version 2.0'
+                url = 'http://www.apache.org/licenses/LICENSE-2.0.txt'
+            }
+        }
+        developers {
+            developer {
+                id 'robertroeser'
+                name 'Robert Roeser'
+                email '[email protected]'
+            }
+            developer {
+                id 'rdegnan'
+                name 'Ryland Degnan'
+                email '[email protected]'
+            }
+            developer {
+                id 'yschimke'
+                name 'Yuri Schimke'
+                email '[email protected]'
+            }
+            developer {
+                id 'OlegDokuka'
+                name 'Oleh Dokuka'
+                email '[email protected]'
+            }
+            developer {
+                id 'mostroverkhov'
+                name 'Maksym Ostroverkhov'
+                email '[email protected]'
+            }
+        }
+        scm {
+            connection = 'scm:git:git://github.com/rsocket/rsocket-java.git'
+            developerConnection = 'scm:git:ssh://github.com/rsocket/rsocket-java.git'
+            url = 'http://github.com/rsocket/rsocket-java/'
+        }
+        issueManagement {
+            system = "GitHub Issues"
+            url = "https://github.com/rsocket/rsocket-java/issues"
+        }
+    }
+
+    generatedPom.writeTo("$buildDir/poms/rsocket-bom-${version}.xml")
+}
+plugins.withType(MavenPublishPlugin) {
+    publishing {
+        publications {
+            mavenJava(MavenPublication) {
+                pom.withXml {
+                    def sb = asString()
+                    sb.setLength 0
+                    sb.append file("$buildDir/poms/rsocket-bom-${version}.xml").text
+                    println(sb.toString())
+                }
+            }
+        }
+    }
+}
+
+publish.dependsOn("build")
+publishToMavenLocal.dependsOn("build")

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

@@ -31,11 +31,11 @@
 import io.rsocket.internal.ClientSetup;
 import io.rsocket.internal.ServerSetup;
 import io.rsocket.keepalive.KeepAliveHandler;
-import io.rsocket.lease.*;
-import io.rsocket.plugins.DuplexConnectionInterceptor;
-import io.rsocket.plugins.PluginRegistry;
-import io.rsocket.plugins.Plugins;
-import io.rsocket.plugins.RSocketInterceptor;
+import io.rsocket.lease.LeaseStats;
+import io.rsocket.lease.Leases;
+import io.rsocket.lease.RequesterLeaseHandler;
+import io.rsocket.lease.ResponderLeaseHandler;
+import io.rsocket.plugins.*;
 import io.rsocket.resume.*;
 import io.rsocket.transport.ClientTransport;
 import io.rsocket.transport.ServerTransport;
@@ -44,7 +44,10 @@
 import io.rsocket.util.MultiSubscriberRSocket;
 import java.time.Duration;
 import java.util.Objects;
-import java.util.function.*;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
 import reactor.core.publisher.Mono;
 
 /** Factory for creating RSocket clients and servers. */
@@ -93,10 +96,7 @@ default <T extends Closeable> Start<T> transport(ServerTransport<T> transport) {
   public static class ClientRSocketFactory implements ClientTransportAcceptor {
     private static final String CLIENT_TAG = "client";
 
-    private Supplier<Function<RSocket, RSocket>> acceptor =
-        () -> rSocket -> new AbstractRSocket() {};
-
-    private BiFunction<ConnectionSetupPayload, RSocket, RSocket> biAcceptor;
+    private SocketAcceptor acceptor = (setup, sendingSocket) -> Mono.just(new AbstractRSocket() {});
 
     private Consumer<Throwable> errorConsumer = Throwable::printStackTrace;
     private int mtu = 0;
@@ -161,6 +161,11 @@ public ClientRSocketFactory addResponderPlugin(RSocketInterceptor interceptor) {
       return this;
     }
 
+    public ClientRSocketFactory addSocketAcceptorPlugin(SocketAcceptorInterceptor interceptor) {
+      plugins.addSocketAcceptorPlugin(interceptor);
+      return this;
+    }
+
     /**
      * Deprecated as Keep-Alive is not optional according to spec
      *
@@ -268,18 +273,25 @@ public Start<RSocket> transport(Supplier<ClientTransport> transportClient) {
     }
 
     public ClientTransportAcceptor acceptor(Function<RSocket, RSocket> acceptor) {
-      this.acceptor = () -> acceptor;
-      return StartClient::new;
+      return acceptor(() -> acceptor);
     }
 
     public ClientTransportAcceptor acceptor(Supplier<Function<RSocket, RSocket>> acceptor) {
-      this.acceptor = acceptor;
-      return StartClient::new;
+      return acceptor(
+          (SocketAcceptor)
+              (setup, sendingSocket) -> Mono.just(acceptor.get().apply(sendingSocket)));
     }
 
+    @Deprecated
     public ClientTransportAcceptor acceptor(
         BiFunction<ConnectionSetupPayload, RSocket, RSocket> biAcceptor) {
-      this.biAcceptor = biAcceptor;
+      return acceptor(
+          (SocketAcceptor)
+              (setup, sendingSocket) -> Mono.just(biAcceptor.apply(setup, sendingSocket)));
+    }
+
+    public ClientTransportAcceptor acceptor(SocketAcceptor acceptor) {
+      this.acceptor = acceptor;
       return StartClient::new;
     }
 
@@ -346,6 +358,8 @@ public Mono<RSocket> start() {
                     rSocketRequester = new MultiSubscriberRSocket(rSocketRequester);
                   }
 
+                  RSocket wrappedRSocketRequester = plugins.applyRequester(rSocketRequester);
+
                   ByteBuf setupFrame =
                       SetupFrameFlyweight.encode(
                           allocator,
@@ -357,34 +371,38 @@ public Mono<RSocket> start() {
                           dataMimeType,
                           setupPayload);
 
-                  RSocket wrappedRSocketRequester = plugins.applyRequester(rSocketRequester);
-
-                  RSocket rSocketHandler;
-                  if (biAcceptor != null) {
-                    ConnectionSetupPayload setup = ConnectionSetupPayload.create(setupFrame);
-                    rSocketHandler = biAcceptor.apply(setup, wrappedRSocketRequester);
-                  } else {
-                    rSocketHandler = acceptor.get().apply(wrappedRSocketRequester);
-                  }
-
-                  RSocket wrappedRSocketHandler = plugins.applyResponder(rSocketHandler);
-
-                  ResponderLeaseHandler responderLeaseHandler =
-                      isLeaseEnabled
-                          ? new ResponderLeaseHandler.Impl<>(
-                              CLIENT_TAG, allocator, leases.sender(), errorConsumer, leases.stats())
-                          : ResponderLeaseHandler.None;
-
-                  RSocket rSocketResponder =
-                      new RSocketResponder(
-                          allocator,
-                          multiplexer.asServerConnection(),
-                          wrappedRSocketHandler,
-                          payloadDecoder,
-                          errorConsumer,
-                          responderLeaseHandler);
+                  ConnectionSetupPayload setup = ConnectionSetupPayload.create(setupFrame);
+
+                  return plugins
+                      .applySocketAcceptorInterceptor(acceptor)
+                      .accept(setup, wrappedRSocketRequester)
+                      .flatMap(
+                          rSocketHandler -> {
+                            RSocket wrappedRSocketHandler = plugins.applyResponder(rSocketHandler);
+
+                            ResponderLeaseHandler responderLeaseHandler =
+                                isLeaseEnabled
+                                    ? new ResponderLeaseHandler.Impl<>(
+                                        CLIENT_TAG,
+                                        allocator,
+                                        leases.sender(),
+                                        errorConsumer,
+                                        leases.stats())
+                                    : ResponderLeaseHandler.None;
+
+                            RSocket rSocketResponder =
+                                new RSocketResponder(
+                                    allocator,
+                                    multiplexer.asServerConnection(),
+                                    wrappedRSocketHandler,
+                                    payloadDecoder,
+                                    errorConsumer,
+                                    responderLeaseHandler);
 
-                  return wrappedConnection.sendOne(setupFrame).thenReturn(wrappedRSocketRequester);
+                            return wrappedConnection
+                                .sendOne(setupFrame)
+                                .thenReturn(wrappedRSocketRequester);
+                          });
                 });
       }
 
@@ -476,6 +494,11 @@ public ServerRSocketFactory addResponderPlugin(RSocketInterceptor interceptor) {
       return this;
     }
 
+    public ServerRSocketFactory addSocketAcceptorPlugin(SocketAcceptorInterceptor interceptor) {
+      plugins.addSocketAcceptorPlugin(interceptor);
+      return this;
+    }
+
     public ServerTransportAcceptor acceptor(SocketAcceptor acceptor) {
       this.acceptor = acceptor;
       return new ServerStart<>();
@@ -644,7 +667,8 @@ private Mono<Void> acceptSetup(
               }
               RSocket wrappedRSocketRequester = plugins.applyRequester(rSocketRequester);
 
-              return acceptor
+              return plugins
+                  .applySocketAcceptorInterceptor(acceptor)
                   .accept(setupPayload, wrappedRSocketRequester)
                   .onErrorResume(
                       err -> sendError(multiplexer, rejectedSetupError(err)).then(Mono.error(err)))