/*
 * 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.openjpa.kernel;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.lang.reflect.Modifier;
import java.security.AccessController;
import java.util.AbstractCollection;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;

import jakarta.transaction.Status;
import jakarta.transaction.Synchronization;

import org.apache.openjpa.conf.Compatibility;
import org.apache.openjpa.conf.OpenJPAConfiguration;
import org.apache.openjpa.datacache.DataCache;
import org.apache.openjpa.datacache.QueryCache;
import org.apache.openjpa.datacache.TypesChangedEvent;
import org.apache.openjpa.ee.ManagedRuntime;
import org.apache.openjpa.enhance.PCRegistry;
import org.apache.openjpa.enhance.PersistenceCapable;
import org.apache.openjpa.enhance.Reflection;
import org.apache.openjpa.event.LifecycleEvent;
import org.apache.openjpa.event.LifecycleEventManager;
import org.apache.openjpa.event.RemoteCommitEventManager;
import org.apache.openjpa.event.TransactionEvent;
import org.apache.openjpa.event.TransactionEventManager;
import org.apache.openjpa.instrumentation.InstrumentationManager;
import org.apache.openjpa.kernel.exps.ExpressionParser;
import org.apache.openjpa.lib.conf.Configurations;
import org.apache.openjpa.lib.instrumentation.InstrumentationLevel;
import org.apache.openjpa.lib.log.Log;
import org.apache.openjpa.lib.util.J2DoPrivHelper;
import org.apache.openjpa.lib.util.Localizer;
import org.apache.openjpa.lib.util.ReferenceHashMap;
import org.apache.openjpa.lib.util.ReferenceHashSet;
import org.apache.openjpa.lib.util.ReferenceMap;
import org.apache.openjpa.lib.util.StringUtil;
import org.apache.openjpa.lib.util.collections.AbstractReferenceMap;
import org.apache.openjpa.lib.util.collections.IteratorChain;
import org.apache.openjpa.lib.util.collections.LinkedMap;
import org.apache.openjpa.lib.util.collections.MapBackedSet;
import org.apache.openjpa.meta.ClassMetaData;
import org.apache.openjpa.meta.FieldMetaData;
import org.apache.openjpa.meta.MetaDataRepository;
import org.apache.openjpa.meta.SequenceMetaData;
import org.apache.openjpa.meta.ValueMetaData;
import org.apache.openjpa.meta.ValueStrategies;
import org.apache.openjpa.util.ApplicationIds;
import org.apache.openjpa.util.CallbackException;
import org.apache.openjpa.util.Exceptions;
import org.apache.openjpa.util.GeneralException;
import org.apache.openjpa.util.ImplHelper;
import org.apache.openjpa.util.InternalException;
import org.apache.openjpa.util.InvalidStateException;
import org.apache.openjpa.util.NoTransactionException;
import org.apache.openjpa.util.ObjectExistsException;
import org.apache.openjpa.util.ObjectId;
import org.apache.openjpa.util.ObjectNotFoundException;
import org.apache.openjpa.util.OpenJPAException;
import org.apache.openjpa.util.OptimisticException;
import org.apache.openjpa.util.RuntimeExceptionTranslator;
import org.apache.openjpa.util.StoreException;
import org.apache.openjpa.util.UnsupportedException;
import org.apache.openjpa.util.UserException;
import org.apache.openjpa.util.WrappedException;
import org.apache.openjpa.validation.ValidatingLifecycleEventManager;

/**
 * Concrete {@link Broker}. The broker handles object-level behavior,
 * but leaves all interaction with the data store to a {@link StoreManager}
 * that must be supplied at initialization.
 *
 * @author Abe White
 */
public class BrokerImpl implements Broker, FindCallbacks, Cloneable, Serializable {
    private static final long serialVersionUID = 1L;

    /**
     * Incremental flush.
     */
    protected static final int FLUSH_INC = 0;

    /**
     * Flush in preparation of commit.
     */
    protected static final int FLUSH_COMMIT = 1;

    /**
     * Flush to check consistency of cache, then immediately rollback changes.
     */
    protected static final int FLUSH_ROLLBACK = 2;

    /**
     * Run persistence-by-reachability and other flush-time operations without
     * accessing the database.
     */
    protected static final int FLUSH_LOGICAL = 3;

    static final int STATUS_INIT = 0;
    static final int STATUS_TRANSIENT = 1;
    static final int STATUS_OID_ASSIGN = 2;
    static final int STATUS_COMMIT_NEW = 3;

    private static final int FLAG_ACTIVE = 2 << 0;
    private static final int FLAG_STORE_ACTIVE = 2 << 1;
    private static final int FLAG_CLOSE_INVOKED = 2 << 2;
    private static final int FLAG_PRESTORING = 2 << 3;
    private static final int FLAG_DEREFDELETING = 2 << 4;
    private static final int FLAG_FLUSHING = 2 << 5;
    private static final int FLAG_STORE_FLUSHING = 2 << 6;
    private static final int FLAG_FLUSHED = 2 << 7;
    private static final int FLAG_FLUSH_REQUIRED = 2 << 8;
    private static final int FLAG_REMOTE_LISTENER = 2 << 9;
    private static final int FLAG_RETAINED_CONN = 2 << 10;
    private static final int FLAG_TRANS_ENDING = 2 << 11;

    private static final Object[] EMPTY_OBJECTS = new Object[0];

    private String _connectionFactoryName = "";
    private String _connectionFactory2Name = "";

    private static final Localizer _loc =
        Localizer.forPackage(BrokerImpl.class);

    //	the store manager in use; this may be a decorator such as a
    //	data cache store manager around the native store manager
    private transient DelegatingStoreManager _store = null;

    private FetchConfiguration _fc = null;
    private String _user = null;
    private String _pass = null;

    // these must be rebuilt by the facade layer during its deserialization
    private transient Log _log = null;
    private transient Compatibility _compat = null;
    private transient ManagedRuntime _runtime = null;
    private transient LockManager _lm = null;
    private transient InverseManager _im = null;
    private transient ReentrantLock _lock = null;
    private transient OpCallbacks _call = null;
    private transient RuntimeExceptionTranslator _extrans = null;
    private transient InstrumentationManager _instm = null;

    // ref to producing factory and configuration
    private transient AbstractBrokerFactory _factory = null;
    private transient OpenJPAConfiguration _conf = null;
    private transient MetaDataRepository _repo = null;

    // cache class loader associated with the broker
    private transient ClassLoader _loader = null;

    // user state
    private Synchronization _sync = null;
    private Map<Object, Object> _userObjects = null;

    // managed object caches
    private ManagedCache _cache = null;
    private TransactionalCache _transCache = null;
    private Set<StateManagerImpl> _transAdditions = null;
    private Set<StateManagerImpl> _derefCache = null;
    private Set<StateManagerImpl> _derefAdditions = null;

    // these are used for method-internal state only
    private transient Map<Object, StateManagerImpl> _loading = null;
    private transient Set<Object> _operating = null;
    private transient boolean _operatingDirty = true;

    private Set<Class<?>> _persistedClss = null;
    private Set<Class<?>> _updatedClss = null;
    private Set<Class<?>> _deletedClss = null;
    private Set<StateManagerImpl> _pending = null;
    private int findAllDepth = 0;

    // track instances that become transactional after the first savepoint
    // (the first uses the transactional cache)
    private Set<StateManagerImpl> _savepointCache = null;
    private LinkedMap _savepoints = null;
    private transient SavepointManager _spm = null;

    // track open queries and extents so we can free their resources on close
    private transient ReferenceHashSet _queries = null;
    private transient ReferenceHashSet _extents = null;

    // track operation stack depth. Transient because operations cannot
    // span serialization.
    private transient int _operationCount = 0;

    // options
    private boolean _nontransRead = false;
    private boolean _nontransWrite = false;
    private boolean _retainState = false;
    private int _autoClear = CLEAR_DATASTORE;
    private int _restoreState = RESTORE_IMMUTABLE;
    private boolean _optimistic = false;
    private boolean _ignoreChanges = false;
    private boolean _multithreaded = false;
    private boolean _managed = false;
    private boolean _syncManaged = false;
    private int _connRetainMode = CONN_RETAIN_DEMAND;
    private boolean _evictDataCache = false;
    private boolean _populateDataCache = true;
    private boolean _largeTransaction = false;
    private int _autoDetach = 0;
    private int _detachState = DETACH_LOADED;
    private boolean _detachedNew = true;
    private boolean _orderDirty = false;
    private boolean _cachePreparedQuery = true;
    private boolean _cacheFinderQuery = true;
    private boolean _suppressBatchOLELogging = false;
    private boolean _allowReferenceToSiblingContext = false;
    private boolean _postLoadOnMerge = false;

    // status
    private int _flags = 0;

    // this is not in status because it should not be serialized
    private transient boolean _isSerializing = false;

    // transient because closed brokers can't be serialized
    private transient boolean _closed = false;
    private transient RuntimeException _closedException = null;

    // event managers
    private TransactionEventManager _transEventManager = null;
    private int _transCallbackMode = 0;
    private LifecycleEventManager _lifeEventManager = null;
    private int _lifeCallbackMode = 0;

    private transient DetachManagerLite _dmLite;

    private transient boolean _initializeWasInvoked = false;
    private transient boolean _fromWriteBehindCallback = false;
    private LinkedList<FetchConfiguration> _fcs;

    // Set of supported property keys. The keys in this set correspond to bean-style setter methods
    // that can be set by reflection. The keys are not qualified by any prefix.
    private static Set<String> _supportedPropertyNames;
    static {
        _supportedPropertyNames = new HashSet<>();
        _supportedPropertyNames.addAll(Arrays.asList(new String[] {
                "AutoClear",
                "AutoDetach",
                "CacheFinderQuery",
                "CachePreparedQuery",
                "DetachedNew",
                "DetachState",
                "EvictFromDataCache",
                "IgnoreChanges",
                "LifecycleListenerCallbackMode",
                "Multithreaded",
                "NontransactionalRead",
                "NontransactionalWrite",
                "Optimistic",
                "PopulateDataCache",
                "RestoreState",
                "RetainState",
                }));
    }

    private boolean _printParameters = false;
    private static final String PRINT_PARAMETERS_CONFIG_STR = "PrintParameters";

    /**
     * Set the persistence manager's authentication. This is the first
     * method called after construction.
     *
     * @param user the username this broker represents; used when pooling
     * brokers to make sure that a request to the factory for
     * a connection with an explicit user is delegated to a suitable broker
     * @param pass the password for the above user
     */
    public void setAuthentication(String user, String pass) {
        _user = user;
        _pass = pass;
    }

    /**
     * Initialize the persistence manager. This method is called
     * automatically by the factory before use.
     *
     * @param factory the factory used to create this broker
     * @param sm a concrete StoreManager implementation to
     * handle interaction with the data store
     * @param managed the transaction mode
     * @param connMode the connection retain mode
     * @param fromDeserialization whether this call happened because of a
     * deserialization or creation of a new BrokerImpl.
     */
    public void initialize(AbstractBrokerFactory factory,
        DelegatingStoreManager sm, boolean managed, int connMode,
        boolean fromDeserialization) {
        initialize(factory, sm, managed, connMode, fromDeserialization, false);
    }

    public void initialize(AbstractBrokerFactory factory,
        DelegatingStoreManager sm, boolean managed, int connMode,
        boolean fromDeserialization, boolean fromWriteBehindCallback) {
        _fromWriteBehindCallback = fromWriteBehindCallback;
        _initializeWasInvoked = true;
        _loader = AccessController.doPrivileged(
            J2DoPrivHelper.getContextClassLoaderAction());
        if (!fromDeserialization){
            _conf = factory.getConfiguration();
            _repo = _conf.getMetaDataRepositoryInstance();
        }
        _compat = _conf.getCompatibilityInstance();
        _factory = factory;
        _log = _conf.getLog(OpenJPAConfiguration.LOG_RUNTIME);
        if (!fromDeserialization)
            _cache = new ManagedCache(this);
        // Force creation of a new operating set
        _operatingDirty = true;
        initializeOperatingSet();

        _connRetainMode = connMode;
        _managed = managed;
        if (managed)
            _runtime = _conf.getManagedRuntimeInstance();
        else
            _runtime = new LocalManagedRuntime(this);

        if (!fromDeserialization) {
            _lifeEventManager = _conf.getLifecycleEventManagerInstance();
            _transEventManager = new TransactionEventManager();
            int cmode = _repo.getMetaDataFactory().getDefaults().getCallbackMode();
            setLifecycleListenerCallbackMode(cmode);
            setTransactionListenerCallbackMode(cmode);

            // setup default options
            _factory.configureBroker(this);
        }

        // make sure to do this after configuring broker so that store manager
        // can look to broker configuration; we set both store and lock managers
        // before initializing them because they may each try to access the
        // other in their initialization
        _store = sm;
        _lm = _conf.newLockManagerInstance();
        _im = _conf.newInverseManagerInstance();
        _spm = _conf.getSavepointManagerInstance();
        _store.setContext(this);
        _lm.setContext(this);

        if (_connRetainMode == CONN_RETAIN_ALWAYS)
            retainConnection();
        if (!fromDeserialization) {
            _fc = _store.newFetchConfiguration();
            _fc.setContext(this);
        }

        _instm = _conf.getInstrumentationManagerInstance();
        if (_instm != null) {
            _instm.start(InstrumentationLevel.BROKER, this);
        }

        _dmLite = new DetachManagerLite(_conf);
        _printParameters =
            Boolean.parseBoolean(Configurations.parseProperties(_conf.getConnectionFactoryProperties()).getProperty(
                PRINT_PARAMETERS_CONFIG_STR, "false"));

        // do it before begin event otherwise transactional listeners can't use it, see @Auditable
        if (!fromDeserialization)
            _factory.addListeners(this);

        // synch with the global transaction in progress, if any
        if (_factory.syncWithManagedTransaction(this, false))
            beginInternal();
    }

    @SuppressWarnings("unchecked")
    private void initializeOperatingSet() {
        if(_operatingDirty) {
            _operatingDirty = false;
            _operating = MapBackedSet.mapBackedSet(new IdentityHashMap<>());
        }
    }

    /**
     * Gets the unmodifiable set of instances being operated.
     */
    protected Set<Object> getOperatingSet() {
    	return Collections.unmodifiableSet(_operating);
    }

    @Override
    public Object clone()
        throws CloneNotSupportedException {
        if (_initializeWasInvoked)
            throw new CloneNotSupportedException();
        else {
            return super.clone();
        }
    }

    /**
     * Create a {@link Map} to be used for the primary managed object cache.
     * Maps oids to state managers. By default, this creates a
     * {@link ReferenceMap} with soft values.
     */
    protected Map<?,?> newManagedObjectCache() {
        return new ReferenceHashMap(
                AbstractReferenceMap.ReferenceStrength.HARD, AbstractReferenceMap.ReferenceStrength.SOFT);
    }

    //////////////////////////////////
    // Implementation of StoreContext
    //////////////////////////////////

    @Override
    public Broker getBroker() {
        return this;
    }

    //////////////
    // Properties
    //////////////

    @Override
    public void setImplicitBehavior(OpCallbacks call,
        RuntimeExceptionTranslator ex) {
        if (_call == null)
            _call = call;
        if (_extrans == null)
            _extrans = ex;
    }

    RuntimeExceptionTranslator getInstanceExceptionTranslator() {
        return (_operationCount == 0) ? _extrans : null;
    }

    @Override
    public BrokerFactory getBrokerFactory() {
        return _factory;
    }

    @Override
    public OpenJPAConfiguration getConfiguration() {
        return _conf;
    }

    @Override
    public FetchConfiguration getFetchConfiguration() {
        return _fc;
    }

    @Override
    public FetchConfiguration pushFetchConfiguration() {
		return pushFetchConfiguration(null);
    }

    @Override
    public FetchConfiguration pushFetchConfiguration(FetchConfiguration fc) {
        if (_fcs == null)
            _fcs = new LinkedList<>();
        _fcs.add(_fc);
        _fc = (FetchConfiguration) (fc != null ? fc : _fc).clone();
        return _fc;
    }

    @Override
    public void popFetchConfiguration() {
        if (_fcs == null || _fcs.isEmpty())
            throw new UserException(
                    _loc.get("fetch-configuration-stack-empty"));
        _fc = _fcs.removeLast();
    }

    @Override
    public int getConnectionRetainMode() {
        return _connRetainMode;
    }

    @Override
    public boolean isManaged() {
        return _managed;
    }

    @Override
    public ManagedRuntime getManagedRuntime() {
        return _runtime;
    }

    @Override
    public ClassLoader getClassLoader() {
        return _loader;
    }

    @Override
    public DelegatingStoreManager getStoreManager() {
        return _store;
    }

    @Override
    public LockManager getLockManager() {
        return _lm;
    }

    @Override
    public InverseManager getInverseManager() {
        return _im;
    }

    @Override
    public String getConnectionUserName() {
        return _user;
    }

    @Override
    public String getConnectionPassword() {
        return _pass;
    }

    @Override
    public boolean getMultithreaded() {
        return _multithreaded;
    }

    @Override
    public void setMultithreaded(boolean multithreaded) {
        assertOpen();
        _multithreaded = multithreaded;
        if (multithreaded && _lock == null)
            _lock = new ReentrantLock();
        else if (!multithreaded)
            _lock = null;
    }

    @Override
    public boolean getIgnoreChanges() {
        return _ignoreChanges;
    }

    @Override
    public void setIgnoreChanges(boolean val) {
        assertOpen();
        _ignoreChanges = val;
    }

    @Override
    public boolean getNontransactionalRead() {
        return _nontransRead;
    }

    @Override
    public void setNontransactionalRead(boolean val) {
        assertOpen();
        if ((_flags & FLAG_PRESTORING) != 0)
            throw new UserException(_loc.get("illegal-op-in-prestore"));

        // make sure the runtime supports it
        if (val && !_conf.supportedOptions().contains
            (OpenJPAConfiguration.OPTION_NONTRANS_READ))
            throw new UnsupportedException(_loc.get
                ("nontrans-read-not-supported"));

        _nontransRead = val;
    }

    @Override
    public boolean getNontransactionalWrite() {
        return _nontransWrite;
    }

    @Override
    public void setNontransactionalWrite(boolean val) {
        assertOpen();
        if ((_flags & FLAG_PRESTORING) != 0)
            throw new UserException(_loc.get("illegal-op-in-prestore"));

        _nontransWrite = val;
    }

    @Override
    public boolean getOptimistic() {
        return _optimistic;
    }

    @Override
    public void setOptimistic(boolean val) {
        assertOpen();
        if ((_flags & FLAG_ACTIVE) != 0)
            throw new InvalidStateException(_loc.get("trans-active",
                "Optimistic"));

        // make sure the runtime supports it
        if (val && !_conf.supportedOptions().contains(OpenJPAConfiguration.OPTION_OPTIMISTIC))
            throw new UnsupportedException(_loc.get
                ("optimistic-not-supported"));

        _optimistic = val;
    }

    @Override
    public int getRestoreState() {
        return _restoreState;
    }

    @Override
    public void setRestoreState(int val) {
        assertOpen();
        if ((_flags & FLAG_ACTIVE) != 0)
            throw new InvalidStateException(_loc.get("trans-active",
                "Restore"));

        _restoreState = val;
    }

    @Override
    public boolean getRetainState() {
        return _retainState;
    }

    @Override
    public void setRetainState(boolean val) {
        assertOpen();
        if ((_flags & FLAG_PRESTORING) != 0)
            throw new UserException(_loc.get("illegal-op-in-prestore"));
        _retainState = val;
    }

    @Override
    public int getAutoClear() {
        return _autoClear;
    }

    @Override
    public void setAutoClear(int val) {
        assertOpen();
        _autoClear = val;
    }

    @Override
    public int getAutoDetach() {
        return _autoDetach;
    }
    /**
     * Sets automatic detachment option.
     * <br>
     * If the given flag contains {@link AutoDetach#DETACH_NONE} option,
     * then no other option can be specified.
     */
    @Override
    public void setAutoDetach(int detachFlags) {
         assertOpen();
         assertAutoDetachValue(detachFlags);
         _autoDetach = detachFlags;
    }

    @Override
    public void setAutoDetach(int detachFlag, boolean on) {
         assertOpen();
         assertAutoDetachValue(on ? _autoDetach | detachFlag : _autoDetach & ~detachFlag);
         if (on)
             _autoDetach |= detachFlag;
         else
             _autoDetach &= ~detachFlag;
    }

    @Override
    public int getDetachState() {
        return _detachState;
    }

    @Override
    public void setDetachState(int mode) {
        assertOpen();
        _detachState = mode;
    }

    @Override
    public boolean isDetachedNew() {
        return _detachedNew;
    }

    @Override
    public void setDetachedNew(boolean isNew) {
        assertOpen();
        _detachedNew = isNew;
    }

    @Override
    public boolean getSyncWithManagedTransactions() {
        return _syncManaged;
    }

    @Override
    public void setSyncWithManagedTransactions(boolean sync) {
        assertOpen();
        _syncManaged = sync;
    }

    @Override
    public boolean getEvictFromDataCache() {
        return _evictDataCache;
    }

    @Override
    public void setEvictFromDataCache(boolean evict) {
        assertOpen();
        _evictDataCache = evict;
    }

    @Override
    public boolean getPopulateDataCache() {
        return _populateDataCache;
    }

    @Override
    public void setPopulateDataCache(boolean cache) {
        assertOpen();
        _populateDataCache = cache;
    }

    @Override
    public boolean isTrackChangesByType() {
        return _largeTransaction;
    }

    @Override
    public void setTrackChangesByType(boolean largeTransaction) {
        assertOpen();
        _largeTransaction = largeTransaction;
    }

    @Override
    public Object getUserObject(Object key) {
        beginOperation(false);
        try {
            return (_userObjects == null) ? null : _userObjects.get(key);
        } finally {
            endOperation();
        }
    }

    @Override
    public Object putUserObject(Object key, Object val) {
        beginOperation(false);
        try {
            if (val == null)
                return (_userObjects == null) ? null : _userObjects.remove(key);

            if (_userObjects == null)
                _userObjects = new HashMap<>();
            return _userObjects.put(key, val);
        } finally {
            endOperation();
        }
    }

    /**
     * Get current configuration property values used by this instance.
     * This values are combination of the current configuration values
     * overwritten by values maintained by this instance such as
     * Optimistic flag.
     */
    @Override
    public Map<String, Object> getProperties() {
        Map<String, Object> props = _conf.toProperties(true);
        for (String s : _supportedPropertyNames) {
            final Object value = Reflection.getValue(this, s, !"CacheFinderQuery".equals(s));
            if (value != null) {
                props.put("openjpa." + s, value);
            }
        }
        return props;
    }

    /**
     * Gets the property names that can be used to corresponding setter methods of this receiver
     * to set its value.
     */
    @Override
    public Set<String> getSupportedProperties() {
        Set<String> keys = _conf.getPropertyKeys();
        for (String s : _supportedPropertyNames)
            keys.add("openjpa." + s);
        return keys;
    }

    // ////////
    // Events
    // ////////

    @Override
    public void addLifecycleListener(Object listener, Class[] classes) {
        beginOperation(false);
        try {
            _lifeEventManager.addListener(listener, classes);
        } finally {
            endOperation();
        }
    }

    @Override
    public void removeLifecycleListener(Object listener) {
        beginOperation(false);
        try {
            _lifeEventManager.removeListener(listener);
        } finally {
            endOperation();
        }
    }

    @Override
    public int getLifecycleListenerCallbackMode() {
        return _lifeCallbackMode;
    }

    @Override
    public void setLifecycleListenerCallbackMode(int mode) {
        beginOperation(false);
        try {
            _lifeCallbackMode = mode;
            _lifeEventManager.setFailFast((mode & CALLBACK_FAIL_FAST) != 0);
        } finally {
            endOperation();
        }
    }

    /**
     * Give state managers access to the lifecycle event manager.
     */
    @Override
    public LifecycleEventManager getLifecycleEventManager() {
        return _lifeEventManager;
    }

    /**
     * Fire given lifecycle event, handling any exceptions appropriately.
     *
     * @return whether events are being processed at this time
     */
    boolean fireLifecycleEvent(Object src, Object related, ClassMetaData meta,
        int eventType) {
        if (_lifeEventManager == null)
            return false;
        if (!_lifeEventManager.isActive(meta))
            return false;

        lock();
        Exception[] exs;
        try {
            exs = _lifeEventManager.fireEvent(src, related, meta, eventType);
        } finally {
            unlock();
        }
        handleCallbackExceptions(exs, _lifeCallbackMode);
        return true;
    }

    /**
     * Take actions on callback exceptions depending on callback mode.
     */
    private void handleCallbackExceptions(Exception[] exceps, int mode) {
        if (exceps.length == 0 || (mode & CALLBACK_IGNORE) != 0)
            return;

        OpenJPAException ce;
        if (exceps.length == 1) {
            // If the exception is already a wrapped exception throw the
            // exception instead of wrapping it with a callback exception
            if (exceps[0] instanceof WrappedException)
                throw (WrappedException)exceps[0];
            else
                ce = new CallbackException(exceps[0]);
        } else {
            ce = new CallbackException(_loc.get("callback-err")).
                setNestedThrowables(exceps);
        }
        if ((mode & CALLBACK_ROLLBACK) != 0 && (_flags & FLAG_ACTIVE) != 0) {
            ce.setFatal(true);
            setRollbackOnlyInternal(ce);
        }
        if ((mode & CALLBACK_LOG) != 0 && _log.isWarnEnabled())
            _log.warn(ce);
        if ((mode & CALLBACK_RETHROW) != 0)
            throw ce;
    }

    @Override
    public void addTransactionListener(Object tl) {
        beginOperation(false);
        try {
            _transEventManager.addListener(tl);
            if (tl instanceof RemoteCommitEventManager)
                _flags |= FLAG_REMOTE_LISTENER;
        } finally {
            endOperation();
        }
    }

    @Override
    public void removeTransactionListener(Object tl) {
        beginOperation(false);
        try {
            if (_transEventManager.removeListener(tl)
                && (tl instanceof RemoteCommitEventManager))
                _flags &= ~FLAG_REMOTE_LISTENER;
        } finally {
            endOperation();
        }
    }

    @Override
    public Collection<Object> getTransactionListeners() {
        return _transEventManager.getListeners();
    }

    @Override
    public int getTransactionListenerCallbackMode() {
        return _transCallbackMode;
    }

    @Override
    public void setTransactionListenerCallbackMode(int mode) {
        beginOperation(false);
        try {
            _transCallbackMode = mode;
            _transEventManager.setFailFast((mode & CALLBACK_FAIL_FAST) != 0);
        } finally {
            endOperation();
        }
    }

    /**
     * Fire given transaction event, handling any exceptions appropriately.
     */
    private void fireTransactionEvent(TransactionEvent trans) {
        if (_transEventManager != null && _transEventManager.hasListeners())
            handleCallbackExceptions(_transEventManager.fireEvent(trans),
                _transCallbackMode);
    }

    /**
     * Set whether this Broker will generate verbose optimistic lock exceptions when batching
     * operations. Defaults to true.
     */
    public void setSuppressBatchOLELogging(boolean b) {
        _suppressBatchOLELogging = b;
    }

    /**
     * Return whether this Broker will generate verbose optimistic lock exceptions when batching
     * operations.
     */
    public boolean getSuppressBatchOLELogging() {
        return _suppressBatchOLELogging;
    }
    ///////////
    // Lookups
    ///////////

    @Override
    public Object find(Object oid, boolean validate, FindCallbacks call) {
        int flags = OID_COPY | OID_ALLOW_NEW | OID_NODELETED;
        if (!validate)
            flags |= OID_NOVALIDATE;
        return find(oid, _fc, null, null, flags, call);
    }

    @Override
    public Object find(Object oid, FetchConfiguration fetch, BitSet exclude,
        Object edata, int flags) {
        return find(oid, fetch, exclude, edata, flags, null);
    }

    /**
     * Internal finder.
     */
    protected Object find(Object oid, FetchConfiguration fetch, BitSet exclude,
        Object edata, int flags, FindCallbacks call) {
        if (call == null)
            call = this;
        oid = call.processArgument(oid);
        if (oid == null) {
            if ((flags & OID_NOVALIDATE) == 0)
                throw new ObjectNotFoundException(_loc.get("null-oid"));
            return call.processReturn(oid, null);
        }
        if (fetch == null)
            fetch = _fc;

        beginOperation(true);
        try {
            assertNontransactionalRead();

            // cached instance?
            StateManagerImpl sm = getStateManagerImplById(oid,
                (flags & OID_ALLOW_NEW) != 0 || hasFlushed());
            if (sm != null) {
