Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Flush after each WebSocket message by default

Browse Source 
Issue: SPR-14527
This commit is contained in:
Rossen Stoyanchev
2016-12-06 17:07:07 -05:00
parent c097c0284a
commit 2c2de82ffb
2 changed files with 3 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/ReactorNettyWebSocketSession.java
View File

@@ -56,8 +56,9 @@ public class ReactorNettyWebSocketSession
@Override @Override
public Mono<Void> send(Publisher<WebSocketMessage> messages) { public Mono<Void> send(Publisher<WebSocketMessage> messages) {
HttpOutbound outbound = getDelegate().getHttpOutbound();
Flux<WebSocketFrame> frameFlux = Flux.from(messages).map(this::toFrame); Flux<WebSocketFrame> frameFlux = Flux.from(messages).map(this::toFrame);
HttpOutbound outbound = getDelegate().getHttpOutbound();
outbound.flushEach();
return outbound.sendObject(frameFlux); return outbound.sendObject(frameFlux);
} }

2
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/RxNettyWebSocketSession.java
View File

@@ -55,7 +55,7 @@ public class RxNettyWebSocketSession extends NettyWebSocketSessionSupport<WebSoc
@Override @Override
public Mono<Void> send(Publisher<WebSocketMessage> messages) { public Mono<Void> send(Publisher<WebSocketMessage> messages) {
Observable<WebSocketFrame> frames = RxReactiveStreams.toObservable(messages).map(this::toFrame); Observable<WebSocketFrame> frames = RxReactiveStreams.toObservable(messages).map(this::toFrame);
Observable<Void> completion = getDelegate().write(frames); Observable<Void> completion = getDelegate().writeAndFlushOnEach(frames);
return Mono.from(RxReactiveStreams.toPublisher(completion)); return Mono.from(RxReactiveStreams.toPublisher(completion));
} }