| /* |
| * ==================================================================== |
| * 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.client5.http.impl.cache; |
| |
| import java.io.IOException; |
| import java.io.InterruptedIOException; |
| import java.nio.ByteBuffer; |
| import java.time.Instant; |
| import java.util.Collection; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.function.Consumer; |
| |
| import org.apache.hc.client5.http.HttpRoute; |
| import org.apache.hc.client5.http.async.AsyncExecCallback; |
| import org.apache.hc.client5.http.async.AsyncExecChain; |
| import org.apache.hc.client5.http.async.AsyncExecChainHandler; |
| import org.apache.hc.client5.http.async.methods.SimpleBody; |
| import org.apache.hc.client5.http.async.methods.SimpleHttpResponse; |
| import org.apache.hc.client5.http.cache.CacheResponseStatus; |
| import org.apache.hc.client5.http.cache.HttpCacheContext; |
| import org.apache.hc.client5.http.cache.HttpCacheEntry; |
| import org.apache.hc.client5.http.cache.RequestCacheControl; |
| import org.apache.hc.client5.http.cache.ResourceIOException; |
| import org.apache.hc.client5.http.cache.ResponseCacheControl; |
| import org.apache.hc.client5.http.impl.ExecSupport; |
| import org.apache.hc.client5.http.protocol.HttpClientContext; |
| import org.apache.hc.client5.http.schedule.SchedulingStrategy; |
| import org.apache.hc.client5.http.validator.ETag; |
| import org.apache.hc.core5.annotation.Contract; |
| import org.apache.hc.core5.annotation.ThreadingBehavior; |
| 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.http.ContentType; |
| import org.apache.hc.core5.http.EntityDetails; |
| import org.apache.hc.core5.http.Header; |
| import org.apache.hc.core5.http.HttpException; |
| import org.apache.hc.core5.http.HttpHeaders; |
| import org.apache.hc.core5.http.HttpHost; |
| import org.apache.hc.core5.http.HttpRequest; |
| import org.apache.hc.core5.http.HttpResponse; |
| import org.apache.hc.core5.http.HttpStatus; |
| import org.apache.hc.core5.http.impl.BasicEntityDetails; |
| import org.apache.hc.core5.http.message.RequestLine; |
| import org.apache.hc.core5.http.nio.AsyncDataConsumer; |
| import org.apache.hc.core5.http.nio.AsyncEntityProducer; |
| import org.apache.hc.core5.http.nio.CapacityChannel; |
| import org.apache.hc.core5.http.protocol.HttpCoreContext; |
| import org.apache.hc.core5.http.support.BasicRequestBuilder; |
| import org.apache.hc.core5.net.URIAuthority; |
| import org.apache.hc.core5.util.Args; |
| import org.apache.hc.core5.util.ByteArrayBuffer; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * Request executor in the request execution chain that is responsible for |
| * transparent client-side caching. |
| * <p> |
| * The current implementation is conditionally |
| * compliant with HTTP/1.1 (meaning all the MUST and MUST NOTs are obeyed), |
| * although quite a lot, though not all, of the SHOULDs and SHOULD NOTs |
| * are obeyed too. |
| * |
| * @since 5.0 |
| */ |
| @Contract(threading = ThreadingBehavior.SAFE) // So long as the responseCache implementation is threadsafe |
| class AsyncCachingExec extends CachingExecBase implements AsyncExecChainHandler { |
| |
| private static final Logger LOG = LoggerFactory.getLogger(AsyncCachingExec.class); |
| private final HttpAsyncCache responseCache; |
| private final DefaultAsyncCacheRevalidator cacheRevalidator; |
| private final ConditionalRequestBuilder<HttpRequest> conditionalRequestBuilder; |
| |
| AsyncCachingExec(final HttpAsyncCache cache, final DefaultAsyncCacheRevalidator cacheRevalidator, final CacheConfig config) { |
| super(config); |
| this.responseCache = Args.notNull(cache, "Response cache"); |
| this.cacheRevalidator = cacheRevalidator; |
| this.conditionalRequestBuilder = new ConditionalRequestBuilder<>(request -> |
| BasicRequestBuilder.copy(request).build()); |
| } |
| |
| AsyncCachingExec( |
| final HttpAsyncCache cache, |
| final ScheduledExecutorService executorService, |
| final SchedulingStrategy schedulingStrategy, |
| final CacheConfig config) { |
| this(cache, |
| executorService != null ? new DefaultAsyncCacheRevalidator(executorService, schedulingStrategy) : null, |
| config); |
| } |
| |
| private void triggerResponse( |
| final SimpleHttpResponse cacheResponse, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecCallback asyncExecCallback) { |
| scope.execRuntime.releaseEndpoint(); |
| |
| final SimpleBody body = cacheResponse.getBody(); |
| final byte[] content = body != null ? body.getBodyBytes() : null; |
| final ContentType contentType = body != null ? body.getContentType() : null; |
| try { |
| final AsyncDataConsumer dataConsumer = asyncExecCallback.handleResponse( |
| cacheResponse, |
| content != null ? new BasicEntityDetails(content.length, contentType) : null); |
| if (dataConsumer != null) { |
| if (content != null) { |
| dataConsumer.consume(ByteBuffer.wrap(content)); |
| } |
| dataConsumer.streamEnd(null); |
| } |
| asyncExecCallback.completed(); |
| } catch (final HttpException | IOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } |
| |
| static class AsyncExecCallbackWrapper implements AsyncExecCallback { |
| |
| private final Runnable command; |
| private final Consumer<Exception> exceptionConsumer; |
| |
| AsyncExecCallbackWrapper(final Runnable command, final Consumer<Exception> exceptionConsumer) { |
| this.command = command; |
| this.exceptionConsumer = exceptionConsumer; |
| } |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse response, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| return null; |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| } |
| |
| @Override |
| public void completed() { |
| command.run(); |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| if (exceptionConsumer != null) { |
| exceptionConsumer.accept(cause); |
| } |
| } |
| |
| } |
| |
| @Override |
| public void execute( |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) throws HttpException, IOException { |
| Args.notNull(request, "HTTP request"); |
| Args.notNull(scope, "Scope"); |
| |
| final HttpRoute route = scope.route; |
| final HttpClientContext context = scope.clientContext; |
| |
| final URIAuthority authority = request.getAuthority(); |
| final String scheme = request.getScheme(); |
| final HttpHost target = authority != null ? new HttpHost(scheme, authority) : route.getTargetHost(); |
| doExecute(target, |
| request, |
| entityProducer, |
| scope, |
| chain, |
| new AsyncExecCallback() { |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse response, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| context.setAttribute(HttpCoreContext.HTTP_REQUEST, request); |
| context.setAttribute(HttpCoreContext.HTTP_RESPONSE, response); |
| return asyncExecCallback.handleResponse(response, entityDetails); |
| } |
| |
| @Override |
| public void handleInformationResponse( |
| final HttpResponse response) throws HttpException, IOException { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| |
| @Override |
| public void completed() { |
| asyncExecCallback.completed(); |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| asyncExecCallback.failed(cause); |
| } |
| |
| }); |
| } |
| |
| public void doExecute( |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) throws HttpException, IOException { |
| |
| final String exchangeId = scope.exchangeId; |
| final HttpCacheContext context = HttpCacheContext.adapt(scope.clientContext); |
| final CancellableDependency operation = scope.cancellableDependency; |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} request via cache: {}", exchangeId, new RequestLine(request)); |
| } |
| |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MISS); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, null); |
| |
| if (clientRequestsOurOptions(request)) { |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| triggerResponse(SimpleHttpResponse.create(HttpStatus.SC_NOT_IMPLEMENTED), scope, asyncExecCallback); |
| return; |
| } |
| |
| final RequestCacheControl requestCacheControl; |
| if (request.containsHeader(HttpHeaders.CACHE_CONTROL)) { |
| requestCacheControl = CacheControlHeaderParser.INSTANCE.parse(request); |
| context.setRequestCacheControl(requestCacheControl); |
| } else { |
| requestCacheControl = context.getRequestCacheControl(); |
| CacheControlHeaderGenerator.INSTANCE.generate(requestCacheControl, request); |
| } |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} request cache control: {}", exchangeId, requestCacheControl); |
| } |
| |
| if (cacheableRequestPolicy.canBeServedFromCache(requestCacheControl, request)) { |
| operation.setDependency(responseCache.match(target, request, new FutureCallback<CacheMatch>() { |
| |
| @Override |
| public void completed(final CacheMatch result) { |
| final CacheHit hit = result != null ? result.hit : null; |
| final CacheHit root = result != null ? result.root : null; |
| if (hit == null) { |
| handleCacheMiss(requestCacheControl, root, target, request, entityProducer, scope, chain, asyncExecCallback); |
| } else { |
| final ResponseCacheControl responseCacheControl = CacheControlHeaderParser.INSTANCE.parse(hit.entry); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} response cache control: {}", exchangeId, responseCacheControl); |
| } |
| context.setResponseCacheControl(responseCacheControl); |
| handleCacheHit(requestCacheControl, responseCacheControl, hit, target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| asyncExecCallback.failed(cause); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| })); |
| |
| } else { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} request cannot be served from cache", exchangeId); |
| } |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| } |
| |
| void chainProceed( |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) { |
| try { |
| chain.proceed(request, entityProducer, scope, asyncExecCallback); |
| } catch (final HttpException | IOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } |
| |
| void callBackend( |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) { |
| final String exchangeId = scope.exchangeId; |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} calling the backend", exchangeId); |
| } |
| final Instant requestDate = getCurrentDate(); |
| final AtomicReference<AsyncExecCallback> callbackRef = new AtomicReference<>(); |
| chainProceed(request, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse backendResponse, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| final Instant responseDate = getCurrentDate(); |
| final AsyncExecCallback callback = new BackendResponseHandler(target, request, requestDate, responseDate, scope, asyncExecCallback); |
| callbackRef.set(callback); |
| return callback.handleResponse(backendResponse, entityDetails); |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| final AsyncExecCallback callback = callbackRef.getAndSet(null); |
| if (callback != null) { |
| callback.handleInformationResponse(response); |
| } else { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| } |
| |
| @Override |
| public void completed() { |
| final AsyncExecCallback callback = callbackRef.getAndSet(null); |
| if (callback != null) { |
| callback.completed(); |
| } else { |
| asyncExecCallback.completed(); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| final AsyncExecCallback callback = callbackRef.getAndSet(null); |
| if (callback != null) { |
| callback.failed(cause); |
| } else { |
| asyncExecCallback.failed(cause); |
| } |
| } |
| |
| }); |
| } |
| |
| class CachingAsyncDataConsumer implements AsyncDataConsumer { |
| |
| private final String exchangeId; |
| private final AsyncExecCallback fallback; |
| private final HttpResponse backendResponse; |
| private final EntityDetails entityDetails; |
| private final AtomicBoolean writtenThrough; |
| private final AtomicReference<ByteArrayBuffer> bufferRef; |
| private final AtomicReference<AsyncDataConsumer> dataConsumerRef; |
| |
| CachingAsyncDataConsumer( |
| final String exchangeId, |
| final AsyncExecCallback fallback, |
| final HttpResponse backendResponse, |
| final EntityDetails entityDetails) { |
| this.exchangeId = exchangeId; |
| this.fallback = fallback; |
| this.backendResponse = backendResponse; |
| this.entityDetails = entityDetails; |
| this.writtenThrough = new AtomicBoolean(false); |
| this.bufferRef = new AtomicReference<>(entityDetails != null ? new ByteArrayBuffer(1024) : null); |
| this.dataConsumerRef = new AtomicReference<>(); |
| } |
| |
| @Override |
| public final void updateCapacity(final CapacityChannel capacityChannel) throws IOException { |
| final AsyncDataConsumer dataConsumer = dataConsumerRef.get(); |
| if (dataConsumer != null) { |
| dataConsumer.updateCapacity(capacityChannel); |
| } else { |
| capacityChannel.update(Integer.MAX_VALUE); |
| } |
| } |
| |
| @Override |
| public final void consume(final ByteBuffer src) throws IOException { |
| final ByteArrayBuffer buffer = bufferRef.get(); |
| if (buffer != null) { |
| if (src.hasArray()) { |
| buffer.append(src.array(), src.arrayOffset() + src.position(), src.remaining()); |
| } else { |
| while (src.hasRemaining()) { |
| buffer.append(src.get()); |
| } |
| } |
| if (buffer.length() > cacheConfig.getMaxObjectSize()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} backend response content length exceeds maximum", exchangeId); |
| } |
| // Over the max limit. Stop buffering and forward the response |
| // along with all the data buffered so far to the caller. |
| bufferRef.set(null); |
| try { |
| final AsyncDataConsumer dataConsumer = fallback.handleResponse(backendResponse, entityDetails); |
| if (dataConsumer != null) { |
| dataConsumerRef.set(dataConsumer); |
| writtenThrough.set(true); |
| dataConsumer.consume(ByteBuffer.wrap(buffer.array(), 0, buffer.length())); |
| } |
| } catch (final HttpException ex) { |
| fallback.failed(ex); |
| } |
| } |
| } else { |
| final AsyncDataConsumer dataConsumer = dataConsumerRef.get(); |
| if (dataConsumer != null) { |
| dataConsumer.consume(src); |
| } |
| } |
| } |
| |
| @Override |
| public final void streamEnd(final List<? extends Header> trailers) throws HttpException, IOException { |
| final AsyncDataConsumer dataConsumer = dataConsumerRef.getAndSet(null); |
| if (dataConsumer != null) { |
| dataConsumer.streamEnd(trailers); |
| } |
| } |
| |
| @Override |
| public void releaseResources() { |
| final AsyncDataConsumer dataConsumer = dataConsumerRef.getAndSet(null); |
| if (dataConsumer != null) { |
| dataConsumer.releaseResources(); |
| } |
| } |
| |
| } |
| |
| class BackendResponseHandler implements AsyncExecCallback { |
| |
| private final HttpHost target; |
| private final HttpRequest request; |
| private final Instant requestDate; |
| private final Instant responseDate; |
| private final AsyncExecChain.Scope scope; |
| private final AsyncExecCallback asyncExecCallback; |
| private final AtomicReference<CachingAsyncDataConsumer> cachingConsumerRef; |
| |
| BackendResponseHandler( |
| final HttpHost target, |
| final HttpRequest request, |
| final Instant requestDate, |
| final Instant responseDate, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecCallback asyncExecCallback) { |
| this.target = target; |
| this.request = request; |
| this.requestDate = requestDate; |
| this.responseDate = responseDate; |
| this.scope = scope; |
| this.asyncExecCallback = asyncExecCallback; |
| this.cachingConsumerRef = new AtomicReference<>(); |
| } |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse backendResponse, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| final String exchangeId = scope.exchangeId; |
| responseCache.evictInvalidatedEntries(target, request, backendResponse, new FutureCallback<Boolean>() { |
| |
| @Override |
| public void completed(final Boolean result) { |
| } |
| |
| @Override |
| public void failed(final Exception ex) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} unable to flush invalidated entries from cache", exchangeId, ex); |
| } |
| } |
| |
| @Override |
| public void cancelled() { |
| } |
| |
| }); |
| if (isResponseTooBig(entityDetails)) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} backend response is known to be too big", exchangeId); |
| } |
| return asyncExecCallback.handleResponse(backendResponse, entityDetails); |
| } |
| |
| final ResponseCacheControl responseCacheControl = CacheControlHeaderParser.INSTANCE.parse(backendResponse); |
| final boolean cacheable = responseCachingPolicy.isResponseCacheable(responseCacheControl, request, backendResponse); |
| if (cacheable) { |
| storeRequestIfModifiedSinceFor304Response(request, backendResponse); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} caching backend response", exchangeId); |
| } |
| final CachingAsyncDataConsumer cachingDataConsumer = new CachingAsyncDataConsumer( |
| exchangeId, asyncExecCallback, backendResponse, entityDetails); |
| cachingConsumerRef.set(cachingDataConsumer); |
| return cachingDataConsumer; |
| } else { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} backend response is not cacheable", exchangeId); |
| } |
| return asyncExecCallback.handleResponse(backendResponse, entityDetails); |
| } |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| |
| void triggerNewCacheEntryResponse(final HttpResponse backendResponse, final Instant responseDate, final ByteArrayBuffer buffer) { |
| final String exchangeId = scope.exchangeId; |
| final HttpClientContext context = scope.clientContext; |
| final CancellableDependency operation = scope.cancellableDependency; |
| operation.setDependency(responseCache.store( |
| target, |
| request, |
| backendResponse, |
| buffer, |
| requestDate, |
| responseDate, |
| new FutureCallback<CacheHit>() { |
| |
| @Override |
| public void completed(final CacheHit hit) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} backend response successfully cached", exchangeId); |
| } |
| try { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, hit.entry); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, hit.entry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception ex) { |
| asyncExecCallback.failed(ex); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| })); |
| |
| } |
| |
| void triggerCachedResponse(final HttpCacheEntry entry) { |
| final HttpClientContext context = scope.clientContext; |
| try { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, entry); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, entry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } |
| |
| @Override |
| public void completed() { |
| final String exchangeId = scope.exchangeId; |
| final CachingAsyncDataConsumer cachingDataConsumer = cachingConsumerRef.getAndSet(null); |
| if (cachingDataConsumer == null || cachingDataConsumer.writtenThrough.get()) { |
| asyncExecCallback.completed(); |
| return; |
| } |
| final HttpResponse backendResponse = cachingDataConsumer.backendResponse; |
| final ByteArrayBuffer buffer = cachingDataConsumer.bufferRef.getAndSet(null); |
| |
| // Handle 304 Not Modified responses |
| if (backendResponse.getCode() == HttpStatus.SC_NOT_MODIFIED) { |
| responseCache.match(target, request, new FutureCallback<CacheMatch>() { |
| |
| @Override |
| public void completed(final CacheMatch result) { |
| final CacheHit hit = result != null ? result.hit : null; |
| if (hit != null) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} existing cache entry found, updating cache entry", exchangeId); |
| } |
| responseCache.update( |
| hit, |
| target, |
| request, |
| backendResponse, |
| requestDate, |
| responseDate, |
| new FutureCallback<CacheHit>() { |
| |
| @Override |
| public void completed(final CacheHit updated) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache entry updated, generating response from updated entry", exchangeId); |
| } |
| triggerCachedResponse(updated.entry); |
| } |
| @Override |
| public void failed(final Exception cause) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} request failed: {}", exchangeId, cause.getMessage()); |
| } |
| asyncExecCallback.failed(cause); |
| } |
| |
| @Override |
| public void cancelled() { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache entry updated aborted", exchangeId); |
| } |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| }); |
| } else { |
| triggerNewCacheEntryResponse(backendResponse, responseDate, buffer); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| asyncExecCallback.failed(cause); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| }); |
| } else { |
| if (cacheConfig.isFreshnessCheckEnabled()) { |
| final CancellableDependency operation = scope.cancellableDependency; |
| operation.setDependency(responseCache.match(target, request, new FutureCallback<CacheMatch>() { |
| |
| @Override |
| public void completed(final CacheMatch result) { |
| final CacheHit hit = result != null ? result.hit : null; |
| if (HttpCacheEntry.isNewer(hit != null ? hit.entry : null, backendResponse)) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} backend already contains fresher cache entry", exchangeId); |
| } |
| triggerCachedResponse(hit.entry); |
| } else { |
| triggerNewCacheEntryResponse(backendResponse, responseDate, buffer); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| asyncExecCallback.failed(cause); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| })); |
| } else { |
| triggerNewCacheEntryResponse(backendResponse, responseDate, buffer); |
| } |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| asyncExecCallback.failed(cause); |
| } |
| |
| } |
| |
| private void handleCacheHit( |
| final RequestCacheControl requestCacheControl, |
| final ResponseCacheControl responseCacheControl, |
| final CacheHit hit, |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) { |
| final HttpClientContext context = scope.clientContext; |
| final String exchangeId = scope.exchangeId; |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache hit: {}", exchangeId, new RequestLine(request)); |
| } |
| |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_HIT); |
| cacheHits.getAndIncrement(); |
| |
| final Instant now = getCurrentDate(); |
| |
| final CacheSuitability cacheSuitability = suitabilityChecker.assessSuitability(requestCacheControl, responseCacheControl, request, hit.entry, now); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache suitability: {}", exchangeId, cacheSuitability); |
| } |
| if (cacheSuitability == CacheSuitability.FRESH || cacheSuitability == CacheSuitability.FRESH_ENOUGH) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache hit is fresh enough", exchangeId); |
| } |
| try { |
| final SimpleHttpResponse cacheResponse = generateCachedResponse(request, hit.entry, now); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, hit.entry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| if (requestCacheControl.isOnlyIfCached()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} request marked only-if-cached", exchangeId); |
| } |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| final SimpleHttpResponse cacheResponse = generateGatewayTimeout(); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } else { |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.FAILURE); |
| try { |
| chain.proceed(request, entityProducer, scope, asyncExecCallback); |
| } catch (final HttpException | IOException ex2) { |
| asyncExecCallback.failed(ex2); |
| } |
| } |
| } |
| } else { |
| if (requestCacheControl.isOnlyIfCached()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache entry not is not fresh and only-if-cached requested", exchangeId); |
| } |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| final SimpleHttpResponse cacheResponse = generateGatewayTimeout(); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } else if (cacheSuitability == CacheSuitability.MISMATCH) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache entry does not match the request; calling backend", exchangeId); |
| } |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } else if (entityProducer != null && !entityProducer.isRepeatable()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} request is not repeatable; calling backend", exchangeId); |
| } |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } else if (hit.entry.getStatus() == HttpStatus.SC_NOT_MODIFIED && !suitabilityChecker.isConditional(request)) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} non-modified cache entry does not match the non-conditional request; calling backend", exchangeId); |
| } |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } else if (cacheSuitability == CacheSuitability.REVALIDATION_REQUIRED) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} revalidation required; revalidating cache entry", exchangeId); |
| } |
| revalidateCacheEntryWithoutFallback(responseCacheControl, hit, target, request, entityProducer, scope, chain, asyncExecCallback); |
| } else if (cacheSuitability == CacheSuitability.STALE_WHILE_REVALIDATED) { |
| if (cacheRevalidator != null) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} serving stale with asynchronous revalidation", exchangeId); |
| } |
| try { |
| final String revalidationExchangeId = ExecSupport.getNextExchangeId(); |
| context.setExchangeId(revalidationExchangeId); |
| final AsyncExecChain.Scope fork = new AsyncExecChain.Scope( |
| revalidationExchangeId, |
| scope.route, |
| scope.originalRequest, |
| new ComplexFuture<>(null), |
| HttpClientContext.create(), |
| scope.execRuntime.fork(), |
| scope.scheduler, |
| scope.execCount); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} starting asynchronous revalidation exchange {}", exchangeId, revalidationExchangeId); |
| } |
| cacheRevalidator.revalidateCacheEntry( |
| hit.getEntryKey(), |
| asyncExecCallback, |
| c -> revalidateCacheEntry(responseCacheControl, hit, target, request, entityProducer, fork, chain, c)); |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, hit.entry); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, hit.entry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final IOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } else { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} revalidating stale cache entry (asynchronous revalidation disabled)", exchangeId); |
| } |
| revalidateCacheEntryWithFallback(requestCacheControl, responseCacheControl, hit, target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| } else if (cacheSuitability == CacheSuitability.STALE) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} revalidating stale cache entry", exchangeId); |
| } |
| revalidateCacheEntryWithFallback(requestCacheControl, responseCacheControl, hit, target, request, entityProducer, scope, chain, asyncExecCallback); |
| } else { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache entry not usable; calling backend", exchangeId); |
| } |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| } |
| } |
| |
| void revalidateCacheEntry( |
| final ResponseCacheControl responseCacheControl, |
| final CacheHit hit, |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) { |
| final Instant requestDate = getCurrentDate(); |
| final HttpRequest conditionalRequest = conditionalRequestBuilder.buildConditionalRequest( |
| responseCacheControl, |
| BasicRequestBuilder.copy(request).build(), |
| hit.entry); |
| final HttpClientContext context = scope.clientContext; |
| chainProceed(conditionalRequest, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| final AtomicReference<AsyncExecCallback> callbackRef = new AtomicReference<>(); |
| |
| void triggerUpdatedCacheEntryResponse(final HttpResponse backendResponse, final Instant responseDate) { |
| final CancellableDependency operation = scope.cancellableDependency; |
| operation.setDependency(responseCache.update( |
| hit, |
| target, |
| request, |
| backendResponse, |
| requestDate, |
| responseDate, |
| new FutureCallback<CacheHit>() { |
| |
| @Override |
| public void completed(final CacheHit updated) { |
| try { |
| final SimpleHttpResponse cacheResponse = generateCachedResponse(request, updated.entry, responseDate); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, hit.entry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception ex) { |
| asyncExecCallback.failed(ex); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| })); |
| } |
| |
| AsyncExecCallback evaluateResponse(final HttpResponse backendResponse, final Instant responseDate) { |
| final int statusCode = backendResponse.getCode(); |
| if (statusCode == HttpStatus.SC_NOT_MODIFIED || statusCode == HttpStatus.SC_OK) { |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.VALIDATED); |
| cacheUpdates.getAndIncrement(); |
| } |
| if (statusCode == HttpStatus.SC_NOT_MODIFIED) { |
| return new AsyncExecCallbackWrapper(() -> triggerUpdatedCacheEntryResponse(backendResponse, responseDate), asyncExecCallback::failed); |
| } |
| return new BackendResponseHandler(target, conditionalRequest, requestDate, responseDate, scope, asyncExecCallback); |
| } |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse backendResponse1, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| |
| final Instant responseDate = getCurrentDate(); |
| |
| final AsyncExecCallback callback1; |
| if (HttpCacheEntry.isNewer(hit.entry, backendResponse1)) { |
| |
| final HttpRequest unconditional = conditionalRequestBuilder.buildUnconditionalRequest( |
| BasicRequestBuilder.copy(scope.originalRequest).build()); |
| |
| callback1 = new AsyncExecCallbackWrapper(() -> chainProceed(unconditional, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse backendResponse2, |
| final EntityDetails entityDetails1) throws HttpException, IOException { |
| final Instant responseDate2 = getCurrentDate(); |
| final AsyncExecCallback callback2 = evaluateResponse(backendResponse2, responseDate2); |
| callbackRef.set(callback2); |
| return callback2.handleResponse(backendResponse2, entityDetails1); |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| final AsyncExecCallback callback2 = callbackRef.getAndSet(null); |
| if (callback2 != null) { |
| callback2.handleInformationResponse(response); |
| } else { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| } |
| |
| @Override |
| public void completed() { |
| final AsyncExecCallback callback2 = callbackRef.getAndSet(null); |
| if (callback2 != null) { |
| callback2.completed(); |
| } else { |
| asyncExecCallback.completed(); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| final AsyncExecCallback callback2 = callbackRef.getAndSet(null); |
| if (callback2 != null) { |
| callback2.failed(cause); |
| } else { |
| asyncExecCallback.failed(cause); |
| } |
| } |
| |
| }), asyncExecCallback::failed); |
| } else { |
| callback1 = evaluateResponse(backendResponse1, responseDate); |
| } |
| callbackRef.set(callback1); |
| return callback1.handleResponse(backendResponse1, entityDetails); |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| final AsyncExecCallback callback1 = callbackRef.getAndSet(null); |
| if (callback1 != null) { |
| callback1.handleInformationResponse(response); |
| } else { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| } |
| |
| @Override |
| public void completed() { |
| final AsyncExecCallback callback1 = callbackRef.getAndSet(null); |
| if (callback1 != null) { |
| callback1.completed(); |
| } else { |
| asyncExecCallback.completed(); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| final AsyncExecCallback callback1 = callbackRef.getAndSet(null); |
| if (callback1 != null) { |
| callback1.failed(cause); |
| } else { |
| asyncExecCallback.failed(cause); |
| } |
| } |
| |
| }); |
| |
| } |
| |
| void revalidateCacheEntryWithoutFallback( |
| final ResponseCacheControl responseCacheControl, |
| final CacheHit hit, |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) { |
| final String exchangeId = scope.exchangeId; |
| final HttpClientContext context = scope.clientContext; |
| revalidateCacheEntry(responseCacheControl, hit, target, request, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| private final AtomicBoolean committed = new AtomicBoolean(); |
| |
| @Override |
| public AsyncDataConsumer handleResponse(final HttpResponse response, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| committed.set(true); |
| return asyncExecCallback.handleResponse(response, entityDetails); |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| |
| @Override |
| public void completed() { |
| asyncExecCallback.completed(); |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| if (!committed.get() && cause instanceof IOException) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} I/O error while revalidating cache entry", exchangeId, cause); |
| } |
| final SimpleHttpResponse cacheResponse = generateGatewayTimeout(); |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } else { |
| asyncExecCallback.failed(cause); |
| } |
| } |
| |
| }); |
| } |
| |
| void revalidateCacheEntryWithFallback( |
| final RequestCacheControl requestCacheControl, |
| final ResponseCacheControl responseCacheControl, |
| final CacheHit hit, |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) { |
| final String exchangeId = scope.exchangeId; |
| final HttpClientContext context = scope.clientContext; |
| revalidateCacheEntry(responseCacheControl, hit, target, request, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| private final AtomicReference<HttpResponse> committed = new AtomicReference<>(); |
| |
| @Override |
| public AsyncDataConsumer handleResponse(final HttpResponse response, final EntityDetails entityDetails) throws HttpException, IOException { |
| final int status = response.getCode(); |
| if (staleIfErrorAppliesTo(status) && |
| suitabilityChecker.isSuitableIfError(requestCacheControl, responseCacheControl, hit.entry, getCurrentDate())) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} serving stale response due to {} status and stale-if-error enabled", exchangeId, status); |
| } |
| return null; |
| } else { |
| committed.set(response); |
| return asyncExecCallback.handleResponse(response, entityDetails); |
| } |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| |
| @Override |
| public void completed() { |
| final HttpResponse response = committed.get(); |
| if (response == null) { |
| try { |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, hit.entry); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, hit.entry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final IOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } else { |
| asyncExecCallback.completed(); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| final HttpResponse response = committed.get(); |
| if (response == null) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} I/O error while revalidating cache entry", exchangeId, cause); |
| } |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| if (cause instanceof IOException && |
| suitabilityChecker.isSuitableIfError(requestCacheControl, responseCacheControl, hit.entry, getCurrentDate())) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} serving stale response due to IOException and stale-if-error enabled", exchangeId); |
| } |
| try { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, hit.entry); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, hit.entry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final IOException ex) { |
| asyncExecCallback.failed(cause); |
| } |
| } else { |
| final SimpleHttpResponse cacheResponse = generateGatewayTimeout(); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } |
| } else { |
| asyncExecCallback.failed(cause); |
| } |
| } |
| |
| }); |
| } |
| private void handleCacheMiss( |
| final RequestCacheControl requestCacheControl, |
| final CacheHit partialMatch, |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) { |
| final String exchangeId = scope.exchangeId; |
| |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} cache miss: {}", exchangeId, new RequestLine(request)); |
| } |
| cacheMisses.getAndIncrement(); |
| |
| final CancellableDependency operation = scope.cancellableDependency; |
| if (requestCacheControl.isOnlyIfCached()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} request marked only-if-cached", exchangeId); |
| } |
| final HttpClientContext context = scope.clientContext; |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| final SimpleHttpResponse cacheResponse = generateGatewayTimeout(); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } |
| |
| if (partialMatch != null && partialMatch.entry.hasVariants() && entityProducer == null) { |
| operation.setDependency(responseCache.getVariants( |
| partialMatch, |
| new FutureCallback<Collection<CacheHit>>() { |
| |
| @Override |
| public void completed(final Collection<CacheHit> variants) { |
| if (variants != null && !variants.isEmpty()) { |
| negotiateResponseFromVariants(target, request, entityProducer, scope, chain, asyncExecCallback, variants); |
| } else { |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception ex) { |
| asyncExecCallback.failed(ex); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| })); |
| } else { |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| } |
| |
| void negotiateResponseFromVariants( |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback, |
| final Collection<CacheHit> variants) { |
| final String exchangeId = scope.exchangeId; |
| final CancellableDependency operation = scope.cancellableDependency; |
| final Map<ETag, CacheHit> variantMap = new HashMap<>(); |
| for (final CacheHit variant : variants) { |
| final ETag eTag = variant.entry.getETag(); |
| if (eTag != null) { |
| variantMap.put(eTag, variant); |
| } |
| } |
| |
| final HttpRequest conditionalRequest = conditionalRequestBuilder.buildConditionalRequestFromVariants( |
| request, |
| variantMap.keySet()); |
| |
| final Instant requestDate = getCurrentDate(); |
| chainProceed(conditionalRequest, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| final AtomicReference<AsyncExecCallback> callbackRef = new AtomicReference<>(); |
| |
| void updateVariantCacheEntry(final HttpResponse backendResponse, final Instant responseDate, final CacheHit match) { |
| final HttpClientContext context = scope.clientContext; |
| context.setAttribute(HttpCacheContext.CACHE_RESPONSE_STATUS, CacheResponseStatus.VALIDATED); |
| cacheUpdates.getAndIncrement(); |
| |
| operation.setDependency(responseCache.storeFromNegotiated( |
| match, |
| target, |
| request, |
| backendResponse, |
| requestDate, |
| responseDate, |
| new FutureCallback<CacheHit>() { |
| |
| @Override |
| public void completed(final CacheHit hit) { |
| try { |
| final SimpleHttpResponse cacheResponse = generateCachedResponse(request, hit.entry, responseDate); |
| context.setAttribute(HttpCacheContext.CACHE_ENTRY, hit.entry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception ex) { |
| asyncExecCallback.failed(ex); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| })); |
| } |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse backendResponse, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| final HttpClientContext context = scope.clientContext; |
| final Instant responseDate = getCurrentDate(); |
| final AsyncExecCallback callback; |
| if (backendResponse.getCode() != HttpStatus.SC_NOT_MODIFIED) { |
| callback = new BackendResponseHandler(target, request, requestDate, responseDate, scope, asyncExecCallback); |
| } else { |
| final ETag resultEtag = ETag.get(backendResponse); |
| if (resultEtag == null) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} 304 response did not contain ETag", exchangeId); |
| } |
| callback = new AsyncExecCallbackWrapper(() -> callBackend(target, request, entityProducer, scope, chain, asyncExecCallback), asyncExecCallback::failed); |
| } else { |
| final CacheHit match = variantMap.get(resultEtag); |
| if (match == null) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("{} 304 response did not contain ETag matching one sent in If-None-Match", exchangeId); |
| } |
| callback = new AsyncExecCallbackWrapper(() -> callBackend(target, request, entityProducer, scope, chain, asyncExecCallback), asyncExecCallback::failed); |
| } else { |
| if (HttpCacheEntry.isNewer(match.entry, backendResponse)) { |
| final HttpRequest unconditional = conditionalRequestBuilder.buildUnconditionalRequest( |
| BasicRequestBuilder.copy(request).build()); |
| callback = new AsyncExecCallbackWrapper(() -> callBackend(target, unconditional, entityProducer, scope, chain, asyncExecCallback), asyncExecCallback::failed); |
| } else { |
| callback = new AsyncExecCallbackWrapper(() -> updateVariantCacheEntry(backendResponse, responseDate, match), asyncExecCallback::failed); |
| } |
| } |
| } |
| } |
| callbackRef.set(callback); |
| return callback.handleResponse(backendResponse, entityDetails); |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| final AsyncExecCallback callback = callbackRef.getAndSet(null); |
| if (callback != null) { |
| callback.handleInformationResponse(response); |
| } else { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| } |
| |
| @Override |
| public void completed() { |
| final AsyncExecCallback callback = callbackRef.getAndSet(null); |
| if (callback != null) { |
| callback.completed(); |
| } else { |
| asyncExecCallback.completed(); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| final AsyncExecCallback callback = callbackRef.getAndSet(null); |
| if (callback != null) { |
| callback.failed(cause); |
| } else { |
| asyncExecCallback.failed(cause); |
| } |
| } |
| |
| }); |
| |
| } |
| |
| } |
