Move opperation bblocking up
@@ -190,14 +190,27 @@ class HttpRequestStrategy implements RequestStrategy, AutoCloseable {
else // ... or send when the previous inflight is done.
previous.thenRun(() -> dispatch.accept(request, vessel));
- handleAttempt(vessel, dispatch, blocker, request, result, documentId, 1);
+ handleAttempt(vessel, dispatch, request, result, 1);
+
+ result.thenRun(() -> {
+ CompletableFuture<Void> current;
+ synchronized (monitor) {
+ current = inflightById.get(documentId);
+ if (current == blocker) { // Release slot and clear map if no other operations enqueued for this doc-id ...
+ releaseSlot();
+ inflightById.put(documentId, null);
+ }
+ }
+ if (current != blocker) // ... or trigger sending the next enqueued operation.
+ blocker.complete(null);
+ });
+
return result;
}
/** Handles the result of one attempt at the given operation, retrying if necessary. */
private void handleAttempt(CompletableFuture<SimpleHttpResponse> vessel, BiConsumer<SimpleHttpRequest, CompletableFuture<SimpleHttpResponse>> dispatch,
- CompletableFuture<Void> blocker, SimpleHttpRequest request, CompletableFuture<SimpleHttpResponse> result,
- DocumentId documentId, int attempt) {
+ SimpleHttpRequest request, CompletableFuture<SimpleHttpResponse> result, int attempt) {
vessel.whenComplete((response, thrown) -> {
// Retry the operation if it failed with a transient error ...
if (thrown != null ? retry(request, thrown, attempt)
@@ -208,22 +221,11 @@ class HttpRequestStrategy implements RequestStrategy, AutoCloseable {
delayed.add(new CompletableFuture<>().thenRun(() -> dispatch.accept(request, retry)));
else
dispatch.accept(request, retry);
- handleAttempt(retry, dispatch, blocker, request, result, documentId, attempt + (hasFailed ? 0 : 1));
+ handleAttempt(retry, dispatch, request, result, attempt + (hasFailed ? 0 : 1));
return;
}
// ... or accept the outcome and mark the operation as complete.
- CompletableFuture<Void> current;
- synchronized (monitor) {
- current = inflightById.get(documentId);
- if (current == blocker) { // Release slot and clear map if no other operations enqueued for this doc-id ...
- releaseSlot();
- inflightById.put(documentId, null);
- }
- }
- if (current != blocker) // ... or trigger sending the next enqueued operation.
- blocker.complete(null);
-
if (thrown == null) result.complete(response);
else result.completeExceptionally(thrown);
});
