From 34f0586741a14cf225934793743c23469b411cfd Mon Sep 17 00:00:00 2001
From: Nitesh Kant <nitesh1706@gmail.com>
Date: Fri, 3 Jun 2016 15:55:19 -0700
Subject: [PATCH] Frame instance was referenced longer than required.

Problem:

`handleXXX` methods in `Responder` were closing over the passed `requestFrame` and using it later in the lifecycle of request processing.
 `Frame` objects and the underlying buffers are not designed to be retained after the scope of the parent method as these objects are threadlocal and reused. This causes issues when the frame object is referenced later in the request processing (eg: `cleanup()`)

 Solution:

 The only reason frame object was retained was to get the stream Id. This change pre-fetches the `streamId` and uses that from within the processing closure.

 Result:

 No more issue with frame access.
---
 .../io/reactivesocket/internal/Responder.java | 32 +++++++++----------
 1 file changed, 15 insertions(+), 17 deletions(-)

diff --git a/src/main/java/io/reactivesocket/internal/Responder.java b/src/main/java/io/reactivesocket/internal/Responder.java
index 4a24cd594..0af49c4ff 100644
--- a/src/main/java/io/reactivesocket/internal/Responder.java
+++ b/src/main/java/io/reactivesocket/internal/Responder.java
@@ -259,7 +259,7 @@ public void onNext(Frame requestFrame) {
 						} else if (requestFrame.getType() == FrameType.CANCEL) {
 							Subscription s;
 							synchronized (Responder.this) {
-								s = cancellationSubscriptions.get(requestFrame.getStreamId());
+								s = cancellationSubscriptions.get(streamId);
 							}
 							if (s != null) {
 								s.cancel();
@@ -268,7 +268,7 @@ public void onNext(Frame requestFrame) {
 						} else if (requestFrame.getType() == FrameType.REQUEST_N) {
 							SubscriptionArbiter inFlightSubscription;
 							synchronized (Responder.this) {
-								inFlightSubscription = inFlight.get(requestFrame.getStreamId());
+								inFlightSubscription = inFlight.get(streamId);
 							}
 							if (inFlightSubscription != null) {
 								long requestN = Frame.RequestN.requestN(requestFrame);
@@ -399,6 +399,7 @@ private Publisher<Frame> handleRequestResponse(
 			final RequestHandler requestHandler,
 			final Int2ObjectHashMap<Subscription> cancellationSubscriptions) {
 
+		final int streamId = requestFrame.getStreamId();
 		return child -> {
 			Subscription s = new Subscription() {
 
@@ -408,8 +409,6 @@ private Publisher<Frame> handleRequestResponse(
 				@Override
 				public void request(long n) {
 					if (n > 0 && started.compareAndSet(false, true)) {
-						final int streamId = requestFrame.getStreamId();
-
 						try {
 							Publisher<Payload> responsePublisher =
 									requestHandler.handleRequestResponse(requestFrame);
@@ -477,13 +476,13 @@ public void cancel() {
 
 				private void cleanup() {
 					synchronized(Responder.this) {
-						cancellationSubscriptions.remove(requestFrame.getStreamId());
+						cancellationSubscriptions.remove(streamId);
 					}
 				}
 
 			};
 			synchronized(Responder.this) {
-				cancellationSubscriptions.put(requestFrame.getStreamId(), s);
+				cancellationSubscriptions.put(streamId, s);
 			}
 			child.onSubscribe(s);
 		};
@@ -541,7 +540,7 @@ private Publisher<Frame> _handleRequestStream(
 			final Int2ObjectHashMap<Subscription> cancellationSubscriptions,
 			final Int2ObjectHashMap<SubscriptionArbiter> inFlight,
 			final boolean allowCompletion) {
-
+		final int streamId = requestFrame.getStreamId();
 		return child -> {
 			Subscription s = new Subscription() {
 
@@ -556,7 +555,6 @@ public void request(long n) {
 					}
 					if (started.compareAndSet(false,  true)) {
 						arbiter.addTransportRequest(n);
-						final int streamId = requestFrame.getStreamId();
 
 						try {
 							Publisher<Payload> responses =
@@ -630,14 +628,14 @@ public void cancel() {
 
 				private void cleanup() {
 					synchronized(Responder.this) {
-						inFlight.remove(requestFrame.getStreamId());
-						cancellationSubscriptions.remove(requestFrame.getStreamId());
+						inFlight.remove(streamId);
+						cancellationSubscriptions.remove(streamId);
 					}
 				}
 
 			};
 			synchronized(Responder.this) {
-				cancellationSubscriptions.put(requestFrame.getStreamId(), s);
+				cancellationSubscriptions.put(streamId, s);
 			}
 			child.onSubscribe(s);
 
@@ -704,8 +702,9 @@ private Publisher<Frame> handleRequestChannel(Frame requestFrame,
 			Int2ObjectHashMap<SubscriptionArbiter> inFlight) {
 
 		UnicastSubject<Payload> channelSubject;
+		final int streamId = requestFrame.getStreamId();
 		synchronized(Responder.this) {
-			channelSubject = channels.get(requestFrame.getStreamId());
+			channelSubject = channels.get(streamId);
 		}
 		if (channelSubject == null) {
 			return child -> {
@@ -722,7 +721,6 @@ public void request(long n) {
 						}
 						if (started.compareAndSet(false, true)) {
 							arbiter.addTransportRequest(n);
-							final int streamId = requestFrame.getStreamId();
 
 							// first request on this channel
 							UnicastSubject<Payload> channelRequests =
@@ -816,14 +814,14 @@ public void cancel() {
 
 					private void cleanup() {
 						synchronized(Responder.this) {
-							inFlight.remove(requestFrame.getStreamId());
-							cancellationSubscriptions.remove(requestFrame.getStreamId());
+							inFlight.remove(streamId);
+							cancellationSubscriptions.remove(streamId);
 						}
 					}
 
 				};
 				synchronized(Responder.this) {
-					cancellationSubscriptions.put(requestFrame.getStreamId(), s);
+					cancellationSubscriptions.put(streamId, s);
 				}
 				child.onSubscribe(s);
 
@@ -848,7 +846,7 @@ private void cleanup() {
                 // handle time-gap issues like this?
                 // TODO validate with unit tests.
 				return PublisherUtils.errorFrame(
-                    requestFrame.getStreamId(), new RuntimeException("Channel unavailable"));
+						streamId, new RuntimeException("Channel unavailable"));
 			}
 		}
 	}