Upgrade to RSocket 0.12.1-RC3 and update tests

1) Tests use a timeout to avoid hanging issues
2) Some tests adjusted to work around potential rsocket-java issue
rsocket/rsocket-java#613

Author: rstoyanchev

spring-messaging/spring-messaging.gradle

@@ -7,7 +7,7 @@ dependencyManagement {
 	}
 }
 
-def rsocketVersion = "0.12.1-RC3-SNAPSHOT"
+def rsocketVersion = "0.12.1-RC3"
 
 dependencies {
 	compile(project(":spring-beans"))

spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketBufferLeakTests.java

@@ -138,7 +138,7 @@ public void assemblyTimeErrorForHandleAndReply() {
 		StepVerifier.create(result).expectErrorMatches(ex -> {
 			String prefix = "Ambiguous handler methods mapped for destination 'A.B':";
 			return ex.getMessage().startsWith(prefix);
-		}).verify();
+		}).verify(Duration.ofSeconds(5));
 	}
 
 	@Test
@@ -147,25 +147,25 @@ public void subscriptionTimeErrorForHandleAndReply() {
 		StepVerifier.create(result).expectErrorMatches(ex -> {
 			String prefix = "Cannot decode to [org.springframework.core.io.Resource]";
 			return ex.getMessage().contains(prefix);
-		}).verify();
+		}).verify(Duration.ofSeconds(5));
 	}
 
 	@Test
 	public void errorSignalWithExceptionHandler() {
 		Mono<String> result = requester.route("error-signal").data("foo").retrieveMono(String.class);
-		StepVerifier.create(result).expectNext("Handled 'bad input'").verifyComplete();
+		StepVerifier.create(result).expectNext("Handled 'bad input'").expectComplete().verify(Duration.ofSeconds(5));
 	}
 
 	@Test
 	public void ignoreInput() {
 		Flux<String> result = requester.route("ignore-input").data("a").retrieveFlux(String.class);
-		StepVerifier.create(result).expectNext("bar").verifyComplete();
+		StepVerifier.create(result).expectNext("bar").thenCancel().verify(Duration.ofSeconds(5));
 	}
 
 	@Test
 	public void retrieveMonoFromFluxResponderMethod() {
 		Mono<String> result = requester.route("request-stream").data("foo").retrieveMono(String.class);
-		StepVerifier.create(result).expectNext("foo-1").verifyComplete();
+		StepVerifier.create(result).expectNext("foo-1").expectComplete().verify(Duration.ofSeconds(5));
 	}
 
 

spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketClientToServerIntegrationTests.java

@@ -121,7 +121,8 @@ public void echo() {
 
 		StepVerifier.create(result)
 				.expectNext("Hello 1").expectNext("Hello 2").expectNext("Hello 3")
-				.verifyComplete();
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
 	}
 
 	@Test
@@ -131,7 +132,8 @@ public void echoAsync() {
 
 		StepVerifier.create(result)
 				.expectNext("Hello 1 async").expectNext("Hello 2 async").expectNext("Hello 3 async")
-				.verifyComplete();
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
 	}
 
 	@Test
@@ -141,7 +143,7 @@ public void echoStream() {
 		StepVerifier.create(result)
 				.expectNext("Hello 0").expectNextCount(6).expectNext("Hello 7")
 				.thenCancel()
-				.verify();
+				.verify(Duration.ofSeconds(5));
 	}
 
 	@Test
@@ -152,37 +154,46 @@ public void echoChannel() {
 
 		StepVerifier.create(result)
 				.expectNext("Hello 1 async").expectNextCount(8).expectNext("Hello 10 async")
-				.verifyComplete();
+				.thenCancel()  // https://github.com/rsocket/rsocket-java/issues/613
+				.verify(Duration.ofSeconds(5));
 	}
 
 	@Test
 	public void voidReturnValue() {
 		Flux<String> result = requester.route("void-return-value").data("Hello").retrieveFlux(String.class);
-		StepVerifier.create(result).verifyComplete();
+		StepVerifier.create(result).expectComplete().verify(Duration.ofSeconds(5));
 	}
 
 	@Test
 	public void voidReturnValueFromExceptionHandler() {
 		Flux<String> result = requester.route("void-return-value").data("bad").retrieveFlux(String.class);
-		StepVerifier.create(result).verifyComplete();
+		StepVerifier.create(result).expectComplete().verify(Duration.ofSeconds(5));
 	}
 
 	@Test
 	public void handleWithThrownException() {
 		Mono<String> result = requester.route("thrown-exception").data("a").retrieveMono(String.class);
-		StepVerifier.create(result).expectNext("Invalid input error handled").verifyComplete();
+		StepVerifier.create(result)
+				.expectNext("Invalid input error handled")
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
 	}
 
 	@Test
 	public void handleWithErrorSignal() {
 		Mono<String> result = requester.route("error-signal").data("a").retrieveMono(String.class);
-		StepVerifier.create(result).expectNext("Invalid input error handled").verifyComplete();
+		StepVerifier.create(result)
+				.expectNext("Invalid input error handled")
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
 	}
 
 	@Test
 	public void noMatchingRoute() {
 		Mono<String> result = requester.route("invalid").data("anything").retrieveMono(String.class);
-		StepVerifier.create(result).verifyErrorMessage("No handler for destination 'invalid'");
+		StepVerifier.create(result)
+				.expectErrorMessage("No handler for destination 'invalid'")
+				.verify(Duration.ofSeconds(5));
 	}
 
 

spring-messaging/src/test/java/org/springframework/messaging/rsocket/RSocketServerToClientIntegrationTests.java

@@ -151,7 +151,8 @@ void echo(RSocketRequester requester) {
 						.expectNext("Hello 1")
 						.expectNext("Hello 2")
 						.expectNext("Hello 3")
-						.verifyComplete();
+						.expectComplete()
+						.verify(Duration.ofSeconds(5));
 			});
 		}
 
@@ -165,7 +166,8 @@ void echoAsync(RSocketRequester requester) {
 						.expectNext("Hello 1 async")
 						.expectNext("Hello 2 async")
 						.expectNext("Hello 3 async")
-						.verifyComplete();
+						.expectComplete()
+						.verify(Duration.ofSeconds(5));
 			});
 		}
 
@@ -180,7 +182,7 @@ void echoStream(RSocketRequester requester) {
 						.expectNext("Hello 6")
 						.expectNext("Hello 7")
 						.thenCancel()
-						.verify();
+						.verify(Duration.ofSeconds(5));
 			});
 		}
 
@@ -196,7 +198,8 @@ void echoChannel(RSocketRequester requester) {
 						.expectNextCount(7)
 						.expectNext("Hello 9 async")
 						.expectNext("Hello 10 async")
-						.verifyComplete();
+						.thenCancel()  // https://github.com/rsocket/rsocket-java/issues/613
+						.verify(Duration.ofSeconds(5));
 			});
 		}