[jboss-cvs] JBossCache/src/org/jboss/cache/notifications ...

[Manik Surtani]

  User: msurtani
  Date: 07/06/14 11:19:24

  Modified:    src/org/jboss/cache/notifications  Notifier.java
  Log:
  JBCACHE-1107 and JBCACHE-1109
  
  Revision  Changes    Path
  1.29      +95 -36    JBossCache/src/org/jboss/cache/notifications/Notifier.java
  
  (In the diff below, changes in quantity of whitespace are not shown.)
  
  Index: Notifier.java
  ===================================================================
  RCS file: /cvsroot/jboss/JBossCache/src/org/jboss/cache/notifications/Notifier.java,v
  retrieving revision 1.28
  retrieving revision 1.29
  diff -u -b -r1.28 -r1.29
  --- Notifier.java	8 Jun 2007 15:52:15 -0000	1.28
  +++ Notifier.java	14 Jun 2007 15:19:24 -0000	1.29
  @@ -24,6 +24,9 @@
   import java.util.List;
   import java.util.Map;
   import java.util.Set;
  +import java.util.concurrent.Executor;
  +import java.util.concurrent.ExecutorService;
  +import java.util.concurrent.Executors;
   
   /**
    * Helper class that handles all notifications to registered listeners.
  @@ -55,6 +58,9 @@
      private static Class emptyMap = Collections.emptyMap().getClass();
      private static Class singletonMap = Collections.singletonMap(null, null).getClass();
   
  +   private ExecutorService executor;
  +   private int numExecutors;
  +
      static
      {
         try
  @@ -77,6 +83,15 @@
      public Notifier(Cache cache)
      {
         this.cache = cache;
  +      numExecutors = cache.getConfiguration().getNumberOfNotifierThreads();
  +   }
  +
  +   public void stop()
  +   {
  +      if (executor != null)
  +      {
  +         executor.shutdown();
  +      }
      }
   
      /**
  @@ -216,25 +231,31 @@
       * @param pre
       * @param ctx context of invocation
       */
  -   public void notifyNodeEvicted(Fqn fqn, boolean pre, InvocationContext ctx)
  +   public void notifyNodeEvicted(final Fqn fqn, final boolean pre, InvocationContext ctx)
      {
  -      boolean originLocal = ctx.isOriginLocal();
  +      final boolean originLocal = ctx.isOriginLocal();
         // always send eviction calls immediately - they do not go on the queue
         if (hasListenersOrEvictionListener)
         {
  -         InvocationContext backup = resetInvocationContext(ctx);
  +//         InvocationContext backup = resetInvocationContext(ctx);
            if (evictionPolicyListener != null)
            {
               evictionPolicyListener.nodeEvicted(fqn, pre, originLocal);
            }
            if (hasListeners)
            {
  +            getExecutor().execute(new Runnable()
  +            {
  +               public void run()
  +               {
               for (CacheListener listener : listeners)
               {
                  listener.nodeEvicted(fqn, pre, originLocal);
               }
            }
  -         restoreInvocationContext(backup);
  +            });
  +         }
  +//         restoreInvocationContext(backup);
         }
      }
   
  @@ -290,23 +311,29 @@
       * @param cache cache instance to notify
       * @param ctx   context of invocation
       */
  -   public void notifyCacheStarted(CacheSPI cache, InvocationContext ctx)
  +   public void notifyCacheStarted(final CacheSPI cache, InvocationContext ctx)
      {
         if (hasListenersOrEvictionListener)
         {
  -         InvocationContext backup = resetInvocationContext(ctx);
  +//         InvocationContext backup = resetInvocationContext(ctx);
            if (evictionPolicyListener != null)
            {
               evictionPolicyListener.cacheStarted(cache);
            }
            if (hasListeners)
            {
  +            getExecutor().execute(new Runnable()
  +            {
  +               public void run()
  +               {
               for (CacheListener listener : listeners)
               {
                  listener.cacheStarted(cache);
               }
            }
  -         restoreInvocationContext(backup);
  +            });
  +         }
  +//         restoreInvocationContext(backup);
         }
      }
   
  @@ -316,23 +343,30 @@
       * @param cache cache instance to notify
       * @param ctx   context of invocation
       */
  -   public void notifyCacheStopped(CacheSPI cache, InvocationContext ctx)
  +   public void notifyCacheStopped(final CacheSPI cache, InvocationContext ctx)
      {
         if (hasListenersOrEvictionListener)
         {
  -         InvocationContext backup = resetInvocationContext(ctx);
  +//         InvocationContext backup = resetInvocationContext(ctx);
            if (evictionPolicyListener != null)
            {
               evictionPolicyListener.cacheStopped(cache);
            }
            if (hasListeners)
            {
  +            getExecutor().execute(new Runnable()
  +            {
  +               public void run()
  +               {
               for (CacheListener listener : listeners)
               {
                  listener.cacheStopped(cache);
               }
            }
  -         restoreInvocationContext(backup);
  +            });
  +
  +         }
  +//         restoreInvocationContext(backup);
         }
      }
   
  @@ -343,23 +377,29 @@
       * @param new_view
       * @param ctx      context of invocation
       */
  -   public void notifyViewChange(View new_view, InvocationContext ctx)
  +   public void notifyViewChange(final View new_view, InvocationContext ctx)
      {
         if (hasListenersOrEvictionListener)
         {
  -         InvocationContext backup = resetInvocationContext(ctx);
  +//         InvocationContext backup = resetInvocationContext(ctx);
            if (evictionPolicyListener != null)
            {
               evictionPolicyListener.viewChange(new_view);
            }
            if (hasListeners)
            {
  +            getExecutor().execute(new Runnable()
  +            {
  +               public void run()
  +               {
               for (CacheListener listener : listeners)
               {
                  listener.viewChange(new_view);
               }
            }
  -         restoreInvocationContext(backup);
  +            });
  +         }
  +//         restoreInvocationContext(backup);
         }
      }
   
  @@ -413,8 +453,9 @@
       */
      public void invokeQueuedNotifications(InvocationContext ctx, List<MethodCall> queue)
      {
  -      InvocationContext backup = resetInvocationContext(ctx);
  -      for (MethodCall c : queue)
  +//      InvocationContext backup = resetInvocationContext(ctx);
  +
  +      for (final MethodCall c : queue)
         {
            if (log.isTraceEnabled()) log.trace("Invoking queued notification " + c);
            if (evictionPolicyListener != null)
  @@ -430,6 +471,10 @@
            }
            if (hasListeners)
            {
  +            getExecutor().execute(new Runnable()
  +            {
  +               public void run()
  +               {
               for (CacheListener listener : listeners)
               {
                  try
  @@ -442,8 +487,10 @@
                  }
               }
            }
  +            });
  +         }
         }
  -      restoreInvocationContext(backup);
  +//      restoreInvocationContext(backup);
      }
   
      /**
  @@ -455,4 +502,16 @@
      {
         invokeQueuedNotifications(ctx, ctx.getCacheListenerEvents());
      }
  +
  +   private Executor getExecutor()
  +   {
  +      if (executor == null)
  +      {
  +         synchronized (this)
  +         {
  +            if (executor == null) executor = Executors.newFixedThreadPool(numExecutors);
  +         }
  +      }
  +      return executor;
  +   }
   }