/*
      Copyright 2008 Ramon Servadei
   
      Licensed under the Apache License, Version 2.0 (the "License");
      you may not use this file except in compliance with the License.
      You may obtain a copy of the License at
   
          http://www.apache.org/licenses/LICENSE-2.0
   
     Unless required by applicable law or agreed to in writing, software
     distributed under the License is distributed on an "AS IS" BASIS,
     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     See the License for the specific language governing permissions and
     limitations under the License.
   */
  package fulmine.event.subscription;
  
  import static fulmine.util.Utils.EMPTY_STRING;
  import static fulmine.util.Utils.logException;
  import static fulmine.util.Utils.nullCheck;
  import static fulmine.util.Utils.safeToString;
  
  import java.util.Collection;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  
  import fulmine.AbstractLifeCycle;
  import fulmine.IAddressable;
  import fulmine.context.IFrameworkContext;
  import fulmine.event.IEventSource;
  import fulmine.event.listener.IEventListener;
  import fulmine.event.system.SubscribeEvent;
  import fulmine.event.system.UnsubscribeEvent;
  import fulmine.util.collection.CollectionFactory;
  import fulmine.util.collection.CollectionUtils;
  import fulmine.util.log.AsyncLog;
  
  /**
   * Base implementation of a subscription manager.
   * <p>
   * This is thread safe. Access on the internal collection of
   * {@link ISubscription} instances is synchronized just enough to get the
   * subscription. The subscription implementations are assumed to be thread safe.
   * 
   * @author Ramon Servadei
   */
  public abstract class SubscriptionManager extends AbstractLifeCycle implements
      ISubscriptionManager
  {
      private final static AsyncLog LOG = new AsyncLog(SubscriptionManager.class);
  
      /** The factory to build {@link ISubscription} instances */
      private final ISubscriptionFactory factory;
  
      /** The subscriptions being managed */
      private final Map<ISubscriptionParameters, ISubscription> subscriptions;
  
      /** The context */
      private final IFrameworkContext context;
  
      /**
       * Standard constructor
       * 
       * @param context
       *            the context to use
       * @param factory
       *            the container subscription factory to use
       */
      public SubscriptionManager(IFrameworkContext context,
          ISubscriptionFactory factory)
      {
          super();
          this.subscriptions = CollectionFactory.newMap();
          nullCheck(factory, "No subscription factory provided");
          this.factory = factory;
          this.context = context;
      }
  
      public final boolean subscribe(ISubscriptionParameters parameters)
      {
          checkActive();
          if (getLog().isInfoEnabled())
          {
              getLog().info("[subscribe] " + safeToString(parameters));
          }
          final ISubscription subscription;
          synchronized (getSubscriptions())
          {
              if (getSubscriptions().containsKey(parameters))
              {
                  if (getLog().isInfoEnabled())
                  {
                      getLog().info(
                          "[subscribe] already contains "
                              + safeToString(parameters));
                  }
                  return false;
              }
             subscription = this.factory.createSubscription(parameters);
             getSubscriptions().put(parameters, subscription);
             // raise a new subscription event
             doRaiseSubscribeEvent(parameters);
         }
         // now find any matches
         Collection<? extends IEventSource> currentEntities =
             doGetEventSources();
         for (IEventSource source : currentEntities)
         {
             try
             {
                 if (subscription.includes(source))
                 {
                     subscription.addMatch(source);
                 }
             }
             catch (Exception e)
             {
                 logException(getLog(), source, e);
             }
         }
         return true;
     }
 
     protected final ISubscription getSubscription(
         ISubscriptionParameters parameters)
     {
         final ISubscription subscription;
         synchronized (getSubscriptions())
         {
             subscription = getSubscriptions().get(parameters);
         }
         return subscription;
     }
 
     public final boolean addListener(ISubscriptionParameters parameters,
         IEventListener listener)
     {
         checkActive();
         boolean added = false;
         final ISubscription subscription;
         subscription = getSubscription(parameters);
         if (subscription != null)
         {
             added = subscription.addListener(listener);
         }
         else
         {
             if (getLog().isInfoEnabled())
             {
                 getLog().info(
                     "[add listener] no subscription for "
                         + safeToString(parameters));
             }
         }
         if (added)
         {
             doAddListener(parameters, listener);
         }
         return added;
     }
 
     public final boolean removeListener(ISubscriptionParameters parameters,
         IEventListener listener)
     {
         checkActive();
         boolean removed = false;
         final ISubscription subscription = getSubscription(parameters);
         if (subscription != null)
         {
             removed = subscription.removeListener(listener);
         }
         else
         {
             if (getLog().isInfoEnabled())
             {
                 getLog().info(
                     "[remove listener] no subscription for "
                         + safeToString(parameters));
             }
         }
         return removed;
     }
 
     public final List<IEventListener> getListeners(
         ISubscriptionParameters parameters)
     {
         checkActive();
         final ISubscription subscription = getSubscription(parameters);
         if (subscription != null)
         {
             return CollectionFactory.newList(subscription.getListeners());
         }
         if (getLog().isInfoEnabled())
         {
             getLog().info(
                 "[get listeners] no subscription for "
                     + safeToString(parameters));
         }
         return Collections.emptyList();
     }
 
     public final boolean unsubscribe(ISubscriptionParameters parameters)
     {
         if (getLog().isInfoEnabled())
         {
             getLog().info("[unsubscribe] " + safeToString(parameters));
         }
         // don't check for active because this is called during destroy!
         // checkActive();
         final ISubscription subscription;
         synchronized (getSubscriptions())
         {
             subscription = getSubscriptions().remove(parameters);
         }
         if (subscription != null)
         {
             doRaiseUnsubscribeEvent(parameters);
         }
         else
         {
             if (getLog().isInfoEnabled())
             {
                 getLog().info(
                     "[unsubscribe] does not contain "
                         + safeToString(parameters));
             }
             return false;
         }
         try
         {
             subscription.destroy();
         }
         catch (Exception e)
         {
             logException(getLog(), subscription, e);
         }
         return true;
     }
 
     public final boolean isSubscribed(IEventSource source)
     {
         checkActive();
         synchronized (getSubscriptions())
         {
             for (ISubscription subscription : getSubscriptions().values())
             {
                 try
                 {
                     if (subscription.getMatches().contains(source))
                     {
                         return true;
                     }
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), subscription, e);
                 }
             }
         }
         return false;
     }
 
     public final Collection<IEventSource> getSubscribedSources(
         ISubscriptionParameters parameters)
     {
         checkActive();
         synchronized (getSubscriptions())
         {
             Collection<IEventSource> matches = CollectionFactory.newSet();
             for (ISubscription subscription : getSubscriptions().values())
             {
                 try
                 {
                     // check if the subscription would include the parameters
                     if (subscription.includes(parameters))
                     {
                         // now check the internal matches - if the subscription
                         // was more generalised than the parameters we need to
                         // filter out non-matching sources for the parameters
                         final Collection<IEventSource> potentialMatches =
                             subscription.getMatches();
                         for (IEventSource potentialMatch : potentialMatches)
                         {
                             if (parameters.includes(potentialMatch))
                             {
                                 matches.add(potentialMatch);
                             }
                         }
                     }
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), subscription, e);
                 }
             }
             return matches;
         }
     }
 
     public final Collection<IEventSource> getSubscribedSources()
     {
         checkActive();
         Collection<IEventSource> subscribed = CollectionFactory.newSet();
         synchronized (getSubscriptions())
         {
             for (ISubscription subscription : getSubscriptions().values())
             {
                 try
                 {
                     subscribed.addAll(subscription.getMatches());
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), subscription, e);
                 }
             }
         }
         return subscribed;
     }
 
     public final boolean includes(IAddressable parameters)
     {
         checkActive();
         synchronized (getSubscriptions())
         {
             for (ISubscription subscription : getSubscriptions().values())
             {
                 try
                 {
                     if (subscription.includes(parameters))
                     {
                         return true;
                     }
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), subscription, e);
                 }
             }
         }
         return false;
     }
 
     public final void eventSourceCreated(IAddressable identity)
     {
         if (getLog().isDebugEnabled())
         {
             getLog().debug(
                 "Checking subscriptions for CREATED event source "
                     + safeToString(identity));
         }
         int subscriptionMatchCount = 0;
         final IEventSource eventSource = doGetEventSource(identity);
         synchronized (this.getSubscriptions())
         {
             // Note: this is rather long-running as it has a time scale
             // proportional to n subscriptions
             final SubscriptionParameters subscriptionParameters =
                 new SubscriptionParameters(identity);
             for (ISubscription subscription : this.getSubscriptions().values())
             {
                 try
                 {
                     if (subscription.includes(subscriptionParameters))
                     {
                         subscriptionMatchCount++;
                         subscription.addMatch(eventSource);
                     }
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), subscription, e);
                 }
             }
         }
         if (subscriptionMatchCount > 0)
         {
             doEventSourceCreated(eventSource);
             if (getLog().isDebugEnabled())
             {
                 getLog().debug(
                     EMPTY_STRING + subscriptionMatchCount
                         + " matches found for CREATED "
                         + safeToString(identity));
             }
         }
     }
 
     public void eventSourceDestroyed(IAddressable identity)
     {
         if (getLog().isDebugEnabled())
         {
             getLog().debug(
                 "Checking subscriptions for DESTROYED "
                     + safeToString(identity));
         }
         int subscriptionMatchCount = 0;
         synchronized (this.getSubscriptions())
         {
             // Note: this is rather long-running as it has a time scale
             // proportional to n subscriptions
             for (ISubscription subscription : this.getSubscriptions().values())
             {
                 try
                 {
                     final SubscriptionParameters subscriptionParameters =
                         new SubscriptionParameters(identity);
                     if (subscription.includes(subscriptionParameters))
                     {
                         subscriptionMatchCount++;
                         subscription.removeMatch(subscriptionParameters);
                     }
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), subscription, e);
                 }
             }
         }
         if (subscriptionMatchCount > 0)
         {
             if (getLog().isDebugEnabled())
             {
                 getLog().debug(
                     EMPTY_STRING + subscriptionMatchCount
                         + " matches found for DESTROYED "
                         + safeToString(identity));
             }
         }
     }
 
     protected void doRaiseSubscribeEvent(ISubscriptionParameters parameters)
     {
         this.context.queueEvent(new SubscribeEvent(this.context, parameters));
     }
 
     protected void doRaiseUnsubscribeEvent(ISubscriptionParameters parameters)
     {
         this.context.queueEvent(new UnsubscribeEvent(this.context, parameters));
     }
 
     /**
      * Get the current subscriptions parameters to subscription. <b>Accessing
      * this must be synchronized</b>
      * 
      * @return the map of subscriptions parameters to subscriptions.
      */
     protected final Map<ISubscriptionParameters, ISubscription> getSubscriptions()
     {
         return this.subscriptions;
     }
 
     /**
      * Template method to get the collection of {@link IEventSource} instances
      * the manager will operate over.
      * 
      * @return a collection of {@link IEventSource} instances
      */
     protected abstract Collection<? extends IEventSource> doGetEventSources();
 
     /**
      * Template method to get the {@link IEventSource} identified by the
      * {@link IAddressable} argument.
      * 
      * @param id
      *            identifies the event source
      * @return the {@link IEventSource} implementation to use
      */
     protected abstract IEventSource doGetEventSource(IAddressable id);
 
     /**
      * Template method to identify if the {@link IEventSource} identified by the
      * {@link IAddressable} argument exists.
      * 
      * @param id
      *            identifies the event source
      * 
      * @return <code>true</code> if the container exists in the current context
      *         scope
      */
     protected abstract boolean doEventSourceExists(IAddressable id);
 
     /**
      * Called as the final step in the {@link #eventSourceCreated(IAddressable)}
      * method. This provides sub-classes with an opportunity to perform any post
      * creation work.
      * 
      * @param identity
      *            identifies the event source
      */
     protected abstract void doEventSourceCreated(IAddressable id);
 
     /**
      * Called as the final step in
      * {@link #addListener(ISubscriptionParameters, IEventListener)}.
      * Sub-classes can perform necessary closure work when the listener is
      * added.
      * 
      * @param parameters
      *            the subscription parameters identifying the subscripion(s)
      * @param listener
      *            the listener added
      */
     protected abstract void doAddListener(ISubscriptionParameters parameters,
         IEventListener listener);
 
     @Override
     protected void doDestroy()
     {
         synchronized (getSubscriptions())
         {
             Map<ISubscriptionParameters, ISubscription> copy =
                 CollectionFactory.newMap(getSubscriptions());
             for (ISubscription subscription : copy.values())
             {
                 try
                 {
                     unsubscribe(subscription);
                 }
                 catch (Exception e)
                 {
                     logException(getLog(), subscription, e);
                 }
             }
             getSubscriptions().clear();
         }
     }
 
     @Override
     protected AsyncLog getLog()
     {
         return LOG;
     }
 
     protected IFrameworkContext getContext()
     {
         return context;
     }
 
     @Override
     public final String toString()
     {
         StringBuilder sb;
         synchronized (getSubscriptions())
         {
             sb = CollectionUtils.toFormattedString(getSubscriptions());
         }
         return getClass().getSimpleName() + " subscriptions=" + sb.toString();
     }
 }