svn commit: r1293855 - /camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java

[davsclaus]

Author: davsclaus
Date: Sun Feb 26 15:43:25 2012
New Revision: 1293855

URL: http://svn.apache.org/viewvc?rev=1293855&view=rev
Log:
CAMEL-5045: Fixed memory leak when removing routes.

Modified:
    
camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java

Modified: 
camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
URL: 
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java?rev=1293855&r1=1293854&r2=1293855&view=diff
==============================================================================
--- 
camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
 (original)
+++ 
camel/trunk/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
 Sun Feb 26 15:43:25 2012
@@ -20,6 +20,7 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -63,6 +64,7 @@ import org.apache.camel.model.OnCompleti
 import org.apache.camel.model.OnExceptionDefinition;
 import org.apache.camel.model.PolicyDefinition;
 import org.apache.camel.model.ProcessorDefinition;
+import org.apache.camel.model.ProcessorDefinitionHelper;
 import org.apache.camel.model.RouteDefinition;
 import org.apache.camel.processor.interceptor.Tracer;
 import org.apache.camel.spi.EventNotifier;
@@ -93,6 +95,8 @@ import org.slf4j.LoggerFactory;
 public class DefaultManagementLifecycleStrategy extends ServiceSupport 
implements LifecycleStrategy, CamelContextAware {
 
     private static final Logger LOG = 
LoggerFactory.getLogger(DefaultManagementLifecycleStrategy.class);
+    // the wrapped processors is for performance counters, which are in use 
for the created routes
+    // when a route is removed, we should remove the associated processors 
from this map
     private final Map<Processor, KeyValueHolder<ProcessorDefinition<?>, 
InstrumentationProcessor>> wrappedProcessors =
             new HashMap<Processor, KeyValueHolder<ProcessorDefinition<?>, 
InstrumentationProcessor>>();
     private final List<PreRegisterService> preServices = new 
ArrayList<PreRegisterService>();
@@ -520,7 +524,14 @@ public class DefaultManagementLifecycleS
             } catch (Exception e) {
                 LOG.warn("Could not unregister Route MBean", e);
             }
+
+            // remove from known routes ids, as the route has been removed
+            knowRouteIds.remove(route.getId());
         }
+
+        // after the routes has been removed, we should clear the wrapped 
processors as we no longer need them
+        // as they were just a provisional map used during creation of routes
+        removeWrappedProcessorsForRoutes(routes);
     }
 
     public void onErrorHandlerAdd(RouteContext routeContext, Processor 
errorHandler, ErrorHandlerFactory errorHandlerBuilder) {
@@ -592,6 +603,30 @@ public class DefaultManagementLifecycleS
         routeContext.setManagedInterceptStrategy(new 
InstrumentationInterceptStrategy(registeredCounters, wrappedProcessors));
     }
 
+    /**
+     * Removes the wrapped processors for the given routes, as they are no 
longer in use.
+     * <p/>
+     * This is needed to avoid accumulating memory, if a lot of routes is 
being added and removed.
+     *
+     * @param routes the routes
+     */
+    private void removeWrappedProcessorsForRoutes(Collection<Route> routes) {
+        // loop the routes, and remove the route associated wrapped 
processors, as they are no longer in use
+        for (Route route : routes) {
+            String id = route.getId();
+
+            Iterator<KeyValueHolder<ProcessorDefinition<?>, 
InstrumentationProcessor>> it = wrappedProcessors.values().iterator();
+            while (it.hasNext()) {
+                KeyValueHolder<ProcessorDefinition<?>, 
InstrumentationProcessor> holder = it.next();
+                RouteDefinition def = 
ProcessorDefinitionHelper.getRoute(holder.getKey());
+                if (def != null && id.equals(def.getId())) {
+                    it.remove();
+                }
+            }
+        }
+        
+    }
+
     private void registerPerformanceCounters(RouteContext routeContext, 
ProcessorDefinition<?> processor,
                                              Map<ProcessorDefinition<?>, 
PerformanceCounter> registeredCounters) {
 
@@ -758,6 +793,8 @@ public class DefaultManagementLifecycleS
         initialized = false;
         knowRouteIds.clear();
         preServices.clear();
+        wrappedProcessors.clear();
+        managedTracers.clear();
         ServiceHelper.stopService(timerListenerManager);
     }