Fixed sending complete signal for mono.empty() (#521)

segabriel · robertroeser · commit 64def74f7ed6 · 2018-07-25T15:08:22.000-07:00
diff --git a/rsocket-core/src/main/java/io/rsocket/RSocketServer.java b/rsocket-core/src/main/java/io/rsocket/RSocketServer.java
@@ -302,6 +302,7 @@ private void handleRequestResponse(int streamId, Mono<Payload> response) {
               payload.release();
               return frame;
             })
+        .switchIfEmpty(Mono.fromCallable(() -> Frame.PayloadFrame.from(streamId, FrameType.COMPLETE)))
         .doFinally(signalType -> sendingSubscriptions.remove(streamId))
         .subscribe(sendProcessor::onNext, t -> handleError(streamId, t));
   }

Original file line number	Diff line number	Diff line change
`@@ -302,6 +302,7 @@ private void handleRequestResponse(int streamId, Mono<Payload> response) {`
`302`	`302`	`payload.release();`
`303`	`303`	`return frame;`
`304`	`304`	`})`
	`305`	`+ .switchIfEmpty(Mono.fromCallable(() -> Frame.PayloadFrame.from(streamId, FrameType.COMPLETE)))`
`305`	`306`	`.doFinally(signalType -> sendingSubscriptions.remove(streamId))`
`306`	`307`	`.subscribe(sendProcessor::onNext, t -> handleError(streamId, t));`
`307`	`308`	`}`