You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hc.apache.org by ol...@apache.org on 2020/11/05 21:14:01 UTC
[httpcomponents-core] 02/03: Convenience base classes for
FutureCallback implementations
This is an automated email from the ASF dual-hosted git repository.
olegk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/httpcomponents-core.git
commit 979714c350b394018e9cde3ec8f90f4188330124
Author: Oleg Kalnichevski <ol...@apache.org>
AuthorDate: Fri Oct 30 23:09:07 2020 +0100
Convenience base classes for FutureCallback implementations
---
.../nio/bootstrap/H2MultiplexingRequester.java | 28 ++++-------
.../apache/hc/core5/http2/nio/pool/H2ConnPool.java | 53 ++++++++++----------
.../core5/testing/nio/ClientSessionEndpoint.java | 13 +----
.../apache/hc/core5/testing/nio/H2TestClient.java | 12 +----
.../hc/core5/testing/nio/Http1TestClient.java | 12 +----
.../hc/core5/concurrent/CallbackContribution.java | 58 ++++++++++++++++++++++
.../hc/core5/concurrent/FutureContribution.java | 58 ++++++++++++++++++++++
.../http/impl/bootstrap/HttpAsyncRequester.java | 26 ++++------
.../hc/core5/http/nio/AsyncClientEndpoint.java | 13 +----
.../support/AbstractAsyncRequesterConsumer.java | 13 +----
.../nio/support/AbstractAsyncResponseConsumer.java | 17 +------
.../http/nio/support/BasicRequestConsumer.java | 17 +------
.../http/nio/support/BasicResponseConsumer.java | 17 +------
.../hc/core5/reactor/AbstractIOSessionPool.java | 13 +----
.../hc/core5/concurrent/TestComplexFuture.java | 12 +----
15 files changed, 181 insertions(+), 181 deletions(-)
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
index 0d86cc7..552b4d3 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/bootstrap/H2MultiplexingRequester.java
@@ -39,6 +39,7 @@ import org.apache.hc.core5.concurrent.Cancellable;
import org.apache.hc.core5.concurrent.CancellableDependency;
import org.apache.hc.core5.concurrent.ComplexFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.concurrent.FutureContribution;
import org.apache.hc.core5.function.Callback;
import org.apache.hc.core5.function.Decorator;
import org.apache.hc.core5.function.Resolver;
@@ -260,24 +261,17 @@ public class H2MultiplexingRequester extends AsyncRequester{
Args.notNull(responseConsumer, "Response consumer");
Args.notNull(timeout, "Timeout");
final ComplexFuture<T> future = new ComplexFuture<>(callback);
- final AsyncClientExchangeHandler exchangeHandler = new BasicClientExchangeHandler<>(requestProducer, responseConsumer, new FutureCallback<T>() {
-
- @Override
- public void completed(final T result) {
- future.completed(result);
- }
-
- @Override
- public void failed(final Exception ex) {
- future.failed(ex);
- }
-
- @Override
- public void cancelled() {
- future.cancel();
- }
+ final AsyncClientExchangeHandler exchangeHandler = new BasicClientExchangeHandler<>(
+ requestProducer,
+ responseConsumer,
+ new FutureContribution<T>(future) {
+
+ @Override
+ public void completed(final T result) {
+ future.completed(result);
+ }
- });
+ });
execute(exchangeHandler, pushHandlerFactory, future, timeout, context != null ? context : HttpCoreContext.create());
return future;
}
diff --git a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java
index 6c7d761..8b8f32b 100644
--- a/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java
+++ b/httpcore5-h2/src/main/java/org/apache/hc/core5/http2/nio/pool/H2ConnPool.java
@@ -31,6 +31,7 @@ import java.util.concurrent.Future;
import org.apache.hc.core5.annotation.Contract;
import org.apache.hc.core5.annotation.ThreadingBehavior;
+import org.apache.hc.core5.concurrent.CallbackContribution;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.function.Callback;
import org.apache.hc.core5.function.Resolver;
@@ -100,36 +101,32 @@ public final class H2ConnPool extends AbstractIOSessionPool<HttpHost> {
final Timeout connectTimeout,
final FutureCallback<IOSession> callback) {
final InetSocketAddress remoteAddress = addressResolver.resolve(namedEndpoint);
- return connectionInitiator.connect(namedEndpoint, remoteAddress, null, connectTimeout, null, new FutureCallback<IOSession>() {
-
- @Override
- public void completed(final IOSession ioSession) {
- if (tlsStrategy != null
- && URIScheme.HTTPS.same(namedEndpoint.getSchemeName())
- && ioSession instanceof TransportSecurityLayer) {
- tlsStrategy.upgrade(
- (TransportSecurityLayer) ioSession,
- namedEndpoint,
- ioSession.getLocalAddress(),
- ioSession.getRemoteAddress(),
- null,
- connectTimeout);
- ioSession.setSocketTimeout(connectTimeout);
- }
- callback.completed(ioSession);
- }
-
- @Override
- public void failed(final Exception ex) {
- callback.failed(ex);
- }
+ return connectionInitiator.connect(
+ namedEndpoint,
+ remoteAddress,
+ null,
+ connectTimeout,
+ null,
+ new CallbackContribution<IOSession>(callback) {
- @Override
- public void cancelled() {
- callback.cancelled();
- }
+ @Override
+ public void completed(final IOSession ioSession) {
+ if (tlsStrategy != null
+ && URIScheme.HTTPS.same(namedEndpoint.getSchemeName())
+ && ioSession instanceof TransportSecurityLayer) {
+ tlsStrategy.upgrade(
+ (TransportSecurityLayer) ioSession,
+ namedEndpoint,
+ ioSession.getLocalAddress(),
+ ioSession.getRemoteAddress(),
+ null,
+ connectTimeout);
+ ioSession.setSocketTimeout(connectTimeout);
+ }
+ callback.completed(ioSession);
+ }
- });
+ });
}
@Override
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java
index 6686f59..824708d 100644
--- a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java
+++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/ClientSessionEndpoint.java
@@ -35,6 +35,7 @@ import org.apache.hc.core5.annotation.Contract;
import org.apache.hc.core5.annotation.ThreadingBehavior;
import org.apache.hc.core5.concurrent.BasicFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.concurrent.FutureContribution;
import org.apache.hc.core5.http.ConnectionClosedException;
import org.apache.hc.core5.http.nio.AsyncClientExchangeHandler;
import org.apache.hc.core5.http.nio.AsyncPushConsumer;
@@ -102,23 +103,13 @@ public final class ClientSessionEndpoint implements ModalCloseable {
Asserts.check(!closed.get(), "Connection is already closed");
final BasicFuture<T> future = new BasicFuture<>(callback);
execute(new BasicClientExchangeHandler<>(requestProducer, responseConsumer,
- new FutureCallback<T>() {
+ new FutureContribution<T>(future) {
@Override
public void completed(final T result) {
future.completed(result);
}
- @Override
- public void failed(final Exception ex) {
- future.failed(ex);
- }
-
- @Override
- public void cancelled() {
- future.cancel();
- }
-
}),
pushHandlerFactory, context);
return future;
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/H2TestClient.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/H2TestClient.java
index 2e62caf..48b3682 100644
--- a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/H2TestClient.java
+++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/H2TestClient.java
@@ -32,6 +32,7 @@ import java.util.concurrent.Future;
import javax.net.ssl.SSLContext;
+import org.apache.hc.core5.concurrent.FutureContribution;
import org.apache.hc.core5.concurrent.BasicFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.function.Supplier;
@@ -129,22 +130,13 @@ public class H2TestClient extends AsyncRequester {
final Timeout timeout,
final FutureCallback<ClientSessionEndpoint> callback) {
final BasicFuture<ClientSessionEndpoint> future = new BasicFuture<>(callback);
- requestSession(host, timeout, new FutureCallback<IOSession>() {
+ requestSession(host, timeout, new FutureContribution<IOSession>(future) {
@Override
public void completed(final IOSession session) {
future.completed(new ClientSessionEndpoint(session));
}
- @Override
- public void failed(final Exception cause) {
- future.failed(cause);
- }
-
- @Override
- public void cancelled() {
- future.cancel();
- }
});
return future;
}
diff --git a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/Http1TestClient.java b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/Http1TestClient.java
index a8e7fad..c3de8b2 100644
--- a/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/Http1TestClient.java
+++ b/httpcore5-testing/src/main/java/org/apache/hc/core5/testing/nio/Http1TestClient.java
@@ -32,6 +32,7 @@ import java.util.concurrent.Future;
import javax.net.ssl.SSLContext;
+import org.apache.hc.core5.concurrent.FutureContribution;
import org.apache.hc.core5.concurrent.BasicFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.http.HttpHost;
@@ -93,22 +94,13 @@ public class Http1TestClient extends AsyncRequester {
final Timeout timeout,
final FutureCallback<ClientSessionEndpoint> callback) {
final BasicFuture<ClientSessionEndpoint> future = new BasicFuture<>(callback);
- requestSession(host, timeout, new FutureCallback<IOSession>() {
+ requestSession(host, timeout, new FutureContribution<IOSession>(future) {
@Override
public void completed(final IOSession session) {
future.completed(new ClientSessionEndpoint(session));
}
- @Override
- public void failed(final Exception cause) {
- future.failed(cause);
- }
-
- @Override
- public void cancelled() {
- future.cancel();
- }
});
return future;
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/concurrent/CallbackContribution.java b/httpcore5/src/main/java/org/apache/hc/core5/concurrent/CallbackContribution.java
new file mode 100644
index 0000000..4dc602f
--- /dev/null
+++ b/httpcore5/src/main/java/org/apache/hc/core5/concurrent/CallbackContribution.java
@@ -0,0 +1,58 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.core5.concurrent;
+
+/**
+ * Convenience base class for {@link FutureCallback}s that contribute a result
+ * of the operation to another {@link FutureCallback}.
+ *
+ * @param <T> the future result type of an asynchronous operation.
+ * @since 5.1
+ */
+public abstract class CallbackContribution<T> implements FutureCallback<T> {
+
+ private final FutureCallback<?> callback;
+
+ public CallbackContribution(final FutureCallback<?> callback) {
+ this.callback = callback;
+ }
+
+ @Override
+ public final void failed(final Exception ex) {
+ if (callback != null) {
+ callback.failed(ex);
+ }
+ }
+
+ @Override
+ public final void cancelled() {
+ if (callback != null) {
+ callback.cancelled();
+ }
+ }
+
+}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/concurrent/FutureContribution.java b/httpcore5/src/main/java/org/apache/hc/core5/concurrent/FutureContribution.java
new file mode 100644
index 0000000..8960b4a
--- /dev/null
+++ b/httpcore5/src/main/java/org/apache/hc/core5/concurrent/FutureContribution.java
@@ -0,0 +1,58 @@
+/*
+ * ====================================================================
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation. For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.core5.concurrent;
+
+/**
+ * Convenience base class for {@link FutureCallback}s that contribute a result
+ * of the operation to another {@link BasicFuture}.
+ *
+ * @param <T> the future result type of an asynchronous operation.
+ * @since 5.1
+ */
+public abstract class FutureContribution<T> implements FutureCallback<T> {
+
+ private final BasicFuture<?> future;
+
+ public FutureContribution(final BasicFuture<?> future) {
+ this.future = future;
+ }
+
+ @Override
+ public final void failed(final Exception ex) {
+ if (future != null) {
+ future.failed(ex);
+ }
+ }
+
+ @Override
+ public final void cancelled() {
+ if (future != null) {
+ future.cancel();
+ }
+ }
+
+}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
index abafada..7657b2a 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/impl/bootstrap/HttpAsyncRequester.java
@@ -38,6 +38,7 @@ import org.apache.hc.core5.annotation.Internal;
import org.apache.hc.core5.concurrent.BasicFuture;
import org.apache.hc.core5.concurrent.ComplexFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.concurrent.FutureContribution;
import org.apache.hc.core5.function.Callback;
import org.apache.hc.core5.function.Decorator;
import org.apache.hc.core5.http.ConnectionClosedException;
@@ -376,24 +377,17 @@ public class HttpAsyncRequester extends AsyncRequester implements ConnPoolContro
Args.notNull(responseConsumer, "Response consumer");
Args.notNull(timeout, "Timeout");
final BasicFuture<T> future = new BasicFuture<>(callback);
- final AsyncClientExchangeHandler exchangeHandler = new BasicClientExchangeHandler<>(requestProducer, responseConsumer, new FutureCallback<T>() {
+ final AsyncClientExchangeHandler exchangeHandler = new BasicClientExchangeHandler<>(
+ requestProducer,
+ responseConsumer,
+ new FutureContribution<T>(future) {
- @Override
- public void completed(final T result) {
- future.completed(result);
- }
-
- @Override
- public void failed(final Exception ex) {
- future.failed(ex);
- }
-
- @Override
- public void cancelled() {
- future.cancel();
- }
+ @Override
+ public void completed(final T result) {
+ future.completed(result);
+ }
- });
+ });
execute(exchangeHandler, pushHandlerFactory, timeout, context != null ? context : HttpCoreContext.create());
return future;
}
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/AsyncClientEndpoint.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/AsyncClientEndpoint.java
index 2912bf4..749bf32 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/AsyncClientEndpoint.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/AsyncClientEndpoint.java
@@ -33,6 +33,7 @@ import org.apache.hc.core5.annotation.Contract;
import org.apache.hc.core5.annotation.ThreadingBehavior;
import org.apache.hc.core5.concurrent.BasicFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.concurrent.FutureContribution;
import org.apache.hc.core5.http.nio.support.BasicClientExchangeHandler;
import org.apache.hc.core5.http.protocol.HttpContext;
import org.apache.hc.core5.http.protocol.HttpCoreContext;
@@ -100,23 +101,13 @@ public abstract class AsyncClientEndpoint {
final FutureCallback<T> callback) {
final BasicFuture<T> future = new BasicFuture<>(callback);
execute(new BasicClientExchangeHandler<>(requestProducer, responseConsumer,
- new FutureCallback<T>() {
+ new FutureContribution<T>(future) {
@Override
public void completed(final T result) {
future.completed(result);
}
- @Override
- public void failed(final Exception ex) {
- future.failed(ex);
- }
-
- @Override
- public void cancelled() {
- future.cancel();
- }
-
}),
pushHandlerFactory, context != null ? context : HttpCoreContext.create());
return future;
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncRequesterConsumer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncRequesterConsumer.java
index da6b861..02447ca 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncRequesterConsumer.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncRequesterConsumer.java
@@ -32,6 +32,7 @@ import java.nio.charset.UnsupportedCharsetException;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hc.core5.concurrent.CallbackContribution;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.function.Supplier;
import org.apache.hc.core5.http.ContentType;
@@ -95,7 +96,7 @@ public abstract class AbstractAsyncRequesterConsumer<T, E> implements AsyncReque
throw new HttpException("Supplied data consumer is null");
}
dataConsumerRef.set(dataConsumer);
- dataConsumer.streamStart(entityDetails, new FutureCallback<E>() {
+ dataConsumer.streamStart(entityDetails, new CallbackContribution<E>(resultCallback) {
@Override
public void completed(final E entity) {
@@ -109,16 +110,6 @@ public abstract class AbstractAsyncRequesterConsumer<T, E> implements AsyncReque
}
}
- @Override
- public void failed(final Exception ex) {
- resultCallback.failed(ex);
- }
-
- @Override
- public void cancelled() {
- resultCallback.cancelled();
- }
-
});
} else {
resultCallback.completed(buildResult(request, null, null));
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncResponseConsumer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncResponseConsumer.java
index 9012ee2..3b55735 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncResponseConsumer.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/AbstractAsyncResponseConsumer.java
@@ -32,6 +32,7 @@ import java.nio.charset.UnsupportedCharsetException;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hc.core5.concurrent.CallbackContribution;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.function.Supplier;
import org.apache.hc.core5.http.ContentType;
@@ -95,7 +96,7 @@ public abstract class AbstractAsyncResponseConsumer<T, E> implements AsyncRespon
throw new HttpException("Supplied data consumer is null");
}
dataConsumerRef.set(dataConsumer);
- dataConsumer.streamStart(entityDetails, new FutureCallback<E>() {
+ dataConsumer.streamStart(entityDetails, new CallbackContribution<E>(resultCallback) {
@Override
public void completed(final E entity) {
@@ -113,20 +114,6 @@ public abstract class AbstractAsyncResponseConsumer<T, E> implements AsyncRespon
}
}
- @Override
- public void failed(final Exception ex) {
- if (resultCallback != null) {
- resultCallback.failed(ex);
- }
- }
-
- @Override
- public void cancelled() {
- if (resultCallback != null) {
- resultCallback.cancelled();
- }
- }
-
});
} else {
final T result = buildResult(response, null, null);
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicRequestConsumer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicRequestConsumer.java
index bec2232..6eaf64c 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicRequestConsumer.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicRequestConsumer.java
@@ -31,6 +31,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hc.core5.concurrent.CallbackContribution;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.function.Supplier;
import org.apache.hc.core5.http.EntityDetails;
@@ -85,7 +86,7 @@ public class BasicRequestConsumer<T> implements AsyncRequestConsumer<Message<Htt
}
dataConsumerRef.set(dataConsumer);
- dataConsumer.streamStart(entityDetails, new FutureCallback<T>() {
+ dataConsumer.streamStart(entityDetails, new CallbackContribution<T>(resultCallback) {
@Override
public void completed(final T body) {
@@ -95,20 +96,6 @@ public class BasicRequestConsumer<T> implements AsyncRequestConsumer<Message<Htt
}
}
- @Override
- public void failed(final Exception ex) {
- if (resultCallback != null) {
- resultCallback.failed(ex);
- }
- }
-
- @Override
- public void cancelled() {
- if (resultCallback != null) {
- resultCallback.cancelled();
- }
- }
-
});
} else {
final Message<HttpRequest, T> result = new Message<>(request, null);
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicResponseConsumer.java b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicResponseConsumer.java
index d6a2ff8..c9d48fa 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicResponseConsumer.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/http/nio/support/BasicResponseConsumer.java
@@ -31,6 +31,7 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.hc.core5.concurrent.CallbackContribution;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.function.Supplier;
import org.apache.hc.core5.http.EntityDetails;
@@ -85,7 +86,7 @@ public class BasicResponseConsumer<T> implements AsyncResponseConsumer<Message<H
throw new HttpException("Supplied data consumer is null");
}
dataConsumerRef.set(dataConsumer);
- dataConsumer.streamStart(entityDetails, new FutureCallback<T>() {
+ dataConsumer.streamStart(entityDetails, new CallbackContribution<T>(resultCallback) {
@Override
public void completed(final T body) {
@@ -95,20 +96,6 @@ public class BasicResponseConsumer<T> implements AsyncResponseConsumer<Message<H
}
}
- @Override
- public void failed(final Exception ex) {
- if (resultCallback != null) {
- resultCallback.failed(ex);
- }
- }
-
- @Override
- public void cancelled() {
- if (resultCallback != null) {
- resultCallback.cancelled();
- }
- }
-
});
} else {
final Message<HttpResponse, T> result = new Message<>(response, null);
diff --git a/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java b/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java
index 08049be..cce68d3 100644
--- a/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java
+++ b/httpcore5/src/main/java/org/apache/hc/core5/reactor/AbstractIOSessionPool.java
@@ -39,6 +39,7 @@ import org.apache.hc.core5.annotation.Contract;
import org.apache.hc.core5.annotation.ThreadingBehavior;
import org.apache.hc.core5.concurrent.ComplexFuture;
import org.apache.hc.core5.concurrent.FutureCallback;
+import org.apache.hc.core5.concurrent.FutureContribution;
import org.apache.hc.core5.function.Callback;
import org.apache.hc.core5.http.ConnectionClosedException;
import org.apache.hc.core5.io.CloseMode;
@@ -140,23 +141,13 @@ public abstract class AbstractIOSessionPool<T> implements ModalCloseable {
future.completed(ioSession);
} else {
getSessionInternal(poolEntry, true, endpoint, connectTimeout,
- new FutureCallback<IOSession>() {
+ new FutureContribution<IOSession>(future) {
@Override
public void completed(final IOSession ioSession) {
future.completed(ioSession);
}
- @Override
- public void failed(final Exception ex) {
- future.failed(ex);
- }
-
- @Override
- public void cancelled() {
- future.cancel();
- }
-
});
}
}
diff --git a/httpcore5/src/test/java/org/apache/hc/core5/concurrent/TestComplexFuture.java b/httpcore5/src/test/java/org/apache/hc/core5/concurrent/TestComplexFuture.java
index e342867..ed04a82 100644
--- a/httpcore5/src/test/java/org/apache/hc/core5/concurrent/TestComplexFuture.java
+++ b/httpcore5/src/test/java/org/apache/hc/core5/concurrent/TestComplexFuture.java
@@ -74,22 +74,12 @@ public class TestComplexFuture {
public void testCancelledWithCallback() throws Exception {
final ComplexFuture<Object> future = new ComplexFuture<>(null);
- final Future<Object> dependency1 = new BasicFuture<>(new FutureCallback<Object>() {
+ final Future<Object> dependency1 = new BasicFuture<>(new FutureContribution<Object>(future) {
@Override
public void completed(final Object result) {
}
- @Override
- public void failed(final Exception ex) {
- future.failed(ex);
- }
-
- @Override
- public void cancelled() {
- future.cancel();
- }
-
});
future.setDependency(dependency1);