Switch to Reactor snapshots

See gh-943

Signed-off-by: Rossen Stoyanchev <[email protected]>

Author: rstoyanchev

build.gradle

@@ -32,7 +32,7 @@ subprojects {
     apply plugin: 'io.spring.dependency-management'
     apply plugin: 'com.github.sherter.google-java-format'
 
-    ext['reactor-bom.version'] = '2020.0.0-RC1'
+    ext['reactor-bom.version'] = '2020.0.0-SNAPSHOT'
     ext['logback.version'] = '1.2.3'
     ext['netty-bom.version'] = '4.1.52.Final'
     ext['netty-boringssl.version'] = '2.0.34.Final'
@@ -96,7 +96,7 @@ subprojects {
         mavenCentral()
 
         maven {
-            url 'https://repo.spring.io/libs-snapshot'
+            url 'https://repo.spring.io/snapshot'
             content {
                 includeGroup "io.projectreactor"
                 includeGroup "io.projectreactor.netty"

rsocket-core/src/main/java/io/rsocket/resume/ResumableDuplexConnection.java

@@ -68,7 +68,7 @@ public class ResumableDuplexConnection extends Flux<ByteBuf>
   public ResumableDuplexConnection(
       String tag, DuplexConnection initialConnection, ResumableFramesStore resumableFramesStore) {
     this.tag = tag;
-    this.onConnectionClosedSink = Sinks.many().unsafe().unicast().onBackpressureBuffer();
+    this.onConnectionClosedSink = Sinks.unsafe().many().unicast().onBackpressureBuffer();
     this.resumableFramesStore = resumableFramesStore;
     this.savableFramesSender = new UnboundedProcessor();
     this.framesSaverDisposable = resumableFramesStore.saveFrames(savableFramesSender).subscribe();
@@ -114,7 +114,10 @@ void initConnection(DuplexConnection nextConnection) {
             __ -> {
               frameReceivingSubscriber.dispose();
               disposable.dispose();
-              onConnectionClosedSink.emitNext(currentConnectionIndex);
+              Sinks.Emission emission = onConnectionClosedSink.tryEmitNext(currentConnectionIndex);
+              if (emission.equals(Sinks.Emission.OK)) {
+                logger.error("Failed to notify session of closed connection: {}", emission);
+              }
             })
         .subscribe();
   }
@@ -160,13 +163,13 @@ public void sendErrorAndClose(RSocketErrorException rSocketErrorException) {
             t -> {
               framesSaverDisposable.dispose();
               savableFramesSender.dispose();
-              onConnectionClosedSink.emitComplete();
+              onConnectionClosedSink.tryEmitComplete();
               onClose.onError(t);
             },
             () -> {
               framesSaverDisposable.dispose();
               savableFramesSender.dispose();
-              onConnectionClosedSink.emitComplete();
+              onConnectionClosedSink.tryEmitComplete();
               final Throwable cause = rSocketErrorException.getCause();
               if (cause == null) {
                 onClose.onComplete();
@@ -206,7 +209,7 @@ public void dispose() {
     framesSaverDisposable.dispose();
     activeReceivingSubscriber.dispose();
     savableFramesSender.dispose();
-    onConnectionClosedSink.emitComplete();
+    onConnectionClosedSink.tryEmitComplete();
     onClose.onComplete();
   }