ChangeBroker.java
/*
** Module : ChangeBroker.java
** Abstract : Clearinghouse for DMO property change events
**
** Copyright (c) 2004-2024, Golden Code Development Corporation.
**
** -#- -I- --Date-- --JPRM-- ----------------------------Description----------------------------
** 001 ECF 20060719 @28114 Created initial version. Brokers notification
** of DMO state changes to registered listeners.
** 002 ECF 20060722 @28134 Fixed listener registration defect. Listeners
** registered at inner scopes were hiding those
** registered for the same DMO types at outer
** scopes.
** 003 ECF 20060802 @28348 Make addListener() tolerant of multiple calls
** within the same scope. Listeners are now
** stored in a HashSet, so duplicates are
** prevented.
** 004 ECF 20060918 @29665 Added forcePendingFlush() method. Forces a
** pending session flush for a particular DMO
** type.
** 005 ECF 20060919 @29696 Fixed handling of pending flush state. The
** pending flush state was being tracked
** globally, across all databases, rather than
** per database. This led to errors in proper
** detection of whether a session flush was
** needed, because the state was being cleared
** too aggressively.
** 006 ECF 20061010 @30308 Made changes to accommodate auto-commit temp
** tables. These were primarily side effects of
** changes to the Persistence class to support
** this feature.
** 007 ECF 20070306 @32286 Added findDirty() implementation to session
** interceptor. This allows us to optimize dirty
** checking in some cases by avoiding the
** default dirty check comparisons which
** Hibernate performs.
** 008 ECF 20070322 @32603 Added forcePendingFlush() variant. Allows
** forcing a pending flush when a RecordBuffer
** instance is not available.
** 009 ECF 20070416 @33019 Integrated user interrupt handling.
** 010 ECF 20080103 @36869 Minor change in stateChanged(). Don't force
** a pending flush for a record that is still
** transient. Also integrated generics.
** 011 ECF 20080227 @37222 Fixed SessionInterceptor.findDirty(). Was
** improperly short-circuiting Hiberate dirty
** check in non-legacy mode.
** 012 ECF 20080314 @37471 Added debug logic. Number of DMO dirty checks
** performed since last flush are now tracked
** and reported when trace level logging is
** enabled.
** 013 ECF 20080403 @38138 Use entity name instead of DMO class to test
** if flush is required.
** 014 ECF 20080822 @39554 Modified stateChanged() to accept an event
** type parameter. Uses inner enum Type from
** RecordChangeEvent.
** 015 CA 20080815 @39452 Support API change in DBUtils.
** 016 SVL 20081024 @40217 Added addListener(.., int depth) function.
** 017 ECF 20090212 @41294 Implemented eviction of unused DMOs after a
** session flush. Temp table DMOs may be left
** behind after a bulk delete, causing more
** flushing effort than is necessary. We now
** evict these objects after a flush, if they
** are not in use.
** 018 ECF 20090610 @42971 Minor change to SessionInterceptor.postFlush().
** Log database info in trace mode.
** 019 ECF 20090716 @43214 Minor change to minimize memory used by listener
** collections.
** 020 CA 20131013 Added no-op scopeDeleted() method, required by the changes in
** Scopeable interface.
** 021 CA 20140513 Added a weight for the context-local var, to ensure predetermined
** order during context reset.
** 022 OM 20150112 Added support for storing listeners in the GLOBAl scope for
** persistent RUN procedure.
** 023 OM 20150213 Prevented a NPE in removeFromGlobal() when listeners is empty.
** 024 ECF 20150906 Changed SessionInterceptor to match changes in BufferManager API.
** 025 ECF 20160607 Reimplemented listener registration and notification to be more
** efficient. Reduced session flushing when processing state changes.
** 026 ECF 20160615 Added support needed for custom dirty entity strategy. Replaced
** Apache commons logging with J2SE logging.
** 027 ECF 20160907 Improved session flush management for better performance.
** 028 CA 20161010 Fixed stateChanged - all listeners, from all scopes, must be notified
** for a certain DMO interface.
** 029 ECF 20180311 Fixed listener registration. Cleanup interceptor state after
** transactions on commit as well rollback.
** 030 ECF 20180709 Refactored pending eviction processing.
** 031 ECF 20190302 Fixed insertions of listeners into scopes between the global and the
** current scope.
** 032 CA 20190326 Implemented ProcedureHelper and TransactionHelper instead of direct
** usage of the static API. This allows the elimination of context
** local usage in ProcedureManager and TransactionManager.
** 033 OM 20190411 Removed compile-time ambiguity.
** 034 OM 20190624 Small javadoc update.
** 035 ECF 20200419 Removed Hibernate dependencies.
** CA 20200722 Performance improvements.
** 036 CA 20200924 Small performance optimization - do not use get in a map usage.
** CA 20200930 Use a lambda instead of an inner class for the scopeable factory.
** CA 20201003 Allow dictionaries with an identity map.
** Use an identity HashSet where possible.
** CA 20220309 Added isQueried(), which checks if a temp-table has any listeners registered for
** events which are a P2JQuery.
** CA 20220901 Refactored scope notification support: ScopeableFactory was removed, and the
** registration is now specific to each type of scopeable. For each case, the block
** will be registered for scope support (for that particular scopeable) only when
** the scopeable is 'active' (i.e. unnamed streams or accumulators are used). This
** allows a lazy registration of scopeables, to avoid the unnecessary overhead of
** processing all the scopeables for each and every block.
** CA 20220906 Extracted all BufferManager state which requires notification for
** 'isImportantBlock' to an external TxWrapper class (including the private class
** here), to allow the scopeable BufferManager state related only to buffers (and
** not transaction related) to be registered for scope notifications in a lazy
** manner, when buffers are accessed/used/opened.
** CA 20221010 Performance improvements - avoid the ProcedureData lookup, by keeping a parallel
** stack of this data for THIS-PROCEDURE. Refs #6826
** 037 GBB 20230512 Logging methods replaced by CentralLogger/ConversionStatus.
** 038 AL2 20230614 Do local context look-up for the bufferManager only once per ChangeBroker.
** 039 CA 20231026 A persistent procedure with a pending delete is no longer persistent, so
** scopeDeleted needs to be executed.
** CA 20231031 Added 'BlockDefinition' parameter to 'scopeStart'.
** 040 CA 20240809 Avoid iterator usage, for performance improvement.
** 041 AB2 20250415 Changed stateChanged to notify all matching listeners from all scopes. See #9722.
** 042 RNC 20250423 Reverted commit with entry 041. See #9914.
** RNC 20250428 Assure local to global linking for fresh listener sets.
*/
/*
** This program is free software: you can redistribute it and/or modify
** it under the terms of the GNU Affero General Public License as
** published by the Free Software Foundation, either version 3 of the
** License, or (at your option) any later version.
**
** This program is distributed in the hope that it will be useful,
** but WITHOUT ANY WARRANTY; without even the implied warranty of
** MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
** GNU Affero General Public License for more details.
**
** You may find a copy of the GNU Affero GPL version 3 at the following
** location: https://www.gnu.org/licenses/agpl-3.0.en.html
**
** Additional terms under GNU Affero GPL version 3 section 7:
**
** Under Section 7 of the GNU Affero GPL version 3, the following additional
** terms apply to the works covered under the License. These additional terms
** are non-permissive additional terms allowed under Section 7 of the GNU
** Affero GPL version 3 and may not be removed by you.
**
** 0. Attribution Requirement.
**
** You must preserve all legal notices or author attributions in the covered
** work or Appropriate Legal Notices displayed by works containing the covered
** work. You may not remove from the covered work any author or developer
** credit already included within the covered work.
**
** 1. No License To Use Trademarks.
**
** This license does not grant any license or rights to use the trademarks
** Golden Code, FWD, any Golden Code or FWD logo, or any other trademarks
** of Golden Code Development Corporation. You are not authorized to use the
** name Golden Code, FWD, or the names of any author or contributor, for
** publicity purposes without written authorization.
**
** 2. No Misrepresentation of Affiliation.
**
** You may not represent yourself as Golden Code Development Corporation or FWD.
**
** You may not represent yourself for publicity purposes as associated with
** Golden Code Development Corporation, FWD, or any author or contributor to
** the covered work, without written authorization.
**
** 3. No Misrepresentation of Source or Origin.
**
** You may not represent the covered work as solely your work. All modified
** versions of the covered work must be marked in a reasonable way to make it
** clear that the modified work is not originating from Golden Code Development
** Corporation or FWD. All modified versions must contain the notices of
** attribution required in this license.
*/
package com.goldencode.p2j.persist;
import java.lang.ref.*;
import java.util.*;
import java.util.function.*;
import java.util.logging.*;
import com.goldencode.p2j.persist.event.*;
import com.goldencode.p2j.security.*;
import com.goldencode.p2j.util.*;
import com.goldencode.p2j.util.logging.*;
/**
* A clearing house which notifies interested parties of changes made to
* active records within the current client context. These notifications
* are used for various purposes:
* <ul>
* <li>to enable {@link RecordBuffer} to defer creating a placeholder
* snapshot as long as possible;
* <li>to permit {@link Persistence} query methods to disable auto-flushing
* of the active Hibernate session when possible;
* <li>to allow {@link AdaptiveQuery} to invalidate its current result set
* when necessary.
* </ul>
* <p>
* {@link com.goldencode.p2j.persist.event.RecordChangeListener
* RecordChangeListener} objects register to receive notifications for one or
* more DMO types, using the {@link #addListener} method. A listener will
* only receive notifications for changes to DMOs of the types for which they
* registered. Beyond this, no other filtering is performed. Listeners are
* maintained in a scoped dictionary, so there is never a need to remove a
* listener explicitly; when the scope in which the listener was registered
* is closed, it is released automatically.
* <p>
* Change notifications are driven by the {@link #stateChanged} method. It
* is anticipated that this only will be invoked by <code>RecordBuffer</code>
* objects, as user code calls setter methods on DMO proxies. A state change
* causes a {@link com.goldencode.p2j.persist.event.RecordChangeEvent
* RecordChangeEvent} to be broadcast to all listeners registered against the
* type of DMO which was modified. "Type" in this sense corresponds to the
* DMO interface.
* <p>
* Because it relies upon the <code>RecordBuffer</code> as the source for all
* state change events, <code>ChangeBroker</code> is not useful in non-legacy
* client contexts (i.e., where external code accesses the services of the
* <code>Persistence</code> class directly). For the most part, this class
* is transparent for these contexts. One exception is that the auto-flush
* optimization technique described above would be harmful if allowed to
* proceed naturally, because flushing would always be disabled since state
* changes would never be reported, thus it would always seem flushing is not
* required. To prevent this problem, this class detects when it is not in
* legacy mode and disables this optimization accordingly in such cases.
*/
final class ChangeBroker
implements Scopeable
{
/** Logger */
private static final CentralLogger LOG = CentralLogger.get(ChangeBroker.class.getName());
/** Context local instance of this class */
private static final ContextContainer context = new ContextContainer();
/** Sets of listeners which have registered interest in DMO property changes */
private final ScopedDictionary<Class<?>, ListenerSet> listeners = new ScopedDictionary<>();
/** Map of saved listeners, per persistent procedure referent. */
private final Map<Object, Map<Class<?>, ListenerSet>> procListeners = new IdentityHashMap<>();
/** Helper to use the ProcedureManager without any context local lookups */
private ProcedureManager.ProcedureHelper pm;
/** Helper to use the TM without any context local lookups */
private TransactionManager.TransactionHelper tm;
/** Helper to use the buffer manager without any context local lookups */
private BufferManager bufferManager;
/**
* This class can only be constructed privately.
*/
private ChangeBroker()
{
listeners.setIdentityKeys(true);
}
/**
* Retrieve the context-local instance of this class, instantiating it
* first if necessary.
*
* @return Context-local instance of this class.
*/
static ChangeBroker get()
{
return context.obtain();
}
/**
* Add a scope to the listeners dictionary whenever a new runtime scope
* opens.
*
* @param block
* The explicit block definition which required this notification.
*/
@Override
public void scopeStart(BlockDefinition block)
{
listeners.addScope(null);
}
/**
* Pop the current scope from the listeners dictionary whenever a runtime
* scope is closed.
*/
public void scopeFinished()
{
Object referent = null;
boolean saveForProc = false;
if (!tm.isGlobalBlock())
{
BlockType bt = tm.getBlockType();
// if we have a persistent procedure, save the data
if (bt == BlockType.EXTERNAL_PROC && pm.isThisProcedurePersistent())
{
referent = pm._thisProcedure();
saveForProc = true;
}
}
Map<Class<?>, ListenerSet> current = listeners.getDictionaryAtScope(0, false);
if (current != null && !current.isEmpty())
{
current.forEach((cls, set) -> set.unlink());
if (saveForProc)
{
procListeners.put(referent, current);
}
}
listeners.deleteScope();
}
/**
* Notification that the instantiating procedure where the scope was opened was deleted.
* <p>
* This implementation is a no-op.
*/
@Override
public void scopeDeleted()
{
Object referent = pm.getProcessedProcedure();
if (referent == null || pm._isPersistentAndNotPendingDelete(referent))
{
// if no referent or procedure is still persistent, do not clean up yet
return;
}
// this needs to be executed only when the procedure gets deleted.
procListeners.remove(referent);
}
/**
* Get the {@link ScopeId} for the instance.
*
* @return {@link ScopeId#CHANGE_BROKER}.
*/
@Override
public ScopeId getScopeId()
{
return ScopeId.CHANGE_BROKER;
}
/**
* Register a listener to receive DMO property change notifications. The
* listener will receive notifications of any changes to DMOs of the types
* for which it is registered, as they occur in this scope or in scopes
* nested within this scope. The listener is deregistered automatically
* when the scope ends.
* <p>
* This method is tolerant of being called multiple times for the same
* listener in the same scope; the listener will only be registered once,
* and therefore will receive only one notification per distinct event.
*
* @param listener
* Listener object to be added.
*/
void addListener(RecordChangeListener listener)
{
BufferManager.registerScopeable(bufferManager);
addListener(listener, listeners.size());
}
/**
* Register a listener to receive DMO property change notifications. The
* registration is performed in the current scope and the outer scopes,
* which number is specified by the <code>depth</code> parameter. The
* listener will receive notifications of any changes to DMOs of the types
* for which it is registered, as they occur in the specified scopes or in
* the nested ones. The listener is deregistered automatically
* when the scopes end.
* <p>
* When we call this function with a <code>depth</code> less than the
* number of innermost scope, we actually want to register it in some outer
* scope, but since registrations at inner scopes will hide those at outer
* scopes during a lookup, we should register it in the several scopes
* if needed.
* <p>
* This method is tolerant of being called multiple times for the same
* listener in the same scope; the listener will only be registered once,
* and therefore will receive only one notification per distinct event.
* <p>
* If the buffer is opened in a procedure executed persistent, the buffer has to survive
* when the procedure code is finished until the persistent procedure is deleted. In this
* case the buffer is stored in the GLOBAL scope (depth = 0).
*
* @param listener
* Listener object to be added.
*
* @param depth
* 1-based number of the most outer scope in which we want to register the listener
* (starting from the very outermost scope). Use 0 for GLOBAL scope when the buffer
* is opened in a persistent procedure.
*/
void addListener(RecordChangeListener listener, int depth)
{
BufferManager.registerScopeable(bufferManager);
boolean global = (depth == 0);
Iterator<RecordBuffer> iter = listener.recordBuffers();
while (iter.hasNext())
{
RecordBuffer buffer = iter.next();
Class<? extends DataModelObject> dmoIface = buffer.getDMOInterface();
ListenerSet set;
if (global)
{
// insert to global scope
set = listeners.getValueAtScope(dmoIface, -1);
boolean createSet = set == null;
if (createSet)
{
set = new ListenerSet();
if (set.nextInner == null && listeners.size() > 1)
{
ListenerSet nextInner = listeners.reverseLookup(dmoIface, 1);
if (nextInner != null)
{
nextInner.link(set);
}
}
listeners.addEntry(true, dmoIface, set);
}
set.add(listener);
}
else
{
// add the buffer to the specified scope of listeners
int scope = listeners.size() - depth;
set = listeners.getValueAtScope(dmoIface, scope);
boolean createSet = set == null;
if (createSet)
{
set = new ListenerSet();
if (set.nextOuter == null)
{
ListenerSet nextOuter = depth > 1 ? listeners.lookup(dmoIface, depth - 1)
: listeners.getValueAtScope(dmoIface, -1);
if (nextOuter != null)
{
set.link(nextOuter);
}
}
listeners.addEntryAt(depth - 1, dmoIface, set);
}
set.add(listener);
}
}
}
/**
* Remove a listener from global scope. This is only called for listeners whose buffers belong
* to a procedure that was executed as persistent. In this case, opening the buffer registered
* it to the global scope. This buffer is available as long as the procedure survives. When the
* procedure is removed / deleted the buffer has to be removed as well.
* <p>
* If this buffer was not open in global scope (from a persistent procedure) this is a no-op and
* the returned value is <code>false</code>.
*
* @param listener
* The listener to be removed.
*
* @return <code>true</code> if the listener was successfully removed from global scope.
*/
boolean removeFromGlobal(RecordChangeListener listener)
{
boolean ret = false;
Iterator<RecordBuffer> iter = listener.recordBuffers();
while (iter.hasNext())
{
RecordBuffer buffer = iter.next();
Class<? extends DataModelObject> dmoIface = buffer.getDMOInterface();
// get the global scope for the dmoIface:
ListenerSet set = listeners.size() == 0 ? null : listeners.getValueAtScope(dmoIface, -1);
if (set != null)
{
ret |= set.remove(listener);
// do the cleanup:
if (set.isEmpty())
{
// if set of listeners is empty, remove from global scope
listeners.removeEntryAtScope(dmoIface, listeners.size() - 1);
set.unlink();
}
}
}
return ret;
}
/**
* Remove the specified listener from all scopes.
*
* @param listener
* The listener to be removed.
*/
void removeListener(RecordChangeListener listener)
{
Iterator<RecordBuffer> iter = listener.recordBuffers();
while (iter.hasNext())
{
RecordBuffer buffer = iter.next();
Class<? extends DataModelObject> dmoIface = buffer.getDMOInterface();
listeners.apply(dmoIface, (dictionary) ->
{
ListenerSet set = dictionary.get(dmoIface);
if (set.remove(listener))
{
if (set.isEmpty())
{
set.unlink();
dictionary.remove(dmoIface);
}
return true;
}
return false;
});
}
}
/**
* Check if there are any listeners registered with a query, for this temp-table, identified by its DMO
* interface and the multiplex ID.
*
* @param dmoIface
* Data model object interface.
* @param multiplexID
* The multiplex ID for the temp-table reclaiming this key.
*
* @return See above.
*/
boolean isQueried(Class<? extends DataModelObject> dmoIface, Integer multiplexID)
{
Function<ListenerSet, Boolean> queried = (lset) ->
{
for (RecordChangeListener listener : lset)
{
if (!(listener instanceof P2JQuery))
{
continue;
}
Iterator<RecordBuffer> iter = listener.recordBuffers();
while (iter.hasNext())
{
RecordBuffer buf = iter.next();
if (multiplexID.equals(buf.getMultiplexID()))
{
return true;
}
}
}
return false;
};
ListenerSet set = listeners.lookup(dmoIface);
if (set != null)
{
if (queried.apply(set))
{
return true;
}
}
for (Map.Entry<Object, Map<Class<?>, ListenerSet>> entry : procListeners.entrySet())
{
Map<Class<?>, ListenerSet> procScope = entry.getValue();
ListenerSet procSet = procScope.get(dmoIface);
if (procSet != null && queried.apply(procSet))
{
return true;
}
}
return false;
}
/**
* Broadcast a notification that a DMO's state has changed. All listeners
* registered to receive events about DMOs of the type managed by
* <code>buffer</code> will receive the event.
*
* @param type
* Event type: insert, update, or delete.
* @param buffer
* Record buffer which originated the event.
* @param dmo
* <b>Post-modification</b> state of the affected DMO (in the
* case where the modification is the deletion of the DMO from
* the database, this will be the object instance which was
* deleted).
* @param snapshot
* A deep copy of the DMO which reflects the state of the DMO as
* it was first set into the record buffer. For a newly created
* record, this will contain the primary key assigned to the
* record. All other properties will be set to their default
* values.
* @param properties
* A map whose keys are the names of those properties which have
* been modified and whose values are lists of the indexes at
* which those values are stored (indexed properties only). For
* a simple property, the matching value will be
* <code>null</code>.
*
* @throws PersistenceException
* if any error occurs within a listener while processing a
* change event. Exceptions will be deferred until all listeners
* have had a chance to process the event. If more than one
* listener throws an exception, only the first will be rethrown
* by this method; however, all will be logged.
*/
void stateChanged(RecordChangeEvent.Type type,
RecordBuffer buffer,
Record dmo,
Record snapshot,
Map<String, List<Integer>> properties)
throws PersistenceException
{
RecordChangeEvent event = null;
if (dmo != null)
{
event = new RecordChangeEvent(type, buffer, dmo, snapshot, properties);
}
Class<? extends DataModelObject> dmoIface = buffer.getDMOInterface();
ListenerSet set = listeners.lookup(dmoIface);
if (set == null)
{
return;
}
if (event == null)
{
event = new RecordChangeEvent(type, buffer, dmo, snapshot, properties);
}
Exception error = null;
// Notify all registered listeners.
for (RecordChangeListener listener : set)
{
try
{
listener.stateChanged(event);
}
catch (Exception exc)
{
DBUtils.handleException(buffer.getDatabase(), exc);
// Can only throw the first exception caught, but we want to
// give all listeners a chance to process the event.
if (error == null)
{
error = exc;
}
// Log it regardless.
LOG.log(Level.SEVERE, exc.getMessage(), exc);
}
}
// process all listeners saved for persistent procedures, if they have registered for this
// DMO interface
for (Map.Entry<Object, Map<Class<?>, ListenerSet>> entry : procListeners.entrySet())
{
Map<Class<?>, ListenerSet> procScope = entry.getValue();
ListenerSet procSet = procScope.get(dmoIface);
if (procSet != null)
{
for (RecordChangeListener listener : procSet)
{
try
{
listener.stateChanged(event);
}
catch (Exception exc)
{
DBUtils.handleException(buffer.getDatabase(), exc);
// Can only throw the first exception caught, but we want to
// give all listeners a chance to process the event.
if (error == null)
{
error = exc;
}
// Log it regardless.
LOG.log(Level.SEVERE, exc.getMessage(), exc);
}
}
}
}
if (error != null)
{
throw new PersistenceException(error);
}
}
/**
* An ordered set of listeners (iterates in the order added to the set), which optionally
* links to other sets of listeners up and down the stack of open scopes.
* <p>
* When the set is iterated, it first walks its links to other listener sets to the outermost
* in scope and iterates those first. It then walks reverse (weak) links back to more deeply
* nested scopes and iterates listeners in those sets, and so on. Duplicate listeners are
* only visited once.
*/
private static class ListenerSet
implements Iterable<RecordChangeListener>
{
/** Entities which have registered interest in DMO property changes */
private final LinkedHashSet<RecordChangeListener> interested = new LinkedHashSet<>();
/** Next set of listeners registered for the same events in an outer scope */
private ListenerSet nextOuter = null;
/** Next set of listeners registered for the same events in an inner scope */
private WeakReference<ListenerSet> nextInner = null;
/**
* Get an iterator which will iterate over outermost to innermost listeners in all outer
* listener sets linked to and including this listener set.
*
* @return Iterator as described above.
*/
@Override
public Iterator<RecordChangeListener> iterator()
{
return new Iter();
}
/**
* Return a string representation of this object's state, primarily for debugging purposes.
*
* @return Description of this object's state.
*/
public String toString()
{
StringBuilder buf = new StringBuilder(superString());
buf.append("; inner: ");
buf.append(nextInner == null
? "null"
: (nextInner.get() == null ? "null" : nextInner.get().superString()));
buf.append("; outer: ");
buf.append(nextOuter == null ? null : nextOuter.superString());
return buf.toString();
}
/**
* Add a listener.
*
* @param listener
* Listener to be registered.
*
* @return {@code true} if the listener was added; {@code false} if it was already present.
*/
boolean add(RecordChangeListener listener)
{
return interested.add(listener);
}
/**
* Remove a listener.
*
* @param listener
* Listener to be deregistered.
*
* @return {@code true} if the listener was removed; {@code false} if it could not be
* found.
*/
boolean remove(RecordChangeListener listener)
{
return interested.remove(listener);
}
/**
* Indicate whether the set of registered listeners is empty.
*
* @return {@code true} if empty, else {@code false}
*/
boolean isEmpty()
{
return interested.isEmpty();
}
/**
* Link to this object the next related listener set in an outer scope. Listeners in that
* set will receive notifications before listeners in this set.
* <p>
* A link from {@code nextOuter} is made back to this listener set, so iteration across
* these sets can work its way back from the outer set to this one. The link in this
* direction is made with a weak reference, so this set can be garbage collected if its
* scope is popped.
*
* @param nextOuter
* Outer listener set to be linked to this one.
*/
void link(ListenerSet nextOuter)
{
ListenerSet outersInner = nextOuter.nextInner != null ? nextOuter.nextInner.get(): null;
if (outersInner != null)
{
this.nextInner = new WeakReference<>(outersInner);
outersInner.nextOuter = this;
}
this.nextOuter = nextOuter;
nextOuter.nextInner = new WeakReference<>(this);
}
/**
* Unlink this object from its next outer and next inner listener sets, if any, and link
* those sets to each other instead.
*/
void unlink()
{
ListenerSet nextOuter = this.nextOuter;
WeakReference<ListenerSet> nextInner = this.nextInner;
if (nextInner != null)
{
ListenerSet set = nextInner.get();
if (set != null)
{
// replace nextInner's reference to us with reference to our nextOuter
set.nextOuter = nextOuter;
}
}
if (nextOuter != null)
{
// replace nextOuter's reference to us with reference to our nextInner
nextOuter.nextInner = nextInner;
}
this.nextOuter = null;
this.nextInner = null;
}
/**
* Return the parent class' string representation of this object's state.
*
* @return The return value of the parent class' {@code toString()} method.
*/
private String superString()
{
return super.toString();
}
/**
* An iterator which walks all record change listeners registered to receive notifications
* for a DMO type. The walk starts at the outermost scope and works its way to inner scopes,
* ending with the listeners in the enclosing listener set. If a listener is registered in
* more than one scope, it is visited only once, in the outermost scope in which it is
* registered. Listeners are visited in the order in which they were registered. If the
* enclosing listener set has not been linked to any outer listener sets, only the current
* set is walked.
*/
private class Iter
implements Iterator<RecordChangeListener>
{
/** Number of scopes to iterate */
private int scopes = 0;
/** Listener set currently being iterated */
private ListenerSet currentSet;
/** Current worker iterator */
private Iterator<RecordChangeListener> iterator;
/** Set of listeners already visited in this iteration */
private Set<RecordChangeListener> visited = null;
/** Next listener to be visited */
private RecordChangeListener next = null;
/**
* Default constructor.
*/
Iter()
{
// start iterating the listener set at the outermost scope
ListenerSet set = ListenerSet.this;
while (set.nextOuter != null)
{
set = set.nextOuter;
scopes++;
}
currentSet = set;
iterator = currentSet.interested.iterator();
// track visited listeners if we have more than one listener set to iterate
if (scopes > 0)
{
visited = Collections.newSetFromMap(new IdentityHashMap<>());
}
}
/**
* Indicate whether there is a next listener to be visited.
*
* @return {@code true} if there is a next listener, else {@code false}.
*/
@Override
public boolean hasNext()
{
if (next != null)
{
return true;
}
boolean result;
do
{
// reset next listener
next = null;
// if current iterator has no more listeners to visit and currentSet has not been
// reset, try to advance to next inner listener set
while (!(result = iterator.hasNext()) && currentSet != null)
{
// only advance as many scopes as we counted during construction
if (--scopes >= 0)
{
WeakReference<ListenerSet> nextInner = currentSet.nextInner;
if (nextInner == null)
{
currentSet = null;
return false;
}
else
{
currentSet = nextInner.get();
if (currentSet != null)
{
iterator = currentSet.interested.iterator();
}
else
{
return false;
}
}
}
else
{
currentSet = null;
return false;
}
}
if (result)
{
next = iterator.next();
}
} while (visited != null && next != null && visited.contains(next));
return result;
}
/**
* Get the next listener to be visited.
*
* @return Next record change listener.
*/
@Override
public RecordChangeListener next()
{
if (!hasNext())
{
throw new NoSuchElementException();
}
if (visited != null)
{
visited.add(next);
}
RecordChangeListener l = next;
next = null;
return l;
}
/**
* Implementation of listener removal which throws an exception, since removal through
* this iterator type is not supported.
*
* @throws UnsupportedOperationException
* always.
*/
@Override
public void remove()
{
throw new UnsupportedOperationException();
}
}
}
/**
* Convenient implementation that allows the local context to be lazily initialized.
* The change broker depends on the buffer manager, so does the buffer manager depend
* on the change broker. To avoid infinite recursion, the change broker opts to be
* lazily initialized.
*/
static class ContextContainer
extends ContextLocal<ChangeBroker>
{
/**
* Get the weight of this context-local var.
*
* @return Always {@link WeightFactor#LEVEL_6}.
*/
@Override
public WeightFactor getWeight()
{
return WeightFactor.LEVEL_6;
}
/**
* Obtains the context-local instance of the contained global work
* area. The first time this method is invoked in a context, a new
* {@code ChangeBroker} instance is lazily created.
*
* @return The change broker associated with this context.
*/
public ChangeBroker obtain()
{
ChangeBroker broker = get();
if (broker == null)
{
broker = new ChangeBroker();
set(broker);
broker.bufferManager = BufferManager.get();
broker.pm = ProcedureManager.getProcedureHelper();
broker.tm = TransactionManager.getTransactionHelper();
}
return broker;
}
/**
* Return {@code null} rather than providing a true initial value;
* initial value is instead lazily created in {@link #obtain}.
*
* @return {@code null}.
*/
@Override
protected ChangeBroker initialValue()
{
return null;
}
}
}