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 handleRequestResponse( final RequestHandler requestHandler, final Int2ObjectHashMap cancellationSubscriptions) { + final int streamId = requestFrame.getStreamId(); return child -> { Subscription s = new Subscription() { @@ -408,8 +409,6 @@ private Publisher handleRequestResponse( @Override public void request(long n) { if (n > 0 && started.compareAndSet(false, true)) { - final int streamId = requestFrame.getStreamId(); - try { Publisher 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 _handleRequestStream( final Int2ObjectHashMap cancellationSubscriptions, final Int2ObjectHashMap 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 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 handleRequestChannel(Frame requestFrame, Int2ObjectHashMap inFlight) { UnicastSubject 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 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")); } } }