From ee5a93467a57eb72fa59d8419340b6c004cc8448 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Wed, 9 Jun 2021 11:07:56 +0300 Subject: [PATCH 01/17] replaces artifactory release with github packages for snapshots Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- .github/workflows/gradle-all.yml | 5 ++- .github/workflows/gradle-main.yml | 5 ++- build.gradle | 1 - gradle/artifactory.gradle | 47 ---------------------------- gradle/github-pkg.gradle | 4 +-- gradle/publications.gradle | 1 - gradle/sonotype.gradle | 40 +++++++++++------------ rsocket-bom/build.gradle | 1 - rsocket-core/build.gradle | 1 - rsocket-load-balancer/build.gradle | 1 - rsocket-micrometer/build.gradle | 1 - rsocket-test/build.gradle | 1 - rsocket-transport-local/build.gradle | 1 - rsocket-transport-netty/build.gradle | 1 - 14 files changed, 25 insertions(+), 85 deletions(-) delete mode 100644 gradle/artifactory.gradle diff --git a/.github/workflows/gradle-all.yml b/.github/workflows/gradle-all.yml index 8540539bb..a1ad36d32 100644 --- a/.github/workflows/gradle-all.yml +++ b/.github/workflows/gradle-all.yml @@ -37,9 +37,8 @@ jobs: run: ./gradlew clean build - name: Publish Packages to Artifactory if: ${{ matrix.jdk == '1.8' }} - run: ./gradlew -PbintrayUser="${bintrayUser}" -PbintrayKey="${bintrayKey}" -PversionSuffix="-${githubRef#refs/heads/}-SNAPSHOT" -PbuildNumber="${buildNumber}" artifactoryPublish --stacktrace + run: ./gradlew -PversionSuffix="-${githubRef#refs/heads/}-SNAPSHOT" -PbuildNumber="${buildNumber}" publishMavenPublicationToGitHubPackagesRepository --stacktrace env: - bintrayUser: ${{ secrets.bintrayUser }} - bintrayKey: ${{ secrets.bintrayKey }} + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} githubRef: ${{ github.ref }} buildNumber: ${{ github.run_number }} \ No newline at end of file diff --git a/.github/workflows/gradle-main.yml b/.github/workflows/gradle-main.yml index d8ba3c3d5..950dc80ce 100644 --- a/.github/workflows/gradle-main.yml +++ b/.github/workflows/gradle-main.yml @@ -37,10 +37,9 @@ jobs: run: ./gradlew clean build - name: Publish Packages to Artifactory if: ${{ matrix.jdk == '1.8' }} - run: ./gradlew -PbintrayUser="${bintrayUser}" -PbintrayKey="${bintrayKey}" -PversionSuffix="-SNAPSHOT" -PbuildNumber="${buildNumber}" artifactoryPublish --stacktrace + run: ./gradlew -PversionSuffix="-SNAPSHOT" -PbuildNumber="${buildNumber}" publishMavenPublicationToGitHubPackagesRepository --stacktrace env: - bintrayUser: ${{ secrets.bintrayUser }} - bintrayKey: ${{ secrets.bintrayKey }} + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} buildNumber: ${{ github.run_number }} - name: Aggregate test reports with ciMate if: always() diff --git a/build.gradle b/build.gradle index 8a7d74156..00c5d2b9d 100644 --- a/build.gradle +++ b/build.gradle @@ -16,7 +16,6 @@ plugins { id 'com.github.sherter.google-java-format' version '0.8' apply false - id 'com.jfrog.artifactory' version '4.21.0' apply false id 'me.champeau.gradle.jmh' version '0.5.0' apply false id 'io.spring.dependency-management' version '1.0.11.RELEASE' apply false id 'io.morethan.jmhreport' version '0.9.0' apply false diff --git a/gradle/artifactory.gradle b/gradle/artifactory.gradle deleted file mode 100644 index cdffb2741..000000000 --- a/gradle/artifactory.gradle +++ /dev/null @@ -1,47 +0,0 @@ -/* - * 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. - */ - -if (project.hasProperty('bintrayUser') && project.hasProperty('bintrayKey')) { - - subprojects { - plugins.withId('com.jfrog.artifactory') { - artifactory { - publish { - contextUrl = 'https://oss.jfrog.org' - - repository { - repoKey = 'oss-snapshot-local' - - // Credentials for oss.jfrog.org are a user's Bintray credentials - username = project.property('bintrayUser') - password = project.property('bintrayKey') - } - - defaults { - publications(publishing.publications.maven) - } - - if (project.hasProperty('buildNumber')) { - clientConfig.info.setBuildNumber(project.property('buildNumber').toString()) - } - } - } - tasks.named("artifactoryPublish").configure { - onlyIf { System.getenv('SKIP_RELEASE') != "true" } - } - } - } -} diff --git a/gradle/github-pkg.gradle b/gradle/github-pkg.gradle index a9c1a24ff..98a68ecdb 100644 --- a/gradle/github-pkg.gradle +++ b/gradle/github-pkg.gradle @@ -8,8 +8,8 @@ subprojects { name = "GitHubPackages" url = uri("https://maven.pkg.github.com/rsocket/rsocket-java") credentials { - username = project.findProperty("gpr.user") ?: System.getenv("USERNAME") - password = project.findProperty("gpr.key") ?: System.getenv("TOKEN") + username = project.findProperty("gpr.user") ?: System.getenv("GITHUB_ACTOR") + password = project.findProperty("gpr.key") ?: System.getenv("GITHUB_TOKEN") } } } diff --git a/gradle/publications.gradle b/gradle/publications.gradle index 383b81c77..c405a4032 100644 --- a/gradle/publications.gradle +++ b/gradle/publications.gradle @@ -1,4 +1,3 @@ -apply from: "${rootDir}/gradle/artifactory.gradle" apply from: "${rootDir}/gradle/github-pkg.gradle" apply from: "${rootDir}/gradle/sonotype.gradle" diff --git a/gradle/sonotype.gradle b/gradle/sonotype.gradle index dc002d2c9..1effd76b0 100644 --- a/gradle/sonotype.gradle +++ b/gradle/sonotype.gradle @@ -1,31 +1,29 @@ subprojects { if (project.hasProperty('sonatypeUsername') && project.hasProperty('sonatypePassword')) { - plugins.withType(JavaLibraryPlugin) { - plugins.withType(MavenPublishPlugin) { - plugins.withType(SigningPlugin) { + plugins.withType(MavenPublishPlugin) { + plugins.withType(SigningPlugin) { - signing { - //requiring signature if there is a publish task that is not to MavenLocal - required { gradle.taskGraph.allTasks.any { it.name.toLowerCase().contains("publish") && !it.name.contains("MavenLocal") } } - def signingKey = project.findProperty("signingKey") - def signingPassword = project.findProperty("signingPassword") + signing { + //requiring signature if there is a publish task that is not to MavenLocal + required { gradle.taskGraph.allTasks.any { it.name.toLowerCase().contains("publish") && !it.name.contains("MavenLocal") } } + def signingKey = project.findProperty("signingKey") + def signingPassword = project.findProperty("signingPassword") - useInMemoryPgpKeys(signingKey, signingPassword) + useInMemoryPgpKeys(signingKey, signingPassword) - afterEvaluate { - sign publishing.publications.maven - } + afterEvaluate { + sign publishing.publications.maven } + } - publishing { - repositories { - maven { - name = "sonatype" - url = "https://s01.oss.sonatype.org/service/local/staging/deploy/maven2" - credentials { - username project.findProperty("sonatypeUsername") - password project.findProperty("sonatypePassword") - } + publishing { + repositories { + maven { + name = "sonatype" + url = "https://oss.sonatype.org/service/local/staging/deploy/maven2" + credentials { + username project.findProperty("sonatypeUsername") + password project.findProperty("sonatypePassword") } } } diff --git a/rsocket-bom/build.gradle b/rsocket-bom/build.gradle index 3f313b6bb..a75ab3bc8 100755 --- a/rsocket-bom/build.gradle +++ b/rsocket-bom/build.gradle @@ -17,7 +17,6 @@ plugins { id 'java-platform' id 'maven-publish' id 'signing' - id 'com.jfrog.artifactory' } description = 'RSocket Java Bill of materials.' diff --git a/rsocket-core/build.gradle b/rsocket-core/build.gradle index 013c283b2..fbeee37ce 100644 --- a/rsocket-core/build.gradle +++ b/rsocket-core/build.gradle @@ -18,7 +18,6 @@ plugins { id 'java-library' id 'maven-publish' id 'signing' - id 'com.jfrog.artifactory' id 'io.morethan.jmhreport' id 'me.champeau.gradle.jmh' } diff --git a/rsocket-load-balancer/build.gradle b/rsocket-load-balancer/build.gradle index c39967e64..d70e5b2cc 100644 --- a/rsocket-load-balancer/build.gradle +++ b/rsocket-load-balancer/build.gradle @@ -18,7 +18,6 @@ plugins { id 'java-library' id 'maven-publish' id 'signing' - id 'com.jfrog.artifactory' } dependencies { diff --git a/rsocket-micrometer/build.gradle b/rsocket-micrometer/build.gradle index b451235ac..fd89aeae0 100644 --- a/rsocket-micrometer/build.gradle +++ b/rsocket-micrometer/build.gradle @@ -18,7 +18,6 @@ plugins { id 'java-library' id 'maven-publish' id 'signing' - id 'com.jfrog.artifactory' } dependencies { diff --git a/rsocket-test/build.gradle b/rsocket-test/build.gradle index 681e41189..282a65829 100644 --- a/rsocket-test/build.gradle +++ b/rsocket-test/build.gradle @@ -18,7 +18,6 @@ plugins { id 'java-library' id 'maven-publish' id 'signing' - id 'com.jfrog.artifactory' } dependencies { diff --git a/rsocket-transport-local/build.gradle b/rsocket-transport-local/build.gradle index 1f9f972f9..8c855f26c 100644 --- a/rsocket-transport-local/build.gradle +++ b/rsocket-transport-local/build.gradle @@ -18,7 +18,6 @@ plugins { id 'java-library' id 'maven-publish' id 'signing' - id 'com.jfrog.artifactory' } dependencies { diff --git a/rsocket-transport-netty/build.gradle b/rsocket-transport-netty/build.gradle index b03afbc6d..201d56bbf 100644 --- a/rsocket-transport-netty/build.gradle +++ b/rsocket-transport-netty/build.gradle @@ -18,7 +18,6 @@ plugins { id 'java-library' id 'maven-publish' id 'signing' - id 'com.jfrog.artifactory' id "com.google.osdetector" version "1.4.0" } From c0ae0f5f356c005bd81febfd653f02746174c13b Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Wed, 9 Jun 2021 11:20:07 +0300 Subject: [PATCH 02/17] adds bom to ghp releases Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- gradle/github-pkg.gradle | 26 ++++++++++++-------------- 1 file changed, 12 insertions(+), 14 deletions(-) diff --git a/gradle/github-pkg.gradle b/gradle/github-pkg.gradle index 98a68ecdb..f53413766 100644 --- a/gradle/github-pkg.gradle +++ b/gradle/github-pkg.gradle @@ -1,23 +1,21 @@ subprojects { - plugins.withType(JavaLibraryPlugin) { - plugins.withType(MavenPublishPlugin) { - publishing { - repositories { - maven { - name = "GitHubPackages" - url = uri("https://maven.pkg.github.com/rsocket/rsocket-java") - credentials { - username = project.findProperty("gpr.user") ?: System.getenv("GITHUB_ACTOR") - password = project.findProperty("gpr.key") ?: System.getenv("GITHUB_TOKEN") - } + plugins.withType(MavenPublishPlugin) { + publishing { + repositories { + maven { + name = "GitHubPackages" + url = uri("https://maven.pkg.github.com/rsocket/rsocket-java") + credentials { + username = project.findProperty("gpr.user") ?: System.getenv("GITHUB_ACTOR") + password = project.findProperty("gpr.key") ?: System.getenv("GITHUB_TOKEN") } } } + } - tasks.named("publish").configure { - onlyIf { System.getenv('SKIP_RELEASE') != "true" } - } + tasks.named("publish").configure { + onlyIf { System.getenv('SKIP_RELEASE') != "true" } } } } \ No newline at end of file From b12d46c94eac3f9b5fe596079973e41f0642da15 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Wed, 9 Jun 2021 11:46:58 +0300 Subject: [PATCH 03/17] bumps versions Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- README.md | 12 ++++++------ gradle.properties | 4 ++-- 2 files changed, 8 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 3c7e87976..cda6d3c0a 100644 --- a/README.md +++ b/README.md @@ -15,7 +15,7 @@ Learn more at http://rsocket.io ## Build and Binaries -[![Build Status](https://travis-ci.org/rsocket/rsocket-java.svg?branch=develop)](https://travis-ci.org/rsocket/rsocket-java) +[![Main Branches Java CI](https://github.com/rsocket/rsocket-java/actions/workflows/gradle-main.yml/badge.svg?branch=1.0.x)](https://github.com/rsocket/rsocket-java/actions/workflows/gradle-main.yml) Releases and milestones are available via Maven Central. @@ -27,8 +27,8 @@ repositories { maven { url 'https://repo.spring.io/milestone' } // Reactor milestones (if needed) } dependencies { - implementation 'io.rsocket:rsocket-core:1.0.4' - implementation 'io.rsocket:rsocket-transport-netty:1.0.4' + implementation 'io.rsocket:rsocket-core:1.0.5' + implementation 'io.rsocket:rsocket-transport-netty:1.0.5' } ``` @@ -38,12 +38,12 @@ Example: ```groovy repositories { - maven { url 'https://oss.jfrog.org/oss-snapshot-local' } + maven { url 'https://maven.pkg.github.com/rsocket/rsocket-java' } maven { url 'https://repo.spring.io/snapshot' } // Reactor snapshots (if needed) } dependencies { - implementation 'io.rsocket:rsocket-core:1.0.5-SNAPSHOT' - implementation 'io.rsocket:rsocket-transport-netty:1.0.5-SNAPSHOT' + implementation 'io.rsocket:rsocket-core:1.0.6-SNAPSHOT' + implementation 'io.rsocket:rsocket-transport-netty:1.0.6-SNAPSHOT' } ``` diff --git a/gradle.properties b/gradle.properties index f75f86589..4f89f4d88 100644 --- a/gradle.properties +++ b/gradle.properties @@ -11,5 +11,5 @@ # See the License for the specific language governing permissions and # limitations under the License. # -version=1.0.5 -perfBaselineVersion=1.0.4 +version=1.0.6 +perfBaselineVersion=1.0.5 From f02867d1fda2f2050a35191c468aba764c64ec1d Mon Sep 17 00:00:00 2001 From: Oleh Dokuka <5380167+OlegDokuka@users.noreply.github.com> Date: Wed, 9 Jun 2021 14:15:14 +0300 Subject: [PATCH 04/17] sets jcstress mode to quick --- rsocket-core/build.gradle | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rsocket-core/build.gradle b/rsocket-core/build.gradle index 71e64b04e..3d4759af0 100644 --- a/rsocket-core/build.gradle +++ b/rsocket-core/build.gradle @@ -46,7 +46,7 @@ dependencies { } jcstress { - mode = 'default' //quick, default, tough + mode = 'quick' //quick, default, tough jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.7" } @@ -56,4 +56,4 @@ jar { } } -description = "Core functionality for the RSocket library" \ No newline at end of file +description = "Core functionality for the RSocket library" From a87abdfb883a49e815b000d2ddeb493e3deb0966 Mon Sep 17 00:00:00 2001 From: Andrii Rodionov Date: Thu, 12 Aug 2021 18:03:56 +0300 Subject: [PATCH 05/17] fixes typo in LeaseSpec initialisation (#1024) --- rsocket-core/src/main/java/io/rsocket/core/LeaseSpec.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rsocket-core/src/main/java/io/rsocket/core/LeaseSpec.java b/rsocket-core/src/main/java/io/rsocket/core/LeaseSpec.java index 3947f296a..ad4b36e3a 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/LeaseSpec.java +++ b/rsocket-core/src/main/java/io/rsocket/core/LeaseSpec.java @@ -38,7 +38,7 @@ public LeaseSpec sender(LeaseSender sender) { * no leases is available */ public LeaseSpec maxPendingRequests(int maxPendingRequests) { - this.maxPendingRequests = 0; + this.maxPendingRequests = maxPendingRequests; return this; } } From 1f7191456f2961d1d29f682fb609c7d0783ef9a2 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Tue, 9 Nov 2021 13:31:05 +0200 Subject: [PATCH 06/17] relaxes connection dispose to avoid dropped error Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- .../java/io/rsocket/transport/netty/TcpDuplexConnection.java | 1 - .../io/rsocket/transport/netty/WebsocketDuplexConnection.java | 1 - 2 files changed, 2 deletions(-) diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java index f9ac705b1..85874f44d 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java @@ -64,7 +64,6 @@ public SocketAddress remoteAddress() { @Override protected void doOnClose() { - sender.dispose(); connection.dispose(); } diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java index c81f040da..140cfc59f 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/WebsocketDuplexConnection.java @@ -69,7 +69,6 @@ public SocketAddress remoteAddress() { @Override protected void doOnClose() { - sender.dispose(); connection.dispose(); } From efd1269a149e14e1a4b6cfeacea5ccd740fd219c Mon Sep 17 00:00:00 2001 From: Oleh Dokuka <5380167+OlegDokuka@users.noreply.github.com> Date: Tue, 9 Nov 2021 23:27:46 +0200 Subject: [PATCH 07/17] adds first frame handling timeout (#1027) Co-authored-by: Rossen Stoyanchev --- .github/workflows/gradle-all.yml | 5 +++- .../java/io/rsocket/core/RSocketServer.java | 30 +++++++++++++++---- .../java/io/rsocket/core/ServerSetup.java | 13 ++++++++ .../core/SetupHandlingDuplexConnection.java | 1 + .../io/rsocket/core/RSocketServerTest.java | 28 +++++++++++++++++ .../java/io/rsocket/test/TransportTest.java | 21 +++++++++++-- 6 files changed, 89 insertions(+), 9 deletions(-) diff --git a/.github/workflows/gradle-all.yml b/.github/workflows/gradle-all.yml index 03e6a4e68..8826f511a 100644 --- a/.github/workflows/gradle-all.yml +++ b/.github/workflows/gradle-all.yml @@ -142,7 +142,10 @@ jobs: run: chmod +x gradlew - name: Publish Packages to Artifactory if: ${{ matrix.jdk == '1.8' }} - run: ./gradlew -PversionSuffix="-${githubRef#refs/heads/}-SNAPSHOT" -PbuildNumber="${buildNumber}" publishMavenPublicationToGitHubPackagesRepository --no-daemon --stacktrace + run: | + githubRef="${githubRef#refs/heads/}" + githubRef="${githubRef////-}" + ./gradlew -PversionSuffix="-${githubRef}-SNAPSHOT" -PbuildNumber="${buildNumber}" publishMavenPublicationToGitHubPackagesRepository --no-daemon --stacktrace env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} githubRef: ${{ github.ref }} diff --git a/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java b/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java index 5ec33e76f..3208bb4fd 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RSocketServer.java @@ -41,6 +41,7 @@ import io.rsocket.plugins.RequestInterceptor; import io.rsocket.resume.SessionManager; import io.rsocket.transport.ServerTransport; +import java.time.Duration; import java.util.Objects; import java.util.function.Consumer; import java.util.function.Supplier; @@ -70,6 +71,7 @@ public final class RSocketServer { private int mtu = 0; private int maxInboundPayloadSize = Integer.MAX_VALUE; private PayloadDecoder payloadDecoder = PayloadDecoder.DEFAULT; + private Duration timeout = Duration.ofMinutes(1); private RSocketServer() {} @@ -223,6 +225,23 @@ public RSocketServer maxInboundPayloadSize(int maxInboundPayloadSize) { return this; } + /** + * Specify the max time to wait for the first frame (e.g. {@code SETUP}) on an accepted + * connection. + * + *

By default this is set to 1 minute. + * + * @param timeout duration + * @return the same instance for method chaining + */ + public RSocketServer maxTimeToFirstFrame(Duration timeout) { + if (timeout.isNegative() || timeout.isZero()) { + throw new IllegalArgumentException("Setup Handling Timeout should be greater than zero"); + } + this.timeout = timeout; + return this; + } + /** * When this is set, frames larger than the given maximum transmission unit (mtu) size value are * fragmented. @@ -287,7 +306,7 @@ public RSocketServer payloadDecoder(PayloadDecoder decoder) { public Mono bind(ServerTransport transport) { return Mono.defer( new Supplier>() { - final ServerSetup serverSetup = serverSetup(); + final ServerSetup serverSetup = serverSetup(timeout); @Override public Mono get() { @@ -326,7 +345,7 @@ public ServerTransport.ConnectionAcceptor asConnectionAcceptor() { public ServerTransport.ConnectionAcceptor asConnectionAcceptor(int maxFrameLength) { assertValidateSetup(maxFrameLength, maxInboundPayloadSize, mtu); return new ServerTransport.ConnectionAcceptor() { - private final ServerSetup serverSetup = serverSetup(); + private final ServerSetup serverSetup = serverSetup(timeout); @Override public Mono apply(DuplexConnection connection) { @@ -469,12 +488,13 @@ private Mono acceptSetup( }); } - private ServerSetup serverSetup() { - return resume != null ? createSetup() : new ServerSetup.DefaultServerSetup(); + private ServerSetup serverSetup(Duration timeout) { + return resume != null ? createSetup(timeout) : new ServerSetup.DefaultServerSetup(timeout); } - ServerSetup createSetup() { + ServerSetup createSetup(Duration timeout) { return new ServerSetup.ResumableServerSetup( + timeout, new SessionManager(), resume.getSessionDuration(), resume.getStreamTimeout(), diff --git a/rsocket-core/src/main/java/io/rsocket/core/ServerSetup.java b/rsocket-core/src/main/java/io/rsocket/core/ServerSetup.java index e716b8fcb..2d367bd73 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/ServerSetup.java +++ b/rsocket-core/src/main/java/io/rsocket/core/ServerSetup.java @@ -36,9 +36,16 @@ abstract class ServerSetup { + final Duration timeout; + + protected ServerSetup(Duration timeout) { + this.timeout = timeout; + } + Mono> init(DuplexConnection connection) { return Mono.>create( sink -> sink.onRequest(__ -> new SetupHandlingDuplexConnection(connection, sink))) + .timeout(this.timeout) .or(connection.onClose().then(Mono.error(ClosedChannelException::new))); } @@ -57,6 +64,10 @@ void sendError(DuplexConnection duplexConnection, RSocketErrorException exceptio static class DefaultServerSetup extends ServerSetup { + DefaultServerSetup(Duration timeout) { + super(timeout); + } + @Override public Mono acceptRSocketSetup( ByteBuf frame, @@ -86,11 +97,13 @@ static class ResumableServerSetup extends ServerSetup { private final boolean cleanupStoreOnKeepAlive; ResumableServerSetup( + Duration timeout, SessionManager sessionManager, Duration resumeSessionDuration, Duration resumeStreamTimeout, Function resumeStoreFactory, boolean cleanupStoreOnKeepAlive) { + super(timeout); this.sessionManager = sessionManager; this.resumeSessionDuration = resumeSessionDuration; this.resumeStreamTimeout = resumeStreamTimeout; diff --git a/rsocket-core/src/main/java/io/rsocket/core/SetupHandlingDuplexConnection.java b/rsocket-core/src/main/java/io/rsocket/core/SetupHandlingDuplexConnection.java index b6bc87513..2da572de3 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/SetupHandlingDuplexConnection.java +++ b/rsocket-core/src/main/java/io/rsocket/core/SetupHandlingDuplexConnection.java @@ -96,6 +96,7 @@ public void request(long n) { @Override public void cancel() { + source.dispose(); s.cancel(); } diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java b/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java index 08555740c..24bf95215 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RSocketServerTest.java @@ -29,11 +29,13 @@ import io.rsocket.test.util.TestServerTransport; import java.time.Duration; import java.util.Random; +import org.assertj.core.api.Assertions; import org.junit.jupiter.api.Test; import reactor.core.Scannable; import reactor.core.publisher.Mono; import reactor.core.publisher.Sinks; import reactor.test.StepVerifier; +import reactor.test.scheduler.VirtualTimeScheduler; public class RSocketServerTest { @@ -60,6 +62,32 @@ public void unexpectedFramesBeforeSetupFrame() { .hasNoLeaks(); } + @Test + public void timeoutOnNoFirstFrame() { + final VirtualTimeScheduler scheduler = VirtualTimeScheduler.getOrSet(); + try { + TestServerTransport transport = new TestServerTransport(); + RSocketServer.create().maxTimeToFirstFrame(Duration.ofMinutes(2)).bind(transport).block(); + + final TestDuplexConnection duplexConnection = transport.connect(); + + scheduler.advanceTimeBy(Duration.ofMinutes(1)); + + Assertions.assertThat(duplexConnection.isDisposed()).isFalse(); + + scheduler.advanceTimeBy(Duration.ofMinutes(1)); + + StepVerifier.create(duplexConnection.onClose()) + .expectSubscription() + .expectComplete() + .verify(Duration.ofSeconds(10)); + + FrameAssert.assertThat(duplexConnection.pollFrame()).isNull(); + } finally { + VirtualTimeScheduler.reset(); + } + } + @Test public void ensuresMaxFrameLengthCanNotBeLessThenMtu() { RSocketServer.create() diff --git a/rsocket-test/src/main/java/io/rsocket/test/TransportTest.java b/rsocket-test/src/main/java/io/rsocket/test/TransportTest.java index 5384c7e8d..902a5844c 100644 --- a/rsocket-test/src/main/java/io/rsocket/test/TransportTest.java +++ b/rsocket-test/src/main/java/io/rsocket/test/TransportTest.java @@ -39,6 +39,7 @@ import java.io.InputStreamReader; import java.net.SocketAddress; import java.time.Duration; +import java.util.Arrays; import java.util.concurrent.CancellationException; import java.util.concurrent.Executors; import java.util.concurrent.ThreadLocalRandom; @@ -96,7 +97,18 @@ default void close() { getTransportPair().responder.awaitAllInteractionTermination(getTimeout()); getTransportPair().dispose(); getTransportPair().awaitClosed(); - RuntimeException throwable = new RuntimeException(); + RuntimeException throwable = + new RuntimeException() { + @Override + public synchronized Throwable fillInStackTrace() { + return this; + } + + @Override + public String getMessage() { + return Arrays.toString(getSuppressed()); + } + }; try { getTransportPair().byteBufAllocator2.assertHasNoLeaks(); @@ -776,8 +788,11 @@ public void onSubscribe(Subscription s) { @Override public void onNext(ByteBuf buf) { - actual.onNext(buf); - buf.release(); + try { + actual.onNext(buf); + } finally { + buf.release(); + } } Mono onClose() { From 1927bf40164f99cb7bcedd9bde79548193f71aca Mon Sep 17 00:00:00 2001 From: Oleh Dokuka <5380167+OlegDokuka@users.noreply.github.com> Date: Sat, 13 Nov 2021 14:57:03 +0200 Subject: [PATCH 08/17] bumps libs versions and provides few UnboundedProcessor fixes (#1028) --- build.gradle | 25 ++++++++++-------- gradle/wrapper/gradle-wrapper.properties | 2 +- .../rsocket/internal/UnboundedProcessor.java | 11 +++----- .../rsocket/resume/ClientRSocketSession.java | 18 ++++++++++++- .../resume/ResumableDuplexConnection.java | 2 +- .../java/io/rsocket/test/TransportTest.java | 26 ++++++++++++------- 6 files changed, 53 insertions(+), 31 deletions(-) diff --git a/build.gradle b/build.gradle index e8dbaedd0..5be434a4a 100644 --- a/build.gradle +++ b/build.gradle @@ -16,10 +16,11 @@ plugins { id 'com.github.sherter.google-java-format' version '0.9' apply false - id 'me.champeau.jmh' version '0.6.4' apply false + id 'me.champeau.jmh' version '0.6.6' apply false id 'io.spring.dependency-management' version '1.0.11.RELEASE' apply false id 'io.morethan.jmhreport' version '0.9.0' apply false - id "io.github.reyerizo.gradle.jcstress" version "0.8.11" apply false + id 'io.github.reyerizo.gradle.jcstress' version '0.8.11' apply false + id 'com.github.vlsi.gradle-extensions' version '1.76' apply false } boolean isCiServer = ["CI", "CONTINUOUS_INTEGRATION", "TRAVIS", "CIRCLECI", "bamboo_planKey", "GITHUB_ACTION"].with { @@ -30,19 +31,20 @@ boolean isCiServer = ["CI", "CONTINUOUS_INTEGRATION", "TRAVIS", "CIRCLECI", "bam subprojects { apply plugin: 'io.spring.dependency-management' apply plugin: 'com.github.sherter.google-java-format' + apply plugin: 'com.github.vlsi.gradle-extensions' - ext['reactor-bom.version'] = '2020.0.7' + ext['reactor-bom.version'] = '2020.0.12' ext['logback.version'] = '1.2.3' - ext['netty-bom.version'] = '4.1.64.Final' - ext['netty-boringssl.version'] = '2.0.39.Final' + ext['netty-bom.version'] = '4.1.70.Final' + ext['netty-boringssl.version'] = '2.0.45.Final' ext['hdrhistogram.version'] = '2.1.12' - ext['mockito.version'] = '3.10.0' + ext['mockito.version'] = '3.12.4' ext['slf4j.version'] = '1.7.30' - ext['jmh.version'] = '1.31' - ext['junit.version'] = '5.7.2' + ext['jmh.version'] = '1.33' + ext['junit.version'] = '5.8.1' ext['hamcrest.version'] = '1.3' - ext['micrometer.version'] = '1.6.7' - ext['assertj.version'] = '3.19.0' + ext['micrometer.version'] = '1.7.5' + ext['assertj.version'] = '3.21.0' ext['netflix.limits.version'] = '0.3.6' ext['bouncycastle-bcpkix.version'] = '1.68' @@ -123,6 +125,7 @@ subprojects { } plugins.withType(JavaPlugin) { + compileJava { sourceCompatibility = 1.8 @@ -198,7 +201,7 @@ subprojects { if (JavaVersion.current().isJava9Compatible()) { println "Java 9+: lowering MaxGCPauseMillis to 20ms in ${project.name} ${name}" println "Java 9+: enabling leak detection [ADVANCED]" - jvmArgs = ["-XX:MaxGCPauseMillis=20", "-Dio.netty.leakDetection.level=ADVANCED"] + jvmArgs = ["-XX:MaxGCPauseMillis=20", "-Dio.netty.leakDetection.level=ADVANCED", "-Dio.netty.leakDetection.samplingInterval=32"] } systemProperty("java.awt.headless", "true") diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties index 0f80bbf51..e750102e0 100644 --- a/gradle/wrapper/gradle-wrapper.properties +++ b/gradle/wrapper/gradle-wrapper.properties @@ -1,5 +1,5 @@ distributionBase=GRADLE_USER_HOME distributionPath=wrapper/dists -distributionUrl=https\://services.gradle.org/distributions/gradle-7.0.2-bin.zip +distributionUrl=https\://services.gradle.org/distributions/gradle-7.3-bin.zip zipStoreBase=GRADLE_USER_HOME zipStorePath=wrapper/dists diff --git a/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java b/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java index c3278a09c..c529b615d 100644 --- a/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java +++ b/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java @@ -146,10 +146,7 @@ public void onNextPrioritized(ByteBuf t) { return; } - if (isWorkInProgress(previousState) - || isCancelled(previousState) - || isDisposed(previousState) - || isTerminated(previousState)) { + if (isWorkInProgress(previousState)) { return; } @@ -185,10 +182,7 @@ public void onNext(ByteBuf t) { return; } - if (isWorkInProgress(previousState) - || isCancelled(previousState) - || isDisposed(previousState) - || isTerminated(previousState)) { + if (isWorkInProgress(previousState)) { return; } @@ -449,6 +443,7 @@ public void subscribe(CoreSubscriber actual) { if (isCancelled(previousState)) { clearAndFinalize(this); + return; } if (isDisposed(previousState)) { diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java b/rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java index 8dcf67a0b..2f2f29001 100644 --- a/rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java +++ b/rsocket-core/src/main/java/io/rsocket/resume/ClientRSocketSession.java @@ -36,6 +36,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import reactor.core.CoreSubscriber; +import reactor.core.Disposable; import reactor.core.publisher.Mono; import reactor.core.publisher.Operators; import reactor.util.function.Tuple2; @@ -58,6 +59,7 @@ public class ClientRSocketSession final boolean cleanupStoreOnKeepAlive; final ByteBuf resumeToken; final String session; + final Disposable reconnectDisposable; volatile Subscription s; static final AtomicReferenceFieldUpdater S = @@ -110,10 +112,22 @@ public ClientRSocketSession( this.resumableConnection = resumableDuplexConnection; resumableDuplexConnection.onClose().doFinally(__ -> dispose()).subscribe(); - resumableDuplexConnection.onActiveConnectionClosed().subscribe(this::reconnect); + + this.reconnectDisposable = + resumableDuplexConnection.onActiveConnectionClosed().subscribe(this::reconnect); } void reconnect(int index) { + if (this.s == Operators.cancelledSubscription()) { + if (logger.isDebugEnabled()) { + logger.debug( + "Side[client]|Session[{}]. Connection[{}] is lost. Reconnecting rejected since session is closed", + session, + index); + } + return; + } + keepAliveSupport.stop(); if (logger.isDebugEnabled()) { logger.debug( @@ -147,6 +161,8 @@ public void onImpliedPosition(long remoteImpliedPos) { @Override public void dispose() { Operators.terminate(S, this); + + reconnectDisposable.dispose(); resumableConnection.dispose(); resumableFramesStore.dispose(); diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java b/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java index 933ac09ca..7ade3e59b 100644 --- a/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java +++ b/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java @@ -253,7 +253,7 @@ void dispose(@Nullable Throwable e) { framesSaverDisposable.dispose(); activeReceivingSubscriber.dispose(); - savableFramesSender.dispose(); + savableFramesSender.onComplete(); onConnectionClosedSink.tryEmitComplete(); if (e != null) { diff --git a/rsocket-test/src/main/java/io/rsocket/test/TransportTest.java b/rsocket-test/src/main/java/io/rsocket/test/TransportTest.java index 902a5844c..570a7de2f 100644 --- a/rsocket-test/src/main/java/io/rsocket/test/TransportTest.java +++ b/rsocket-test/src/main/java/io/rsocket/test/TransportTest.java @@ -57,6 +57,7 @@ import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; import reactor.core.Disposable; +import reactor.core.Disposables; import reactor.core.Exceptions; import reactor.core.Fuseable; import reactor.core.publisher.Flux; @@ -641,7 +642,11 @@ public String expectedPayloadMetadata() { } public void awaitClosed() { - server.onClose().and(client.onClose()).block(Duration.ofMinutes(1)); + server + .onClose() + .onErrorResume(__ -> Mono.empty()) + .and(client.onClose().onErrorResume(__ -> Mono.empty())) + .block(Duration.ofMinutes(1)); } private static class AsyncDuplexConnection implements DuplexConnection { @@ -706,6 +711,7 @@ private static class DisconnectingDuplexConnection implements DuplexConnection { private final String tag; final DuplexConnection source; final Duration delay; + final Disposable.Swap disposables = Disposables.swap(); DisconnectingDuplexConnection(String tag, DuplexConnection source, Duration delay) { this.tag = tag; @@ -715,6 +721,7 @@ private static class DisconnectingDuplexConnection implements DuplexConnection { @Override public void dispose() { + disposables.dispose(); source.dispose(); } @@ -743,14 +750,15 @@ public Flux receive() { bb -> { if (!receivedFirst) { receivedFirst = true; - Mono.delay(delay) - .takeUntilOther(source.onClose()) - .subscribe( - __ -> { - logger.warn( - "Tag {}. Disposing Connection[{}]", tag, source.hashCode()); - source.dispose(); - }); + disposables.replace( + Mono.delay(delay) + .takeUntilOther(source.onClose()) + .subscribe( + __ -> { + logger.warn( + "Tag {}. Disposing Connection[{}]", tag, source.hashCode()); + source.dispose(); + })); } }); } From 16fdb87bada4050c0ff470b50062c48a87ba7e10 Mon Sep 17 00:00:00 2001 From: olme04 <86063649+olme04@users.noreply.github.com> Date: Sat, 13 Nov 2021 16:17:38 +0300 Subject: [PATCH 09/17] eliminate boxing in RequesterResponderSupport when using IntObjectMap (#1029) --- .../core/RequesterResponderSupport.java | 21 +++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/rsocket-core/src/main/java/io/rsocket/core/RequesterResponderSupport.java b/rsocket-core/src/main/java/io/rsocket/core/RequesterResponderSupport.java index 52db6e198..bea7dc1aa 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RequesterResponderSupport.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RequesterResponderSupport.java @@ -7,6 +7,7 @@ import io.rsocket.RSocket; import io.rsocket.frame.decoder.PayloadDecoder; import io.rsocket.plugins.RequestInterceptor; +import java.util.Objects; import java.util.function.Function; import reactor.util.annotation.Nullable; @@ -118,9 +119,14 @@ public int addAndGetNextStreamId(FrameHandler frameHandler) { } public synchronized boolean add(int streamId, FrameHandler frameHandler) { - final FrameHandler previousHandler = this.activeStreams.putIfAbsent(streamId, frameHandler); - - return previousHandler == null; + final IntObjectMap activeStreams = this.activeStreams; + // copy of Map.putIfAbsent(key, value) without `streamId` boxing + final FrameHandler previousHandler = activeStreams.get(streamId); + if (previousHandler == null) { + activeStreams.put(streamId, frameHandler); + return true; + } + return false; } /** @@ -143,6 +149,13 @@ public synchronized FrameHandler get(int streamId) { * instance equals to the passed one */ public synchronized boolean remove(int streamId, FrameHandler frameHandler) { - return this.activeStreams.remove(streamId, frameHandler); + final IntObjectMap activeStreams = this.activeStreams; + // copy of Map.remove(key, value) without `streamId` boxing + final FrameHandler curValue = activeStreams.get(streamId); + if (!Objects.equals(curValue, frameHandler)) { + return false; + } + activeStreams.remove(streamId); + return true; } } From 4fba9d4e72bea4fab64a59ed9329e02b185bdd7d Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Tue, 25 Jan 2022 14:50:19 +0200 Subject: [PATCH 10/17] adds tests for WeightedLoadbalanceStrategy Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- .../rsocket/loadbalance/FrugalQuantile.java | 11 +- .../java/io/rsocket/loadbalance/Median.java | 9 + .../WeightedLoadbalanceStrategy.java | 5 +- .../WeightedLoadbalanceStrategyTest.java | 237 ++++++++++++++++++ 4 files changed, 260 insertions(+), 2 deletions(-) create mode 100644 rsocket-core/src/test/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategyTest.java diff --git a/rsocket-core/src/main/java/io/rsocket/loadbalance/FrugalQuantile.java b/rsocket-core/src/main/java/io/rsocket/loadbalance/FrugalQuantile.java index a15d88529..cdbdc19b3 100644 --- a/rsocket-core/src/main/java/io/rsocket/loadbalance/FrugalQuantile.java +++ b/rsocket-core/src/main/java/io/rsocket/loadbalance/FrugalQuantile.java @@ -65,7 +65,8 @@ public synchronized void insert(double x) { estimate = x; sign = 1; } else { - double v = rnd.nextDouble(); + final double v = rnd.nextDouble(); + final double estimate = this.estimate; if (x > estimate && v > (1 - quantile)) { higher(x); @@ -76,6 +77,8 @@ public synchronized void insert(double x) { } private void higher(double x) { + double estimate = this.estimate; + step += sign * increment; if (step > 0) { @@ -94,9 +97,13 @@ private void higher(double x) { } sign = 1; + + this.estimate = estimate; } private void lower(double x) { + double estimate = this.estimate; + step -= sign * increment; if (step > 0) { @@ -115,6 +122,8 @@ private void lower(double x) { } sign = -1; + + this.estimate = estimate; } @Override diff --git a/rsocket-core/src/main/java/io/rsocket/loadbalance/Median.java b/rsocket-core/src/main/java/io/rsocket/loadbalance/Median.java index 42b125b41..5319706f9 100644 --- a/rsocket-core/src/main/java/io/rsocket/loadbalance/Median.java +++ b/rsocket-core/src/main/java/io/rsocket/loadbalance/Median.java @@ -33,6 +33,7 @@ public synchronized void insert(double x) { estimate = x; sign = 1; } else { + final double estimate = this.estimate; if (x > estimate) { greaterThanZero(x); } else if (x < estimate) { @@ -42,6 +43,8 @@ public synchronized void insert(double x) { } private void greaterThanZero(double x) { + double estimate = this.estimate; + step += sign; if (step > 0) { @@ -60,9 +63,13 @@ private void greaterThanZero(double x) { } sign = 1; + + this.estimate = estimate; } private void lessThanZero(double x) { + double estimate = this.estimate; + step -= sign; if (step > 0) { @@ -81,6 +88,8 @@ private void lessThanZero(double x) { } sign = -1; + + this.estimate = estimate; } @Override diff --git a/rsocket-core/src/main/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategy.java b/rsocket-core/src/main/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategy.java index c401818f9..c30c8ad6b 100644 --- a/rsocket-core/src/main/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategy.java +++ b/rsocket-core/src/main/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategy.java @@ -124,7 +124,10 @@ public RSocket select(List sockets) { private static double algorithmicWeight( RSocket rSocket, @Nullable final WeightedStats weightedStats) { - if (weightedStats == null || rSocket.isDisposed() || rSocket.availability() == 0.0) { + if (weightedStats == null) { + return 1.0; + } + if (rSocket.isDisposed() || rSocket.availability() == 0.0) { return 0.0; } final int pending = weightedStats.pending(); diff --git a/rsocket-core/src/test/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategyTest.java b/rsocket-core/src/test/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategyTest.java new file mode 100644 index 000000000..6640aea4e --- /dev/null +++ b/rsocket-core/src/test/java/io/rsocket/loadbalance/WeightedLoadbalanceStrategyTest.java @@ -0,0 +1,237 @@ +package io.rsocket.loadbalance; + +import io.rsocket.Payload; +import io.rsocket.RSocket; +import io.rsocket.RaceTestConstants; +import io.rsocket.core.RSocketConnector; +import io.rsocket.transport.ClientTransport; +import io.rsocket.util.Clock; +import io.rsocket.util.EmptyPayload; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; +import org.assertj.core.api.Assertions; +import org.assertj.core.data.Offset; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; +import reactor.core.publisher.Hooks; +import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; +import reactor.test.publisher.TestPublisher; + +public class WeightedLoadbalanceStrategyTest { + + @BeforeEach + void setUp() { + Hooks.onErrorDropped((__) -> {}); + } + + @AfterAll + static void afterAll() { + Hooks.resetOnErrorDropped(); + } + + @Test + public void allRequestsShouldGoToTheSocketWithHigherWeight() { + final AtomicInteger counter1 = new AtomicInteger(); + final AtomicInteger counter2 = new AtomicInteger(); + final ClientTransport mockTransport = Mockito.mock(ClientTransport.class); + final RSocketConnector rSocketConnectorMock = Mockito.mock(RSocketConnector.class); + final WeightedTestRSocket rSocket1 = + new WeightedTestRSocket( + new RSocket() { + @Override + public Mono fireAndForget(Payload payload) { + counter1.incrementAndGet(); + return Mono.empty(); + } + }); + final WeightedTestRSocket rSocket2 = + new WeightedTestRSocket( + new RSocket() { + @Override + public Mono fireAndForget(Payload payload) { + counter2.incrementAndGet(); + return Mono.empty(); + } + }); + Mockito.when(rSocketConnectorMock.connect(Mockito.any(ClientTransport.class))) + .then(im -> Mono.just(rSocket1)) + .then(im -> Mono.just(rSocket2)); + + final TestPublisher> source = TestPublisher.create(); + final RSocketPool rSocketPool = + new RSocketPool( + rSocketConnectorMock, + source, + WeightedLoadbalanceStrategy.builder() + .weightedStatsResolver(r -> r instanceof WeightedStats ? (WeightedStats) r : null) + .build()); + + for (int j = 0; j < RaceTestConstants.REPEATS; j++) { + rSocketPool.select().fireAndForget(EmptyPayload.INSTANCE).subscribe(); + } + + source.next( + Arrays.asList( + LoadbalanceTarget.from("1", mockTransport), + LoadbalanceTarget.from("2", mockTransport))); + + Assertions.assertThat(counter1.get()).isCloseTo(1000, Offset.offset(1)); + Assertions.assertThat(counter2.get()).isCloseTo(0, Offset.offset(1)); + } + + @Test + public void shouldDeliverValuesToTheSocketWithTheHighestCalculatedWeight() { + final AtomicInteger counter1 = new AtomicInteger(); + final AtomicInteger counter2 = new AtomicInteger(); + final ClientTransport mockTransport1 = Mockito.mock(ClientTransport.class); + final ClientTransport mockTransport2 = Mockito.mock(ClientTransport.class); + final RSocketConnector rSocketConnectorMock = Mockito.mock(RSocketConnector.class); + final WeightedTestRSocket rSocket1 = + new WeightedTestRSocket( + new RSocket() { + @Override + public Mono fireAndForget(Payload payload) { + counter1.incrementAndGet(); + return Mono.empty(); + } + }); + final WeightedTestRSocket rSocket2 = + new WeightedTestRSocket( + new RSocket() { + @Override + public Mono fireAndForget(Payload payload) { + counter1.incrementAndGet(); + return Mono.empty(); + } + }); + final WeightedTestRSocket rSocket3 = + new WeightedTestRSocket( + new RSocket() { + @Override + public Mono fireAndForget(Payload payload) { + counter2.incrementAndGet(); + return Mono.empty(); + } + }); + + Mockito.when(rSocketConnectorMock.connect(Mockito.any(ClientTransport.class))) + .then(im -> Mono.just(rSocket1)) + .then(im -> Mono.just(rSocket2)) + .then(im -> Mono.just(rSocket3)); + + final TestPublisher> source = TestPublisher.create(); + final RSocketPool rSocketPool = + new RSocketPool( + rSocketConnectorMock, + source, + WeightedLoadbalanceStrategy.builder() + .weightedStatsResolver(r -> r instanceof WeightedStats ? (WeightedStats) r : null) + .build()); + + for (int j = 0; j < RaceTestConstants.REPEATS; j++) { + rSocketPool.select().fireAndForget(EmptyPayload.INSTANCE).subscribe(); + } + + source.next(Collections.singletonList(LoadbalanceTarget.from("1", mockTransport1))); + + Assertions.assertThat(counter1.get()).isCloseTo(RaceTestConstants.REPEATS, Offset.offset(1)); + + source.next(Collections.emptyList()); + + for (int j = 0; j < RaceTestConstants.REPEATS; j++) { + rSocketPool.select().fireAndForget(EmptyPayload.INSTANCE).subscribe(); + } + + rSocket1.updateAvailability(0.0); + + source.next(Collections.singletonList(LoadbalanceTarget.from("1", mockTransport1))); + + Assertions.assertThat(counter1.get()) + .isCloseTo(RaceTestConstants.REPEATS * 2, Offset.offset(1)); + + source.next( + Arrays.asList( + LoadbalanceTarget.from("1", mockTransport1), + LoadbalanceTarget.from("2", mockTransport2))); + + for (int j = 0; j < RaceTestConstants.REPEATS; j++) { + final RSocket rSocket = rSocketPool.select(); + rSocket.fireAndForget(EmptyPayload.INSTANCE).subscribe(); + } + + Assertions.assertThat(counter1.get()) + .isCloseTo(RaceTestConstants.REPEATS * 3, Offset.offset(100)); + Assertions.assertThat(counter2.get()).isCloseTo(0, Offset.offset(100)); + + rSocket2.updateAvailability(0.0); + + source.next(Collections.singletonList(LoadbalanceTarget.from("2", mockTransport1))); + + for (int j = 0; j < RaceTestConstants.REPEATS; j++) { + rSocketPool.select().fireAndForget(EmptyPayload.INSTANCE).subscribe(); + } + + Assertions.assertThat(counter1.get()) + .isCloseTo(RaceTestConstants.REPEATS * 3, Offset.offset(100)); + Assertions.assertThat(counter2.get()).isCloseTo(RaceTestConstants.REPEATS, Offset.offset(100)); + + source.next( + Arrays.asList( + LoadbalanceTarget.from("1", mockTransport1), + LoadbalanceTarget.from("2", mockTransport2))); + + for (int j = 0; j < RaceTestConstants.REPEATS; j++) { + final RSocket rSocket = rSocketPool.select(); + rSocket.fireAndForget(EmptyPayload.INSTANCE).subscribe(); + } + + Assertions.assertThat(counter1.get()) + .isCloseTo(RaceTestConstants.REPEATS * 3, Offset.offset(100)); + Assertions.assertThat(counter2.get()) + .isCloseTo(RaceTestConstants.REPEATS * 2, Offset.offset(100)); + } + + static class WeightedTestRSocket extends BaseWeightedStats implements RSocket { + + final Sinks.Empty sink = Sinks.empty(); + + final RSocket rSocket; + + public WeightedTestRSocket(RSocket rSocket) { + this.rSocket = rSocket; + } + + @Override + public Mono fireAndForget(Payload payload) { + startRequest(); + final long startTime = Clock.now(); + return this.rSocket + .fireAndForget(payload) + .doFinally( + __ -> { + stopRequest(startTime); + record(Clock.now() - startTime); + updateAvailability(1.0); + }); + } + + @Override + public Mono onClose() { + return sink.asMono(); + } + + @Override + public void dispose() { + sink.tryEmitEmpty(); + } + + public RSocket source() { + return rSocket; + } + } +} From 21852cacfed67541683ec1e0e11c5b99adc217cc Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Tue, 25 Jan 2022 15:58:36 +0200 Subject: [PATCH 11/17] fixes test Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- build.gradle | 2 ++ rsocket-core/build.gradle | 1 + .../src/test/java/io/rsocket/core/ReconnectMonoTests.java | 2 ++ 3 files changed, 5 insertions(+) diff --git a/build.gradle b/build.gradle index 5be434a4a..f535c84f6 100644 --- a/build.gradle +++ b/build.gradle @@ -47,6 +47,7 @@ subprojects { ext['assertj.version'] = '3.21.0' ext['netflix.limits.version'] = '0.3.6' ext['bouncycastle-bcpkix.version'] = '1.68' + ext['awaitility.version'] = '4.1.1' group = "io.rsocket" @@ -80,6 +81,7 @@ subprojects { dependency "org.assertj:assertj-core:${ext['assertj.version']}" dependency "org.hdrhistogram:HdrHistogram:${ext['hdrhistogram.version']}" dependency "org.slf4j:slf4j-api:${ext['slf4j.version']}" + dependency "org.awaitility:awaitility:${ext['awaitility.version']}" dependencySet(group: 'org.mockito', version: ext['mockito.version']) { entry 'mockito-junit-jupiter' entry 'mockito-core' diff --git a/rsocket-core/build.gradle b/rsocket-core/build.gradle index 3d4759af0..ecd23296a 100644 --- a/rsocket-core/build.gradle +++ b/rsocket-core/build.gradle @@ -35,6 +35,7 @@ dependencies { testImplementation 'org.junit.jupiter:junit-jupiter-api' testImplementation 'org.junit.jupiter:junit-jupiter-params' testImplementation 'org.mockito:mockito-core' + testImplementation 'org.awaitility:awaitility' testRuntimeOnly 'ch.qos.logback:logback-classic' testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine' diff --git a/rsocket-core/src/test/java/io/rsocket/core/ReconnectMonoTests.java b/rsocket-core/src/test/java/io/rsocket/core/ReconnectMonoTests.java index 672141eaa..3112a0943 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/ReconnectMonoTests.java +++ b/rsocket-core/src/test/java/io/rsocket/core/ReconnectMonoTests.java @@ -17,6 +17,7 @@ package io.rsocket.core; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; import io.rsocket.RaceTestConstants; import io.rsocket.internal.subscriber.AssertSubscriber; @@ -321,6 +322,7 @@ public String get() { assertThat(expired).hasSize(1).containsOnly("value_to_expire" + i); if (reconnectMono.resolvingInner.subscribers == ResolvingOperator.READY) { + await().atMost(Duration.ofSeconds(5)).until(() -> received.size() == 2); assertThat(received) .hasSize(2) .containsExactly( From 37fc68c68f4b61d826084330a7b0476a456b63da Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Tue, 25 Jan 2022 19:50:42 +0200 Subject: [PATCH 12/17] removes hamcrest from test dependencies Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- build.gradle | 2 - rsocket-core/build.gradle | 2 - .../io/rsocket/core/RSocketRequesterTest.java | 125 +++++++++--------- .../io/rsocket/core/RSocketResponderTest.java | 77 +++++------ .../io/rsocket/test/util/MockRSocket.java | 7 +- rsocket-examples/build.gradle | 1 - rsocket-load-balancer/build.gradle | 1 - .../io/rsocket/client/TimeoutClientTest.java | 8 +- 8 files changed, 107 insertions(+), 116 deletions(-) diff --git a/build.gradle b/build.gradle index f535c84f6..98298af9a 100644 --- a/build.gradle +++ b/build.gradle @@ -42,7 +42,6 @@ subprojects { ext['slf4j.version'] = '1.7.30' ext['jmh.version'] = '1.33' ext['junit.version'] = '5.8.1' - ext['hamcrest.version'] = '1.3' ext['micrometer.version'] = '1.7.5' ext['assertj.version'] = '3.21.0' ext['netflix.limits.version'] = '0.3.6' @@ -86,7 +85,6 @@ subprojects { entry 'mockito-junit-jupiter' entry 'mockito-core' } - dependency "org.hamcrest:hamcrest-library:${ext['hamcrest.version']}" dependencySet(group: 'org.openjdk.jmh', version: ext['jmh.version']) { entry 'jmh-core' entry 'jmh-generator-annprocess' diff --git a/rsocket-core/build.gradle b/rsocket-core/build.gradle index ecd23296a..7a0ebfb32 100644 --- a/rsocket-core/build.gradle +++ b/rsocket-core/build.gradle @@ -40,8 +40,6 @@ dependencies { testRuntimeOnly 'ch.qos.logback:logback-classic' testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine' - testImplementation 'org.hamcrest:hamcrest-library' - jcstressImplementation(project(":rsocket-test")) jcstressImplementation "ch.qos.logback:logback-classic" } diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java b/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java index f31b74800..6a1d07a67 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java @@ -32,10 +32,7 @@ import static io.rsocket.frame.FrameType.REQUEST_FNF; import static io.rsocket.frame.FrameType.REQUEST_RESPONSE; import static io.rsocket.frame.FrameType.REQUEST_STREAM; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.greaterThanOrEqualTo; -import static org.hamcrest.Matchers.hasSize; -import static org.hamcrest.Matchers.is; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.verify; @@ -131,7 +128,7 @@ public void tearDown() { @Timeout(2_000) public void testInvalidFrameOnStream0ShouldNotTerminateRSocket() { rule.connection.addToReceivedBuffer(RequestNFrameCodec.encode(rule.alloc(), 0, 10)); - Assertions.assertThat(rule.socket.isDisposed()).isFalse(); + assertThat(rule.socket.isDisposed()).isFalse(); rule.assertHasNoLeaks(); } @@ -149,19 +146,21 @@ protected void hookOnSubscribe(Subscription subscription) { }; stream.subscribe(subscriber); - Assertions.assertThat(rule.connection.getSent()).isEmpty(); + assertThat(rule.connection.getSent()).isEmpty(); subscriber.request(5); List sent = new ArrayList<>(rule.connection.getSent()); - assertThat("sent frame count", sent.size(), is(1)); + assertThat(sent.size()).describedAs("sent frame count").isEqualTo(1); ByteBuf f = sent.get(0); - assertThat("initial frame", frameType(f), is(REQUEST_STREAM)); - assertThat("initial request n", RequestStreamFrameCodec.initialRequestN(f), is(5L)); - assertThat("should be released", f.release(), is(true)); + assertThat(frameType(f)).describedAs("initial frame").isEqualTo(REQUEST_STREAM); + assertThat(RequestStreamFrameCodec.initialRequestN(f)) + .describedAs("initial request n") + .isEqualTo(5L); + assertThat(f.release()).describedAs("should be released").isEqualTo(true); rule.assertHasNoLeaks(); } @@ -189,7 +188,7 @@ public void testHandleApplicationException() { verify(responseSub).onError(any(ApplicationErrorException.class)); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) // requestResponseFrame .hasSize(1) .allMatch(ReferenceCounted::release); @@ -210,7 +209,7 @@ public void testHandleValidFrame() { rule.alloc(), streamId, EmptyPayload.INSTANCE)); verify(sub).onComplete(); - Assertions.assertThat(rule.connection.getSent()).hasSize(1).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).hasSize(1).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); } @@ -226,10 +225,13 @@ public void testRequestReplyWithCancel() { List sent = new ArrayList<>(rule.connection.getSent()); - assertThat( - "Unexpected frame sent on the connection.", frameType(sent.get(0)), is(REQUEST_RESPONSE)); - assertThat("Unexpected frame sent on the connection.", frameType(sent.get(1)), is(CANCEL)); - Assertions.assertThat(sent).hasSize(2).allMatch(ReferenceCounted::release); + assertThat(frameType(sent.get(0))) + .describedAs("Unexpected frame sent on the connection.") + .isEqualTo(REQUEST_RESPONSE); + assertThat(frameType(sent.get(1))) + .describedAs("Unexpected frame sent on the connection.") + .isEqualTo(CANCEL); + assertThat(sent).hasSize(2).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); } @@ -282,7 +284,7 @@ public void testChannelRequestCancellation2() { Flux.error(new IllegalStateException("Channel request not cancelled")) .delaySubscription(Duration.ofSeconds(1))) .blockFirst(); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); } @@ -303,10 +305,9 @@ public void testChannelRequestServerSideCancellation() { .delaySubscription(Duration.ofSeconds(1))) .blockFirst(); - Assertions.assertThat( - request.scan(Scannable.Attr.TERMINATED) || request.scan(Scannable.Attr.CANCELLED)) + assertThat(request.scan(Scannable.Attr.TERMINATED) || request.scan(Scannable.Attr.CANCELLED)) .isTrue(); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(1) .first() .matches(bb -> frameType(bb) == REQUEST_CHANNEL) @@ -336,14 +337,13 @@ protected void hookOnSubscribe(Subscription subscription) {} ByteBuf initialFrame = iterator.next(); - Assertions.assertThat(FrameHeaderCodec.frameType(initialFrame)).isEqualTo(REQUEST_CHANNEL); - Assertions.assertThat(RequestChannelFrameCodec.initialRequestN(initialFrame)) - .isEqualTo(Long.MAX_VALUE); - Assertions.assertThat(RequestChannelFrameCodec.data(initialFrame).toString(CharsetUtil.UTF_8)) + assertThat(FrameHeaderCodec.frameType(initialFrame)).isEqualTo(REQUEST_CHANNEL); + assertThat(RequestChannelFrameCodec.initialRequestN(initialFrame)).isEqualTo(Long.MAX_VALUE); + assertThat(RequestChannelFrameCodec.data(initialFrame).toString(CharsetUtil.UTF_8)) .isEqualTo("0"); - Assertions.assertThat(initialFrame.release()).isTrue(); + assertThat(initialFrame.release()).isTrue(); - Assertions.assertThat(iterator.hasNext()).isFalse(); + assertThat(iterator.hasNext()).isFalse(); rule.assertHasNoLeaks(); } @@ -364,7 +364,7 @@ public void shouldThrownExceptionIfGivenPayloadIsExitsSizeAllowanceWithNoFragmen .expectSubscription() .expectErrorSatisfies( t -> - Assertions.assertThat(t) + assertThat(t) .isInstanceOf(IllegalArgumentException.class) .hasMessage( String.format(INVALID_PAYLOAD_ERROR_MESSAGE, maxFrameLength))) @@ -406,11 +406,11 @@ static Stream>> prepareCalls() { }) .expectErrorSatisfies( t -> - Assertions.assertThat(t) + assertThat(t) .isInstanceOf(IllegalArgumentException.class) .hasMessage(String.format(INVALID_PAYLOAD_ERROR_MESSAGE, maxFrameLength))) .verify(); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) // expect to be sent RequestChannelFrame // expect to be sent CancelFrame .hasSize(2) @@ -439,8 +439,7 @@ public void checkNoLeaksOnRacing( runner.accept(assertSubscriber, clientSocketRule); - Assertions.assertThat(clientSocketRule.connection.getSent()) - .allMatch(ReferenceCounted::release); + assertThat(clientSocketRule.connection.getSent()).allMatch(ReferenceCounted::release); clientSocketRule.assertHasNoLeaks(); } @@ -501,8 +500,8 @@ private static Stream racingCases() { RaceTestUtils.race(() -> as.request(1), as::cancel); // ensures proper frames order if (rule.connection.getSent().size() > 0) { - Assertions.assertThat(rule.connection.getSent()).hasSize(2); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()).hasSize(2); + assertThat(rule.connection.getSent()) .element(0) .matches( bb -> frameType(bb) == REQUEST_STREAM, @@ -511,7 +510,7 @@ private static Stream racingCases() { + "} but was {" + frameType(rule.connection.getSent().stream().findFirst().get()) + "}"); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .element(1) .matches( bb -> frameType(bb) == CANCEL, @@ -548,8 +547,8 @@ private static Stream racingCases() { int size = rule.connection.getSent().size(); if (size > 0) { - Assertions.assertThat(size).isLessThanOrEqualTo(3).isGreaterThanOrEqualTo(2); - Assertions.assertThat(rule.connection.getSent()) + assertThat(size).isLessThanOrEqualTo(3).isGreaterThanOrEqualTo(2); + assertThat(rule.connection.getSent()) .element(0) .matches( bb -> frameType(bb) == REQUEST_CHANNEL, @@ -559,7 +558,7 @@ private static Stream racingCases() { + frameType(rule.connection.getSent().stream().findFirst().get()) + "}"); if (size == 2) { - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .element(1) .matches( bb -> frameType(bb) == CANCEL, @@ -570,7 +569,7 @@ private static Stream racingCases() { rule.connection.getSent().stream().skip(1).findFirst().get()) + "}"); } else { - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .element(1) .matches( bb -> frameType(bb) == COMPLETE || frameType(bb) == CANCEL, @@ -582,7 +581,7 @@ private static Stream racingCases() { + frameType( rule.connection.getSent().stream().skip(1).findFirst().get()) + "}"); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .element(2) .matches( bb -> frameType(bb) == CANCEL || frameType(bb) == COMPLETE, @@ -720,7 +719,7 @@ public void simpleOnDiscardRequestChannelTest() { assertSubscriber.cancel(); - Assertions.assertThat(rule.connection.getSent()).allMatch(ByteBuf::release); + assertThat(rule.connection.getSent()).allMatch(ByteBuf::release); rule.assertHasNoLeaks(); } @@ -744,7 +743,7 @@ public void simpleOnDiscardRequestChannelTest2() { ErrorFrameCodec.encode( allocator, streamId, new CustomRSocketException(0x00000404, "test"))); - Assertions.assertThat(rule.connection.getSent()).allMatch(ByteBuf::release); + assertThat(rule.connection.getSent()).allMatch(ByteBuf::release); rule.assertHasNoLeaks(); } @@ -815,18 +814,18 @@ public void verifiesThatFrameWithNoMetadataHasDecodedCorrectlyIntoPayload( testPublisher.next(ByteBufPayload.create("d" + i)); } - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .describedAs( "Interaction Type :[%s]. Expected to observe %s frames sent", frameType, framesCnt) .hasSize(framesCnt) .allMatch(bb -> !FrameHeaderCodec.hasMetadata(bb)) .allMatch(ByteBuf::release); - Assertions.assertThat(assertSubscriber.isTerminated()) + assertThat(assertSubscriber.isTerminated()) .describedAs("Interaction Type :[%s]. Expected to be terminated", frameType) .isTrue(); - Assertions.assertThat(assertSubscriber.values()) + assertThat(assertSubscriber.values()) .describedAs( "Interaction Type :[%s]. Expected to observe %s frames received", frameType, responsesCnt) @@ -889,12 +888,12 @@ public void ensuresThatNoOpsMustHappenUntilSubscriptionInCaseOfFnfCall() { Payload payload2 = ByteBufPayload.create("abc2"); Mono fnf2 = rule.socket.fireAndForget(payload2); - Assertions.assertThat(rule.connection.getSent()).isEmpty(); + assertThat(rule.connection.getSent()).isEmpty(); // checks that fnf2 should have id 1 even though it was generated later than fnf1 AssertSubscriber voidAssertSubscriber2 = fnf2.subscribeWith(AssertSubscriber.create(0)); voidAssertSubscriber2.assertTerminated().assertNoError(); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(1) .first() .matches(bb -> frameType(bb) == REQUEST_FNF) @@ -912,7 +911,7 @@ public void ensuresThatNoOpsMustHappenUntilSubscriptionInCaseOfFnfCall() { // checks that fnf1 should have id 3 even though it was generated earlier AssertSubscriber voidAssertSubscriber1 = fnf1.subscribeWith(AssertSubscriber.create(0)); voidAssertSubscriber1.assertTerminated().assertNoError(); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(1) .first() .matches(bb -> frameType(bb) == REQUEST_FNF) @@ -936,7 +935,7 @@ public void ensuresThatNoOpsMustHappenUntilFirstRequestN( Payload payload2 = ByteBufPayload.create("abc2"); Publisher interaction2 = interaction.apply(rule, payload2); - Assertions.assertThat(rule.connection.getSent()).isEmpty(); + assertThat(rule.connection.getSent()).isEmpty(); AssertSubscriber assertSubscriber1 = AssertSubscriber.create(0); interaction1.subscribe(assertSubscriber1); @@ -945,12 +944,12 @@ public void ensuresThatNoOpsMustHappenUntilFirstRequestN( assertSubscriber1.assertNotTerminated().assertNoError(); assertSubscriber2.assertNotTerminated().assertNoError(); // even though we subscribed, nothing should happen until the first requestN - Assertions.assertThat(rule.connection.getSent()).isEmpty(); + assertThat(rule.connection.getSent()).isEmpty(); // first request on the second interaction to ensure that stream id issuing on the first request assertSubscriber2.request(1); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(frameType == REQUEST_CHANNEL ? 2 : 1) .first() .matches(bb -> frameType(bb) == frameType) @@ -979,7 +978,7 @@ public void ensuresThatNoOpsMustHappenUntilFirstRequestN( .matches(ReferenceCounted::release); if (frameType == REQUEST_CHANNEL) { - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .element(1) .matches(bb -> frameType(bb) == COMPLETE) .matches( @@ -993,7 +992,7 @@ public void ensuresThatNoOpsMustHappenUntilFirstRequestN( rule.connection.clearSendReceiveBuffers(); assertSubscriber1.request(1); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(frameType == REQUEST_CHANNEL ? 2 : 1) .first() .matches(bb -> frameType(bb) == frameType) @@ -1022,7 +1021,7 @@ public void ensuresThatNoOpsMustHappenUntilFirstRequestN( .matches(ReferenceCounted::release); if (frameType == REQUEST_CHANNEL) { - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .element(1) .matches(bb -> frameType(bb) == COMPLETE) .matches( @@ -1068,7 +1067,7 @@ public void ensuresCorrectOrderOfStreamIdIssuingInCaseOfRacing( () -> publisher1.subscribe(AssertSubscriber.create()), () -> publisher2.subscribe(AssertSubscriber.create())); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .extracting(FrameHeaderCodec::streamId) .containsExactly(i, i + 2); rule.connection.getSent().forEach(bb -> bb.release()); @@ -1180,11 +1179,11 @@ public void shouldTerminateAllStreamsIfThereRacingBetweenDisposeAndRequests( } } - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); rule.connection.getSent().clear(); - Assertions.assertThat(payload1.refCnt()).isZero(); - Assertions.assertThat(payload2.refCnt()).isZero(); + assertThat(payload1.refCnt()).isZero(); + assertThat(payload2.refCnt()).isZero(); } } @@ -1199,13 +1198,13 @@ public void testWorkaround858() { rule.connection.addToReceivedBuffer( ErrorFrameCodec.encode(rule.alloc(), 1, new RuntimeException("test"))); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(1) .first() .matches(bb -> FrameHeaderCodec.frameType(bb) == REQUEST_RESPONSE) .matches(ByteBuf::release); - Assertions.assertThat(rule.socket.isDisposed()).isFalse(); + assertThat(rule.socket.isDisposed()).isFalse(); rule.assertHasNoLeaks(); } @@ -1382,9 +1381,9 @@ public void testWorkaround959(String type) { assertSubscriber.request(1); }); - Assertions.assertThat(rule.connection.getSent()).allMatch(ByteBuf::release); + assertThat(rule.connection.getSent()).allMatch(ByteBuf::release); - Assertions.assertThat(rule.socket.isDisposed()).isFalse(); + assertThat(rule.socket.isDisposed()).isFalse(); assertSubscriber.values().forEach(ReferenceCountUtil::safeRelease); assertSubscriber.assertNoError(); @@ -1412,7 +1411,9 @@ protected RSocketRequester newRSocket() { } public int getStreamIdForRequestType(FrameType expectedFrameType) { - assertThat("Unexpected frames sent.", connection.getSent(), hasSize(greaterThanOrEqualTo(1))); + assertThat(connection.getSent().size()) + .describedAs("Unexpected frames sent.") + .isGreaterThanOrEqualTo(1); List framesFound = new ArrayList<>(); for (ByteBuf frame : connection.getSent()) { FrameType frameType = frameType(frame); diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketResponderTest.java b/rsocket-core/src/test/java/io/rsocket/core/RSocketResponderTest.java index 4c44d827d..c0f64469c 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RSocketResponderTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RSocketResponderTest.java @@ -34,10 +34,7 @@ import static io.rsocket.frame.FrameType.REQUEST_N; import static io.rsocket.frame.FrameType.REQUEST_RESPONSE; import static io.rsocket.frame.FrameType.REQUEST_STREAM; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.anyOf; -import static org.hamcrest.Matchers.empty; -import static org.hamcrest.Matchers.is; +import static org.assertj.core.api.Assertions.assertThat; import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufAllocator; @@ -76,7 +73,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Stream; -import org.assertj.core.api.Assertions; import org.assertj.core.api.Assumptions; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -126,12 +122,13 @@ public void testHandleKeepAlive() { rule.connection.addToReceivedBuffer( KeepAliveFrameCodec.encode(rule.alloc(), true, 0, Unpooled.EMPTY_BUFFER)); ByteBuf sent = rule.connection.awaitFrame(); - assertThat("Unexpected frame sent.", frameType(sent), is(FrameType.KEEPALIVE)); + assertThat(frameType(sent)) + .describedAs("Unexpected frame sent.") + .isEqualTo(FrameType.KEEPALIVE); /*Keep alive ack must not have respond flag else, it will result in infinite ping-pong of keep alive frames.*/ - assertThat( - "Unexpected keep-alive frame respond flag.", - KeepAliveFrameCodec.respondFlag(sent), - is(false)); + assertThat(KeepAliveFrameCodec.respondFlag(sent)) + .describedAs("Unexpected keep-alive frame respond flag.") + .isEqualTo(false); } @Test @@ -149,10 +146,9 @@ public Mono requestResponse(Payload payload) { }); rule.sendRequest(streamId, FrameType.REQUEST_RESPONSE); testPublisher.complete(); - assertThat( - "Unexpected frame sent.", - frameType(rule.connection.awaitFrame()), - anyOf(is(FrameType.COMPLETE), is(FrameType.NEXT_COMPLETE))); + assertThat(frameType(rule.connection.awaitFrame())) + .describedAs("Unexpected frame sent.") + .isIn(FrameType.COMPLETE, FrameType.NEXT_COMPLETE); testPublisher.assertWasNotCancelled(); } @@ -162,8 +158,9 @@ public void testHandlerEmitsError() { final int streamId = 4; rule.prefetch = 1; rule.sendRequest(streamId, FrameType.REQUEST_STREAM); - assertThat( - "Unexpected frame sent.", frameType(rule.connection.awaitFrame()), is(FrameType.ERROR)); + assertThat(frameType(rule.connection.awaitFrame())) + .describedAs("Unexpected frame sent.") + .isEqualTo(FrameType.ERROR); } @Test @@ -182,12 +179,12 @@ public Mono requestResponse(Payload payload) { }); rule.sendRequest(streamId, FrameType.REQUEST_RESPONSE); - assertThat("Unexpected frame sent.", rule.connection.getSent(), is(empty())); + assertThat(rule.connection.getSent()).describedAs("Unexpected frame sent.").isEmpty(); rule.connection.addToReceivedBuffer(CancelFrameCodec.encode(allocator, streamId)); - assertThat("Unexpected frame sent.", rule.connection.getSent(), is(empty())); - assertThat("Subscription not cancelled.", cancelled.get(), is(true)); + assertThat(rule.connection.getSent()).describedAs("Unexpected frame sent.").isEmpty(); + assertThat(cancelled.get()).describedAs("Subscription not cancelled.").isTrue(); rule.assertHasNoLeaks(); } @@ -243,7 +240,7 @@ protected void hookOnSubscribe(Subscription subscription) { for (Runnable runnable : runnables) { rule.connection.clearSendReceiveBuffers(); runnable.run(); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(1) .first() .matches(bb -> FrameHeaderCodec.frameType(bb) == FrameType.ERROR) @@ -253,7 +250,7 @@ protected void hookOnSubscribe(Subscription subscription) { .contains(String.format(INVALID_PAYLOAD_ERROR_MESSAGE, maxFrameLength))) .matches(ReferenceCounted::release); - assertThat("Subscription not cancelled.", cancelled.get(), is(true)); + assertThat(cancelled.get()).describedAs("Subscription not cancelled.").isTrue(); } rule.assertHasNoLeaks(); @@ -308,9 +305,9 @@ public Flux requestChannel(Publisher payloads) { sink.tryEmitEmpty(); }); - Assertions.assertThat(assertSubscriber.values()).allMatch(ReferenceCounted::release); + assertThat(assertSubscriber.values()).allMatch(ReferenceCounted::release); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); testRequestInterceptor.expectOnStart(1, REQUEST_CHANNEL).expectOnComplete(1).expectNothing(); @@ -353,7 +350,7 @@ public Flux requestChannel(Publisher payloads) { sink.complete(); }); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); testRequestInterceptor.expectOnStart(1, REQUEST_CHANNEL).expectOnCancel(1).expectNothing(); @@ -398,7 +395,7 @@ public Flux requestChannel(Publisher payloads) { sink.complete(); }); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); testRequestInterceptor.expectOnStart(1, REQUEST_CHANNEL).expectOnCancel(1).expectNothing(); rule.assertHasNoLeaks(); } @@ -483,13 +480,13 @@ public Flux requestChannel(Publisher payloads) { sink.error(new RuntimeException()); }); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); assertSubscriber .assertTerminated() .assertError(CancellationException.class) .assertErrorMessage("Outbound has terminated with an error"); - Assertions.assertThat(assertSubscriber.values()) + assertThat(assertSubscriber.values()) .allMatch( msg -> { ReferenceCountUtil.safeRelease(msg); @@ -531,7 +528,7 @@ public Flux requestStream(Payload payload) { sink.next(ByteBufPayload.create("d3", "m3")); }); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); @@ -573,7 +570,7 @@ public void subscribe(CoreSubscriber actual) { sources[0].complete(ByteBufPayload.create("d1", "m1")); }); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); @@ -581,7 +578,7 @@ public void subscribe(CoreSubscriber actual) { .expectOnStart(1, REQUEST_RESPONSE) .assertNext( e -> - Assertions.assertThat(e.eventType) + assertThat(e.eventType) .isIn( TestRequestInterceptor.EventType.ON_COMPLETE, TestRequestInterceptor.EventType.ON_CANCEL)) @@ -614,7 +611,7 @@ public Flux requestStream(Payload payload) { sink.next(ByteBufPayload.create("d3", "m3")); rule.connection.addToReceivedBuffer(cancelFrame); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); } @@ -660,7 +657,7 @@ public Flux requestChannel(Publisher payloads) { rule.connection.addToReceivedBuffer(cancelFrame); - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); rule.assertHasNoLeaks(); } @@ -730,8 +727,7 @@ public Flux requestChannel(Publisher payloads) { } if (responsesCnt > 0) { - Assertions.assertThat( - rule.connection.getSent().stream().filter(bb -> frameType(bb) != REQUEST_N)) + assertThat(rule.connection.getSent().stream().filter(bb -> frameType(bb) != REQUEST_N)) .describedAs( "Interaction Type :[%s]. Expected to observe %s frames sent", frameType, responsesCnt) .hasSize(responsesCnt) @@ -739,8 +735,7 @@ public Flux requestChannel(Publisher payloads) { } if (framesCnt > 1) { - Assertions.assertThat( - rule.connection.getSent().stream().filter(bb -> frameType(bb) == REQUEST_N)) + assertThat(rule.connection.getSent().stream().filter(bb -> frameType(bb) == REQUEST_N)) .describedAs( "Interaction Type :[%s]. Expected to observe single RequestN(%s) frame", frameType, framesCnt - 1) @@ -749,9 +744,9 @@ public Flux requestChannel(Publisher payloads) { .matches(bb -> RequestNFrameCodec.requestN(bb) == (framesCnt - 1)); } - Assertions.assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); + assertThat(rule.connection.getSent()).allMatch(ReferenceCounted::release); - Assertions.assertThat(assertSubscriber.awaitAndAssertNextValueCount(framesCnt).values()) + assertThat(assertSubscriber.awaitAndAssertNextValueCount(framesCnt).values()) .hasSize(framesCnt) .allMatch(p -> !p.hasMetadata()) .allMatch(ReferenceCounted::release); @@ -796,7 +791,7 @@ public Flux requestChannel(Publisher payloads) { rule.sendRequest(1, frameType); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(1) .first() .matches( @@ -837,13 +832,13 @@ public Flux requestChannel(Publisher payloads) { rule.connection.addToReceivedBuffer( ErrorFrameCodec.encode(rule.alloc(), 1, new RuntimeException("test"))); - Assertions.assertThat(rule.connection.getSent()) + assertThat(rule.connection.getSent()) .hasSize(1) .first() .matches(bb -> FrameHeaderCodec.frameType(bb) == REQUEST_N) .matches(ReferenceCounted::release); - Assertions.assertThat(rule.socket.isDisposed()).isFalse(); + assertThat(rule.socket.isDisposed()).isFalse(); testPublisher.assertWasCancelled(); rule.assertHasNoLeaks(); diff --git a/rsocket-core/src/test/java/io/rsocket/test/util/MockRSocket.java b/rsocket-core/src/test/java/io/rsocket/test/util/MockRSocket.java index 179afff58..a33c4c4b3 100644 --- a/rsocket-core/src/test/java/io/rsocket/test/util/MockRSocket.java +++ b/rsocket-core/src/test/java/io/rsocket/test/util/MockRSocket.java @@ -16,8 +16,7 @@ package io.rsocket.test.util; -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.is; +import static org.assertj.core.api.Assertions.assertThat; import io.rsocket.Payload; import io.rsocket.RSocket; @@ -116,6 +115,8 @@ public void assertMetadataPushCount(int expected) { } private static void assertCount(int expected, String type, AtomicInteger counter) { - assertThat("Unexpected invocations for " + type + '.', counter.get(), is(expected)); + assertThat(counter.get()) + .describedAs("Unexpected invocations for " + type + '.') + .isEqualTo(expected); } } diff --git a/rsocket-examples/build.gradle b/rsocket-examples/build.gradle index e5d74494f..d03524cd9 100644 --- a/rsocket-examples/build.gradle +++ b/rsocket-examples/build.gradle @@ -34,7 +34,6 @@ dependencies { testImplementation 'org.assertj:assertj-core' testImplementation 'io.projectreactor:reactor-test' - testImplementation 'org.hamcrest:hamcrest-library' testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine' } diff --git a/rsocket-load-balancer/build.gradle b/rsocket-load-balancer/build.gradle index c247486be..6d91324ae 100644 --- a/rsocket-load-balancer/build.gradle +++ b/rsocket-load-balancer/build.gradle @@ -32,7 +32,6 @@ dependencies { testImplementation 'org.assertj:assertj-core' testImplementation 'io.projectreactor:reactor-test' - testImplementation 'org.hamcrest:hamcrest-library' testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine' testRuntimeOnly 'ch.qos.logback:logback-classic' } diff --git a/rsocket-load-balancer/src/test/java/io/rsocket/client/TimeoutClientTest.java b/rsocket-load-balancer/src/test/java/io/rsocket/client/TimeoutClientTest.java index e6c8aa313..b8866b1f6 100644 --- a/rsocket-load-balancer/src/test/java/io/rsocket/client/TimeoutClientTest.java +++ b/rsocket-load-balancer/src/test/java/io/rsocket/client/TimeoutClientTest.java @@ -16,14 +16,13 @@ package io.rsocket.client; -import static org.hamcrest.Matchers.instanceOf; +import static org.assertj.core.api.Assertions.assertThat; import io.rsocket.Payload; import io.rsocket.RSocket; import io.rsocket.client.filter.RSockets; import io.rsocket.util.EmptyPayload; import java.time.Duration; -import org.hamcrest.MatcherAssert; import org.junit.jupiter.api.Test; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; @@ -50,8 +49,9 @@ public void onNext(Payload payload) { @Override public void onError(Throwable t) { - MatcherAssert.assertThat( - "Unexpected exception in onError", t, instanceOf(TimeoutException.class)); + assertThat(t) + .describedAs("Unexpected exception in onError") + .isInstanceOf(TimeoutException.class); } @Override From bf0c60850b7b4a217082c1bf950a68dbe8954331 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Fri, 18 Mar 2022 11:37:28 +0200 Subject: [PATCH 13/17] migrates from deprecated api, updates dependencies Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- build.gradle | 20 +- rsocket-core/build.gradle | 5 +- .../io/rsocket/core/StressSubscriber.java | 9 + .../UnboundedProcessorStressTest.java | 304 +++++++++++++++++- .../core/RequestChannelRequesterFlux.java | 4 +- .../rsocket/internal/UnboundedProcessor.java | 43 +-- .../resume/ResumableDuplexConnection.java | 2 +- .../io/rsocket/core/RSocketRequesterTest.java | 2 +- .../rsocket/frame/ByteBufRepresentation.java | 11 +- .../org.junit.jupiter.api.extension.Extension | 1 + .../server/BaseWebsocketServerTransport.java | 5 +- .../server/WebsocketServerTransportTest.java | 2 +- 12 files changed, 350 insertions(+), 58 deletions(-) create mode 100644 rsocket-core/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension diff --git a/build.gradle b/build.gradle index 98298af9a..a3ff46a4f 100644 --- a/build.gradle +++ b/build.gradle @@ -19,7 +19,7 @@ plugins { id 'me.champeau.jmh' version '0.6.6' apply false id 'io.spring.dependency-management' version '1.0.11.RELEASE' apply false id 'io.morethan.jmhreport' version '0.9.0' apply false - id 'io.github.reyerizo.gradle.jcstress' version '0.8.11' apply false + id 'io.github.reyerizo.gradle.jcstress' version '0.8.13' apply false id 'com.github.vlsi.gradle-extensions' version '1.76' apply false } @@ -33,19 +33,19 @@ subprojects { apply plugin: 'com.github.sherter.google-java-format' apply plugin: 'com.github.vlsi.gradle-extensions' - ext['reactor-bom.version'] = '2020.0.12' - ext['logback.version'] = '1.2.3' - ext['netty-bom.version'] = '4.1.70.Final' - ext['netty-boringssl.version'] = '2.0.45.Final' + ext['reactor-bom.version'] = '2020.0.17' + ext['logback.version'] = '1.2.10' + ext['netty-bom.version'] = '4.1.75.Final' + ext['netty-boringssl.version'] = '2.0.51.Final' ext['hdrhistogram.version'] = '2.1.12' - ext['mockito.version'] = '3.12.4' - ext['slf4j.version'] = '1.7.30' + ext['mockito.version'] = '4.4.0' + ext['slf4j.version'] = '1.7.36' ext['jmh.version'] = '1.33' ext['junit.version'] = '5.8.1' - ext['micrometer.version'] = '1.7.5' - ext['assertj.version'] = '3.21.0' + ext['micrometer.version'] = '1.8.4' + ext['assertj.version'] = '3.22.0' ext['netflix.limits.version'] = '0.3.6' - ext['bouncycastle-bcpkix.version'] = '1.68' + ext['bouncycastle-bcpkix.version'] = '1.70' ext['awaitility.version'] = '4.1.1' group = "io.rsocket" diff --git a/rsocket-core/build.gradle b/rsocket-core/build.gradle index 7a0ebfb32..cd8595216 100644 --- a/rsocket-core/build.gradle +++ b/rsocket-core/build.gradle @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2022 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. @@ -42,11 +42,12 @@ dependencies { jcstressImplementation(project(":rsocket-test")) jcstressImplementation "ch.qos.logback:logback-classic" + jcstressImplementation 'io.projectreactor:reactor-test' } jcstress { mode = 'quick' //quick, default, tough - jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.7" + jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.15" } jar { diff --git a/rsocket-core/src/jcstress/java/io/rsocket/core/StressSubscriber.java b/rsocket-core/src/jcstress/java/io/rsocket/core/StressSubscriber.java index 31fd44374..883077f77 100644 --- a/rsocket-core/src/jcstress/java/io/rsocket/core/StressSubscriber.java +++ b/rsocket-core/src/jcstress/java/io/rsocket/core/StressSubscriber.java @@ -224,6 +224,11 @@ public void onError(Throwable throwable) { } else { GUARD.compareAndSet(this, Operation.ON_ERROR, null); } + + if (done) { + throw new IllegalStateException("Already done"); + } + error = throwable; done = true; q.offer(throwable); @@ -241,6 +246,10 @@ public void onComplete() { } else { GUARD.compareAndSet(this, Operation.ON_COMPLETE, null); } + if (done) { + throw new IllegalStateException("Already done"); + } + done = true; ON_COMPLETE_CALLS.incrementAndGet(this); diff --git a/rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java b/rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java index 39ed2e4cb..bdbdc7a3b 100644 --- a/rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java +++ b/rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java @@ -13,9 +13,14 @@ import org.openjdk.jcstress.infra.results.LLL_Result; import org.openjdk.jcstress.infra.results.L_Result; import reactor.core.Fuseable; +import reactor.core.publisher.Hooks; public abstract class UnboundedProcessorStressTest { + static { + Hooks.onErrorDropped(t -> {}); + } + final UnboundedProcessor unboundedProcessor = new UnboundedProcessor(); @JCStressTest @@ -95,7 +100,6 @@ public void request() { stressSubscriber.request(1); stressSubscriber.request(1); stressSubscriber.request(1); - stressSubscriber.request(1); } @Actor @@ -221,7 +225,6 @@ public void request() { stressSubscriber.request(1); stressSubscriber.request(1); stressSubscriber.request(1); - stressSubscriber.request(1); } @Actor @@ -328,6 +331,110 @@ public void subscribeAndRequest() { stressSubscriber.request(1); stressSubscriber.request(1); stressSubscriber.request(1); + } + + @Actor + public void dispose() { + unboundedProcessor.dispose(); + } + + @Actor + public void next1() { + unboundedProcessor.onNext(byteBuf1); + unboundedProcessor.onNextPrioritized(byteBuf2); + } + + @Actor + public void next2() { + unboundedProcessor.onNextPrioritized(byteBuf3); + unboundedProcessor.onNext(byteBuf4); + } + + @Actor + public void complete() { + unboundedProcessor.onComplete(); + } + + @Actor + public void error() { + unboundedProcessor.onError(testException); + } + + @Arbiter + public void arbiter(LLL_Result r) { + r.r1 = stressSubscriber.onNextCalls; + r.r2 = + stressSubscriber.onCompleteCalls + + stressSubscriber.onErrorCalls * 2 + + stressSubscriber.droppedErrors.size() * 3; + + if (stressSubscriber.onCompleteCalls > 0 && stressSubscriber.onErrorCalls > 0) { + throw new RuntimeException("boom"); + } + + stressSubscriber.values.forEach(ByteBuf::release); + + r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + } + } + + @JCStressTest + @Outcome( + id = { + "0, 1, 0", + "1, 1, 0", + "2, 1, 0", + "3, 1, 0", + "4, 1, 0", + + // dropped error scenarios + "0, 4, 0", + "1, 4, 0", + "2, 4, 0", + "3, 4, 0", + "4, 4, 0", + }, + expect = Expect.ACCEPTABLE, + desc = "onComplete() before dispose() || onError()") + @Outcome( + id = { + "0, 2, 0", "1, 2, 0", "2, 2, 0", "3, 2, 0", "4, 2, 0", + }, + expect = Expect.ACCEPTABLE, + desc = "onError() before dispose() || onComplete()") + @Outcome( + id = { + "0, 2, 0", + "1, 2, 0", + "2, 2, 0", + "3, 2, 0", + "4, 2, 0", + // dropped error + "0, 5, 0", + "1, 5, 0", + "2, 5, 0", + "3, 5, 0", + "4, 5, 0", + }, + expect = Expect.ACCEPTABLE, + desc = "dispose() before onError() || onComplete()") + @State + public static class Smoke24StressTest extends UnboundedProcessorStressTest { + + static final RuntimeException testException = new RuntimeException("test"); + + final StressSubscriber stressSubscriber = new StressSubscriber<>(0, Fuseable.NONE); + final ByteBuf byteBuf1 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(1); + final ByteBuf byteBuf2 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(2); + final ByteBuf byteBuf3 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(3); + final ByteBuf byteBuf4 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(4); + + @Actor + public void subscribeAndRequest() { + unboundedProcessor.subscribe(stressSubscriber); + stressSubscriber.request(1); + stressSubscriber.request(1); + stressSubscriber.request(1); stressSubscriber.request(1); } @@ -597,6 +704,197 @@ public void arbiter(LLL_Result r) { } } + @JCStressTest + @Outcome( + id = { + "0, 1, 0", "1, 1, 0", "2, 1, 0", "3, 1, 0", "4, 1, 0", + }, + expect = Expect.ACCEPTABLE, + desc = "onComplete()") + @Outcome( + id = { + "0, 0, 0", + "1, 0, 0", + "2, 0, 0", + "3, 0, 0", + "4, 0, 0", + // interleave with error or complete happened first but dispose suppressed them + "0, 3, 0", + "1, 3, 0", + "2, 3, 0", + "3, 3, 0", + "4, 3, 0", + }, + expect = Expect.ACCEPTABLE, + desc = "cancel() before or interleave with onComplete()") + @State + public static class Smoke30StressTest extends UnboundedProcessorStressTest { + + final StressSubscriber stressSubscriber = new StressSubscriber<>(0, Fuseable.NONE); + final ByteBuf byteBuf1 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(1); + final ByteBuf byteBuf2 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(2); + final ByteBuf byteBuf3 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(3); + final ByteBuf byteBuf4 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(4); + + { + unboundedProcessor.subscribe(stressSubscriber); + } + + @Actor + public void subscribeAndRequest() { + stressSubscriber.request(1); + stressSubscriber.request(1); + stressSubscriber.request(1); + stressSubscriber.request(1); + } + + @Actor + public void cancel() { + stressSubscriber.cancel(); + } + + @Actor + public void next1() { + unboundedProcessor.onNext(byteBuf1); + unboundedProcessor.onNextPrioritized(byteBuf2); + } + + @Actor + public void next2() { + unboundedProcessor.onNextPrioritized(byteBuf3); + unboundedProcessor.onNext(byteBuf4); + } + + @Actor + public void complete() { + unboundedProcessor.onComplete(); + } + + @Arbiter + public void arbiter(LLL_Result r) { + r.r1 = stressSubscriber.onNextCalls; + r.r2 = + stressSubscriber.onCompleteCalls + + stressSubscriber.onErrorCalls * 2 + + stressSubscriber.droppedErrors.size() * 3; + + stressSubscriber.values.forEach(ByteBuf::release); + + r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + } + } + + @JCStressTest + @Outcome( + id = { + "0, 1, 0", "1, 1, 0", "2, 1, 0", "3, 1, 0", "4, 1, 0", + }, + expect = Expect.ACCEPTABLE, + desc = "onComplete()") + @State + public static class Smoke31StressTest extends UnboundedProcessorStressTest { + + final StressSubscriber stressSubscriber = new StressSubscriber<>(0, Fuseable.NONE); + final ByteBuf byteBuf1 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(1); + final ByteBuf byteBuf2 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(2); + final ByteBuf byteBuf3 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(3); + final ByteBuf byteBuf4 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(4); + + { + unboundedProcessor.subscribe(stressSubscriber); + } + + @Actor + public void subscribeAndRequest() { + stressSubscriber.request(1); + stressSubscriber.request(1); + stressSubscriber.request(1); + stressSubscriber.request(1); + } + + @Actor + public void next1() { + unboundedProcessor.onNext(byteBuf1); + unboundedProcessor.onNextPrioritized(byteBuf2); + } + + @Actor + public void next2() { + unboundedProcessor.onNextPrioritized(byteBuf3); + unboundedProcessor.onNext(byteBuf4); + } + + @Actor + public void complete() { + unboundedProcessor.onComplete(); + } + + @Arbiter + public void arbiter(LLL_Result r) { + r.r1 = stressSubscriber.onNextCalls; + r.r2 = + stressSubscriber.onCompleteCalls + + stressSubscriber.onErrorCalls * 2 + + stressSubscriber.droppedErrors.size() * 3; + + stressSubscriber.values.forEach(ByteBuf::release); + + r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + } + } + + @JCStressTest + @Outcome( + id = { + "0, 1, 0", "1, 1, 0", "2, 1, 0", "3, 1, 0", "4, 1, 0", + }, + expect = Expect.ACCEPTABLE, + desc = "onComplete()") + @State + public static class Smoke32StressTest extends UnboundedProcessorStressTest { + + final StressSubscriber stressSubscriber = + new StressSubscriber<>(Long.MAX_VALUE, Fuseable.NONE); + final ByteBuf byteBuf1 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(1); + final ByteBuf byteBuf2 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(2); + final ByteBuf byteBuf3 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(3); + final ByteBuf byteBuf4 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(4); + + { + unboundedProcessor.subscribe(stressSubscriber); + } + + @Actor + public void next1() { + unboundedProcessor.onNext(byteBuf1); + unboundedProcessor.onNextPrioritized(byteBuf2); + } + + @Actor + public void next2() { + unboundedProcessor.onNextPrioritized(byteBuf3); + unboundedProcessor.onNext(byteBuf4); + } + + @Actor + public void complete() { + unboundedProcessor.onComplete(); + } + + @Arbiter + public void arbiter(LLL_Result r) { + r.r1 = stressSubscriber.onNextCalls; + r.r2 = + stressSubscriber.onCompleteCalls + + stressSubscriber.onErrorCalls * 2 + + stressSubscriber.droppedErrors.size() * 3; + + stressSubscriber.values.forEach(ByteBuf::release); + + r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt(); + } + } + @JCStressTest @Outcome( id = { @@ -997,7 +1295,6 @@ public void subscribeAndRequest() { stressSubscriber.request(1); stressSubscriber.request(1); stressSubscriber.request(1); - stressSubscriber.request(1); } @Actor @@ -1130,7 +1427,6 @@ public void subscribeAndRequest() { stressSubscriber.request(1); stressSubscriber.request(1); stressSubscriber.request(1); - stressSubscriber.request(1); } @Actor diff --git a/rsocket-core/src/main/java/io/rsocket/core/RequestChannelRequesterFlux.java b/rsocket-core/src/main/java/io/rsocket/core/RequestChannelRequesterFlux.java index eee1346eb..809125402 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/RequestChannelRequesterFlux.java +++ b/rsocket-core/src/main/java/io/rsocket/core/RequestChannelRequesterFlux.java @@ -48,6 +48,7 @@ import reactor.util.annotation.NonNull; import reactor.util.annotation.Nullable; import reactor.util.context.Context; +import reactor.util.context.ContextView; final class RequestChannelRequesterFlux extends Flux implements RequesterFrameHandler, @@ -763,7 +764,8 @@ public Context currentContext() { if (isSubscribedOrTerminated(state)) { Context cachedContext = this.cachedContext; if (cachedContext == null) { - cachedContext = this.inboundSubscriber.currentContext().putAll(DISCARD_CONTEXT); + cachedContext = + this.inboundSubscriber.currentContext().putAll((ContextView) DISCARD_CONTEXT); this.cachedContext = cachedContext; } return cachedContext; diff --git a/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java b/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java index c529b615d..520ff318a 100644 --- a/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java +++ b/rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java @@ -26,10 +26,11 @@ import java.util.stream.Stream; import org.reactivestreams.Subscription; import reactor.core.CoreSubscriber; +import reactor.core.Disposable; import reactor.core.Exceptions; import reactor.core.Fuseable; import reactor.core.Scannable; -import reactor.core.publisher.FluxProcessor; +import reactor.core.publisher.Flux; import reactor.core.publisher.Operators; import reactor.util.annotation.Nullable; import reactor.util.concurrent.Queues; @@ -40,8 +41,12 @@ * *

The implementation keeps the order of signals. */ -public final class UnboundedProcessor extends FluxProcessor - implements Fuseable.QueueSubscription, Fuseable { +public final class UnboundedProcessor extends Flux + implements Scannable, + Disposable, + CoreSubscriber, + Fuseable.QueueSubscription, + Fuseable { final Queue queue; final Queue priorityQueue; @@ -98,11 +103,6 @@ public UnboundedProcessor(Runnable onFinalizedHook) { this.priorityQueue = new MpscUnboundedArrayQueue<>(Queues.SMALL_BUFFER_SIZE); } - @Override - public int getBufferSize() { - return Integer.MAX_VALUE; - } - @Override public Stream inners() { return hasDownstreams() ? Stream.of(Scannable.from(this.actual)) : Stream.empty(); @@ -118,7 +118,7 @@ public Object scanUnsafe(Attr key) { return isCancelled(state) || isDisposed(state); } - return super.scanUnsafe(key); + return null; } public void onNextPrioritized(ByteBuf t) { @@ -622,30 +622,7 @@ public boolean isDisposed() { return isFinalized(this.state); } - @Override - public boolean isTerminated() { - return this.done || isTerminated(this.state); - } - - @Override - @Nullable - public Throwable getError() { - //noinspection unused - final long state = this.state; - if (this.done) { - return this.error; - } else { - return null; - } - } - - @Override - public long downstreamCount() { - return hasDownstreams() ? 1L : 0L; - } - - @Override - public boolean hasDownstreams() { + boolean hasDownstreams() { final long state = this.state; return !isTerminated(state) && isSubscriberReady(state); } diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java b/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java index 7ade3e59b..933ac09ca 100644 --- a/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java +++ b/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java @@ -253,7 +253,7 @@ void dispose(@Nullable Throwable e) { framesSaverDisposable.dispose(); activeReceivingSubscriber.dispose(); - savableFramesSender.onComplete(); + savableFramesSender.dispose(); onConnectionClosedSink.tryEmitComplete(); if (e != null) { diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java b/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java index 6a1d07a67..327172255 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java @@ -1265,7 +1265,7 @@ void reassembleMetadata( .then(() -> rule.connection.addToReceivedBuffer(fragments.toArray(new ByteBuf[0]))) .assertNext( responsePayload -> { - PayloadAssert.assertThat(requestPayload).isEqualTo(metadataOnlyPayload).hasNoLeaks(); + PayloadAssert.assertThat(responsePayload).isEqualTo(metadataOnlyPayload).hasNoLeaks(); metadataOnlyPayload.release(); }) .thenCancel() diff --git a/rsocket-core/src/test/java/io/rsocket/frame/ByteBufRepresentation.java b/rsocket-core/src/test/java/io/rsocket/frame/ByteBufRepresentation.java index 75aa2a5b2..b12d72b51 100644 --- a/rsocket-core/src/test/java/io/rsocket/frame/ByteBufRepresentation.java +++ b/rsocket-core/src/test/java/io/rsocket/frame/ByteBufRepresentation.java @@ -18,9 +18,18 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufUtil; import io.netty.util.IllegalReferenceCountException; +import org.assertj.core.api.Assertions; import org.assertj.core.presentation.StandardRepresentation; +import org.junit.jupiter.api.extension.BeforeAllCallback; +import org.junit.jupiter.api.extension.ExtensionContext; -public final class ByteBufRepresentation extends StandardRepresentation { +public final class ByteBufRepresentation extends StandardRepresentation + implements BeforeAllCallback { + + @Override + public void beforeAll(ExtensionContext context) { + Assertions.useRepresentation(this); + } @Override protected String fallbackToStringOf(Object object) { diff --git a/rsocket-core/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension b/rsocket-core/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension new file mode 100644 index 000000000..2b51ba0de --- /dev/null +++ b/rsocket-core/src/test/resources/META-INF/services/org.junit.jupiter.api.extension.Extension @@ -0,0 +1 @@ +io.rsocket.frame.ByteBufRepresentation \ No newline at end of file diff --git a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/BaseWebsocketServerTransport.java b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/BaseWebsocketServerTransport.java index 5f04eb575..33cff28b4 100644 --- a/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/BaseWebsocketServerTransport.java +++ b/rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/server/BaseWebsocketServerTransport.java @@ -24,10 +24,7 @@ abstract class BaseWebsocketServerTransport< private static final ChannelHandler pongHandler = new PongHandler(); static Function serverConfigurer = - server -> - server.tcpConfiguration( - tcpServer -> - tcpServer.doOnConnection(connection -> connection.addHandlerLast(pongHandler))); + server -> server.doOnConnection(connection -> connection.addHandlerLast(pongHandler)); final WebsocketServerSpec.Builder specBuilder = WebsocketServerSpec.builder().maxFramePayloadLength(FRAME_LENGTH_MASK); diff --git a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketServerTransportTest.java b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketServerTransportTest.java index b9b6201b8..540076704 100644 --- a/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketServerTransportTest.java +++ b/rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketServerTransportTest.java @@ -41,7 +41,7 @@ public void testThatSetupWithUnSpecifiedFrameSizeShouldSetMaxFrameSize() { ArgumentCaptor httpHandlerCaptor = ArgumentCaptor.forClass(BiFunction.class); HttpServer server = Mockito.spy(HttpServer.create()); Mockito.doAnswer(a -> server).when(server).handle(httpHandlerCaptor.capture()); - Mockito.doAnswer(a -> server).when(server).tcpConfiguration(any()); + Mockito.doAnswer(a -> server).when(server).doOnConnection(any()); Mockito.doAnswer(a -> Mono.empty()).when(server).bind(); WebsocketServerTransport serverTransport = WebsocketServerTransport.create(server); From d8cccbe4e69e5ff358b0f75321fc7589e693dff9 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka <5380167+OlegDokuka@users.noreply.github.com> Date: Fri, 25 Mar 2022 11:36:42 +0200 Subject: [PATCH 14/17] adds routing example with TaggingMetadata and CompositeMetadata (#1021) --- .../routing/CompositeMetadataExample.java | 102 ++++++++++++++++++ .../routing/RoutingMetadataExample.java | 83 ++++++++++++++ 2 files changed, 185 insertions(+) create mode 100644 rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/metadata/routing/CompositeMetadataExample.java create mode 100644 rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/metadata/routing/RoutingMetadataExample.java diff --git a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/metadata/routing/CompositeMetadataExample.java b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/metadata/routing/CompositeMetadataExample.java new file mode 100644 index 000000000..a0a02a946 --- /dev/null +++ b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/metadata/routing/CompositeMetadataExample.java @@ -0,0 +1,102 @@ +/* + * Copyright 2015-Present 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.examples.transport.tcp.metadata.routing; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.buffer.ByteBufUtil; +import io.netty.buffer.CompositeByteBuf; +import io.rsocket.RSocket; +import io.rsocket.SocketAcceptor; +import io.rsocket.core.RSocketConnector; +import io.rsocket.core.RSocketServer; +import io.rsocket.metadata.CompositeMetadata; +import io.rsocket.metadata.CompositeMetadataCodec; +import io.rsocket.metadata.RoutingMetadata; +import io.rsocket.metadata.TaggingMetadataCodec; +import io.rsocket.metadata.WellKnownMimeType; +import io.rsocket.transport.netty.client.TcpClientTransport; +import io.rsocket.transport.netty.server.TcpServerTransport; +import io.rsocket.util.ByteBufPayload; +import java.util.Collections; +import java.util.Objects; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Mono; + +public class CompositeMetadataExample { + static final Logger logger = LoggerFactory.getLogger(CompositeMetadataExample.class); + + public static void main(String[] args) { + RSocketServer.create( + SocketAcceptor.forRequestResponse( + payload -> { + final String route = decodeRoute(payload.sliceMetadata()); + + logger.info("Received RequestResponse[route={}]", route); + + payload.release(); + + if ("my.test.route".equals(route)) { + return Mono.just(ByteBufPayload.create("Hello From My Test Route")); + } + + return Mono.error(new IllegalArgumentException("Route " + route + " not found")); + })) + .bindNow(TcpServerTransport.create("localhost", 7000)); + + RSocket socket = + RSocketConnector.create() + // here we specify that every metadata payload will be encoded using + // CompositeMetadata layout as specified in the following subspec + // https://github.com/rsocket/rsocket/blob/master/Extensions/CompositeMetadata.md + .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_COMPOSITE_METADATA.getString()) + .connect(TcpClientTransport.create("localhost", 7000)) + .block(); + + final ByteBuf routeMetadata = + TaggingMetadataCodec.createTaggingContent( + ByteBufAllocator.DEFAULT, Collections.singletonList("my.test.route")); + final CompositeByteBuf compositeMetadata = ByteBufAllocator.DEFAULT.compositeBuffer(); + + CompositeMetadataCodec.encodeAndAddMetadata( + compositeMetadata, + ByteBufAllocator.DEFAULT, + WellKnownMimeType.MESSAGE_RSOCKET_ROUTING, + routeMetadata); + + socket + .requestResponse( + ByteBufPayload.create( + ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT, "HelloWorld"), compositeMetadata)) + .log() + .block(); + } + + static String decodeRoute(ByteBuf metadata) { + final CompositeMetadata compositeMetadata = new CompositeMetadata(metadata, false); + + for (CompositeMetadata.Entry metadatum : compositeMetadata) { + if (Objects.requireNonNull(metadatum.getMimeType()) + .equals(WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString())) { + return new RoutingMetadata(metadatum.getContent()).iterator().next(); + } + } + + return null; + } +} diff --git a/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/metadata/routing/RoutingMetadataExample.java b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/metadata/routing/RoutingMetadataExample.java new file mode 100644 index 000000000..2aee18bf9 --- /dev/null +++ b/rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/metadata/routing/RoutingMetadataExample.java @@ -0,0 +1,83 @@ +/* + * Copyright 2015-Present 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.examples.transport.tcp.metadata.routing; + +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufAllocator; +import io.netty.buffer.ByteBufUtil; +import io.rsocket.RSocket; +import io.rsocket.SocketAcceptor; +import io.rsocket.core.RSocketConnector; +import io.rsocket.core.RSocketServer; +import io.rsocket.metadata.RoutingMetadata; +import io.rsocket.metadata.TaggingMetadataCodec; +import io.rsocket.metadata.WellKnownMimeType; +import io.rsocket.transport.netty.client.TcpClientTransport; +import io.rsocket.transport.netty.server.TcpServerTransport; +import io.rsocket.util.ByteBufPayload; +import java.util.Collections; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import reactor.core.publisher.Mono; + +public class RoutingMetadataExample { + static final Logger logger = LoggerFactory.getLogger(RoutingMetadataExample.class); + + public static void main(String[] args) { + RSocketServer.create( + SocketAcceptor.forRequestResponse( + payload -> { + final String route = decodeRoute(payload.sliceMetadata()); + + logger.info("Received RequestResponse[route={}]", route); + + payload.release(); + + if ("my.test.route".equals(route)) { + return Mono.just(ByteBufPayload.create("Hello From My Test Route")); + } + + return Mono.error(new IllegalArgumentException("Route " + route + " not found")); + })) + .bindNow(TcpServerTransport.create("localhost", 7000)); + + RSocket socket = + RSocketConnector.create() + // here we specify that route will be encoded using + // Routing&Tagging Metadata layout specified at this + // subspec https://github.com/rsocket/rsocket/blob/master/Extensions/Routing.md + .metadataMimeType(WellKnownMimeType.MESSAGE_RSOCKET_ROUTING.getString()) + .connect(TcpClientTransport.create("localhost", 7000)) + .block(); + + final ByteBuf routeMetadata = + TaggingMetadataCodec.createTaggingContent( + ByteBufAllocator.DEFAULT, Collections.singletonList("my.test.route")); + socket + .requestResponse( + ByteBufPayload.create( + ByteBufUtil.writeUtf8(ByteBufAllocator.DEFAULT, "HelloWorld"), routeMetadata)) + .log() + .block(); + } + + static String decodeRoute(ByteBuf metadata) { + final RoutingMetadata routingMetadata = new RoutingMetadata(metadata); + + return routingMetadata.iterator().next(); + } +} From 80a05f873c21f4e557d1980b1734c8a18fea411f Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Sat, 26 Mar 2022 11:03:04 +0200 Subject: [PATCH 15/17] adds jdk 17 instead of 16 in build matrix Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- .github/workflows/gradle-all.yml | 10 +++++----- .github/workflows/gradle-main.yml | 10 +++++----- .github/workflows/gradle-pr.yml | 8 ++++---- 3 files changed, 14 insertions(+), 14 deletions(-) diff --git a/.github/workflows/gradle-all.yml b/.github/workflows/gradle-all.yml index 8826f511a..f2df20620 100644 --- a/.github/workflows/gradle-all.yml +++ b/.github/workflows/gradle-all.yml @@ -15,7 +15,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -42,7 +42,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -69,7 +69,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -96,7 +96,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -123,7 +123,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: diff --git a/.github/workflows/gradle-main.yml b/.github/workflows/gradle-main.yml index 34d3e65f0..904c45fb7 100644 --- a/.github/workflows/gradle-main.yml +++ b/.github/workflows/gradle-main.yml @@ -15,7 +15,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -42,7 +42,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -69,7 +69,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -96,7 +96,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -123,7 +123,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: diff --git a/.github/workflows/gradle-pr.yml b/.github/workflows/gradle-pr.yml index fd88ad76f..cecca085f 100644 --- a/.github/workflows/gradle-pr.yml +++ b/.github/workflows/gradle-pr.yml @@ -9,7 +9,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -36,7 +36,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -63,7 +63,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: @@ -90,7 +90,7 @@ jobs: strategy: matrix: os: [ ubuntu-latest ] - jdk: [ 1.8, 11, 16 ] + jdk: [ 1.8, 11, 17 ] fail-fast: false steps: From 9a504a882a6d43bb26298660c5b0b7a2f71caeb2 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka <5380167+OlegDokuka@users.noreply.github.com> Date: Mon, 28 Mar 2022 18:16:07 +0300 Subject: [PATCH 16/17] fixes `block()` in MetadataPushRequesterMono/FnfRequesterMono (#1044) --- .../core/FireAndForgetRequesterMono.java | 5 ++ .../core/MetadataPushRequesterMono.java | 12 +++- .../io/rsocket/core/RSocketRequesterTest.java | 63 ++++++++++++++++++- 3 files changed, 75 insertions(+), 5 deletions(-) diff --git a/rsocket-core/src/main/java/io/rsocket/core/FireAndForgetRequesterMono.java b/rsocket-core/src/main/java/io/rsocket/core/FireAndForgetRequesterMono.java index eceb0976c..a5d527f5c 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/FireAndForgetRequesterMono.java +++ b/rsocket-core/src/main/java/io/rsocket/core/FireAndForgetRequesterMono.java @@ -185,6 +185,11 @@ public Void block(Duration m) { return block(); } + /** + * This method is deliberately non-blocking regardless it is named as `.block`. The main intent to + * keep this method along with the {@link #subscribe()} is to eliminate redundancy which comes + * with a default block method implementation. + */ @Override @Nullable public Void block() { diff --git a/rsocket-core/src/main/java/io/rsocket/core/MetadataPushRequesterMono.java b/rsocket-core/src/main/java/io/rsocket/core/MetadataPushRequesterMono.java index 226e9a0af..e2512e995 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/MetadataPushRequesterMono.java +++ b/rsocket-core/src/main/java/io/rsocket/core/MetadataPushRequesterMono.java @@ -120,6 +120,11 @@ public Void block(Duration m) { return block(); } + /** + * This method is deliberately non-blocking regardless it is named as `.block`. The main intent to + * keep this method along with the {@link #subscribe()} is to eliminate redundancy which comes + * with a default block method implementation. + */ @Override @Nullable public Void block() { @@ -133,15 +138,16 @@ public Void block() { try { final boolean hasMetadata = p.hasMetadata(); metadata = p.metadata(); - if (hasMetadata) { + if (!hasMetadata) { lazyTerminate(STATE, this); p.release(); - throw new IllegalArgumentException("Metadata push does not support metadata field"); + throw new IllegalArgumentException("Metadata push should have metadata field present"); } if (!isValidMetadata(this.maxFrameLength, metadata)) { lazyTerminate(STATE, this); p.release(); - throw new IllegalArgumentException("Too Big Payload size"); + throw new IllegalArgumentException( + String.format(INVALID_PAYLOAD_ERROR_MESSAGE, this.maxFrameLength)); } } catch (IllegalReferenceCountException e) { lazyTerminate(STATE, this); diff --git a/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java b/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java index 327172255..183785d2f 100644 --- a/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java +++ b/rsocket-core/src/test/java/io/rsocket/core/RSocketRequesterTest.java @@ -33,6 +33,7 @@ import static io.rsocket.frame.FrameType.REQUEST_RESPONSE; import static io.rsocket.frame.FrameType.REQUEST_STREAM; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.verify; @@ -81,7 +82,6 @@ import java.util.function.BiFunction; import java.util.function.Function; import java.util.stream.Stream; -import org.assertj.core.api.Assertions; import org.assertj.core.api.Assumptions; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -169,7 +169,7 @@ protected void hookOnSubscribe(Subscription subscription) { public void testHandleSetupException() { rule.connection.addToReceivedBuffer( ErrorFrameCodec.encode(rule.alloc(), 0, new RejectedSetupException("boom"))); - Assertions.assertThatThrownBy(() -> rule.socket.onClose().block()) + assertThatThrownBy(() -> rule.socket.onClose().block()) .isInstanceOf(RejectedSetupException.class); rule.assertHasNoLeaks(); } @@ -373,6 +373,65 @@ public void shouldThrownExceptionIfGivenPayloadIsExitsSizeAllowanceWithNoFragmen }); } + @ParameterizedTest + @ValueSource(ints = {128, 256, FRAME_LENGTH_MASK}) + public void shouldThrownExceptionIfGivenPayloadIsExitsSizeAllowanceWithNoFragmentation1( + int maxFrameLength) { + rule.setMaxFrameLength(maxFrameLength); + prepareCalls() + .forEach( + generator -> { + byte[] metadata = new byte[maxFrameLength]; + byte[] data = new byte[maxFrameLength]; + ThreadLocalRandom.current().nextBytes(metadata); + ThreadLocalRandom.current().nextBytes(data); + + assertThatThrownBy( + () -> { + final Publisher source = + generator.apply(rule.socket, DefaultPayload.create(data, metadata)); + + if (source instanceof Mono) { + ((Mono) source).block(); + } else { + ((Flux) source).blockLast(); + } + }) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage(String.format(INVALID_PAYLOAD_ERROR_MESSAGE, maxFrameLength)); + + rule.assertHasNoLeaks(); + }); + } + + @Test + public void shouldRejectCallOfNoMetadataPayload() { + final ByteBuf data = rule.allocator.buffer(10); + final Payload payload = ByteBufPayload.create(data); + StepVerifier.create(rule.socket.metadataPush(payload)) + .expectSubscription() + .expectErrorSatisfies( + t -> + assertThat(t) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Metadata push should have metadata field present")) + .verify(); + PayloadAssert.assertThat(payload).isReleased(); + rule.assertHasNoLeaks(); + } + + @Test + public void shouldRejectCallOfNoMetadataPayloadBlocking() { + final ByteBuf data = rule.allocator.buffer(10); + final Payload payload = ByteBufPayload.create(data); + + assertThatThrownBy(() -> rule.socket.metadataPush(payload).block()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Metadata push should have metadata field present"); + PayloadAssert.assertThat(payload).isReleased(); + rule.assertHasNoLeaks(); + } + static Stream>> prepareCalls() { return Stream.of( RSocket::fireAndForget, From 571af15e2c9c6c89964113d2e0c6b528ffc662d5 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Mon, 28 Mar 2022 18:39:29 +0300 Subject: [PATCH 17/17] eliminates dropped error Signed-off-by: Oleh Dokuka Signed-off-by: Oleh Dokuka --- .../io/rsocket/resume/ResumableDuplexConnection.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java b/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java index 933ac09ca..f061857ff 100644 --- a/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java +++ b/rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java @@ -192,7 +192,8 @@ public void sendErrorAndClose(RSocketErrorException rSocketErrorException) { t -> { framesSaverDisposable.dispose(); activeReceivingSubscriber.dispose(); - savableFramesSender.dispose(); + savableFramesSender.onComplete(); + savableFramesSender.cancel(); onConnectionClosedSink.tryEmitComplete(); onClose.tryEmitError(t); @@ -200,7 +201,8 @@ public void sendErrorAndClose(RSocketErrorException rSocketErrorException) { () -> { framesSaverDisposable.dispose(); activeReceivingSubscriber.dispose(); - savableFramesSender.dispose(); + savableFramesSender.onComplete(); + savableFramesSender.cancel(); onConnectionClosedSink.tryEmitComplete(); final Throwable cause = rSocketErrorException.getCause(); @@ -253,7 +255,8 @@ void dispose(@Nullable Throwable e) { framesSaverDisposable.dispose(); activeReceivingSubscriber.dispose(); - savableFramesSender.dispose(); + savableFramesSender.onComplete(); + savableFramesSender.cancel(); onConnectionClosedSink.tryEmitComplete(); if (e != null) {