| /* |
| * 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. |
| */ |
| package org.apache.cocoon.transformation.helpers; |
| |
| import java.io.IOException; |
| import java.net.URL; |
| |
| import org.apache.avalon.framework.activity.Disposable; |
| import org.apache.avalon.framework.component.Component; |
| import org.apache.avalon.framework.logger.AbstractLogEnabled; |
| import org.apache.avalon.framework.parameters.ParameterException; |
| import org.apache.avalon.framework.parameters.Parameterizable; |
| import org.apache.avalon.framework.parameters.Parameters; |
| import org.apache.avalon.framework.service.ServiceException; |
| import org.apache.avalon.framework.service.ServiceManager; |
| import org.apache.avalon.framework.service.Serviceable; |
| import org.apache.avalon.framework.thread.ThreadSafe; |
| import org.apache.cocoon.ProcessingException; |
| import org.apache.cocoon.caching.CachedResponse; |
| import org.apache.cocoon.components.sax.XMLDeserializer; |
| import org.apache.cocoon.components.sax.XMLSerializer; |
| import org.apache.cocoon.components.sax.XMLTeePipe; |
| import org.apache.cocoon.components.source.SourceUtil; |
| import org.apache.cocoon.components.thread.RunnableManager; |
| import org.apache.cocoon.environment.CocoonRunnable; |
| import org.apache.cocoon.xml.XMLConsumer; |
| import org.apache.excalibur.source.Source; |
| import org.apache.excalibur.source.SourceException; |
| import org.apache.excalibur.source.SourceResolver; |
| import org.apache.excalibur.source.SourceValidity; |
| import org.apache.excalibur.store.Store; |
| import org.xml.sax.SAXException; |
| import EDU.oswego.cs.dl.util.concurrent.CountDown; |
| |
| /** |
| * Default implementation of a {@link IncludeCacheManager}. |
| * |
| * This implementation requires a configuration, if preemptive |
| * loading is used: |
| * <parameter name="preemptive-loader-url" value="some url"/> |
| * |
| * This is a url inside cocoon, that contains the preemptive loader |
| * url; it must be specified absolute (with http://...) |
| * If this loader cannot be started, only an error is logged into the |
| * log, so actually cached content is never updated! |
| * |
| * @author <a href="mailto:cziegeler@apache.org">Carsten Ziegeler</a> |
| * @version CVS $Id$ |
| * @since 2.1 |
| */ |
| public final class DefaultIncludeCacheManager |
| extends AbstractLogEnabled |
| implements IncludeCacheManager, |
| ThreadSafe, |
| Serviceable, |
| Disposable, |
| Parameterizable, |
| Component { |
| |
| private ServiceManager manager; |
| |
| private SourceResolver resolver; |
| |
| private Store store; |
| |
| private IncludeCacheStorageProxy defaultCacheStorage; |
| |
| private String preemptiveLoaderURI; |
| |
| /** |
| * @see IncludeCacheManager#getSession(org.apache.avalon.framework.parameters.Parameters) |
| */ |
| public IncludeCacheManagerSession getSession(Parameters pars) { |
| String sourceURI = pars.getParameter("source", null); |
| IncludeCacheManagerSession session; |
| if ( null == sourceURI ) { |
| session = new IncludeCacheManagerSession(pars, this.defaultCacheStorage); |
| } else { |
| Source source = null; |
| try { |
| source = this.resolver.resolveURI(sourceURI); |
| IncludeCacheStorageProxy proxy = new ModifiableSourceIncludeCacheStorageProxy(this.resolver, source.getURI(), this.getLogger()); |
| session = new IncludeCacheManagerSession(pars, proxy); |
| } catch (Exception local) { |
| session = new IncludeCacheManagerSession(pars, this.defaultCacheStorage); |
| this.getLogger().warn("Error creating writeable source.", local); |
| } finally { |
| this.resolver.release(source); |
| } |
| } |
| if (session.isPreemptive()) { |
| if ( null == this.preemptiveLoaderURI ) { |
| this.getLogger().error("Preemptive loading is turned off because the preemptive-loader-url is not configured."); |
| session.setPreemptive(false); |
| } else { |
| if ( !PreemptiveLoader.getInstance().alive ) { |
| |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Booting preemptive loader: " + this.preemptiveLoaderURI); |
| } |
| PreemptiveBooter thread = new PreemptiveBooter( this.preemptiveLoaderURI ); |
| try |
| { |
| final RunnableManager runnableManager = (RunnableManager)this.manager.lookup( RunnableManager.ROLE ); |
| runnableManager.execute( thread ); |
| this.manager.release( runnableManager ); |
| } |
| catch( final ServiceException se ) |
| { |
| getLogger().error( "Cannot lookup RunnableManager", se ); |
| } |
| } |
| } |
| } |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Creating cache manager session: " + session); |
| } |
| return session; |
| } |
| |
| /** |
| * @see IncludeCacheManager#load(java.lang.String, IncludeCacheManagerSession) |
| */ |
| public String load(String uri, |
| IncludeCacheManagerSession session) |
| throws IOException, SourceException { |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Load " + uri + " for session " + session); |
| } |
| |
| // first make the URI absolute |
| if ( uri.indexOf("://") == -1) { |
| final Source source = session.resolveURI(uri, this.resolver); |
| uri = source.getURI(); |
| } |
| |
| // if we are not processing in parallel (or do preemptive) |
| // then we don't have to do anything in this method - everything |
| // is done in the stream method. |
| |
| // if we are processing in parallel (and not preemptive) then.... |
| if ( session.isParallel() && !session.isPreemptive()) { |
| |
| // first look-up if we have a valid stored response |
| IncludeCacheStorageProxy storage = session.getCacheStorageProxy(); |
| CachedResponse response = (CachedResponse)storage.get(uri); |
| if ( null != response) { |
| SourceValidity[] validities = response.getValidityObjects(); |
| |
| // if we are valid and do not purging |
| if ( !session.isPurging() |
| && validities[0].isValid() == SourceValidity.VALID) { |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Using cached response for parallel processing."); |
| } |
| session.add(uri, response.getResponse()); |
| return uri; |
| } else { |
| // response is not used |
| storage.remove(uri); |
| } |
| } |
| |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Starting parallel thread for loading " + uri); |
| } |
| // now we start a parallel thread, this thread gets all required avalon components |
| // so it does not have to lookup them by itself |
| try { |
| XMLSerializer serializer = (XMLSerializer)this.manager.lookup(XMLSerializer.ROLE); |
| Source source = session.resolveURI(uri, this.resolver); |
| |
| LoaderThread loader = new LoaderThread(source, serializer, this.manager); |
| // Load the included content in a thread that inherits the current thread's environment |
| Thread thread = new Thread(new CocoonRunnable(loader)); |
| session.add(uri, loader); |
| thread.start(); |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Thread started for " + uri); |
| } |
| } catch (ServiceException ce) { |
| throw new SourceException("Unable to lookup thread pool or xml serializer.", ce); |
| } catch (Exception e) { |
| throw new SourceException("Unable to get pooled thread.", e); |
| } |
| } |
| return uri; |
| } |
| |
| /** |
| * @see IncludeCacheManager#stream(java.lang.String, IncludeCacheManagerSession, XMLConsumer) |
| */ |
| public void stream(String uri, |
| IncludeCacheManagerSession session, |
| XMLConsumer handler) |
| throws IOException, SourceException, SAXException { |
| |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Stream " + uri + " for session " + session); |
| } |
| |
| // if we are processing in parallel (and not preemptive) then.... |
| if ( session.isParallel() && !session.isPreemptive()) { |
| |
| // get either the cached content or the pooled thread |
| Object object = session.get(uri); |
| |
| if ( null == object ) { |
| // this should never happen! |
| throw new SAXException("No pooled thread found for " + uri); |
| } |
| byte[] result; |
| |
| // is this a pooled thread? |
| if (object instanceof LoaderThread) { |
| LoaderThread loader = (LoaderThread)object; |
| |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Waiting for pooled thread to finish loading."); |
| } |
| |
| // wait for it |
| loader.join(); |
| |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Pooled thread finished loading."); |
| } |
| |
| // did an exception occur? Then reraise it |
| if ( null != loader.exception) { |
| if ( loader.exception instanceof SAXException ) { |
| throw (SAXException)loader.exception; |
| } else if (loader.exception instanceof SourceException ) { |
| throw (SourceException)loader.exception; |
| } else if (loader.exception instanceof IOException) { |
| throw (IOException)loader.exception; |
| } else { |
| throw new SAXException("Exception.", loader.exception); |
| } |
| } |
| |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Streaming from pooled thread."); |
| } |
| result = loader.content; |
| |
| // cache the response (remember preemptive is off) |
| if (session.getExpires() > 0) { |
| SourceValidity[] validities = new SourceValidity[1]; |
| validities[0] = session.getExpiresValidity(); |
| CachedResponse response = new CachedResponse(validities, result); |
| session.getCacheStorageProxy().put(uri, response); |
| } |
| } else { |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Streaming from cached response."); |
| } |
| |
| // use the response from the cache |
| result = (byte[])object; |
| } |
| |
| // stream the content |
| XMLDeserializer deserializer = null; |
| try { |
| deserializer = (XMLDeserializer)this.manager.lookup( XMLDeserializer.ROLE ); |
| deserializer.setConsumer(handler); |
| deserializer.deserialize(result); |
| } catch (ServiceException ce) { |
| throw new SAXException("Unable to lookup xml deserializer.", ce); |
| } finally { |
| this.manager.release( deserializer ); |
| } |
| return; |
| |
| } else { |
| // we are not processing parallel |
| |
| // first: test for a cached response |
| IncludeCacheStorageProxy storage = session.getCacheStorageProxy(); |
| CachedResponse response = (CachedResponse)storage.get(uri); |
| if ( null != response) { |
| SourceValidity[] validities = response.getValidityObjects(); |
| // if purging is turned off and either the cached response is valid or |
| // we are loading preemptive, then use the cached response |
| if ( !session.isPurging() |
| && (session.isPreemptive() || validities[0].isValid() == SourceValidity.VALID)) { |
| |
| // stream the content |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Streaming from cached response."); |
| } |
| XMLDeserializer deserializer = null; |
| try { |
| deserializer = (XMLDeserializer)this.manager.lookup( XMLDeserializer.ROLE ); |
| deserializer.setConsumer(handler); |
| deserializer.deserialize(response.getResponse()); |
| } catch (ServiceException ce) { |
| throw new SAXException("Unable to lookup xml deserializer.", ce); |
| } finally { |
| this.manager.release( deserializer ); |
| } |
| |
| // load preemptive if the response is not valid |
| if ( session.getExpires() > 0 |
| && session.isPreemptive() |
| && validities[0].isValid() != SourceValidity.VALID) { |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Add uri to preemptive loader list " + uri); |
| } |
| if (!PreemptiveLoader.getInstance().alive) { |
| this.getLogger().error("Preemptive loader has not started yet."); |
| } |
| PreemptiveLoader.getInstance().add(session.getCacheStorageProxy(), uri, session.getExpires()); |
| } |
| return; |
| |
| } else { |
| // cached response is not valid |
| storage.remove(uri); |
| } |
| } |
| } |
| |
| // we are not processing in parallel and have no (valid) cached response |
| XMLSerializer serializer = null; |
| try { |
| final Source source = session.resolveURI(uri, this.resolver); |
| |
| // stream directly (and cache the response) |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Streaming directly from source."); |
| } |
| if (session.getExpires() > 0) { |
| serializer = (XMLSerializer)this.manager.lookup(XMLSerializer.ROLE); |
| XMLTeePipe tee = new XMLTeePipe(handler, serializer); |
| |
| SourceUtil.toSAX(source, tee); |
| |
| SourceValidity[] validities = new SourceValidity[1]; |
| validities[0] = session.getExpiresValidity(); |
| CachedResponse response = new CachedResponse(validities, |
| (byte[])serializer.getSAXFragment()); |
| session.getCacheStorageProxy().put(uri, response); |
| } else { |
| SourceUtil.toSAX(source, handler); |
| } |
| |
| } catch (ProcessingException pe) { |
| throw new SAXException("ProcessingException", pe); |
| } catch (ServiceException e) { |
| throw new SAXException("Unable to lookup xml serializer.", e); |
| } finally { |
| this.manager.release(serializer); |
| } |
| } |
| |
| /** |
| * @see IncludeCacheManager#terminateSession(IncludeCacheManagerSession) |
| */ |
| public void terminateSession(IncludeCacheManagerSession session) { |
| if (this.getLogger().isDebugEnabled()) { |
| this.getLogger().debug("Terminating cache manager session " + session); |
| } |
| session.cleanup(this.resolver); |
| } |
| |
| /** |
| * @see org.apache.avalon.framework.service.Serviceable#service(org.apache.avalon.framework.service.ServiceManager) |
| */ |
| public void service(ServiceManager manager) throws ServiceException { |
| this.manager = manager; |
| this.resolver = (SourceResolver)this.manager.lookup(SourceResolver.ROLE); |
| } |
| |
| /** |
| * @see org.apache.avalon.framework.activity.Disposable#dispose() |
| */ |
| public void dispose() { |
| // stop preemptive loader (if running) |
| PreemptiveLoader.getInstance().stop(); |
| if ( null != this.manager ) { |
| this.manager.release( this.resolver); |
| this.manager.release(this.store); |
| this.store = null; |
| this.resolver = null; |
| this.manager = null; |
| this.defaultCacheStorage = null; |
| } |
| } |
| |
| /** |
| * @see org.apache.avalon.framework.parameters.Parameterizable#parameterize(org.apache.avalon.framework.parameters.Parameters) |
| */ |
| public void parameterize(Parameters parameters) throws ParameterException { |
| this.preemptiveLoaderURI = parameters.getParameter("preemptive-loader-url", null); |
| if ( null != this.preemptiveLoaderURI |
| && this.preemptiveLoaderURI.indexOf("://") == -1) { |
| throw new ParameterException("The preemptive-loader-url must be absolute: " + this.preemptiveLoaderURI); |
| } |
| final String storeRole = parameters.getParameter("use-store", Store.ROLE); |
| try { |
| this.store = (Store)this.manager.lookup(storeRole); |
| } catch (ServiceException e) { |
| throw new ParameterException("Unable to lookup store with role " + storeRole, e); |
| } |
| this.defaultCacheStorage = new StoreIncludeCacheStorageProxy(this.store, this.getLogger()); |
| } |
| |
| final private static class LoaderThread implements Runnable { |
| |
| final private Source source; |
| final private XMLSerializer serializer; |
| final private CountDown finished; |
| Exception exception; |
| byte[] content; |
| final private ServiceManager manager; |
| |
| public LoaderThread(Source source, |
| XMLSerializer serializer, |
| ServiceManager manager) { |
| this.source = source; |
| this.serializer = serializer; |
| this.finished = new CountDown( 1 ); |
| this.manager = manager; |
| } |
| |
| public void run() { |
| try { |
| SourceUtil.toSAX(this.source, this.serializer); |
| this.content = (byte[])this.serializer.getSAXFragment(); |
| } catch (Exception local) { |
| this.exception = local; |
| } finally { |
| this.manager.release( this.serializer ); |
| this.finished.release(); |
| } |
| } |
| |
| void join() { |
| try { |
| this.finished.acquire(); |
| } catch ( final InterruptedException ie) { |
| // ignore |
| } |
| } |
| } |
| |
| final static class PreemptiveBooter implements Runnable { |
| |
| private final String uri; |
| |
| public PreemptiveBooter( final String uri ) { |
| this.uri = uri; |
| } |
| |
| public void run() { |
| try { |
| URL url = new URL(this.uri); |
| url.getContent(); |
| } catch (Exception ignore) { |
| } |
| } |
| } |
| } |
