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