0001: /*
0002: * Licensed to the Apache Software Foundation (ASF) under one or more
0003: * contributor license agreements. See the NOTICE file distributed with
0004: * this work for additional information regarding copyright ownership.
0005: * The ASF licenses this file to You under the Apache License, Version 2.0
0006: * (the "License"); you may not use this file except in compliance with
0007: * the License. You may obtain a copy of the License at
0008: *
0009: * http://www.apache.org/licenses/LICENSE-2.0
0010: *
0011: * Unless required by applicable law or agreed to in writing, software
0012: * distributed under the License is distributed on an "AS IS" BASIS,
0013: * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
0014: * See the License for the specific language governing permissions and
0015: * limitations under the License.
0016: */
0017: package org.apache.cocoon.components.pipeline.impl;
0018:
0019: import java.io.ByteArrayOutputStream;
0020: import java.io.IOException;
0021: import java.io.OutputStream;
0022: import java.io.Serializable;
0023: import java.util.ArrayList;
0024: import java.util.Date;
0025:
0026: import org.apache.avalon.framework.component.ComponentException;
0027: import org.apache.avalon.framework.parameters.ParameterException;
0028: import org.apache.avalon.framework.parameters.Parameters;
0029: import org.apache.cocoon.ProcessingException;
0030: import org.apache.cocoon.caching.CacheValidity;
0031: import org.apache.cocoon.caching.CacheValidityToSourceValidity;
0032: import org.apache.cocoon.caching.Cacheable;
0033: import org.apache.cocoon.caching.CacheableProcessingComponent;
0034: import org.apache.cocoon.caching.CachedResponse;
0035: import org.apache.cocoon.caching.CachingOutputStream;
0036: import org.apache.cocoon.caching.ComponentCacheKey;
0037: import org.apache.cocoon.caching.PipelineCacheKey;
0038: import org.apache.cocoon.environment.Environment;
0039: import org.apache.cocoon.transformation.Transformer;
0040: import org.apache.cocoon.util.HashUtil;
0041: import org.apache.excalibur.source.SourceValidity;
0042: import org.apache.excalibur.source.impl.validity.AggregatedValidity;
0043: import org.apache.excalibur.source.impl.validity.DeferredValidity;
0044: import org.apache.excalibur.source.impl.validity.NOPValidity;
0045: import org.apache.excalibur.store.Store;
0046:
0047: /**
0048: * This is the base class for all caching pipeline implementations
0049: * that check different pipeline components.
0050: *
0051: * @since 2.1
0052: * @author <a href="mailto:cziegeler@apache.org">Carsten Ziegeler</a>
0053: * @author <a href="mailto:Michael.Melhem@managesoft.com">Michael Melhem</a>
0054: * @version $Id: AbstractCachingProcessingPipeline.java 498470 2007-01-21 22:34:30Z anathaniel $
0055: */
0056: public abstract class AbstractCachingProcessingPipeline extends
0057: BaseCachingProcessingPipeline {
0058:
0059: public static final String PIPELOCK_PREFIX = "PIPELOCK:";
0060:
0061: /** The role name of the generator */
0062: protected String generatorRole;
0063:
0064: /** The role names of the transfomrers */
0065: protected ArrayList transformerRoles = new ArrayList();
0066:
0067: /** The role name of the serializer */
0068: protected String serializerRole;
0069:
0070: /** The role name of the reader */
0071: protected String readerRole;
0072:
0073: /** The cached response */
0074: protected CachedResponse cachedResponse;
0075:
0076: /** The index indicating the first transformer getting input from the cache */
0077: protected int firstProcessedTransformerIndex;
0078:
0079: /** Complete response is cached */
0080: protected boolean completeResponseIsCached;
0081:
0082: /** This key indicates the response that is fetched from the cache */
0083: protected PipelineCacheKey fromCacheKey;
0084:
0085: /** This key indicates the response that will get into the cache */
0086: protected PipelineCacheKey toCacheKey;
0087:
0088: /** The source validities used for caching */
0089: protected SourceValidity[] toCacheSourceValidities;
0090:
0091: /** The index indicating to the first transformer which is not cacheable */
0092: protected int firstNotCacheableTransformerIndex;
0093:
0094: /** Cache complete response */
0095: protected boolean cacheCompleteResponse;
0096:
0097: protected boolean generatorIsCacheableProcessingComponent;
0098: protected boolean serializerIsCacheableProcessingComponent;
0099: protected boolean[] transformerIsCacheableProcessingComponent;
0100:
0101: protected Store transientStore = null;
0102:
0103: /** Abstract method defined in subclasses */
0104: protected abstract void cacheResults(Environment environment,
0105: OutputStream os) throws Exception;
0106:
0107: /** Abstract method defined in subclasses */
0108: protected abstract ComponentCacheKey newComponentCacheKey(int type,
0109: String role, Serializable key);
0110:
0111: /** Abstract method defined in subclasses */
0112: protected abstract void connectCachingPipeline(
0113: Environment environment) throws ProcessingException;
0114:
0115: /**
0116: * Parameterizable Interface - Configuration
0117: */
0118: public void parameterize(Parameters params)
0119: throws ParameterException {
0120: super .parameterize(params);
0121:
0122: String storeRole = params.getParameter("store-role",
0123: Store.TRANSIENT_STORE);
0124:
0125: try {
0126: transientStore = (Store) manager.lookup(storeRole);
0127: } catch (ComponentException e) {
0128: if (getLogger().isDebugEnabled()) {
0129: getLogger()
0130: .debug(
0131: "Could not look up transient store, synchronizing requests will not work!",
0132: e);
0133: }
0134: }
0135: }
0136:
0137: /**
0138: * Set the generator.
0139: */
0140: public void setGenerator(String role, String source,
0141: Parameters param, Parameters hintParam)
0142: throws ProcessingException {
0143: super .setGenerator(role, source, param, hintParam);
0144: this .generatorRole = role;
0145: }
0146:
0147: /**
0148: * Add a transformer.
0149: */
0150: public void addTransformer(String role, String source,
0151: Parameters param, Parameters hintParam)
0152: throws ProcessingException {
0153: super .addTransformer(role, source, param, hintParam);
0154: this .transformerRoles.add(role);
0155: }
0156:
0157: /**
0158: * Set the serializer.
0159: */
0160: public void setSerializer(String role, String source,
0161: Parameters param, Parameters hintParam, String mimeType)
0162: throws ProcessingException {
0163: super .setSerializer(role, source, param, hintParam, mimeType);
0164: this .serializerRole = role;
0165: }
0166:
0167: /**
0168: * Set the Reader.
0169: */
0170: public void setReader(String role, String source, Parameters param,
0171: String mimeType) throws ProcessingException {
0172: super .setReader(role, source, param, mimeType);
0173: this .readerRole = role;
0174: }
0175:
0176: protected boolean waitForLock(Object key) {
0177: if (transientStore != null) {
0178: Object lock = null;
0179: synchronized (transientStore) {
0180: String lockKey = PIPELOCK_PREFIX + key;
0181: if (transientStore.containsKey(lockKey)) {
0182: // cache content is currently being generated, wait for other thread
0183: lock = transientStore.get(lockKey);
0184: }
0185: }
0186: // Avoid deadlock with self (see JIRA COCOON-1985).
0187: if (lock != null && lock != Thread.currentThread()) {
0188: try {
0189: // become owner of monitor
0190: synchronized (lock) {
0191: lock.wait();
0192: }
0193: } catch (InterruptedException e) {
0194: if (getLogger().isDebugEnabled()) {
0195: getLogger()
0196: .debug(
0197: "Got interrupted waiting for other pipeline to finish processing, retrying...",
0198: e);
0199: }
0200: return false;
0201: }
0202: if (getLogger().isDebugEnabled()) {
0203: getLogger()
0204: .debug(
0205: "Other pipeline finished processing, retrying to get cached response.");
0206: }
0207: return false;
0208: }
0209: }
0210: return true;
0211: }
0212:
0213: /**
0214: * makes the lock (instantiates a new object and puts it into the store)
0215: */
0216: protected boolean generateLock(Object key) {
0217: boolean succeeded = true;
0218:
0219: if (transientStore != null && key != null) {
0220: String lockKey = PIPELOCK_PREFIX + key;
0221: synchronized (transientStore) {
0222: if (transientStore.containsKey(lockKey)) {
0223: succeeded = false;
0224: if (getLogger().isDebugEnabled()) {
0225: getLogger().debug(
0226: "Lock already present in the store!");
0227: }
0228: } else {
0229: Object lock = Thread.currentThread();
0230: try {
0231: transientStore.store(lockKey, lock);
0232: } catch (IOException e) {
0233: if (getLogger().isDebugEnabled()) {
0234: getLogger().debug(
0235: "Could not put lock in the store!",
0236: e);
0237: }
0238: succeeded = false;
0239: }
0240: }
0241: }
0242: }
0243:
0244: return succeeded;
0245: }
0246:
0247: /**
0248: * releases the lock (notifies it and removes it from the store)
0249: */
0250: protected boolean releaseLock(Object key) {
0251: boolean succeeded = true;
0252:
0253: if (transientStore != null && key != null) {
0254: String lockKey = PIPELOCK_PREFIX + key;
0255: Object lock = null;
0256: synchronized (transientStore) {
0257: if (!transientStore.containsKey(lockKey)) {
0258: succeeded = false;
0259: if (getLogger().isDebugEnabled()) {
0260: getLogger().debug(
0261: "Lock not present in the store!");
0262: }
0263: } else {
0264: try {
0265: lock = transientStore.get(lockKey);
0266: transientStore.remove(lockKey);
0267: } catch (Exception e) {
0268: if (getLogger().isDebugEnabled()) {
0269: getLogger()
0270: .debug(
0271: "Could not get lock from the store!",
0272: e);
0273: }
0274: succeeded = false;
0275: }
0276: }
0277: }
0278: if (succeeded && lock != null) {
0279: // become monitor owner
0280: synchronized (lock) {
0281: lock.notifyAll();
0282: }
0283: }
0284: }
0285:
0286: return succeeded;
0287: }
0288:
0289: /**
0290: * Process the given <code>Environment</code>, producing the output.
0291: */
0292: protected boolean processXMLPipeline(Environment environment)
0293: throws ProcessingException {
0294: if (this .toCacheKey == null && this .cachedResponse == null) {
0295: return super .processXMLPipeline(environment);
0296: }
0297:
0298: if (this .cachedResponse != null
0299: && this .completeResponseIsCached) {
0300:
0301: // Allow for 304 (not modified) responses in dynamic content
0302: if (checkIfModified(environment, this .cachedResponse
0303: .getLastModified())) {
0304: return true;
0305: }
0306:
0307: // Set mime-type
0308: if (this .cachedResponse.getContentType() != null) {
0309: environment.setContentType(this .cachedResponse
0310: .getContentType());
0311: } else {
0312: setMimeTypeForSerializer(environment);
0313: }
0314:
0315: // Write response out
0316: try {
0317: final OutputStream outputStream = environment
0318: .getOutputStream(0);
0319: final byte[] content = this .cachedResponse
0320: .getResponse();
0321: if (content.length > 0) {
0322: environment.setContentLength(content.length);
0323: outputStream.write(content);
0324: }
0325: } catch (Exception e) {
0326: handleException(e);
0327: }
0328: } else {
0329: setMimeTypeForSerializer(environment);
0330: if (getLogger().isDebugEnabled() && this .toCacheKey != null) {
0331: getLogger().debug(
0332: "processXMLPipeline: caching content for further"
0333: + " requests of '"
0334: + environment.getURI() + "' using key "
0335: + this .toCacheKey);
0336: }
0337:
0338: generateLock(this .toCacheKey);
0339:
0340: try {
0341: OutputStream os = null;
0342:
0343: if (this .cacheCompleteResponse
0344: && this .toCacheKey != null) {
0345: os = new CachingOutputStream(environment
0346: .getOutputStream(this .outputBufferSize));
0347: }
0348:
0349: if (super .serializer != super .lastConsumer) {
0350: if (os == null) {
0351: os = environment
0352: .getOutputStream(this .outputBufferSize);
0353: }
0354:
0355: // internal processing
0356: if (this .xmlDeserializer != null) {
0357: this .xmlDeserializer
0358: .deserialize(this .cachedResponse
0359: .getResponse());
0360: } else {
0361: this .generator.generate();
0362: }
0363:
0364: } else {
0365: if (this .serializer.shouldSetContentLength()) {
0366: if (os == null) {
0367: os = environment.getOutputStream(0);
0368: }
0369:
0370: // Set the output stream
0371: ByteArrayOutputStream baos = new ByteArrayOutputStream();
0372: this .serializer.setOutputStream(baos);
0373:
0374: // Execute the pipeline
0375: if (this .xmlDeserializer != null) {
0376: this .xmlDeserializer
0377: .deserialize(this .cachedResponse
0378: .getResponse());
0379: } else {
0380: this .generator.generate();
0381: }
0382:
0383: environment.setContentLength(baos.size());
0384: baos.writeTo(os);
0385: } else {
0386: if (os == null) {
0387: os = environment
0388: .getOutputStream(this .outputBufferSize);
0389: }
0390:
0391: // Set the output stream
0392: this .serializer.setOutputStream(os);
0393:
0394: // Execute the pipeline
0395: if (this .xmlDeserializer != null) {
0396: this .xmlDeserializer
0397: .deserialize(this .cachedResponse
0398: .getResponse());
0399: } else {
0400: this .generator.generate();
0401: }
0402: }
0403: }
0404:
0405: //
0406: // Now that we have processed the pipeline,
0407: // we do the actual caching
0408: //
0409: cacheResults(environment, os);
0410:
0411: } catch (Exception e) {
0412: handleException(e);
0413: } finally {
0414: releaseLock(this .toCacheKey);
0415: }
0416:
0417: return true;
0418: }
0419:
0420: return true;
0421: }
0422:
0423: /**
0424: * The components of the pipeline are checked if they are Cacheable.
0425: */
0426: protected void generateCachingKey(Environment environment)
0427: throws ProcessingException {
0428:
0429: this .toCacheKey = null;
0430:
0431: this .generatorIsCacheableProcessingComponent = false;
0432: this .serializerIsCacheableProcessingComponent = false;
0433: this .transformerIsCacheableProcessingComponent = new boolean[this .transformers
0434: .size()];
0435:
0436: this .firstNotCacheableTransformerIndex = 0;
0437: this .cacheCompleteResponse = false;
0438:
0439: // first step is to generate the key:
0440: // All pipeline components starting with the generator
0441: // are tested if they are either a CacheableProcessingComponent
0442: // or Cacheable (deprecated). The returned keys are chained together
0443: // to build a unique key of the request
0444:
0445: // is the generator cacheable?
0446: Serializable key = getGeneratorKey();
0447: if (key != null) {
0448: this .toCacheKey = new PipelineCacheKey();
0449: this .toCacheKey.addKey(this .newComponentCacheKey(
0450: ComponentCacheKey.ComponentType_Generator,
0451: this .generatorRole, key));
0452:
0453: // now testing transformers
0454: final int transformerSize = super .transformers.size();
0455: boolean continueTest = true;
0456:
0457: while (this .firstNotCacheableTransformerIndex < transformerSize
0458: && continueTest) {
0459: final Transformer trans = (Transformer) super .transformers
0460: .get(this .firstNotCacheableTransformerIndex);
0461: key = getTransformerKey(trans);
0462: if (key != null) {
0463: this .toCacheKey
0464: .addKey(this
0465: .newComponentCacheKey(
0466: ComponentCacheKey.ComponentType_Transformer,
0467: (String) this .transformerRoles
0468: .get(this .firstNotCacheableTransformerIndex),
0469: key));
0470:
0471: this .firstNotCacheableTransformerIndex++;
0472: } else {
0473: continueTest = false;
0474: }
0475: }
0476: // all transformers are cacheable => pipeline is cacheable
0477: // test serializer if this is not an internal request
0478: if (this .firstNotCacheableTransformerIndex == transformerSize
0479: && super .serializer == this .lastConsumer) {
0480:
0481: key = getSerializerKey();
0482: if (key != null) {
0483: this .toCacheKey.addKey(this .newComponentCacheKey(
0484: ComponentCacheKey.ComponentType_Serializer,
0485: this .serializerRole, key));
0486: this .cacheCompleteResponse = true;
0487: }
0488: }
0489: }
0490: }
0491:
0492: /**
0493: * Generate validity objects for the new response
0494: */
0495: protected void setupValidities() throws ProcessingException {
0496:
0497: if (this .toCacheKey != null) {
0498: // only update validity objects if we cannot use
0499: // a cached response or when the cached response does
0500: // cache less than now is cacheable
0501: if (this .fromCacheKey == null
0502: || this .fromCacheKey.size() < this .toCacheKey
0503: .size()) {
0504:
0505: this .toCacheSourceValidities = new SourceValidity[this .toCacheKey
0506: .size()];
0507:
0508: int len = this .toCacheSourceValidities.length;
0509: int i = 0;
0510: while (i < len) {
0511: final SourceValidity validity = getValidityForInternalPipeline(i);
0512:
0513: if (validity == null) {
0514: if (i > 0
0515: && (this .fromCacheKey == null || i > this .fromCacheKey
0516: .size())) {
0517: // shorten key
0518: for (int m = i; m < this .toCacheSourceValidities.length; m++) {
0519: this .toCacheKey.removeLastKey();
0520: if (!this .cacheCompleteResponse) {
0521: this .firstNotCacheableTransformerIndex--;
0522: }
0523: this .cacheCompleteResponse = false;
0524: }
0525: SourceValidity[] copy = new SourceValidity[i];
0526: System.arraycopy(
0527: this .toCacheSourceValidities, 0,
0528: copy, 0, copy.length);
0529: this .toCacheSourceValidities = copy;
0530: len = this .toCacheSourceValidities.length;
0531: } else {
0532: // caching is not possible!
0533: this .toCacheKey = null;
0534: this .toCacheSourceValidities = null;
0535: this .cacheCompleteResponse = false;
0536: len = 0;
0537: }
0538: } else {
0539: this .toCacheSourceValidities[i] = validity;
0540: }
0541: i++;
0542: }
0543: } else {
0544: // we don't have to cache
0545: this .toCacheKey = null;
0546: this .cacheCompleteResponse = false;
0547: }
0548: }
0549: }
0550:
0551: /**
0552: * Calculate the key that can be used to get something from the cache, and
0553: * handle expires properly.
0554: */
0555: protected void validatePipeline(Environment environment)
0556: throws ProcessingException {
0557: this .completeResponseIsCached = this .cacheCompleteResponse;
0558: this .fromCacheKey = this .toCacheKey.copy();
0559: this .firstProcessedTransformerIndex = this .firstNotCacheableTransformerIndex;
0560:
0561: boolean finished = false;
0562: while (this .fromCacheKey != null && !finished) {
0563: finished = true;
0564:
0565: final CachedResponse response = this .cache
0566: .get(this .fromCacheKey);
0567:
0568: // now test validity
0569: if (response != null) {
0570: if (getLogger().isDebugEnabled()) {
0571: getLogger().debug(
0572: "Found cached response for '"
0573: + environment.getURI()
0574: + "' using key: "
0575: + this .fromCacheKey);
0576: }
0577:
0578: boolean responseIsValid = true;
0579: boolean responseIsUsable = true;
0580:
0581: // See if we have an explicit "expires" setting. If so,
0582: // and if it's still fresh, we're done.
0583: Long responseExpires = response.getExpires();
0584:
0585: if (responseExpires != null) {
0586: if (getLogger().isDebugEnabled()) {
0587: getLogger().debug(
0588: "Expires time found for "
0589: + environment.getURI());
0590: }
0591:
0592: if (responseExpires.longValue() > System
0593: .currentTimeMillis()) {
0594: if (getLogger().isDebugEnabled()) {
0595: getLogger()
0596: .debug(
0597: "Expires time still fresh for "
0598: + environment
0599: .getURI()
0600: + ", ignoring all other cache settings. This entry expires on "
0601: + new Date(
0602: responseExpires
0603: .longValue()));
0604: }
0605: this .cachedResponse = response;
0606: return;
0607: } else {
0608: if (getLogger().isDebugEnabled()) {
0609: getLogger()
0610: .debug(
0611: "Expires time has expired for "
0612: + environment
0613: .getURI()
0614: + ", regenerating content.");
0615: }
0616:
0617: // If an expires parameter was provided, use it. If this parameter is not available
0618: // it means that the sitemap was modified, and the old expires value is not valid
0619: // anymore.
0620: if (expires != 0) {
0621: if (this .getLogger().isDebugEnabled())
0622: this
0623: .getLogger()
0624: .debug(
0625: "Refreshing expires informations");
0626: response.setExpires(new Long(expires
0627: + System.currentTimeMillis()));
0628: } else {
0629: if (this .getLogger().isDebugEnabled())
0630: this
0631: .getLogger()
0632: .debug(
0633: "No expires defined anymore for this object, setting it to no expires");
0634: response.setExpires(null);
0635: }
0636: }
0637: } else {
0638: // The response had no expires informations. See if it needs to be set (i.e. because the configuration has changed)
0639: if (expires != 0) {
0640: if (this .getLogger().isDebugEnabled())
0641: this
0642: .getLogger()
0643: .debug(
0644: "Setting a new expires object for this resource");
0645: response.setExpires(new Long(expires
0646: + System.currentTimeMillis()));
0647: }
0648: }
0649:
0650: SourceValidity[] fromCacheValidityObjects = response
0651: .getValidityObjects();
0652:
0653: int i = 0;
0654: while (responseIsValid
0655: && i < fromCacheValidityObjects.length) {
0656: boolean isValid = false;
0657:
0658: // BH Check if validities[i] is null, may happen
0659: // if exception was thrown due to malformed content
0660: SourceValidity validity = fromCacheValidityObjects[i];
0661: int valid = validity == null ? SourceValidity.INVALID
0662: : validity.isValid();
0663: if (valid == SourceValidity.UNKNOWN) {
0664: // Don't know if valid, make second test
0665: validity = getValidityForInternalPipeline(i);
0666: if (validity != null) {
0667: valid = fromCacheValidityObjects[i]
0668: .isValid(validity);
0669: if (valid == SourceValidity.UNKNOWN) {
0670: validity = null;
0671: } else {
0672: isValid = (valid == SourceValidity.VALID);
0673: }
0674: }
0675: } else {
0676: isValid = (valid == SourceValidity.VALID);
0677: }
0678:
0679: if (!isValid) {
0680: responseIsValid = false;
0681: // update validity
0682: if (validity == null) {
0683: responseIsUsable = false;
0684: if (getLogger().isDebugEnabled()) {
0685: getLogger().debug(
0686: "validatePipeline: responseIsUsable is false, valid="
0687: + valid + " at index "
0688: + i);
0689: }
0690: } else {
0691: if (getLogger().isDebugEnabled()) {
0692: getLogger().debug(
0693: "validatePipeline: responseIsValid is false due to "
0694: + validity);
0695: }
0696: }
0697: } else {
0698: i++;
0699: }
0700: }
0701:
0702: if (responseIsValid) {
0703: if (getLogger().isDebugEnabled()) {
0704: getLogger().debug(
0705: "validatePipeline: using valid cached content for '"
0706: + environment.getURI() + "'.");
0707: }
0708:
0709: // we are valid, ok that's it
0710: this .cachedResponse = response;
0711: this .toCacheSourceValidities = fromCacheValidityObjects;
0712: } else {
0713: if (getLogger().isDebugEnabled()) {
0714: getLogger().debug(
0715: "validatePipeline: cached content is invalid for '"
0716: + environment.getURI() + "'.");
0717: }
0718: // we are not valid!
0719:
0720: if (!responseIsUsable) {
0721: // we could not compare, because we got no
0722: // validity object, so shorten pipeline key
0723: if (i > 0) {
0724: int deleteCount = fromCacheValidityObjects.length
0725: - i;
0726: if (i > 0
0727: && i <= firstNotCacheableTransformerIndex + 1) {
0728: this .firstNotCacheableTransformerIndex = i - 1;
0729: }
0730: for (int x = 0; x < deleteCount; x++) {
0731: this .toCacheKey.removeLastKey();
0732: }
0733: finished = false;
0734: } else {
0735: this .toCacheKey = null;
0736: }
0737: this .cacheCompleteResponse = false;
0738: } else {
0739: // the entry is invalid, remove it
0740: this .cache.remove(this .fromCacheKey);
0741: }
0742:
0743: // try a shorter key
0744: if (i > 0) {
0745: this .fromCacheKey.removeLastKey();
0746: if (!this .completeResponseIsCached) {
0747: this .firstProcessedTransformerIndex--;
0748: }
0749: } else {
0750: this .fromCacheKey = null;
0751: }
0752: finished = false;
0753: this .completeResponseIsCached = false;
0754: }
0755: } else {
0756:
0757: // check if there might be one being generated
0758: if (!waitForLock(this .fromCacheKey)) {
0759: finished = false;
0760: continue;
0761: }
0762:
0763: // no cached response found
0764: if (this .getLogger().isDebugEnabled()) {
0765: this .getLogger().debug(
0766: "Cached response not found for '"
0767: + environment.getURI()
0768: + "' using key: "
0769: + this .fromCacheKey);
0770: }
0771:
0772: finished = setupFromCacheKey();
0773: this .completeResponseIsCached = false;
0774: }
0775: }
0776:
0777: }
0778:
0779: boolean setupFromCacheKey() {
0780: // stop on longest key for smart caching
0781: this .fromCacheKey = null;
0782: return true;
0783: }
0784:
0785: /**
0786: * Setup the evenet pipeline.
0787: * The components of the pipeline are checked if they are
0788: * Cacheable.
0789: */
0790: protected void setupPipeline(Environment environment)
0791: throws ProcessingException {
0792: super .setupPipeline(environment);
0793:
0794: // Generate the key to fill the cache
0795: generateCachingKey(environment);
0796:
0797: // Test the cache for a valid response
0798: if (this .toCacheKey != null) {
0799: validatePipeline(environment);
0800: }
0801:
0802: setupValidities();
0803: }
0804:
0805: /**
0806: * Connect the pipeline.
0807: */
0808: protected void connectPipeline(Environment environment)
0809: throws ProcessingException {
0810: if (this .toCacheKey == null && this .cachedResponse == null) {
0811: super .connectPipeline(environment);
0812: return;
0813: } else if (this .completeResponseIsCached) {
0814: // do nothing
0815: return;
0816: } else {
0817: this .connectCachingPipeline(environment);
0818: }
0819: }
0820:
0821: /** Process the pipeline using a reader.
0822: * @throws ProcessingException if an error occurs
0823: */
0824: protected boolean processReader(Environment environment)
0825: throws ProcessingException {
0826: try {
0827: boolean usedCache = false;
0828: OutputStream outputStream = null;
0829: SourceValidity readerValidity = null;
0830: PipelineCacheKey pcKey = null;
0831:
0832: // test if reader is cacheable
0833: Serializable readerKey = null;
0834: boolean isCacheableProcessingComponent = false;
0835: if (super .reader instanceof CacheableProcessingComponent) {
0836: readerKey = ((CacheableProcessingComponent) super .reader)
0837: .getKey();
0838: isCacheableProcessingComponent = true;
0839: } else if (super .reader instanceof Cacheable) {
0840: readerKey = new Long(((Cacheable) super .reader)
0841: .generateKey());
0842: }
0843:
0844: boolean finished = false;
0845:
0846: if (readerKey != null) {
0847: // response is cacheable, build the key
0848: pcKey = new PipelineCacheKey();
0849: pcKey.addKey(new ComponentCacheKey(
0850: ComponentCacheKey.ComponentType_Reader,
0851: this .readerRole, readerKey));
0852:
0853: while (!finished) {
0854: finished = true;
0855: // now we have the key to get the cached object
0856: CachedResponse cachedObject = this .cache.get(pcKey);
0857: if (cachedObject != null) {
0858: if (getLogger().isDebugEnabled()) {
0859: getLogger().debug(
0860: "Found cached response for '"
0861: + environment.getURI()
0862: + "' using key: " + pcKey);
0863: }
0864:
0865: SourceValidity[] validities = cachedObject
0866: .getValidityObjects();
0867: if (validities == null
0868: || validities.length != 1) {
0869: // to avoid getting here again and again, we delete it
0870: this .cache.remove(pcKey);
0871: if (getLogger().isDebugEnabled()) {
0872: getLogger().debug(
0873: "Cached response for '"
0874: + environment.getURI()
0875: + "' using key: "
0876: + pcKey
0877: + " is invalid.");
0878: }
0879: this .cachedResponse = null;
0880: } else {
0881: SourceValidity cachedValidity = validities[0];
0882: boolean isValid = false;
0883: int valid = cachedValidity.isValid();
0884: if (valid == SourceValidity.UNKNOWN) {
0885: // get reader validity and compare
0886: if (isCacheableProcessingComponent) {
0887: readerValidity = ((CacheableProcessingComponent) super .reader)
0888: .getValidity();
0889: } else {
0890: CacheValidity cv = ((Cacheable) super .reader)
0891: .generateValidity();
0892: if (cv != null) {
0893: readerValidity = CacheValidityToSourceValidity
0894: .createValidity(cv);
0895: }
0896: }
0897: if (readerValidity != null) {
0898: valid = cachedValidity
0899: .isValid(readerValidity);
0900: if (valid == SourceValidity.UNKNOWN) {
0901: readerValidity = null;
0902: } else {
0903: isValid = (valid == SourceValidity.VALID);
0904: }
0905: }
0906: } else {
0907: isValid = (valid == SourceValidity.VALID);
0908: }
0909:
0910: if (isValid) {
0911: if (getLogger().isDebugEnabled()) {
0912: getLogger().debug(
0913: "processReader: using valid cached content for '"
0914: + environment
0915: .getURI()
0916: + "'.");
0917: }
0918: byte[] response = cachedObject
0919: .getResponse();
0920: if (response.length > 0) {
0921: usedCache = true;
0922: if (cachedObject.getContentType() != null) {
0923: environment
0924: .setContentType(cachedObject
0925: .getContentType());
0926: } else {
0927: setMimeTypeForReader(environment);
0928: }
0929: outputStream = environment
0930: .getOutputStream(0);
0931: environment
0932: .setContentLength(response.length);
0933: outputStream.write(response);
0934: }
0935: } else {
0936: if (getLogger().isDebugEnabled()) {
0937: getLogger().debug(
0938: "processReader: cached content is invalid for '"
0939: + environment
0940: .getURI()
0941: + "'.");
0942: }
0943: // remove invalid cached object
0944: this .cache.remove(pcKey);
0945: }
0946: }
0947: } else {
0948: // check if something is being generated right now
0949: if (!waitForLock(pcKey)) {
0950: finished = false;
0951: continue;
0952: }
0953: }
0954: }
0955: }
0956:
0957: if (!usedCache) {
0958: // make sure lock will be released
0959: try {
0960: if (pcKey != null) {
0961: if (getLogger().isDebugEnabled()) {
0962: getLogger().debug(
0963: "processReader: caching content for further requests of '"
0964: + environment.getURI()
0965: + "'.");
0966: }
0967: generateLock(pcKey);
0968:
0969: if (readerValidity == null) {
0970: if (isCacheableProcessingComponent) {
0971: readerValidity = ((CacheableProcessingComponent) super .reader)
0972: .getValidity();
0973: } else {
0974: CacheValidity cv = ((Cacheable) super .reader)
0975: .generateValidity();
0976: if (cv != null) {
0977: readerValidity = CacheValidityToSourceValidity
0978: .createValidity(cv);
0979: }
0980: }
0981: }
0982:
0983: if (readerValidity != null) {
0984: outputStream = environment
0985: .getOutputStream(this .outputBufferSize);
0986: outputStream = new CachingOutputStream(
0987: outputStream);
0988: }
0989: }
0990:
0991: setMimeTypeForReader(environment);
0992: if (this .reader.shouldSetContentLength()) {
0993: ByteArrayOutputStream os = new ByteArrayOutputStream();
0994: this .reader.setOutputStream(os);
0995: this .reader.generate();
0996: environment.setContentLength(os.size());
0997: if (outputStream == null) {
0998: outputStream = environment
0999: .getOutputStream(0);
1000: }
1001: os.writeTo(outputStream);
1002: } else {
1003: if (outputStream == null) {
1004: outputStream = environment
1005: .getOutputStream(this .outputBufferSize);
1006: }
1007: this .reader.setOutputStream(outputStream);
1008: this .reader.generate();
1009: }
1010:
1011: // store the response
1012: if (pcKey != null && readerValidity != null) {
1013: final CachedResponse res = new CachedResponse(
1014: new SourceValidity[] { readerValidity },
1015: ((CachingOutputStream) outputStream)
1016: .getContent());
1017: res
1018: .setContentType(environment
1019: .getContentType());
1020: this .cache.store(pcKey, res);
1021: }
1022:
1023: } finally {
1024: if (pcKey != null) {
1025: releaseLock(pcKey);
1026: }
1027: }
1028:
1029: }
1030: } catch (Exception e) {
1031: handleException(e);
1032: }
1033:
1034: return true;
1035: }
1036:
1037: /**
1038: * Return valid validity objects for the event pipeline.
1039: *
1040: * If the event pipeline (the complete pipeline without the
1041: * serializer) is cacheable and valid, return all validity objects.
1042: *
1043: * Otherwise, return <code>null</code>.
1044: */
1045: public SourceValidity getValidityForEventPipeline() {
1046: if (isInternalError()) {
1047: return null;
1048: }
1049:
1050: if (this .cachedResponse != null) {
1051: if (!this .cacheCompleteResponse
1052: && this .firstNotCacheableTransformerIndex < super .transformers
1053: .size()) {
1054: // Cache contains only partial pipeline.
1055: return null;
1056: }
1057:
1058: if (this .toCacheSourceValidities != null) {
1059: // This means that the pipeline is valid based on the validities
1060: // of the individual components
1061: final AggregatedValidity validity = new AggregatedValidity();
1062: for (int i = 0; i < this .toCacheSourceValidities.length; i++) {
1063: validity.add(this .toCacheSourceValidities[i]);
1064: }
1065:
1066: return validity;
1067: }
1068:
1069: // This means that the pipeline is valid because it has not yet expired
1070: return NOPValidity.SHARED_INSTANCE;
1071: } else {
1072: int vals = 0;
1073:
1074: if (null != this .toCacheKey
1075: && !this .cacheCompleteResponse
1076: && this .firstNotCacheableTransformerIndex == super .transformers
1077: .size()) {
1078: vals = this .toCacheKey.size();
1079: } else if (null != this .fromCacheKey
1080: && !this .completeResponseIsCached
1081: && this .firstProcessedTransformerIndex == super .transformers
1082: .size()) {
1083: vals = this .fromCacheKey.size();
1084: }
1085:
1086: if (vals > 0) {
1087: final AggregatedValidity validity = new AggregatedValidity();
1088: for (int i = 0; i < vals; i++) {
1089: validity.add(getValidityForInternalPipeline(i));
1090: }
1091:
1092: return validity;
1093: }
1094:
1095: return null;
1096: }
1097: }
1098:
1099: /**
1100: * Get generator cache key (null if not cacheable)
1101: */
1102: private Serializable getGeneratorKey() {
1103: Serializable key = null;
1104: if (super .generator instanceof CacheableProcessingComponent) {
1105: key = ((CacheableProcessingComponent) super .generator)
1106: .getKey();
1107: this .generatorIsCacheableProcessingComponent = true;
1108: } else if (super .generator instanceof Cacheable) {
1109: key = new Long(((Cacheable) super .generator).generateKey());
1110: }
1111: return key;
1112: }
1113:
1114: /**
1115: * Get transformer cache key (null if not cacheable)
1116: */
1117: private Serializable getTransformerKey(final Transformer transformer) {
1118: Serializable key = null;
1119: if (transformer instanceof CacheableProcessingComponent) {
1120: key = ((CacheableProcessingComponent) transformer).getKey();
1121: this .transformerIsCacheableProcessingComponent[this .firstNotCacheableTransformerIndex] = true;
1122: } else if (transformer instanceof Cacheable) {
1123: key = new Long(((Cacheable) transformer).generateKey());
1124: }
1125: return key;
1126: }
1127:
1128: /**
1129: * Get serializer cache key (null if not cacheable)
1130: */
1131: private Serializable getSerializerKey() {
1132: Serializable key = null;
1133: if (super .serializer instanceof CacheableProcessingComponent) {
1134: key = ((CacheableProcessingComponent) this .serializer)
1135: .getKey();
1136: this .serializerIsCacheableProcessingComponent = true;
1137: } else if (this .serializer instanceof Cacheable) {
1138: key = new Long(((Cacheable) this .serializer).generateKey());
1139: }
1140: return key;
1141: }
1142:
1143: /* (non-Javadoc)
1144: * @see org.apache.cocoon.components.pipeline.ProcessingPipeline#getKeyForEventPipeline()
1145: */
1146: public String getKeyForEventPipeline() {
1147: if (isInternalError()) {
1148: return null;
1149: }
1150:
1151: if (null != this .toCacheKey
1152: && !this .cacheCompleteResponse
1153: && this .firstNotCacheableTransformerIndex == super .transformers
1154: .size()) {
1155: return String.valueOf(HashUtil.hash(this .toCacheKey
1156: .toString()));
1157: }
1158: if (null != this .fromCacheKey
1159: && !this .completeResponseIsCached
1160: && this .firstProcessedTransformerIndex == super .transformers
1161: .size()) {
1162: return String.valueOf(HashUtil.hash(this .fromCacheKey
1163: .toString()));
1164: }
1165:
1166: return null;
1167: }
1168:
1169: SourceValidity getValidityForInternalPipeline(int index) {
1170: final SourceValidity validity;
1171:
1172: // if debugging try to tell why something is not cacheable
1173: final boolean debug = this .getLogger().isDebugEnabled();
1174: String msg = null;
1175: if (debug)
1176: msg = "getValidityForInternalPipeline(" + index + "): ";
1177:
1178: if (index == 0) {
1179: // test generator
1180: if (this .generatorIsCacheableProcessingComponent) {
1181: validity = ((CacheableProcessingComponent) super .generator)
1182: .getValidity();
1183: if (debug)
1184: msg += "generator: using getValidity";
1185: } else {
1186: validity = CacheValidityToSourceValidity
1187: .createValidity(((Cacheable) super .generator)
1188: .generateValidity());
1189: if (debug)
1190: msg += "generator: using generateValidity";
1191: }
1192: } else if (index <= firstNotCacheableTransformerIndex) {
1193: // test transformer
1194: final Transformer trans = (Transformer) super .transformers
1195: .get(index - 1);
1196: if (this .transformerIsCacheableProcessingComponent[index - 1]) {
1197: validity = ((CacheableProcessingComponent) trans)
1198: .getValidity();
1199: if (debug)
1200: msg += "transformer: using getValidity";
1201: } else {
1202: validity = CacheValidityToSourceValidity
1203: .createValidity(((Cacheable) trans)
1204: .generateValidity());
1205: if (debug)
1206: msg += "transformer: using generateValidity";
1207: }
1208: } else {
1209: // test serializer
1210: if (this .serializerIsCacheableProcessingComponent) {
1211: validity = ((CacheableProcessingComponent) super .serializer)
1212: .getValidity();
1213: if (debug)
1214: msg += "serializer: using getValidity";
1215: } else {
1216: validity = CacheValidityToSourceValidity
1217: .createValidity(((Cacheable) super .serializer)
1218: .generateValidity());
1219: if (debug)
1220: msg += "serializer: using generateValidity";
1221: }
1222: }
1223:
1224: if (debug) {
1225: msg += ", validity==" + validity;
1226: this .getLogger().debug(msg);
1227: }
1228: return validity;
1229: }
1230:
1231: /**
1232: * Recyclable Interface
1233: */
1234: public void recycle() {
1235: this .generatorRole = null;
1236: this .transformerRoles.clear();
1237: this .serializerRole = null;
1238: this .readerRole = null;
1239:
1240: this .fromCacheKey = null;
1241: this .cachedResponse = null;
1242:
1243: this .transformerIsCacheableProcessingComponent = null;
1244: this .toCacheKey = null;
1245: this .toCacheSourceValidities = null;
1246:
1247: super .recycle();
1248: }
1249: }
1250:
1251: final class DeferredPipelineValidity implements DeferredValidity {
1252:
1253: private final AbstractCachingProcessingPipeline pipeline;
1254: private final int index;
1255:
1256: public DeferredPipelineValidity(
1257: AbstractCachingProcessingPipeline pipeline, int index) {
1258: this .pipeline = pipeline;
1259: this .index = index;
1260: }
1261:
1262: /**
1263: * @see org.apache.excalibur.source.impl.validity.DeferredValidity#getValidity()
1264: */
1265: public SourceValidity getValidity() {
1266: return pipeline.getValidityForInternalPipeline(this.index);
1267: }
1268: }
|