Frame instance was referenced longer than required. (#88)

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.

Author: NiteshKant

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"));
 			}
 		}
 	}