From 1443c0808dc1ccd32bf6eaf201c60db6115b704a Mon Sep 17 00:00:00 2001
From: Violeta Georgieva <vgeorgieva@pivotal.io>
Date: Wed, 11 Jan 2017 15:58:02 +0200
Subject: [PATCH] Enable suspend/resume on reactive JettyWebsocketSession

Issue: SPR-15130
---
 .../socket/adapter/JettyWebSocketSession.java       | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/JettyWebSocketSession.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/JettyWebSocketSession.java
index 97549e1a128..8beb4cd290c 100644
--- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/JettyWebSocketSession.java
+++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/JettyWebSocketSession.java
@@ -21,11 +21,13 @@ import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 
 import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.SuspendToken;
 import org.eclipse.jetty.websocket.api.WriteCallback;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
 
 import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 import org.springframework.web.reactive.socket.CloseStatus;
 import org.springframework.web.reactive.socket.HandshakeInfo;
@@ -42,6 +44,8 @@ import org.springframework.web.reactive.socket.WebSocketSession;
  */
 public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Session> {
 
+	private SuspendToken suspendToken;
+
 
 	public JettyWebSocketSession(Session session, HandshakeInfo info, DataBufferFactory factory) {
 		this(session, info, factory, null);
@@ -56,17 +60,20 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Sess
 
 	@Override
 	protected boolean canSuspendReceiving() {
-		return false;
+		return true;
 	}
 
 	@Override
 	protected void suspendReceiving() {
-		// no-op
+		Assert.isNull(this.suspendToken);
+		this.suspendToken = getDelegate().suspend();
 	}
 
 	@Override
 	protected void resumeReceiving() {
-		// no-op
+		Assert.notNull(this.suspendToken);
+		this.suspendToken.resume();
+		this.suspendToken = null;
 	}
 
 	@Override