content: /*****************************************************************
   *   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
   *
   *    https://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.cayenne.event;
  
  import org.apache.cayenne.CayenneRuntimeException;
  import org.apache.cayenne.di.BeforeScopeEnd;
  import org.apache.cayenne.util.Invocation;
  
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.EventObject;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.WeakHashMap;
  
  /**
   * A default implementation of {@link EventManager}.
   * 
   * @since 3.1
   */
  public class DefaultEventManager implements EventManager {
  
      private static final int DEFAULT_DISPATCH_THREAD_COUNT = 5;
  
      // keeps weak references to subjects
      protected final Map<EventSubject, DispatchQueue> subjects;
      protected final List<Dispatch> eventQueue;
      protected final boolean singleThread;
      protected final List<DispatchThread> dispatchThreads;
  
      protected volatile boolean stopped;
  
      /**
       * Creates a multithreaded EventManager using default thread count.
       */
      public DefaultEventManager() {
          this(DEFAULT_DISPATCH_THREAD_COUNT);
      }
  
      /**
       * Creates an EventManager starting the specified number of threads for multithreaded
       * dispatching. To create a single-threaded EventManager, use thread count of zero or
       * less.
       */
      public DefaultEventManager(int dispatchThreadCount) {
          this.subjects = Collections.synchronizedMap(new WeakHashMap<>());
          this.eventQueue = Collections.synchronizedList(new LinkedList<>());
          this.singleThread = dispatchThreadCount <= 0;
  
          if (!singleThread) {
              dispatchThreads = new ArrayList<>(dispatchThreadCount);
  
              String prefix = "cayenne-event-";
  
              // start dispatch threads
              for (int i = 0; i < dispatchThreadCount; i++) {
                  DispatchThread thread = new DispatchThread(prefix + i);
                  dispatchThreads.add(thread);
                  thread.start();
              }
          } else {
              dispatchThreads = Collections.emptyList();
          }
      }
  
      /**
       * Returns true if the EventManager was stopped via {@link #shutdown()} method.
       * 
       * @since 3.1
       */
      public boolean isStopped() {
          return stopped;
      }
  
      /**
       * Returns true if this EventManager is single-threaded. If so it will throw an
       * exception on any attempt to register an unblocking listener or dispatch a
       * non-blocking event.
       * 
       * @since 1.2
       */
      public boolean isSingleThreaded() {
          return singleThread;
      }
  
      /**
       * Stops event threads. After the EventManager is stopped, it can not be restarted and
       * should be discarded.
       * 
       * @since 3.0
       */
      @BeforeScopeEnd
      public void shutdown() {
  
          if (!stopped) {
  
              this.stopped = true;
  
              for (DispatchThread thread : dispatchThreads) {
                  thread.interrupt();
              }
  
              dispatchThreads.clear();
          }
      }
  
      /**
       * Register an <code>EventListener</code> for events sent by any sender.
       * 
       * @throws RuntimeException if <code>methodName</code> is not found
       * @see #addListener(Object, String, Class, EventSubject, Object)
       */
      public void addListener(
              Object listener,
              String methodName,
              Class<?> eventParameterClass,
              EventSubject subject) {
          this.addListener(listener, methodName, eventParameterClass, subject, null, true);
      }
  
      public void addNonBlockingListener(
              Object listener,
              String methodName,
              Class<?> eventParameterClass,
              EventSubject subject) {
  
          if (singleThread) {
              throw new IllegalStateException("DefaultEventManager is configured to be single-threaded.");
          }
  
          this.addListener(listener, methodName, eventParameterClass, subject, null, false);
      }
  
      /**
       * Register an <code>EventListener</code> for events sent by a specific sender.
       * 
       * @param listener the object to be notified about events
       * @param methodName the name of the listener method to be invoked
       * @param eventParameterClass the class of the single event argument passed to
       *            <code>methodName</code>
       * @param subject the event subject that the listener is interested in
       * @param sender the object whose events the listener is interested in;
       *            <code>null</code> means 'any sender'.
       * @throws RuntimeException if <code>methodName</code> is not found
       */
      public void addListener(
              Object listener,
              String methodName,
              Class<?> eventParameterClass,
              EventSubject subject,
              Object sender) {
          addListener(listener, methodName, eventParameterClass, subject, sender, true);
      }
  
      public void addNonBlockingListener(
              Object listener,
              String methodName,
              Class<?> eventParameterClass,
              EventSubject subject,
              Object sender) {
  
          if (singleThread) {
              throw new IllegalStateException("DefaultEventManager is configured to be single-threaded.");
          }
  
          addListener(listener, methodName, eventParameterClass, subject, sender, false);
      }
  
      protected void addListener(
              Object listener,
              String methodName,
              Class<?> eventParameterClass,
              EventSubject subject,
              Object sender,
              boolean blocking) {
  
          if (listener == null) {
              throw new IllegalArgumentException("Listener must not be null.");
          }
  
          if (eventParameterClass == null) {
              throw new IllegalArgumentException("Event class must not be null.");
          }
  
          if (subject == null) {
              throw new IllegalArgumentException("Subject must not be null.");
          }
  
          try {
              Invocation invocation = blocking
                      ? new Invocation(listener, methodName, eventParameterClass)
                      : new NonBlockingInvocation(listener, methodName, eventParameterClass);
              dispatchQueueForSubject(subject, true).addInvocation(invocation, sender);
          } catch (NoSuchMethodException nsm) {
              throw new CayenneRuntimeException("Error adding listener, method name: %s", nsm, methodName);
          }
      }
  
      /**
       * Unregister the specified listener from all event subjects handled by this manager
       * instance.
       * 
       * @param listener the object to be unregistered
       * @return <code>true</code> if <code>listener</code> could be removed for any
       *         existing subjects, else returns <code>false</code>.
       */
      public boolean removeListener(Object listener) {
          if (listener == null) {
              return false;
          }
  
          boolean didRemove = false;
  
          synchronized (subjects) {
              if (!subjects.isEmpty()) {
                  for (EventSubject subject : subjects.keySet()) {
                      didRemove |= this.removeListener(listener, subject);
                  }
              }
          }
  
          return didRemove;
      }
  
      /**
       * Removes all listeners for a given subject.
       */
      public boolean removeAllListeners(EventSubject subject) {
          if (subject != null) {
              synchronized (subjects) {
                  return subjects.remove(subject) != null;
              }
          }
  
          return false;
      }
  
      /**
       * Unregister the specified listener for the events about the given subject.
       * 
       * @param listener the object to be unregistered
       * @param subject the subject from which the listener is to be unregistered
       * @return <code>true</code> if <code>listener</code> could be removed for the given
       *         subject, else returns <code>false</code>.
       */
      public boolean removeListener(Object listener, EventSubject subject) {
          return this.removeListener(listener, subject, null);
      }
  
      /**
       * Unregister the specified listener for the events about the given subject and the
       * given sender.
       * 
       * @param listener the object to be unregistered
       * @param subject the subject from which the listener is to be unregistered
       * @param sender the object whose events the listener was interested in;
       *            <code>null</code> means 'any sender'.
       * @return <code>true</code> if <code>listener</code> could be removed for the given
       *         subject, else returns <code>false</code>.
       */
      public boolean removeListener(Object listener, EventSubject subject, Object sender) {
          if (listener == null || subject == null) {
              return false;
          }
  
          DispatchQueue subjectQueue = dispatchQueueForSubject(subject, false);
          if (subjectQueue == null) {
              return false;
          }
  
          return subjectQueue.removeInvocations(listener, sender);
      }
  
      /**
       * Sends an event to all registered objects about a particular subject. Event is sent
       * synchronously, so the sender thread is blocked until all the listeners finish
       * processing the event.
       * 
       * @param event the event to be posted to the observers
       * @param subject the subject about which observers will be notified
       * @throws IllegalArgumentException if event or subject are null
       */
      public void postEvent(EventObject event, EventSubject subject) {
          dispatchEvent(new Dispatch(event, subject));
      }
  
      /**
       * Sends an event to all registered objects about a particular subject. Event is
       * queued by EventManager, releasing the sender thread, and is later dispatched in a
       * separate thread.
       * 
       * @param event the event to be posted to the observers
       * @param subject the subject about which observers will be notified
       * @throws IllegalArgumentException if event or subject are null
       * @since 1.1
       */
      public void postNonBlockingEvent(EventObject event, EventSubject subject) {
          if (singleThread) {
              throw new IllegalStateException("EventManager is configured to be single-threaded.");
          }
  
          // add dispatch to the queue and return
          synchronized (eventQueue) {
              eventQueue.add(new Dispatch(event, subject));
              eventQueue.notifyAll();
          }
      }
  
      private void dispatchEvent(Dispatch dispatch) {
          DispatchQueue dispatchQueue = dispatchQueueForSubject(dispatch.subject, false);
          if (dispatchQueue != null) {
              dispatchQueue.dispatchEvent(dispatch);
          }
      }
  
      // returns a subject's mapping from senders to registered listener invocations
      private DispatchQueue dispatchQueueForSubject(EventSubject subject, boolean create) {
          synchronized (subjects) {
              DispatchQueue listenersStore = subjects.get(subject);
              if (create && listenersStore == null) {
                  listenersStore = new DispatchQueue();
                  subjects.put(subject, listenersStore);
              }
              return listenersStore;
          }
      }
  
      // represents a posted event
      class Dispatch {
  
          EventObject[] eventArgument;
          EventSubject subject;
  
          Dispatch(EventObject event, EventSubject subject) {
              this(new EventObject[] {event}, subject);
          }
  
          Dispatch(EventObject[] eventArgument, EventSubject subject) {
              this.eventArgument = eventArgument;
              this.subject = subject;
          }
  
          Object getSender() {
              return eventArgument[0].getSource();
          }
  
          void fire() {
              DefaultEventManager.this.dispatchEvent(Dispatch.this);
          }
  
          boolean fire(Invocation invocation) {
              if (invocation instanceof NonBlockingInvocation) {
  
                  // do minimal checks first...
                  if (invocation.getTarget() == null) {
                      return false;
                  }
  
                  // inject single invocation dispatch into the queue
                  synchronized (eventQueue) {
                      eventQueue.add(new InvocationDispatch(eventArgument, subject, invocation));
                      eventQueue.notifyAll();
                  }
  
                  return true;
              } else {
                  return invocation.fire(eventArgument);
              }
          }
      }
  
      // represents a posted event that should be sent to a single known listener
      class InvocationDispatch extends Dispatch {
  
          Invocation target;
  
          InvocationDispatch(EventObject[] eventArgument, EventSubject subject, Invocation target) {
              super(eventArgument, subject);
              this.target = target;
          }
  
          @Override
          void fire() {
              // there is no way to kill the invocation if it is bad...
              // so don't check for status
              target.fire(eventArgument);
          }
      }
  
      // subclass exists only to tag invocations that should be
      // dispatched in a separate thread
      final class NonBlockingInvocation extends Invocation {
          NonBlockingInvocation(Object target, String methodName, Class<?> parameterType) throws NoSuchMethodException {
              super(target, methodName, parameterType);
          }
      }
  
      final class DispatchThread extends Thread {
  
          DispatchThread(String name) {
              super(name);
              setDaemon(true);
          }
  
          @Override
          public void run() {
              while (!stopped) {
  
                  // get event from the queue, if the queue is empty, just wait
                  Dispatch dispatch = null;
  
                  synchronized (DefaultEventManager.this.eventQueue) {
                      if (DefaultEventManager.this.eventQueue.size() > 0) {
                          dispatch = DefaultEventManager.this.eventQueue.remove(0);
                      } else {
                          try {
                              // wake up occasionally to check whether EM has been stopped
                              DefaultEventManager.this.eventQueue.wait(3 * 60 * 1000);
                          } catch (InterruptedException e) {
                              // ignore interrupts...
                          }
                      }
                  }
  
                  // dispatch outside of synchronized block
                  if (!stopped && dispatch != null) {
                      // this try/catch is needed to prevent DispatchThread
                      // from dying on dispatch errors
                      try {
                          dispatch.fire();
                      } catch (Throwable th) {
                          // ignoring exception
                      }
                  }
              }
          }
      }
  }
  