| /* |
| * ==================================================================== |
| * 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.util.Date; |
| 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 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.HeaderConstants; |
| import org.apache.hc.client5.http.cache.HttpAsyncCacheStorage; |
| import org.apache.hc.client5.http.cache.HttpCacheEntry; |
| import org.apache.hc.client5.http.cache.ResourceFactory; |
| import org.apache.hc.client5.http.cache.ResourceIOException; |
| import org.apache.hc.client5.http.impl.ExecSupport; |
| import org.apache.hc.client5.http.impl.RequestCopier; |
| import org.apache.hc.client5.http.protocol.HttpClientContext; |
| import org.apache.hc.client5.http.schedule.SchedulingStrategy; |
| import org.apache.hc.client5.http.utils.DateUtils; |
| 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.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.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<>(RequestCopier.INSTANCE); |
| } |
| |
| AsyncCachingExec( |
| final HttpAsyncCache responseCache, |
| final CacheValidityPolicy validityPolicy, |
| final ResponseCachingPolicy responseCachingPolicy, |
| final CachedHttpResponseGenerator responseGenerator, |
| final CacheableRequestPolicy cacheableRequestPolicy, |
| final CachedResponseSuitabilityChecker suitabilityChecker, |
| final ResponseProtocolCompliance responseCompliance, |
| final RequestProtocolCompliance requestCompliance, |
| final DefaultAsyncCacheRevalidator cacheRevalidator, |
| final ConditionalRequestBuilder<HttpRequest> conditionalRequestBuilder, |
| final CacheConfig config) { |
| super(validityPolicy, responseCachingPolicy, responseGenerator, cacheableRequestPolicy, |
| suitabilityChecker, responseCompliance, requestCompliance, config); |
| this.responseCache = responseCache; |
| this.cacheRevalidator = cacheRevalidator; |
| this.conditionalRequestBuilder = conditionalRequestBuilder; |
| } |
| |
| AsyncCachingExec( |
| final HttpAsyncCache cache, |
| final ScheduledExecutorService executorService, |
| final SchedulingStrategy schedulingStrategy, |
| final CacheConfig config) { |
| this(cache, |
| executorService != null ? new DefaultAsyncCacheRevalidator(executorService, schedulingStrategy) : null, |
| config); |
| } |
| |
| AsyncCachingExec( |
| final ResourceFactory resourceFactory, |
| final HttpAsyncCacheStorage storage, |
| final ScheduledExecutorService executorService, |
| final SchedulingStrategy schedulingStrategy, |
| final CacheConfig config) { |
| this(new BasicHttpAsyncCache(resourceFactory, storage), executorService, schedulingStrategy, config); |
| } |
| |
| private void triggerResponse( |
| final SimpleHttpResponse cacheResponse, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecCallback asyncExecCallback) { |
| scope.clientContext.setAttribute(HttpCoreContext.HTTP_RESPONSE, cacheResponse); |
| 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 AsyncExecCallback asyncExecCallback; |
| private final Runnable command; |
| |
| AsyncExecCallbackWrapper(final AsyncExecCallback asyncExecCallback, final Runnable command) { |
| this.asyncExecCallback = asyncExecCallback; |
| this.command = command; |
| } |
| |
| @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) { |
| asyncExecCallback.failed(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 CancellableDependency operation = scope.cancellableDependency; |
| final HttpClientContext context = scope.clientContext; |
| context.setAttribute(HttpClientContext.HTTP_ROUTE, route); |
| context.setAttribute(HttpClientContext.HTTP_REQUEST, request); |
| |
| final URIAuthority authority = request.getAuthority(); |
| final String scheme = request.getScheme(); |
| final HttpHost target = authority != null ? new HttpHost(scheme, authority) : route.getTargetHost(); |
| final String via = generateViaHeader(request); |
| |
| // default response context |
| setResponseStatus(context, CacheResponseStatus.CACHE_MISS); |
| |
| if (clientRequestsOurOptions(request)) { |
| setResponseStatus(context, CacheResponseStatus.CACHE_MODULE_RESPONSE); |
| triggerResponse(SimpleHttpResponse.create(HttpStatus.SC_NOT_IMPLEMENTED), scope, asyncExecCallback); |
| return; |
| } |
| |
| final SimpleHttpResponse fatalErrorResponse = getFatallyNoncompliantResponse(request, context); |
| if (fatalErrorResponse != null) { |
| triggerResponse(fatalErrorResponse, scope, asyncExecCallback); |
| return; |
| } |
| |
| requestCompliance.makeRequestCompliant(request); |
| request.addHeader("Via",via); |
| |
| if (!cacheableRequestPolicy.isServableFromCache(request)) { |
| LOG.debug("Request is not servable from cache"); |
| operation.setDependency(responseCache.flushCacheEntriesInvalidatedByRequest(target, request, new FutureCallback<Boolean>() { |
| |
| @Override |
| public void completed(final Boolean result) { |
| callBackend(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 { |
| operation.setDependency(responseCache.getCacheEntry(target, request, new FutureCallback<HttpCacheEntry>() { |
| |
| @Override |
| public void completed(final HttpCacheEntry entry) { |
| if (entry == null) { |
| LOG.debug("Cache miss"); |
| handleCacheMiss(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } else { |
| handleCacheHit(target, request, entityProducer, scope, chain, asyncExecCallback, entry); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| asyncExecCallback.failed(cause); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| })); |
| |
| } |
| } |
| |
| 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) { |
| LOG.debug("Calling the backend"); |
| final Date 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 Date responseDate = getCurrentDate(); |
| backendResponse.addHeader("Via", generateViaHeader(backendResponse)); |
| |
| 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 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 AsyncExecCallback fallback, |
| final HttpResponse backendResponse, |
| final EntityDetails entityDetails) { |
| 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()) { |
| LOG.debug("Backend response content length exceeds maximum"); |
| // 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 Date requestDate; |
| private final Date responseDate; |
| private final AsyncExecChain.Scope scope; |
| private final AsyncExecCallback asyncExecCallback; |
| private final AtomicReference<CachingAsyncDataConsumer> cachingConsumerRef; |
| |
| BackendResponseHandler( |
| final HttpHost target, |
| final HttpRequest request, |
| final Date requestDate, |
| final Date 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 { |
| responseCompliance.ensureProtocolCompliance(scope.originalRequest, request, backendResponse); |
| responseCache.flushCacheEntriesInvalidatedByExchange(target, request, backendResponse, new FutureCallback<Boolean>() { |
| |
| @Override |
| public void completed(final Boolean result) { |
| } |
| |
| @Override |
| public void failed(final Exception ex) { |
| LOG.warn("Unable to flush invalidated entries from cache", ex); |
| } |
| |
| @Override |
| public void cancelled() { |
| } |
| |
| }); |
| final boolean cacheable = responseCachingPolicy.isResponseCacheable(request, backendResponse); |
| if (cacheable) { |
| cachingConsumerRef.set(new CachingAsyncDataConsumer(asyncExecCallback, backendResponse, entityDetails)); |
| storeRequestIfModifiedSinceFor304Response(request, backendResponse); |
| } else { |
| LOG.debug("Backend response is not cacheable"); |
| responseCache.flushCacheEntriesFor(target, request, new FutureCallback<Boolean>() { |
| |
| @Override |
| public void completed(final Boolean result) { |
| } |
| |
| @Override |
| public void failed(final Exception ex) { |
| LOG.warn("Unable to flush invalidated entries from cache", ex); |
| } |
| |
| @Override |
| public void cancelled() { |
| } |
| |
| }); |
| } |
| final CachingAsyncDataConsumer cachingDataConsumer = cachingConsumerRef.get(); |
| if (cachingDataConsumer != null) { |
| LOG.debug("Caching backend response"); |
| return cachingDataConsumer; |
| } |
| return asyncExecCallback.handleResponse(backendResponse, entityDetails); |
| } |
| |
| @Override |
| public void handleInformationResponse(final HttpResponse response) throws HttpException, IOException { |
| asyncExecCallback.handleInformationResponse(response); |
| } |
| |
| void triggerNewCacheEntryResponse(final HttpResponse backendResponse, final Date responseDate, final ByteArrayBuffer buffer) { |
| final CancellableDependency operation = scope.cancellableDependency; |
| operation.setDependency(responseCache.createCacheEntry( |
| target, |
| request, |
| backendResponse, |
| buffer, |
| requestDate, |
| responseDate, |
| new FutureCallback<HttpCacheEntry>() { |
| |
| @Override |
| public void completed(final HttpCacheEntry newEntry) { |
| LOG.debug("Backend response successfully cached"); |
| try { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, newEntry); |
| 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 void completed() { |
| final CachingAsyncDataConsumer cachingDataConsumer = cachingConsumerRef.getAndSet(null); |
| if (cachingDataConsumer != null && !cachingDataConsumer.writtenThrough.get()) { |
| final ByteArrayBuffer buffer = cachingDataConsumer.bufferRef.getAndSet(null); |
| final HttpResponse backendResponse = cachingDataConsumer.backendResponse; |
| if (cacheConfig.isFreshnessCheckEnabled()) { |
| final CancellableDependency operation = scope.cancellableDependency; |
| operation.setDependency(responseCache.getCacheEntry(target, request, new FutureCallback<HttpCacheEntry>() { |
| |
| @Override |
| public void completed(final HttpCacheEntry existingEntry) { |
| if (DateUtils.isAfter(existingEntry, backendResponse, HttpHeaders.DATE)) { |
| LOG.debug("Backend already contains fresher cache entry"); |
| try { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, existingEntry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } 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); |
| } |
| } else { |
| asyncExecCallback.completed(); |
| } |
| } |
| |
| @Override |
| public void failed(final Exception cause) { |
| asyncExecCallback.failed(cause); |
| } |
| |
| } |
| |
| private void handleCacheHit( |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback, |
| final HttpCacheEntry entry) { |
| final HttpClientContext context = scope.clientContext; |
| recordCacheHit(target, request); |
| final Date now = getCurrentDate(); |
| if (suitabilityChecker.canCachedResponseBeUsed(target, request, entry, now)) { |
| LOG.debug("Cache hit"); |
| try { |
| final SimpleHttpResponse cacheResponse = generateCachedResponse(request, context, entry, now); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| recordCacheFailure(target, request); |
| if (!mayCallBackend(request)) { |
| final SimpleHttpResponse cacheResponse = generateGatewayTimeout(context); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } else { |
| setResponseStatus(scope.clientContext, CacheResponseStatus.FAILURE); |
| try { |
| chain.proceed(request, entityProducer, scope, asyncExecCallback); |
| } catch (final HttpException | IOException ex2) { |
| asyncExecCallback.failed(ex2); |
| } |
| } |
| } |
| } else if (!mayCallBackend(request)) { |
| LOG.debug("Cache entry not suitable but only-if-cached requested"); |
| final SimpleHttpResponse cacheResponse = generateGatewayTimeout(context); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } else if (!(entry.getStatus() == HttpStatus.SC_NOT_MODIFIED && !suitabilityChecker.isConditional(request))) { |
| LOG.debug("Revalidating cache entry"); |
| if (cacheRevalidator != null |
| && !staleResponseNotAllowed(request, entry, now) |
| && validityPolicy.mayReturnStaleWhileRevalidating(entry, now)) { |
| LOG.debug("Serving stale with asynchronous revalidation"); |
| try { |
| final SimpleHttpResponse cacheResponse = generateCachedResponse(request, context, entry, now); |
| final String exchangeId = ExecSupport.getNextExchangeId(); |
| final AsyncExecChain.Scope fork = new AsyncExecChain.Scope( |
| exchangeId, |
| scope.route, |
| scope.originalRequest, |
| new ComplexFuture<>(null), |
| HttpClientContext.create(), |
| scope.execRuntime.fork()); |
| cacheRevalidator.revalidateCacheEntry( |
| responseCache.generateKey(target, request, entry), |
| asyncExecCallback, |
| new DefaultAsyncCacheRevalidator.RevalidationCall() { |
| |
| @Override |
| public void execute(final AsyncExecCallback asyncExecCallback) { |
| revalidateCacheEntry(target, request, entityProducer, fork, chain, asyncExecCallback, entry); |
| } |
| |
| }); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } else { |
| revalidateCacheEntry(target, request, entityProducer, scope, chain, asyncExecCallback, entry); |
| } |
| } else { |
| LOG.debug("Cache entry not usable; calling backend"); |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| } |
| |
| void revalidateCacheEntry( |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback, |
| final HttpCacheEntry cacheEntry) { |
| final Date requestDate = getCurrentDate(); |
| final HttpRequest conditionalRequest = conditionalRequestBuilder.buildConditionalRequest(scope.originalRequest, cacheEntry); |
| chainProceed(conditionalRequest, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| final AtomicReference<AsyncExecCallback> callbackRef = new AtomicReference<>(); |
| |
| void triggerUpdatedCacheEntryResponse(final HttpResponse backendResponse, final Date responseDate) { |
| final CancellableDependency operation = scope.cancellableDependency; |
| recordCacheUpdate(scope.clientContext); |
| operation.setDependency(responseCache.updateCacheEntry( |
| target, |
| request, |
| cacheEntry, |
| backendResponse, |
| requestDate, |
| responseDate, |
| new FutureCallback<HttpCacheEntry>() { |
| |
| @Override |
| public void completed(final HttpCacheEntry updatedEntry) { |
| if (suitabilityChecker.isConditional(request) |
| && suitabilityChecker.allConditionalsMatch(request, updatedEntry, new Date())) { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateNotModifiedResponse(updatedEntry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } else { |
| try { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, updatedEntry); |
| 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 triggerResponseStaleCacheEntry() { |
| try { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, cacheEntry); |
| cacheResponse.addHeader(HeaderConstants.WARNING, "110 localhost \"Response is stale\""); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } catch (final ResourceIOException ex) { |
| asyncExecCallback.failed(ex); |
| } |
| } |
| |
| AsyncExecCallback evaluateResponse(final HttpResponse backendResponse, final Date responseDate) { |
| backendResponse.addHeader(HeaderConstants.VIA, generateViaHeader(backendResponse)); |
| |
| final int statusCode = backendResponse.getCode(); |
| if (statusCode == HttpStatus.SC_NOT_MODIFIED || statusCode == HttpStatus.SC_OK) { |
| recordCacheUpdate(scope.clientContext); |
| } |
| if (statusCode == HttpStatus.SC_NOT_MODIFIED) { |
| return new AsyncExecCallbackWrapper(asyncExecCallback, new Runnable() { |
| |
| @Override |
| public void run() { |
| triggerUpdatedCacheEntryResponse(backendResponse, responseDate); |
| } |
| |
| }); |
| } |
| if (staleIfErrorAppliesTo(statusCode) |
| && !staleResponseNotAllowed(request, cacheEntry, getCurrentDate()) |
| && validityPolicy.mayReturnStaleIfError(request, cacheEntry, responseDate)) { |
| return new AsyncExecCallbackWrapper(asyncExecCallback, new Runnable() { |
| |
| @Override |
| public void run() { |
| triggerResponseStaleCacheEntry(); |
| } |
| |
| }); |
| } |
| return new BackendResponseHandler(target, conditionalRequest, requestDate, responseDate, scope, asyncExecCallback); |
| } |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse backendResponse1, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| |
| final Date responseDate1 = getCurrentDate(); |
| |
| final AsyncExecCallback callback1; |
| if (revalidationResponseIsTooOld(backendResponse1, cacheEntry) |
| && (entityProducer == null || entityProducer.isRepeatable())) { |
| |
| final HttpRequest unconditional = conditionalRequestBuilder.buildUnconditionalRequest( |
| scope.originalRequest); |
| |
| callback1 = new AsyncExecCallbackWrapper(asyncExecCallback, new Runnable() { |
| |
| @Override |
| public void run() { |
| chainProceed(unconditional, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| @Override |
| public AsyncDataConsumer handleResponse( |
| final HttpResponse backendResponse2, |
| final EntityDetails entityDetails) throws HttpException, IOException { |
| final Date responseDate2 = getCurrentDate(); |
| final AsyncExecCallback callback2 = evaluateResponse(backendResponse2, responseDate2); |
| callbackRef.set(callback2); |
| return callback2.handleResponse(backendResponse2, entityDetails); |
| } |
| |
| @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); |
| } |
| } |
| |
| }); |
| |
| } |
| |
| }); |
| } else { |
| callback1 = evaluateResponse(backendResponse1, responseDate1); |
| } |
| 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); |
| } |
| } |
| |
| }); |
| |
| } |
| |
| private void handleCacheMiss( |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback) { |
| recordCacheMiss(target, request); |
| |
| if (mayCallBackend(request)) { |
| final CancellableDependency operation = scope.cancellableDependency; |
| operation.setDependency(responseCache.getVariantCacheEntriesWithEtags( |
| target, |
| request, |
| new FutureCallback<Map<String, Variant>>() { |
| |
| @Override |
| public void completed(final Map<String, Variant> variants) { |
| if (variants != null && !variants.isEmpty() && (entityProducer == null || entityProducer.isRepeatable())) { |
| 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 { |
| final SimpleHttpResponse cacheResponse = SimpleHttpResponse.create(HttpStatus.SC_GATEWAY_TIMEOUT, "Gateway Timeout"); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } |
| } |
| |
| void negotiateResponseFromVariants( |
| final HttpHost target, |
| final HttpRequest request, |
| final AsyncEntityProducer entityProducer, |
| final AsyncExecChain.Scope scope, |
| final AsyncExecChain chain, |
| final AsyncExecCallback asyncExecCallback, |
| final Map<String, Variant> variants) { |
| final CancellableDependency operation = scope.cancellableDependency; |
| final HttpRequest conditionalRequest = conditionalRequestBuilder.buildConditionalRequestFromVariants(request, variants); |
| |
| final Date requestDate = getCurrentDate(); |
| chainProceed(conditionalRequest, entityProducer, scope, chain, new AsyncExecCallback() { |
| |
| final AtomicReference<AsyncExecCallback> callbackRef = new AtomicReference<>(); |
| |
| void updateVariantCacheEntry(final HttpResponse backendResponse, final Date responseDate, final Variant matchingVariant) { |
| recordCacheUpdate(scope.clientContext); |
| operation.setDependency(responseCache.updateVariantCacheEntry( |
| target, |
| conditionalRequest, |
| backendResponse, |
| matchingVariant, |
| requestDate, |
| responseDate, |
| new FutureCallback<HttpCacheEntry>() { |
| |
| @Override |
| public void completed(final HttpCacheEntry responseEntry) { |
| if (shouldSendNotModifiedResponse(request, responseEntry)) { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateNotModifiedResponse(responseEntry); |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } else { |
| try { |
| final SimpleHttpResponse cacheResponse = responseGenerator.generateResponse(request, responseEntry); |
| operation.setDependency(responseCache.reuseVariantEntryFor( |
| target, |
| request, |
| matchingVariant, |
| new FutureCallback<Boolean>() { |
| |
| @Override |
| public void completed(final Boolean result) { |
| triggerResponse(cacheResponse, scope, asyncExecCallback); |
| } |
| |
| @Override |
| public void failed(final Exception ex) { |
| asyncExecCallback.failed(ex); |
| } |
| |
| @Override |
| public void cancelled() { |
| asyncExecCallback.failed(new InterruptedIOException()); |
| } |
| |
| })); |
| } 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 Date responseDate = getCurrentDate(); |
| backendResponse.addHeader("Via", generateViaHeader(backendResponse)); |
| |
| final AsyncExecCallback callback; |
| |
| if (backendResponse.getCode() != HttpStatus.SC_NOT_MODIFIED) { |
| callback = new BackendResponseHandler(target, request, requestDate, responseDate, scope, asyncExecCallback); |
| } else { |
| final Header resultEtagHeader = backendResponse.getFirstHeader(HeaderConstants.ETAG); |
| if (resultEtagHeader == null) { |
| LOG.warn("304 response did not contain ETag"); |
| callback = new AsyncExecCallbackWrapper(asyncExecCallback, new Runnable() { |
| |
| @Override |
| public void run() { |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| |
| }); |
| } else { |
| final String resultEtag = resultEtagHeader.getValue(); |
| final Variant matchingVariant = variants.get(resultEtag); |
| if (matchingVariant == null) { |
| LOG.debug("304 response did not contain ETag matching one sent in If-None-Match"); |
| callback = new AsyncExecCallbackWrapper(asyncExecCallback, new Runnable() { |
| |
| @Override |
| public void run() { |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| |
| }); |
| } else { |
| if (revalidationResponseIsTooOld(backendResponse, matchingVariant.getEntry())) { |
| final HttpRequest unconditional = conditionalRequestBuilder.buildUnconditionalRequest(request); |
| scope.clientContext.setAttribute(HttpCoreContext.HTTP_REQUEST, unconditional); |
| callback = new AsyncExecCallbackWrapper(asyncExecCallback, new Runnable() { |
| |
| @Override |
| public void run() { |
| callBackend(target, request, entityProducer, scope, chain, asyncExecCallback); |
| } |
| |
| }); |
| } else { |
| callback = new AsyncExecCallbackWrapper(asyncExecCallback, new Runnable() { |
| |
| @Override |
| public void run() { |
| updateVariantCacheEntry(backendResponse, responseDate, matchingVariant); |
| } |
| |
| }); |
| } |
| } |
| } |
| } |
| 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); |
| } |
| } |
| |
| }); |
| |
| } |
| |
| } |
