# GlobalEventManager.java

    /*
    ** Module   : GlobalEventManager.java
    ** Abstract : Cross-context clearinghouse for DMO change events
    **
    ** Copyright (c) 2004-2024, Golden Code Development Corporation.
    **
    ** -#- -I- --Date-- --JPRM-- ----------------------------------Description-----------------------------------
    ** 001 ECF 20080813   @39540 Created initial version. Cross-context
    **                           clearinghouse for DMO change events.
    ** 002 ECF 20080926   @39964 Fixed ConcurrentModificationException in
    **                           cleanupContext().
    ** 003 ECF 20090619   @42964 Named PurgeWorker thread for easier debugging.
    ** 004 SVL 20090715   @43193 Added OOME protection into PurgeWorker.
    ** 005 IAS 20160331          Fixed updateEntityRefCount method.
    ** 006 ECF 20160815          Synchronized registration; replaced Apache commons logging with J2SE
    **                           logging.
    ** 007 ECF 20200901          New ORM implementation.
    ** 008 TJD 20220504          Upgrade do Java 11 minor changes
    ** 009 SVL 20230113          Improved performance by replacing some "for-each" loops with indexed "for" loops.
    ** 010 GBB 20230512          Logging methods replaced by CentralLogger/ConversionStatus.
    ** 011 GBB 20230825          SecurityManager session methods calls updated.
    ** 012 TJD 20240123          Java 17 compatibility updates
    */
    /*
    ** 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.dirty;

    import java.util.*;
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.*;
    import java.util.logging.*;
    import com.goldencode.p2j.persist.*;
    import com.goldencode.p2j.persist.Record;
    import com.goldencode.p2j.persist.event.*;
    import com.goldencode.p2j.security.*;
    import com.goldencode.p2j.security.SecurityManager;
    import com.goldencode.p2j.util.logging.*;

    /**
     * A clearinghouse for events which represent changes to records which affect
     * any index of their backing tables.  Events originate in the {@link
     * DirtyShareManager} and are posted here for further handling.  Objects which
     * wish to process such events {@link #register(String) register} and {@link
     * #deregister(long) deregister} via the <code>DirtyShareManager</code>.
     * Events also are {@link #getEvents(long, long) collected} via the
     * <code>DirtyShareManager</code>.  Since all access is managed through that
     * interface, this class is package private.
     * <p>
     * This class is described as a "clearinghouse" rather than a "dispatcher",
     * because events are not dispatched via a "push" model.  Rather, they are
     * collected by an instance of this object (one per physical database), and
     * await {@link #getEvents(long, long) requests} by interested parties (i.e.,
     * registrants).  This is a "pull" model for distributing these events.
     * <p>
     * A registrant registers interest in a particular type of DMO.  Only events
     * which affect DMO types for which interest has been registered are collected
     * in this object's event queue.  All other events are discarded upon receipt.
     * The following methods are used by a <code>DirtyShareManager</code> to
     * enqueue events:
     * <ul>
     *   <li>{@link #enqueueInsert(String, Record)}
     *   <li>{@link #enqueueDelete(String, Long)}
     *   <li>{@link #enqueueUpdate(String, Record, String[])}
     * </ul>
     * <p>
     * When a registrant collects new events, only those events which did not
     * originate in that registrant's own context are provided.  It is assumed
     * that other mechanisms already exist to process "local" events.
     * <p>
     * The "pull" model of event distribution is used to prevent unnecessary event
     * traffic, since in the common case, very few objects are interested in
     * receiving such events.  At the time of writing, this event distribution
     * mechanism is intended primarily to support the need for certain
     * <code>AdaptiveQuery</code> instances to invalidate their results in
     * response to DMO changes in other sessions.  By default, this is limited to
     * instances of <code>AdaptiveFind</code>, which are more sensitive to such
     * changes than common <code>AdaptiveQuery</code> instances.
     * <p>
     * Because a pull-based event model cannot be certain when all interested
     * parties have processed each applicable event (or whether they ever will),
     * some safeguards have been established to prevent both the premature release
     * of needed events, and to prevent the unnecessary accumulation of old events
     * which are no longer useful.
     * <p>
     * Each registrant is assigned a unique, sequential (ascending) registrant
     * identifier upon registration.  Registrants are tracked by these IDs.  Each
     * event is assigned a unique, sequential (ascending) event identifier, using
     * the same sequence.  Thus, IDs are unique across both all registrants and
     * all events, and they are guaranteed to increase sequentially.  When an
     * event is enqueued, its ID is guaranteed to be higher than the registrant(s)
     * interested in it.
     * <p>
     * When a registrant requests events, it provides the ID of the last event it
     * retrieved.  All new events (if any) collected after that ID are retrieved.
     * When a registrant is no longer interested in events, it deregisters itself.
     * Events older than the oldest registrant are purged at various points.
     * <p>
     * To prevent an idle registrant from pinning old events in the queue, an idle
     * timeout check is applied at various points.  If a registrant has been idle
     * longer than the permitted timeout value, it is automatically deregistered.
     * The next time that registrant requests new events (if ever), an exception
     * is thrown, and it is up to the registrant to update its state accordingly.
     * <p>
     * In case a context is interrupted unexpectedly, all of its registrants are
     * automatically deregistered.  Even if this were not the case, the idle
     * timeout would eventually clean up these resources.
     * <p>
     * TODO:  enable directory-based configuration of idle timeout.
     */
    final class GlobalEventManager
    {
       /** Logger */
       private static final CentralLogger log = CentralLogger.get(GlobalEventManager.class.getName());

       /** Default time, in milliseconds, for a registrant to be idle before being expired */
       private static final long DEFAULT_IDLE_TIME = 600000;

       /** Database for which this object manages global DMO change events */
       private final Database database;

       /** Map of interested party (registrant) IDs to registrants */
       private final SortedMap<Long, Registrant> interested = new TreeMap<>();

       /** Map of entity names to reference counts of interested parties */
       private final ConcurrentMap<String, AtomicInteger> entities = new ConcurrentHashMap<>();

       /** Object which tracks registrants by context for cleanup purposes */
       private final ContextLocal<Context> context = new ContextLocal<Context>()
       {
          protected Context initialValue() { return (new Context()); }
          protected void cleanup(Context ctx) { cleanupContext(ctx); }
       };

       /** Time, in milliseconds, for a registrant to be idle before being expired */
       private final long idleTimeout = DEFAULT_IDLE_TIME;

       /** Event queue */
       private List<EventWrapper> eventQueue = null;

       /** Variable from which unique registrant and event IDs are assigned */
       private volatile long nextID = 0L;

       /**
        * Constructor
        *
        * @param   database
        *          Database for which events are tracked by this object.
        */
       GlobalEventManager(Database database)
       {
          this.database = database;

          // Create event queue.
          resetQueue();

          // Launch purge worker.
          new PurgeWorker();
       }

       /**
        * Register interest in index changes for a particular DMO type.
        *
        * @param   entity
        *          DMO entity name.
        *
        * @return  Unique identifier for this registrant.
        *
        * @see     #deregister(long)
        */
       long register(String entity)
       {
          Long id;

          synchronized (this)
          {
             id = Long.valueOf(++nextID);
             Registrant reg = new Registrant(entity);
             interested.put(id, reg);
             updateEntityRefCount(entity, 1);
          }

          context.get().localIDs.add(id);

          return id;
       }

       /**
        * Deregister interest in index changes for a particular DMO type.
        * <p>
        * Invocation of this method triggers a purge sweep in a separate thread.
        *
        * @param   id
        *          Unique identifier previously assigned to the registrant to be
        *          deregistered.
        *
        * @see     #register(String)
        */
       synchronized void deregister(long id)
       {
          deregister(id, true);
       }

       /**
        * Retrieve any events which have been enqueued, which are newer than
        * <code>lastID</code>, and which correspond to the DMO entity for which
        * the specified registrant previously has registered interest.
        * <p>
        * As a side effect of this method, expired events may be detected, which
        * are then purged from the event queue.
        *
        * @param   registerID
        *          A unique identifier for an object which has registered interest
        *          in record-changing events.
        * @param   lastID
        *          A unique identifier representing the last event retrieved by
        *          the caller.  If less than 0, it is assumed no events have yet
        *          been retrieved since the caller registered interest in these
        *          events.
        *
        * @return  Array of applicable, new events, or <code>null</code> if no
        *          such events have been collected.
        *
        * @throws  EventRegistrationException
        *          if <code>registerID</code> is invalid or has expired.
        */
       synchronized GlobalChangeEvent[] getEvents(long registerID, long lastID)
       throws EventRegistrationException
       {
          Registrant reg = interested.get(registerID);

          if (reg == null)
          {
             throw new EventRegistrationException(registerID);
          }

          reg.setLastUse();

          int size = eventQueue.size();
          if (size == 0)
          {
             return null;
          }

          long oldestRegisterID = interested.firstKey();
          List<GlobalChangeEvent> events = null;
          if (lastID < 0)
          {
             lastID = registerID;
          }

          ListIterator<EventWrapper> iter = eventQueue.listIterator(size);
          while (iter.hasPrevious())
          {
             EventWrapper ew = iter.previous();
             GlobalChangeEvent event = ew.getEvent();

             long eventID = event.getEventID();
             if (eventID <= lastID)
             {
                if (eventID < oldestRegisterID)
                {
                   // Event has expired;  purge it.
                   iter.remove();
                }

                continue;
             }

             if (ew.isOfInterest(reg))
             {
                if (events == null)
                {
                   events = new ArrayList<>();
                }
                events.add(event);
             }
          }

          if (events == null)
          {
             return null;
          }

          return events.toArray(new GlobalChangeEvent[events.size()]);
       }

       /**
        * Enqueue a DMO update event, if any registrant is interested in it.
        *
        * @param   entity
        *          DMO entity name;  identifies DMO type affected by update.
        * @param   dmo
        *          The DMO instance in its <i>post-update</i> state.
        * @param   properties
        *          Names of all properties which have been updated.
        */
       synchronized void enqueueUpdate(String entity, Record dmo, String[] properties)
       {
          if (isEntityReferenced(entity))
          {
             GlobalChangeEvent event = new GlobalChangeEvent(++nextID, entity, dmo, properties);
             enqueueEvent(event);
          }
       }

       /**
        * Enqueue a DMO insert event, if any registrant is interested in it.
        *
        * @param   entity
        *          DMO entity name;  identifies DMO type affected by insert.
        * @param   dmo
        *          The new DMO instance.
        */
       synchronized void enqueueInsert(String entity, Record dmo)
       {
          if (isEntityReferenced(entity))
          {
             GlobalChangeEvent event = new GlobalChangeEvent(++nextID, entity, dmo);
             enqueueEvent(event);
          }
       }

       /**
        * Enqueue a DMO delete event, if any registrant is interested in it.
        *
        * @param   entity
        *          DMO entity name;  identifies DMO type affected by update.
        * @param   primaryKey
        *          Primary key ID of the deleted DMO instance.
        */
       synchronized void enqueueDelete(String entity, Long primaryKey)
       {
          if (isEntityReferenced(entity))
          {
             GlobalChangeEvent event = new GlobalChangeEvent(++nextID, entity, primaryKey);
             enqueueEvent(event);
          }
       }

       /**
        * Enqueue a DMO update, insert, or delete event.
        * <p>
        * Invocation of this method triggers a purge sweep in a separate thread.
        *
        * @param   event
        *          Event object which encapsulates information about the change.
        */
       private void enqueueEvent(GlobalChangeEvent event)
       {
          EventWrapper ew = new EventWrapper(event);
          eventQueue.add(ew);

          // Trigger purge sweep.
          notify();
       }

       /**
        * Deregister interest in index changes for a particular DMO type.
        * <p>
        * Invocation of this method may trigger a purge sweep in a separate thread.
        *
        * @param   id
        *          Unique identifier previously assigned to the registrant to be deregistered.
        * @param   purge
        *          <code>true</code> to trigger a purge sweep in a separate thread
        *          after the deregistration takes place;  else <code>false</code>.
        */
       private void deregister(long id, boolean purge)
       {
          Registrant reg = interested.remove(id);
          if (reg != null)
          {
             updateEntityRefCount(reg.getEntity(), -1);
             context.get().localIDs.remove(id);
          }

          if (purge)
          {
             // Trigger purge sweep.
             notify();
          }
       }

       /**
        * Modify a reference count for the given DMO entity.  When a reference
        * count for any entity reaches 0, this indicates that no registrants
        * currently are interested in events about this DMO type.
        *
        * @param   entity
        *          DMO entity name.
        * @param   delta
        *          Amount to adjust the reference count.  Should be 1 to increment
        *          or -1 to decrement.
        */
       private void updateEntityRefCount(String entity, int delta)
       {
          AtomicInteger count = entities.get(entity);
          if (count == null)
          {
             count = new AtomicInteger(0);
             AtomicInteger prevCount = entities.putIfAbsent(entity, count);
             if (prevCount != null)
             {
                count = prevCount;
             }
          }

          int nv = count.addAndGet(delta);
          if (nv == 0)
          {
             entities.remove(entity, count);
          }
       }

       /**
        * Report whether any registrant is interested in events about the given
        * DMO type.
        *
        * @param   entity
        *          DMO entity name.
        *
        * @return  <code>true</code> if any registrant is interested in the given
        *          DMO type, else <code>false</code>.
        */
       private boolean isEntityReferenced(String entity)
       {
          if (interested.isEmpty())
          {
             return false;
          }

          AtomicInteger count = entities.get(entity);

          return count != null && count.get() > 0;
       }

       /**
        * Reset the event queue by instantiating a new, empty queue and discarding
        * the old one, if any.
        */
       private void resetQueue()
       {
          eventQueue = new LinkedList<>();
       }

       /**
        * Invoked when a context is ending, this method deregisters any
        * registrants associated with the local context.  If the context is ending
        * normally, there should be nothing for this method to do.  However, if
        * the context is ending abruptly, some registrants may need to be cleaned
        * up.
        * <p>
        * Invocation of this method triggers a purge sweep in a separate thread if
        * any "orphaned" registrants were detected.
        *
        * @param   ctx
        *          Context requiring cleanup at end of life.
        */
       private void cleanupContext(Context ctx)
       {
          Set<Long> localIDs = ctx.localIDs;
          if (localIDs.isEmpty())
          {
             return;
          }

          synchronized (this)
          {
             Iterator<Long> iter = localIDs.iterator();
             while (iter.hasNext())
             {
                // Update data structures directly, rather than calling
                // deregister(long, boolean), because that method will try to
                // update localIDs, causing ConcurrentModificationException.
                Long id = iter.next();
                Registrant reg = interested.remove(id);
                if (reg != null)
                {
                   updateEntityRefCount(reg.getEntity(), -1);
                }

                iter.remove();
             }

             // Trigger purge sweep.
             notify();
          }
       }

       /**
        * Object which tracks registrants associated with the current session,
        * primarily for end-of-context cleanup purposes.
        */
       private static class Context
       {
          /** Set of IDs registered to current context */
          private final Set<Long> localIDs = new HashSet<>(4);
       }

       /**
        * A party which is interested in tracking index change for a particular
        * DMO type.  Also tracks the timestamp of the last event request for this
        * registrant.
        */
       private static class Registrant
       {
          /** DMO entity name */
          private final String entity;

          /** Millisecond timestamp of last event collection request */
          private long lastUse;

          /**
           * Constructor.
           *
           * @param   entity
           *          DMO entity name.
           */
          Registrant(String entity)
          {
             this.entity = entity;
             setLastUse();
          }

          /**
           * Get name of DMO entity being tracked by this registrant.
           *
           * @return  DMO entity name.
           */
          private String getEntity()
          {
             return entity;
          }

          /**
           * Get timestamp of the last event request made for this registrant.
           *
           * @return  Millisecond timestamp value.
           */
          private long getLastUse()
          {
             return lastUse;
          }

          /**
           * Set the timestamp of the last event request made for this registrant
           * to the current system time (since the epoch) in milliseconds.
           */
          private void setLastUse()
          {
             lastUse = System.currentTimeMillis();
          }
       }

       /**
        * Object which wraps a change event to remember the unique identifier of
        * the session which originated the event.  This allows us to filter events
        * by originating session.  It is assumed registrants are not interested in
        * receiving events from this clearinghouse, which they themselves generated.
        */
       private static class EventWrapper
       {
          /** Security manager which provides session IDs */
          private static final SecurityManager security = SecurityManager.getInstance();

          /** Unique ID of current context */
          private final Integer sessionID;

          /** Event representing a DMO change (insert, update, delete) */
          private final GlobalChangeEvent event;

          /**
           * Constructor.  Records current session ID at the time of construction.
           *
           * @param   event
           *          Event representing a DMO change (insert, update, delete).
           */
          private EventWrapper(GlobalChangeEvent event)
          {
             this.sessionID = security.sessionSm.getSessionId();
             this.event = event;
          }

          /**
           * Get the event wrapped by this object.
           *
           * @return  Event representing a DMO change (insert, update, delete).
           */
          private GlobalChangeEvent getEvent()
          {
             return event;
          }

          /**
           * Determine whether the event wrapped by this object is of interest to
           * the current context.  It is considered of interest if it originated
           * in a different context, and is associated with the given DMO entity.
           *
           * @param   registration
           *          Party interested in DMO changes.
           *
           * @return  <code>true</code> if this event should be delivered to the
           *          current context, else <code>false</code>.
           */
          private boolean isOfInterest(Registrant registration)
          {
             String entity = registration.getEntity();
             if (!entity.equals(event.getEntity()))
             {
                return false;
             }

             Integer currentSessionID = security.sessionSm.getSessionId();

             return (!sessionID.equals(currentSessionID));
          }
       }

       /**
        * A worker which purges idle registrants from the registrant map and
        * expired events from the event queue in a dedicated, daemon thread.  The
        * worker runs in an infinite loop, waiting on each pass for a notification
        * from the event manager to do its work.
        */
       private class PurgeWorker
       extends Thread
       {
          /**
           * Constructor which launches this worker's dedicated thread.
           */
          private PurgeWorker()
          {
             super("GlobalEventManager Queue Purge Thread");

             setDaemon(true);
             start();
          }

          /**
           * Run in an infinite loop, waiting for purge notifications from the
           * event manager at the top of each pass.  When a notification is
           * received, first purge idle registrants, then purge expired events.
           */
          public void run()
          {
             while (true)
             {
                try
                {
                   GlobalEventManager gem = GlobalEventManager.this;
                   synchronized (gem)
                   {
                      gem.wait();
                      purgeRegistrants();
                      purgeEvents();
                   }
                }
                catch (InterruptedException exc)
                {
                   if (log.isLoggable(Level.WARNING))
                   {
                      log.log(Level.WARNING,
                              "Persistence event clearing house purge thread for " +
                              database +
                              " interrupted;  resuming work.",
                              exc);
                   }
                }
                catch (OutOfMemoryError oome)
                {
                   try
                   {
                      log.log(Level.SEVERE, "OOME in purge worker", oome);
                   }
                   catch (OutOfMemoryError e)
                   {
                      // Ignore.
                   }
                }
             }
          }

          /**
           * Walk from oldest to most recent registrant, deregistering those that
           * have been idle more than the required timeout period.
           */
          private void purgeRegistrants()
          {
             ArrayList<Long> dead = null;
             long threshold = System.currentTimeMillis() - idleTimeout;

             Iterator<SortedMap.Entry<Long, Registrant>> iter = interested.entrySet().iterator();
             while (iter.hasNext())
             {
                SortedMap.Entry<Long, Registrant> entry = iter.next();
                Registrant reg = entry.getValue();
                if (reg.getLastUse() < threshold)
                {
                   if (dead == null)
                   {
                      dead = new ArrayList<>();
                   }
                   dead.add(entry.getKey());
                }
                else
                {
                   break;
                }
             }

             // We deregister in a separate pass, because calling deregister() in
             // the loop above would raise ConcurrentModificationException.
             if (dead != null)
             {
                for (int i = 0; i < dead.size(); i++)
                {
                   Long id = dead.get(i);
                   deregister(id, false);
                }
             }
          }

          /**
           * Purge expired events from the event queue, or reset the queue entirely if there are no
           * registrants at the time this method is invoked.
           */
          private void purgeEvents()
          {
             int size = eventQueue.size();

             if (size == 0)
             {
                return;
             }

             if (interested.isEmpty())
             {
                resetQueue();

                return;
             }

             long oldestRegisterID = interested.firstKey();
             ListIterator<EventWrapper> iter = eventQueue.listIterator(size);
             while (iter.hasPrevious())
             {
                EventWrapper ew = iter.previous();
                GlobalChangeEvent event = ew.getEvent();
                if (event.getEventID() > oldestRegisterID)
                {
                   break;
                }

                iter.remove();
             }
          }
       }
    }