/*
 * $Header: /home/cvspublic/jakarta-slide/src/share/org/apache/slide/store/ExtendedStore.java,v 1.32 2005/03/01 15:39:10 luetzkendorf Exp $
 * $Revision: 1.32 $
 * $Date: 2005/03/01 15:39:10 $
 *
 * ====================================================================
 *
 * Copyright 1999-2002 The Apache Software Foundation 
 *
 * Licensed 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.slide.store;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Vector;

import javax.transaction.Transaction;
import javax.transaction.xa.XAException;
import javax.transaction.xa.Xid;

import org.apache.commons.transaction.util.xa.XidWrapper;
import org.apache.slide.common.Service;
import org.apache.slide.common.ServiceAccessException;
import org.apache.slide.common.ServiceParameterErrorException;
import org.apache.slide.common.ServiceParameterMissingException;
import org.apache.slide.common.Uri;
import org.apache.slide.content.NodeRevisionContent;
import org.apache.slide.content.NodeRevisionDescriptor;
import org.apache.slide.content.NodeRevisionDescriptors;
import org.apache.slide.content.NodeRevisionNumber;
import org.apache.slide.content.RevisionAlreadyExistException;
import org.apache.slide.content.RevisionDescriptorNotFoundException;
import org.apache.slide.content.RevisionNotFoundException;
import org.apache.slide.lock.LockTokenNotFoundException;
import org.apache.slide.lock.NodeLock;
import org.apache.slide.security.NodePermission;
import org.apache.slide.structure.LinkNode;
import org.apache.slide.structure.ObjectAlreadyExistsException;
import org.apache.slide.structure.ObjectNode;
import org.apache.slide.structure.ObjectNotFoundException;
import org.apache.slide.util.ByteSizeLimitedObjectCache;
import org.apache.slide.util.EmptyEnumeration;
import org.apache.slide.util.Messages;
import org.apache.slide.util.ObjectCache;
import org.apache.slide.util.TxLRUObjectCache;
import org.apache.slide.util.logger.Logger;

/**
 * Store that allows for transactional caching of data. Takes over much modified code from StandardStore.
 * That's why Remy is listed as author as well.
 *
 * @version $Revision: 1.32 $
 */
public class ExtendedStore extends AbstractStore {

    private static final Vector EMPTY_VECTOR = new Vector(0);
    private static final String LOG_CHANNEL = ExtendedStore.class.getName();

    protected static final int DEFAULT_OBJECT_GLOBAL_CACHE_SIZE = 10000;
    protected static final String GLOBAL_OBJECT_CACHE_SIZE_PARAMETER = "object-cache-size";

    protected static final int DEFAULT_GLOBAL_PERMISSION_CACHE_SIZE = 10000;
    protected static final String GLOBAL_PERMISSION_CACHE_SIZE_PARAMETER = "permission-cache-size";

    protected static final int DEFAULT_GLOBAL_LOCK_CACHE_SIZE = 10000;
    protected static final String GLOBAL_LOCK_CACHE_SIZE_PARAMETER = "lock-cache-size";

    protected static final int DEFAULT_GLOBAL_DESCRIPTORS_CACHE_SIZE = 10000;
    protected static final String GLOBAL_DESCRIPTORS_CACHE_SIZE_PARAMETER = "descriptors-cache-size";

    protected static final int DEFAULT_GLOBAL_DESCRIPTOR_CACHE_SIZE = 10000;
    protected static final String GLOBAL_DESCRIPTOR_CACHE_SIZE_PARAMETER = "descriptor-cache-size";

    protected static final int DEFAULT_GLOBAL_CONTENT_CACHE_SIZE = 10000;
    protected static final String GLOBAL_CONTENT_CACHE_SIZE_PARAMETER = "content-cache-size";

    protected static final boolean DEFAULT_ENABLE_CONTENT_CACHING = false;
    protected static final String ENABLE_CONTENT_CACHING_PARAMETER = "enable-content-caching";

    protected static final int DEFAULT_TX_CONTENT_CACHE_SIZE = 1000;
    protected static final String TX_CONTENT_CACHE_SIZE_PARAMETER = "tx-content-cache-size";

    protected static final long DEFAULT_CONTENT_CACHE_BYTES = 10000000; // 10 M
    protected static final String CONTENT_CACHE_BYTES_PARAMETER = "content-cache-bytes";

    protected static final long DEFAULT_TX_CONTENT_CACHE_BYTES = 1000000; // 1 M
    protected static final String TX_CONTENT_CACHE_BYTES_PARAMETER = "tx-content-cache-bytes";

    protected static final long DEFAULT_MAX_CONTENT_BYTES_PER_ENTRY = 50000; // 50 K
    protected static final String MAX_CONTENT_BYTES_PER_ENTRY_PARAMETER = "max-content-bytes-per-entry";

    protected static final String CACHE_MODE_PARAMETER = "cache-mode";
    protected static final String CACHE_MODE_FULL = "full";
    protected static final String CACHE_MODE_LOCAL = "cluster";
    protected static final String CACHE_MODE_NO_GLOBAL_IN_TX = "isolation-shadow";
    protected static final String CACHE_MODE_OFF = "off";
    protected static final String DEFAULT_CACHE_MODE = CACHE_MODE_FULL;

    protected static final int DEFAULT_CACHE_TIMEOUT = -1; // infinite
    protected static final String CACHE_TIMEOUT_PARAMETER = "cache-timeout";

    protected static final int DEFAULT_TLOCK_TIMEOUT = 20000; // 20 sec
    protected static final String TLOCK_TIMEOUT_PARAMETER = "tlock-timeout";

    // there might be at least one active transaction branch per thread
    protected ThreadLocal activeTransactionBranch = new ThreadLocal();

    protected TxContentCacheWrapper contentCache;
    protected TxCacheWrapper objectsCache;
    protected TxCacheWrapper permissionsCache;
    protected TxCacheWrapper locksCache;
    protected TxCacheWrapper descriptorsCache;
    protected TxCacheWrapper descriptorCache;

    protected long cacheTimeout;

    protected boolean globalCacheOff;

    int globalObjectCacheSize;
    int globalPermissionCacheSize;
    int globalLockCacheSize;
    int globalDescrtiptorsCacheSize;
    int globalDescrtiptorCacheSize;
    boolean contentCachingEnabled;
    int globalContentCacheSize;
    long contentCacheBytes;
    int txContentCacheSize;
    long txContentCacheBytes;
    long maxByteSizePerEntry;
    boolean noGlobalCacheInTx;
    
    public void setParameters(Hashtable parameters)
        throws ServiceParameterErrorException, ServiceParameterMissingException {
        super.setParameters(parameters);

        globalObjectCacheSize = DEFAULT_OBJECT_GLOBAL_CACHE_SIZE;
        globalPermissionCacheSize = DEFAULT_GLOBAL_PERMISSION_CACHE_SIZE;
        globalLockCacheSize = DEFAULT_GLOBAL_LOCK_CACHE_SIZE;
        globalDescrtiptorsCacheSize = DEFAULT_GLOBAL_DESCRIPTORS_CACHE_SIZE;
        globalDescrtiptorCacheSize = DEFAULT_GLOBAL_DESCRIPTOR_CACHE_SIZE;
        contentCachingEnabled = DEFAULT_ENABLE_CONTENT_CACHING;
        globalContentCacheSize = DEFAULT_GLOBAL_CONTENT_CACHE_SIZE;
        contentCacheBytes = DEFAULT_CONTENT_CACHE_BYTES;
        txContentCacheSize = DEFAULT_TX_CONTENT_CACHE_SIZE;
        txContentCacheBytes = DEFAULT_TX_CONTENT_CACHE_BYTES;
        maxByteSizePerEntry = DEFAULT_MAX_CONTENT_BYTES_PER_ENTRY;

        cacheTimeout = DEFAULT_CACHE_TIMEOUT;
        String cacheTimeoutString = (String) parameters.get(CACHE_TIMEOUT_PARAMETER);
        if (cacheTimeoutString != null) {
            try {
                cacheTimeout = Long.parseLong(cacheTimeoutString) * 1000;
            } catch (NumberFormatException nfe) {
                getLogger().log("Cache timeout must be integer! Using default value", LOG_CHANNEL, Logger.WARNING);
            }
        }
        if (cacheTimeout != -1) {
            getLogger().log("Setting cache timeout for store " + getName() + " to " + cacheTimeout / 1000 + " seconds",
                    Logger.INFO);
        } else {
            getLogger().log("Disabling cache timeout for store " + getName(), Logger.INFO);
        }

        String globalObjectCacheSizeString = (String) parameters.get(GLOBAL_OBJECT_CACHE_SIZE_PARAMETER);
        if (globalObjectCacheSizeString != null) {
            try {
                globalObjectCacheSize = Integer.parseInt(globalObjectCacheSizeString);
            } catch (NumberFormatException nfe) {
                getLogger().log("Cache size must be an integer! Using default size", LOG_CHANNEL, Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting object cache size for store " + getName() + " to " + globalObjectCacheSize,
            Logger.INFO);

        String globalPermissionCacheSizeString = (String) parameters.get(GLOBAL_PERMISSION_CACHE_SIZE_PARAMETER);
        if (globalPermissionCacheSizeString != null) {
            try {
                globalPermissionCacheSize = Integer.parseInt(globalPermissionCacheSizeString);
            } catch (NumberFormatException nfe) {
                getLogger().log("Cache size must be an integer! Using default size", LOG_CHANNEL, Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting permission cache size for store " + getName() + " to " + globalPermissionCacheSize,
            LOG_CHANNEL,
            Logger.INFO);

        String globalLockCacheSizeString = (String) parameters.get(GLOBAL_LOCK_CACHE_SIZE_PARAMETER);
        if (globalLockCacheSizeString != null) {
            try {
                globalLockCacheSize = Integer.parseInt(globalLockCacheSizeString);
            } catch (NumberFormatException nfe) {
                getLogger().log("Cache size must be an integer! Using default size", LOG_CHANNEL, Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting lock cache size for store " + getName() + " to " + globalLockCacheSize,
            LOG_CHANNEL,
            Logger.INFO);

        String globalDescriptorsCacheSizeString = (String) parameters.get(GLOBAL_DESCRIPTORS_CACHE_SIZE_PARAMETER);
        if (globalDescriptorsCacheSizeString != null) {
            try {
                globalDescrtiptorsCacheSize = Integer.parseInt(globalDescriptorsCacheSizeString);
            } catch (NumberFormatException nfe) {
                getLogger().log("Cache size must be an integer! Using default size", LOG_CHANNEL, Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting descriptors cache size for store " + getName() + " to " + globalDescrtiptorsCacheSize,
            LOG_CHANNEL,
            Logger.INFO);

        String globalDescriptorCacheSizeString = (String) parameters.get(GLOBAL_DESCRIPTOR_CACHE_SIZE_PARAMETER);
        if (globalDescriptorCacheSizeString != null) {
            try {
                globalDescrtiptorCacheSize = Integer.parseInt(globalDescriptorCacheSizeString);
            } catch (NumberFormatException nfe) {
                getLogger().log("Cache size must be an integer! Using default size", LOG_CHANNEL, Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting descriptor cache size for store " + getName() + " to " + globalDescrtiptorCacheSize,
            LOG_CHANNEL,
            Logger.INFO);

        String enableContentCachingString = (String) parameters.get(ENABLE_CONTENT_CACHING_PARAMETER);
        if (enableContentCachingString != null) {
            contentCachingEnabled = Boolean.valueOf(enableContentCachingString).booleanValue();
        }
        getLogger().log(
            "Setting content caching for store " + getName() + " to " + contentCachingEnabled,
            LOG_CHANNEL,
            Logger.INFO);

        String globalContentCacheSizeString = (String) parameters.get(GLOBAL_CONTENT_CACHE_SIZE_PARAMETER);
        if (globalContentCacheSizeString != null) {
            try {
                globalContentCacheSize = Integer.parseInt(globalContentCacheSizeString);
            } catch (NumberFormatException nfe) {
                getLogger().log("Cache size must be an integer! Using default size", LOG_CHANNEL, Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting content cache size for store " + getName() + " to " + globalContentCacheSize,
            LOG_CHANNEL,
            Logger.INFO);

        String contentCacheBytesString = (String) parameters.get(CONTENT_CACHE_BYTES_PARAMETER);
        if (contentCacheBytesString != null) {
            try {
                contentCacheBytes = Long.parseLong(contentCacheBytesString);
            } catch (NumberFormatException nfe) {
                getLogger().log(
                    "Content cache byte size must be an integer! Using default size",
                    LOG_CHANNEL,
                    Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting content cache byte size for store " + getName() + " to " + contentCacheBytes,
            LOG_CHANNEL,
            Logger.INFO);

        String txContentCacheSizeString = (String) parameters.get(TX_CONTENT_CACHE_SIZE_PARAMETER);
        if (txContentCacheSizeString != null) {
            try {
                txContentCacheSize = Integer.parseInt(txContentCacheSizeString);
            } catch (NumberFormatException nfe) {
                getLogger().log(
                    "Content transaction cache size must be an integer! Using default size",
                    LOG_CHANNEL,
                    Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting transaction content cache size for store " + getName() + " to " + txContentCacheSize,
            LOG_CHANNEL,
            Logger.INFO);

        String txContentCacheBytesString = (String) parameters.get(TX_CONTENT_CACHE_BYTES_PARAMETER);
        if (txContentCacheBytesString != null) {
            try {
                txContentCacheBytes = Long.parseLong(txContentCacheBytesString);
            } catch (NumberFormatException nfe) {
                getLogger().log(
                    "Transaction content cache byte size must be an integer! Using default size",
                    LOG_CHANNEL,
                    Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting transaction content cache byte size for store " + getName() + " to " + txContentCacheBytes,
            LOG_CHANNEL,
            Logger.INFO);

        String maxByteSizePerEntryString = (String) parameters.get(MAX_CONTENT_BYTES_PER_ENTRY_PARAMETER);
        if (maxByteSizePerEntryString != null) {
            try {
                maxByteSizePerEntry = Long.parseLong(maxByteSizePerEntryString);
            } catch (NumberFormatException nfe) {
                getLogger().log(
                    "Maximum byte size for content cache entry must be an integer! Using default size",
                    LOG_CHANNEL,
                    Logger.WARNING);
            }
        }
        getLogger().log(
            "Setting maximum byte size for content cache entry for store " + getName() + " to " + maxByteSizePerEntry,
            LOG_CHANNEL,
            Logger.INFO);

        String cacheModeString = (String) parameters.get(CACHE_MODE_PARAMETER);
        if (cacheModeString == null) cacheModeString = DEFAULT_CACHE_MODE;
        cacheModeString = cacheModeString.trim();

        boolean noGlobalCache;
        if (cacheModeString.equals(CACHE_MODE_FULL)) {
            noGlobalCache = false;
            globalCacheOff = false;
            noGlobalCacheInTx = false;
            getLogger().log("Enabling full caching causing low isolation", LOG_CHANNEL, Logger.INFO);
        } else if (cacheModeString.equals(CACHE_MODE_LOCAL)) {
            noGlobalCache = true;
            globalCacheOff = false;
            noGlobalCacheInTx = false;
            getLogger().log("Disabling global cache to shadow store isolation and allow for clustering", LOG_CHANNEL, Logger.INFO);
        } else if (cacheModeString.equals(CACHE_MODE_NO_GLOBAL_IN_TX)) {
            globalCacheOff = false;
            noGlobalCache = false;
            noGlobalCacheInTx = true;
            getLogger().log("Disabling global cache inside transactions to shadow store isolation", LOG_CHANNEL, Logger.INFO);
        } else if (cacheModeString.equals(CACHE_MODE_OFF)) {
            globalCacheOff = true;
            noGlobalCache = true;
            noGlobalCacheInTx = true;
            getLogger().log("Disabling cache completely", LOG_CHANNEL, Logger.INFO);
        } else {
            noGlobalCache = false;
            globalCacheOff = false;
            noGlobalCacheInTx = false;
            getLogger().log("Unknown cache mode "+cacheModeString+" Using full mode instead...", LOG_CHANNEL, Logger.WARNING);
        }

        // disable global cache if desired        
        if (noGlobalCache) {
            globalObjectCacheSize = -1;
            globalPermissionCacheSize = -1;
            globalLockCacheSize = -1;
            globalDescrtiptorsCacheSize = -1;
            globalDescrtiptorCacheSize = -1;
            contentCachingEnabled = false;
        }

        init(
            globalObjectCacheSize,
            globalPermissionCacheSize,
            globalLockCacheSize,
            globalDescrtiptorsCacheSize,
            globalDescrtiptorCacheSize,
            contentCachingEnabled,
            globalContentCacheSize,
            contentCacheBytes,
            txContentCacheSize,
            txContentCacheBytes,
            maxByteSizePerEntry,
            noGlobalCacheInTx);
    }

    public void resetCache() {
        init(globalObjectCacheSize, globalPermissionCacheSize, globalLockCacheSize, globalDescrtiptorsCacheSize,
                globalDescrtiptorCacheSize, contentCachingEnabled, globalContentCacheSize, contentCacheBytes,
                txContentCacheSize, txContentCacheBytes, maxByteSizePerEntry, noGlobalCacheInTx);
    }
    
    /**
     * Removes an object from all internal caches.
     * 
     * @param key the key under which the object is stored in the caches.
     */
    public void removeObjectFromCache( Object key ) {
        if (getLogger().isEnabled(LOG_CHANNEL, Logger.DEBUG)) {
            getLogger().log( "Removing " + key + " from cache.",
            		LOG_CHANNEL,
        			Logger.DEBUG );
        }
    	if ( contentStore.cacheResults() && contentCachingEnabled ) {
    		contentCache.remove( key.toString(), "_" );
    	}
    	if ( nodeStore.cacheResults() ) {
    		objectsCache.remove( key.toString() );
    	}
    	if ( securityStore.cacheResults() ) {
    		permissionsCache.remove( key.toString() );
    	}
    	// Locks shouldn't be cached, but just in case.
    	if ( lockStore.cacheResults() ) {
    		locksCache.remove( key.toString() );
    	}
    	if ( revisionDescriptorsStore.cacheResults() ) {
    		descriptorsCache.remove( key.toString() );
    	}
    	if ( revisionDescriptorStore.cacheResults() ) {
    		descriptorCache.remove( key.toString(), "-" );
    	}
    }
    
    /**
     * Removes an object hierarchy from all internal caches.
     * 
     * @param key
     *            the key under which the object is stored in the caches.
     */
    public void removeObjectHierachyFromCache(Uri folderUri) {
        if (getLogger().isEnabled(LOG_CHANNEL, Logger.DEBUG)) {
            getLogger().log("Removing hierarchy" + folderUri 
                    + " from cache.", LOG_CHANNEL, Logger.DEBUG);
        }
        if (contentStore.cacheResults() && contentCachingEnabled) {
            contentCache.remove(folderUri.toString(), "/");
        }
        if (nodeStore.cacheResults()) {
            objectsCache.remove(folderUri.toString(), "/");
        }
        if (securityStore.cacheResults()) {
            permissionsCache.remove(folderUri.toString(), "/");
        }
        // Locks shouldn't be cached, but just in case.
        if (lockStore.cacheResults()) {
            locksCache.remove(folderUri.toString(), "/");
        }
        if (revisionDescriptorsStore.cacheResults()) {
            descriptorsCache.remove(folderUri.toString(), "/");
        }
        if (revisionDescriptorStore.cacheResults()) {
            descriptorCache.remove(folderUri.toString(), "/");
        }
    }
    
    //
    // overloaded content methods with caching
    //

    void logAccess(Uri uri, boolean write) {
        if (getLogger().isEnabled(Logger.DEBUG)) {
            getLogger().log((write ? "Writing " : "Reading ") + uri.toString(), LOG_CHANNEL,
                    Logger.DEBUG);
        }
    }
    
    public NodeRevisionContent retrieveRevisionContent(Uri uri, NodeRevisionDescriptor revisionDescriptor)
        throws ServiceAccessException, RevisionNotFoundException {
        logAccess(uri, false);
        if (contentStore.cacheResults() && contentCache != null) {
            if (isForceStoreEnlistment(uri)) {
                enlist(this);
            }
            try {
                String key = uri.toString() + "_" + revisionDescriptor.getRevisionNumber();
                Object result = contentCache.get(key);
                if (result != null) {
                    getLogger().log("Retrieving content at '" + key + "' from cache", LOG_CHANNEL, Logger.DEBUG);
                    // FIXME make a copy?! how?
                    return (NodeRevisionContent) result;
                } else {
                    NodeRevisionContent revisionContent = super.retrieveRevisionContent(uri, revisionDescriptor);
                    long size = revisionDescriptor.getContentLength();
                    // Do not add content to global cache, as this might violate repeatable read.
                    // When we reread the entry later and another thread has committed its stuff
                    // into the global cache we might get a different version on the second read.
                    contentCache.put(key, revisionContent, size);
//                    contentCache.putForRead(key, revisionContent, size);
                    // FIXME make a copy?! how?
                    return revisionContent;
                }
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            } finally {
                if (isForceStoreEnlistment(uri)) {
                    delist(this);
                }
            }
        } else {
            try {
                return super.retrieveRevisionContent(uri, revisionDescriptor);
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            }
        }
    }

    public void createRevisionContent(
        Uri uri,
        NodeRevisionDescriptor revisionDescriptor,
        NodeRevisionContent revisionContent)
        throws ServiceAccessException, RevisionAlreadyExistException {
        logAccess(uri, true);

        long size = -1L;
        if (contentStore.cacheResults() && contentCache != null) {
            size = revisionDescriptor.getContentLength();
            contentCache.precache(revisionContent, size);
        }
        try {
            super.createRevisionContent(uri, revisionDescriptor, revisionContent);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }

        if (contentStore.cacheResults() && contentCache != null) {
            enlist(this);
            try {
                String key = uri.toString() + "_" + revisionDescriptor.getRevisionNumber();
                contentCache.put(key, revisionContent, size);
            } finally {
                delist(this);
            }
        }
    }

    public void storeRevisionContent(
        Uri uri,
        NodeRevisionDescriptor revisionDescriptor,
        NodeRevisionContent revisionContent)
        throws ServiceAccessException, RevisionNotFoundException {
        logAccess(uri, true);

        long size = -1L;
        if (contentStore.cacheResults() && contentCache != null) {
            size = revisionDescriptor.getContentLength();
            contentCache.precache(revisionContent, size);
        }
        try {
            super.storeRevisionContent(uri, revisionDescriptor, revisionContent);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }

        if (contentStore.cacheResults() && contentCache != null) {
            enlist(this);
            try {
                String key = uri.toString() + "_" + revisionDescriptor.getRevisionNumber();
                contentCache.put(key, revisionContent, size);
            } finally {
                delist(this);
            }
        }
    }

    public void removeRevisionContent(Uri uri, NodeRevisionDescriptor revisionDescriptor)
        throws ServiceAccessException {
        logAccess(uri, true);
        if (contentStore.cacheResults() && contentCache != null) {
            enlist(this);
            try {
                String key = uri.toString() + "_" + revisionDescriptor.getRevisionNumber();
                contentCache.remove(key);
            } finally {
                delist(this);
            }
        }
        try {
            super.removeRevisionContent(uri, revisionDescriptor);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
    }

    //
    // overloaded retrieval methods guaranteeing cache enlistment in retrieval
    //

    public ObjectNode retrieveObject(Uri uri) throws ServiceAccessException, ObjectNotFoundException {
        logAccess(uri, false);
        if (nodeStore.cacheResults()) {
            if (isForceStoreEnlistment(uri)) {
                enlist(this);
            }
            try {
                Object tempObject = objectsCache.get(uri.toString());
                if (tempObject != null) {
                    return ((ObjectNode) tempObject).cloneObject();
                } else {
                    ObjectNode objectNode = super.retrieveObject(uri);
                    objectNode.validate(uri.toString());
                    objectsCache.put(uri.toString(), objectNode);
                    return objectNode.cloneObject();
                }
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            } finally {
                if (isForceStoreEnlistment(uri)) {
                    delist(this);
                }
            }
        } else {
            try {
                return super.retrieveObject(uri);
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            }
        }
    }

    public void storeObject(Uri uri, ObjectNode object) throws ServiceAccessException, ObjectNotFoundException {
        logAccess(uri, true);
        try {
            super.storeObject(uri, object);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (nodeStore.cacheResults()) {
            enlist(this);
            try {
                objectsCache.put(uri.toString(), object.cloneObject());
            } finally {
                delist(this);
            }
        }
    }

    public void createObject(Uri uri, ObjectNode object) throws ServiceAccessException, ObjectAlreadyExistsException {
        logAccess(uri, true);
        try {
            super.createObject(uri, object);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (nodeStore.cacheResults()) {
            enlist(this);
            try {
                
                if (object instanceof LinkNode) {
                    // add the link into the "linked" object if it is in the cache
                   String linkedUri = ((LinkNode) object).getLinkedUri();
                   ObjectNode linkedObject = (ObjectNode) objectsCache.get(linkedUri);
                   if (linkedObject != null) {
                       linkedObject.addLink((LinkNode)object);
                   }
                }
                objectsCache.put(uri.toString(), object.cloneObject());
            } finally {
                delist(this);
            }
        }
    }

    public void removeObject(Uri uri, ObjectNode object) throws ServiceAccessException, ObjectNotFoundException {
        logAccess(uri, true);
        try {
            super.removeObject(uri, object);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (nodeStore.cacheResults()) {
            enlist(this);
            try {
                if (object instanceof LinkNode) {
                    // check if the linked object is in the cache, and remove the linknode  fromits link vector
                    ObjectNode linkedObject = (ObjectNode) objectsCache.get(((LinkNode)object).getLinkedUri());
                    if (linkedObject !=null) {
                        linkedObject.removeLink((LinkNode) object);
                    }
                }
                objectsCache.remove(uri.toString());
            } finally {
                delist(this);
            }
        }
    }

    public void grantPermission(Uri uri, NodePermission permission) throws ServiceAccessException {
        logAccess(uri, true);
        try {
            super.grantPermission(uri, permission);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (securityStore.cacheResults()) {
            enlist(this);
            try {
                Vector permissionsVector = fillPermissionsCache(uri);
                Vector tempPermissions; 
                if (permissionsVector == EMPTY_VECTOR) {
                    tempPermissions = new Vector();
                } else {
                    // operate on a copy
                    tempPermissions = (Vector)permissionsVector.clone();
                }
                tempPermissions.addElement(permission);
                permissionsCache.put(uri.toString(), tempPermissions);
            } finally {
                delist(this);
            }
        }
    }

    public void revokePermission(Uri uri, NodePermission permission) throws ServiceAccessException {
        logAccess(uri, true);
        try {
            super.revokePermission(uri, permission);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (securityStore.cacheResults()) {
            enlist(this);
            try {
                Vector permissionsVector = fillPermissionsCache(uri);
                if (permissionsVector != EMPTY_VECTOR) {
                    // operate on a copy
                    Vector tempPermissions = (Vector)permissionsVector.clone();
                    tempPermissions.removeElement(permission);
                    permissionsCache.put(uri.toString(), tempPermissions);
                }
            } finally {
                delist(this);
            }

        }
    }

    public void revokePermissions(Uri uri) throws ServiceAccessException {
        logAccess(uri, true);
        try {
            super.revokePermissions(uri);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (securityStore.cacheResults()) {
            enlist(this);
            try {
                Object value = permissionsCache.get(uri.toString());
                if (value != null) {
                    permissionsCache.put(uri.toString(), EMPTY_VECTOR);
                }
            } finally {
                delist(this);
            }
        }
    }

    public Enumeration enumeratePermissions(Uri uri) throws ServiceAccessException {
        logAccess(uri, false);
        if (securityStore.cacheResults()) {
            if (isForceStoreEnlistment(uri)) {
                enlist(this);
            }
            try {
                Vector permissionsVector = fillPermissionsCache(uri);
                if (permissionsVector == EMPTY_VECTOR) {
                    return EmptyEnumeration.INSTANCE;
                } else {
                    return ((Vector) permissionsVector.clone()).elements();
                }
            } finally {
                if (isForceStoreEnlistment(uri)) {
                    delist(this);
                }
            }
        } else {
            try {
                return super.enumeratePermissions(uri);
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            }
        }
    }

    public void putLock(Uri uri, NodeLock lock) throws ServiceAccessException {
        logAccess(uri, true);
        try {
            super.putLock(uri, lock);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (lockStore.cacheResults()) {
            enlist(this);
            try {
                Vector locks = fillLocksCache(uri);
                Vector tempLocks;
                if (locks == EMPTY_VECTOR) {
                    tempLocks = new Vector();
                } else {
                    // operate on a copy
                    tempLocks = (Vector) locks.clone();
                }
                tempLocks.addElement(lock.cloneObject());
                locksCache.put(uri.toString(), tempLocks);
            } finally {
                delist(this);
            }
        }
    }

    public void renewLock(Uri uri, NodeLock lock) throws ServiceAccessException, LockTokenNotFoundException {
        logAccess(uri, true);
        try {
            super.renewLock(uri, lock);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (lockStore.cacheResults()) {
            enlist(this);
            try {
                Object value = locksCache.get(uri.toString());
                if (value != null) {
                    if (value != EMPTY_VECTOR) {
                        Vector locksVector = new Vector((Vector) value);
                        boolean wasPresent = locksVector.removeElement(lock);
                        if (!wasPresent) {
                            throw new LockTokenNotFoundException(lock);
                        }
                        locksVector.addElement(lock.cloneObject());
                        locksCache.put(uri.toString(), locksVector);
                    } else {
                        throw new LockTokenNotFoundException(lock);
                    }
                } 
            } finally {
                delist(this);
            }
        }
    }

    public void removeLock(Uri uri, NodeLock lock) throws ServiceAccessException, LockTokenNotFoundException {
        logAccess(uri, true);
        try {
            super.removeLock(uri, lock);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (lockStore.cacheResults()) {
            enlist(this);
            try {
                Object value = locksCache.get(uri.toString());
                if (value != null) {
                    if (value != EMPTY_VECTOR) {
                        Vector locksVector = new Vector((Vector) value);
                        boolean wasPresent = locksVector.removeElement(lock);
                        if (!wasPresent) {
                            throw new LockTokenNotFoundException(lock);
                        }
                        locksCache.put(uri.toString(), locksVector);
                    } else {
                        throw new LockTokenNotFoundException(lock);
                    }
                }
            } finally {
                delist(this);
            }
        }
    }

    public void killLock(Uri uri, NodeLock lock) throws ServiceAccessException, LockTokenNotFoundException {
        logAccess(uri, true);
        try {
            super.killLock(uri, lock);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (lockStore.cacheResults()) {
            enlist(this);
            try {
                Object value = locksCache.get(uri.toString());
                if (value != null) {
                    if (value != EMPTY_VECTOR) {
                        Vector locksVector = new Vector((Vector) value);
                        if (!locksVector.removeElement(lock)) {
                            throw new LockTokenNotFoundException(lock);
                        }
                        locksCache.put(uri.toString(), locksVector);
                    } else {
                        throw new LockTokenNotFoundException(lock);
                    }
                }
            } finally {
                delist(this);
            }
        }
    }

    public Enumeration enumerateLocks(Uri uri) throws ServiceAccessException {
        logAccess(uri, false);
        if (lockStore.cacheResults()) {
            if (isForceStoreEnlistment(uri)) {
                enlist(this);
            }
            try {
                Vector locks = fillLocksCache(uri);
                if (locks == EMPTY_VECTOR) {
                    return EmptyEnumeration.INSTANCE;
                } else {
                    return ((Vector) locks.clone()).elements();
                }
            } finally {
                if (isForceStoreEnlistment(uri)) {
                    delist(this);
                }
            }
        } else {
            try {
                return super.enumerateLocks(uri);
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            }
        }
    }
    
    public NodeRevisionDescriptors retrieveRevisionDescriptors(Uri uri)
        throws ServiceAccessException, RevisionDescriptorNotFoundException {
        logAccess(uri, false);
        if (revisionDescriptorsStore.cacheResults()) {
            if (isForceStoreEnlistment(uri)) {
                enlist(this);
            }
            try {
                Object tempObject = descriptorsCache.get(uri.toString());
                if (tempObject != null) {
                    return ((NodeRevisionDescriptors) tempObject).cloneObject();
                } else {
                    NodeRevisionDescriptors revisionDescriptors = super.retrieveRevisionDescriptors(uri);
                    descriptorsCache.put(uri.toString(), revisionDescriptors);
                    revisionDescriptors.validate(uri.toString());
                    return revisionDescriptors.cloneObject();
                }
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            } finally {
                if (isForceStoreEnlistment(uri)) {
                    delist(this);
                }
            }
        } else {
            try {
                return super.retrieveRevisionDescriptors(uri);
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            }
        }
    }

    public void createRevisionDescriptors(Uri uri, NodeRevisionDescriptors revisionDescriptors)
        throws ServiceAccessException {
        logAccess(uri, true);
        try {
            super.createRevisionDescriptors(uri, revisionDescriptors);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (revisionDescriptorsStore.cacheResults()) {
            enlist(this);
            try {
                descriptorsCache.put(uri.toString(), revisionDescriptors.cloneObject());
            } finally {
                delist(this);
            }
        }
    }

    public void storeRevisionDescriptors(Uri uri, NodeRevisionDescriptors revisionDescriptors)
        throws ServiceAccessException, RevisionDescriptorNotFoundException {
        logAccess(uri, true);
        try {
            super.storeRevisionDescriptors(uri, revisionDescriptors);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (revisionDescriptorsStore.cacheResults()) {
            enlist(this);
            try {
                descriptorsCache.put(uri.toString(), revisionDescriptors.cloneObject());
            } finally {
                delist(this);
            }
        }
    }

    public void removeRevisionDescriptors(Uri uri) throws ServiceAccessException {
        logAccess(uri, true);
        if (revisionDescriptorsStore.cacheResults()) {
            enlist(this);
            try {
                descriptorsCache.remove(uri.toString());
            } finally {
                delist(this);
            }
        }
        try {
            super.removeRevisionDescriptors(uri);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
    }

    public NodeRevisionDescriptor retrieveRevisionDescriptor(Uri uri, NodeRevisionNumber revisionNumber)
        throws ServiceAccessException, RevisionDescriptorNotFoundException {
        logAccess(uri, false);
        if (revisionDescriptorStore.cacheResults()) {
            if (isForceStoreEnlistment(uri)) {
                enlist(this);
            }
            try {
                Object result = descriptorCache.get(uri + "-" + revisionNumber);
                if (result != null) {
                    return ((NodeRevisionDescriptor) result).cloneObject();
                } else {
                    NodeRevisionDescriptor revisionDescriptor = super.retrieveRevisionDescriptor(uri, revisionNumber);
                    revisionDescriptor.validate();
                    descriptorCache.put(uri + "-" + revisionNumber, revisionDescriptor);
                    return revisionDescriptor.cloneObject();
                }
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            } finally {
                if (isForceStoreEnlistment(uri)) {
                    delist(this);
                }
            }
        } else {
            try {
                return super.retrieveRevisionDescriptor(uri, revisionNumber);
            } catch (ServiceAccessException sae) {
                invalidateCacheUponError(uri);
                throw sae;
            }
        }
    }

    public void createRevisionDescriptor(Uri uri, NodeRevisionDescriptor revisionDescriptor)
        throws ServiceAccessException {
        logAccess(uri, true);
        try {
            super.createRevisionDescriptor(uri, revisionDescriptor);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (revisionDescriptorStore.cacheResults()) {
            enlist(this);
            try {
                descriptorCache.put(
                    uri + "-" + revisionDescriptor.getRevisionNumber(),
                    revisionDescriptor.cloneObject());
            } finally {
                delist(this);
            }
        }
    }

    public void storeRevisionDescriptor(Uri uri, NodeRevisionDescriptor revisionDescriptor)
        throws ServiceAccessException, RevisionDescriptorNotFoundException {
        logAccess(uri, true);
        try {
            super.storeRevisionDescriptor(uri, revisionDescriptor);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (revisionDescriptorStore.cacheResults()) {
            enlist(this);
            try {
                String key = uri + "-" + revisionDescriptor.getRevisionNumber();
                descriptorCache.put(key, revisionDescriptor.cloneObject());
            } finally {
                delist(this);
            }
        }
    }

    public void removeRevisionDescriptor(Uri uri, NodeRevisionNumber number) throws ServiceAccessException {
        logAccess(uri, true);
        try {
            super.removeRevisionDescriptor(uri, number);
        } catch (ServiceAccessException sae) {
            invalidateCacheUponError(uri);
            throw sae;
        }
        if (revisionDescriptorStore.cacheResults()) {
            enlist(this);
            try {
                descriptorCache.remove(uri + "-" + number);
            } finally {
                delist(this);
            }
        }
    }

    //
    // overloaded XAResource methods for transaction support
    // 

    public void forget(Xid xid) throws XAException {
        Xid txId = XidWrapper.wrap(xid);
        activeTransactionBranch.set(null);

        objectsCache.getTxCache().forget(txId);
        permissionsCache.getTxCache().forget(txId);
        locksCache.getTxCache().forget(txId);
        descriptorsCache.getTxCache().forget(txId);
        descriptorCache.getTxCache().forget(txId);
        if (contentCache != null)
            contentCache.getTxCache().forget(txId);
    }

    public void rollback(Xid xid) throws XAException {
        Xid txId = XidWrapper.wrap(xid);
        activeTransactionBranch.set(null);

        objectsCache.getTxCache().rollback(txId);
        permissionsCache.getTxCache().rollback(txId);
        locksCache.getTxCache().rollback(txId);
        descriptorsCache.getTxCache().rollback(txId);
        descriptorCache.getTxCache().rollback(txId);
        if (contentCache != null)
            contentCache.getTxCache().rollback(txId);
    }

    public void commit(Xid xid, boolean onePhase) throws XAException {
        Xid txId = XidWrapper.wrap(xid);
        activeTransactionBranch.set(null);

        objectsCache.getTxCache().commit(txId);
        permissionsCache.getTxCache().commit(txId);
        locksCache.getTxCache().commit(txId);
        descriptorsCache.getTxCache().commit(txId);
        descriptorCache.getTxCache().commit(txId);
        if (contentCache != null)
            contentCache.getTxCache().commit(txId);
    }

    public int prepare(Xid xid) throws XAException {
        // failure is notifyed directly on transaction, so there is no need to check it here
        return XA_OK;
    }

    // FIXME needs suspend and resume here as well!
    public void start(Xid xid, int flags) throws XAException {
        if (getLogger().isEnabled(LOG_CHANNEL, Logger.DEBUG)) {
            getLogger().log(new StringBuffer(128)
                    .append("Thread ").append(Thread.currentThread())
                    .append(flags == TMNOFLAGS ? " starts"
    								: flags == TMJOIN ? " joins" : " resumes")
                    .append(" work on behalf of transaction branch ")
                    .append(xid).toString(),
    				   LOG_CHANNEL, Logger.DEBUG);
        }
        
        Xid txId = XidWrapper.wrap(xid);
        activeTransactionBranch.set(txId);
        if (flags == TMNOFLAGS || flags == TMJOIN) {
            objectsCache.getTxCache().start(txId);
            permissionsCache.getTxCache().start(txId);
            locksCache.getTxCache().start(txId);
            descriptorsCache.getTxCache().start(txId);
            descriptorCache.getTxCache().start(txId);
            if (contentCache != null)
                contentCache.getTxCache().start(txId);
        }
    }

    public void end(Xid xid, int flags) throws XAException {
        if (getLogger().isEnabled(LOG_CHANNEL, Logger.DEBUG)) {
            getLogger().log(new StringBuffer(128)
                    .append("Thread ").append(Thread.currentThread())
                    .append((flags == TMSUSPEND ? " suspends"
    								: flags == TMFAIL ? " fails" : " ends"))
                    .append(" work on behalf of transaction branch ")
                    .append(xid).toString(),
    				LOG_CHANNEL, Logger.DEBUG);
        }

        activeTransactionBranch.set(null);
    }

    //

    public String toString() {
        return getName() + "(" + getClass().getName() + ")";
    }

    protected Vector fillPermissionsCache(Uri uri) throws ServiceAccessException {
        Vector permissions = (Vector) permissionsCache.get(uri.toString());
        if (permissions != null) {
            return permissions;
        } else {
            Enumeration permissionsList = super.enumeratePermissions(uri);
            if (permissionsList.hasMoreElements()) {
                permissions = new Vector();
                while (permissionsList.hasMoreElements()) {
                    NodePermission tempPermission = (NodePermission) permissionsList.nextElement();
                    tempPermission.validate(uri.toString());
                    permissions.addElement(tempPermission);
                }
            } else {
                permissions = EMPTY_VECTOR;
            }
            permissionsCache.put(uri.toString(), permissions);
            return permissions;
        }
    }

    protected Vector fillLocksCache(Uri uri) throws ServiceAccessException {
        Vector locksVector = (Vector) locksCache.get(uri.toString());
        if (locksVector != null) {
            return locksVector;
        } else {
            Enumeration lockList = super.enumerateLocks(uri);
            if (lockList.hasMoreElements()) {
                locksVector = new Vector();
                while (lockList.hasMoreElements()) {
                    NodeLock tempLock = (NodeLock) lockList.nextElement();
                    tempLock.validate(uri.toString());
                    locksVector.addElement(tempLock);
                }
            } else {
                // store null if no locks are available
                locksVector = EMPTY_VECTOR;
            }
            locksCache.put(uri.toString(), locksVector);
            return locksVector;
        }
    }

    //

    // have it outside ctor to make it overloadable
    protected void init(
        int globalObjectCacheSize,
        int globalPermissionCacheSize,
        int globalLockCacheSize,
        int globalDescrtiptorsCacheSize,
        int globalDescrtiptorCacheSize,
        boolean contentCachingEnabled,
        int globalContentCacheSize,
        long contentCacheBytes,
        int txContentCacheSize,
        long txContentCacheBytes,
        long maxByteSizePerEntry,
        boolean noGlobalCacheInTx) {
        try {
            objectsCache = new TxCacheWrapper(globalObjectCacheSize, "object", noGlobalCacheInTx);
            permissionsCache = new TxCacheWrapper(globalPermissionCacheSize, "permission", noGlobalCacheInTx);
            locksCache = new TxCacheWrapper(globalLockCacheSize, "lock", noGlobalCacheInTx);
            descriptorsCache = new TxCacheWrapper(globalDescrtiptorsCacheSize, "descriptors", noGlobalCacheInTx);
            descriptorCache = new TxCacheWrapper(globalDescrtiptorCacheSize, "descriptor", noGlobalCacheInTx);

            if (contentCachingEnabled) {
                contentCache =
                    new TxContentCacheWrapper(
                        new ByteSizeLimitedObjectCache(
                            globalContentCacheSize,
                            txContentCacheSize,
                            contentCacheBytes,
                            txContentCacheBytes,
                            maxByteSizePerEntry,
                            getName() + ".content",
                            getLogger(),
                            noGlobalCacheInTx));
            } else {
                contentCache = null;
            }
        } catch (Error e) {
            fatalError(e);
        } catch (RuntimeException re) {
            fatalError(re);
        }
    }

    // could be out of memory etc.
    protected void fatalError(Error e) {
        getLogger().log("Fatal error: " + e, LOG_CHANNEL, Logger.CRITICAL);
        // XXX needed to guarantee, stack trace will be displayed
        getLogger().log(e, LOG_CHANNEL, Logger.CRITICAL);
        setRollbackOnly();
        throw e;
    }

    // could be  null pointer or any other programming error
    protected void fatalError(RuntimeException re) {
        getLogger().log("Fatal error: " + re, LOG_CHANNEL, Logger.CRITICAL);
        // XXX needed to guarantee, stack trace will be displayed
        getLogger().log(re, LOG_CHANNEL, Logger.CRITICAL);
        setRollbackOnly();
        throw re;
    }

    protected void invalidateCacheUponError(Uri uri) {
        // this might indicate the physical store has been changed from other sources than Slide
        // and there no longer is an object at this uri
        // so invalidate caches to at least display this in the next request
        Uri parentUri = uri.getParentUri();
        if (parentUri != null) {
            removeObjectHierachyFromCache(parentUri);
        }
    }

    protected void enlist(Service service) throws ServiceAccessException {
        boolean enlisted = false;
        try {
            Transaction transaction = namespace.getTransactionManager().getTransaction();
            if (transaction == null) {
                getLogger().log("WARNING: No active transaction", Logger.WARNING);
                return;
            }
            enlisted = transaction.enlistResource(service);

        } catch (Exception e) {
            // Something went wrong.
            setRollbackOnly();
            throw new ServiceAccessException(this, e);
        }
        if (!enlisted) {
            String exMessage = Messages.format(AbstractStore.class.getName() + ".enlistFail",
                    service);
            setRollbackOnly();
            throw new ServiceAccessException(this, exMessage);
        }
        if (getLogger().isEnabled(LOG_CHANNEL, Logger.DEBUG)) {
            String logMessage = Messages.format(AbstractStore.class.getName() + ".enlist", service);
            getLogger().log(logMessage, LOG_CHANNEL, Logger.DEBUG);
        }
    }
    
    protected class TxCacheWrapper implements ObjectCache {

        private TxLRUObjectCache txCache;

        public TxCacheWrapper(TxLRUObjectCache txCache) {
            this.txCache = txCache;
        }

        public TxCacheWrapper(int globalCacheSize, String name, boolean noGlobalCacheInTx) {
            this(new TxLRUObjectCache(globalCacheSize, getName() + "." + name, getLogger(), noGlobalCacheInTx));
        }

        public Object get(Object key) {
            if (globalCacheOff)
                return null;
            try {
                Xid txId = (Xid) activeTransactionBranch.get();
                return txCache.get(txId, key);
            } catch (Error e) {
                fatalError(e);
            } catch (RuntimeException re) {
                fatalError(re);
            }
            // XXX will never be called, as fatalError will always throw a throwable, just satisfy compiler
            return null;
        }

        public void put(Object key, Object value) {
            put(key, value, ExtendedStore.this.cacheTimeout);
        }
        
        public void put(Object key, Object value, long timeout) {
            if (globalCacheOff)
                return;
            try {
                Xid txId = (Xid) activeTransactionBranch.get();
                txCache.put(txId, key, value, timeout);
            } catch (Error e) {
                fatalError(e);
            } catch (RuntimeException re) {
                fatalError(re);
            }
        }

        public void remove(Object key) {
            if (globalCacheOff)
                return;
            try {
                Xid txId = (Xid) activeTransactionBranch.get();
                txCache.remove(txId, key);
                getLogger().log("Removing content at '" + key + "' from cache", LOG_CHANNEL, Logger.DEBUG);
            } catch (Error e) {
                fatalError(e);
            } catch (RuntimeException re) {
                fatalError(re);
            }
        }

        public void remove(Object key, String delimiter) {
            if (globalCacheOff)
                return;
            try {
                Xid txId = (Xid) activeTransactionBranch.get();
                txCache.remove(txId, key, delimiter);
                getLogger().log("Removing content at '" + key + "' with delimeter '" + delimiter + "' from cache", LOG_CHANNEL, Logger.DEBUG);
            } catch (Error e) {
                fatalError(e);
            } catch (RuntimeException re) {
                fatalError(re);
            }
        }

        public void clear() {
            try {
                txCache.clear();
            } catch (Error e) {
                fatalError(e);
            } catch (RuntimeException re) {
                fatalError(re);
            }
        }

        public TxLRUObjectCache getTxCache() {
            return txCache;
        }

    }

    protected class TxContentCacheWrapper extends TxCacheWrapper {

        public TxContentCacheWrapper(ByteSizeLimitedObjectCache txCache) {
            super(txCache);
        }

        /*
         * Ensures that if revisionContent will be cached later, it will be accessible multiple times.
         * It does so by copying its input stream into a byte array. To avoid excessively large byte 
         * arrays this is only done when revisionContent fits into the cache and will be cached later anyway.
         * <br>
         * <br>
         * <em>Caution</em>: This method must be called before any other method modifies revisionContent.
         *   
         */
        public void precache(NodeRevisionContent revisionContent, long byteSize) {
            if (globalCacheOff)
                return;
            try {
                Xid txId = (Xid) activeTransactionBranch.get();
                if (((ByteSizeLimitedObjectCache) getTxCache()).canCache(txId, byteSize)) {
                    // buffer stream content into byte array for multiple access
                    revisionContent.getContentBytes();
                }
            } catch (Error e) {
                fatalError(e);
            } catch (RuntimeException re) {
                fatalError(re);
            }
        }

        public void putForRead(Object key, NodeRevisionContent revisionContent, long byteSize) {
            putForRead(key, revisionContent, byteSize, ExtendedStore.this.cacheTimeout);
        }
        
        public void putForRead(Object key, NodeRevisionContent revisionContent, long byteSize, long timeout) {
            if (globalCacheOff)
                return;
            try {
                if (((ByteSizeLimitedObjectCache) getTxCache()).canCache(null, byteSize)) {

                    // buffer stream content into byte array for multiple access
                    revisionContent.getContentBytes();

                    ((ByteSizeLimitedObjectCache) getTxCache()).put(null, key, revisionContent, byteSize);
                    getLogger().log(
                        "Globally caching content at '" + key + "' with " + byteSize + " bytes",
                        LOG_CHANNEL,
                        Logger.DEBUG);
                }

            } catch (Error e) {
                fatalError(e);
            } catch (RuntimeException re) {
                fatalError(re);
            }
        }

        public void put(Object key, NodeRevisionContent revisionContent, long byteSize) {
            put(key, revisionContent, byteSize, ExtendedStore.this.cacheTimeout);
        }
        
        public void put(Object key, NodeRevisionContent revisionContent, long byteSize, long timeout) {
            if (globalCacheOff)
                return;
            try {
                Xid txId = (Xid) activeTransactionBranch.get();
                if (((ByteSizeLimitedObjectCache) getTxCache()).canCache(txId, byteSize)) {

                    // buffer stream content into byte array for multiple access
                    revisionContent.getContentBytes();

                    ((ByteSizeLimitedObjectCache) getTxCache()).put(txId, key, revisionContent, byteSize, timeout);
                    getLogger().log(
                        "Caching content at '" + key + "' with " + byteSize + " bytes",
                        LOG_CHANNEL,
                        Logger.DEBUG);
                } else {
                    // if we can not cache it, we need to invalidate global entry upon commit
                    getTxCache().remove(txId, key);
                }

            } catch (Error e) {
                fatalError(e);
            } catch (RuntimeException re) {
                fatalError(re);
            }
        }

    }

    public void macroDelete(Uri targetUri) throws ServiceAccessException, ObjectNotFoundException {
        enlist(this);
        try {
            removeObjectHierachyFromCache(targetUri);
        } finally {
            delist(this);
        }
        super.macroDelete(targetUri);
    }

    /**
     * @see org.apache.slide.store.MacroStore#macroCopy(Uri, Uri, boolean, boolean)
     */
    public void macroCopy(Uri sourceUri, Uri targetUri, boolean overwrite, boolean recursive)
            throws ServiceAccessException, ObjectNotFoundException, ObjectAlreadyExistsException {
        enlist(this);
        try {
            removeObjectHierachyFromCache(targetUri);
        } finally {
            delist(this);
        }
        super.macroCopy(sourceUri, targetUri, overwrite, recursive);
    }
    
    /**
     * @see org.apache.slide.store.MacroStore#macroMove(Uri, Uri, boolean)
     */
    public void macroMove(Uri sourceUri, Uri targetUri, boolean overwrite)
            throws ServiceAccessException, ObjectNotFoundException, ObjectAlreadyExistsException {
        enlist(this);
        try {
            removeObjectHierachyFromCache(sourceUri);
            removeObjectHierachyFromCache(targetUri);
        } finally {
            delist(this);
        }
        super.macroMove(sourceUri, targetUri, overwrite);
    }

}
