You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2015/01/18 12:07:37 UTC
[4/6] camel git commit: CAMEL-7965: EndpointRegistry should remove
endpoints which are no longer in use when removing routes. The trick is to
detect if endpoints is shared and therefore not to be removed yet.
CAMEL-7965: EndpointRegistry should remove endpoints which are no longer in use when removing routes. The trick is to detect if endpoints is shared and therefore not to be removed yet.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/017fd45b
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/017fd45b
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/017fd45b
Branch: refs/heads/master
Commit: 017fd45b0ac350df786db36875df38b0db7a6c12
Parents: f993d18
Author: Claus Ibsen <da...@apache.org>
Authored: Sun Jan 18 09:42:07 2015 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Sun Jan 18 09:42:07 2015 +0100
----------------------------------------------------------------------
.../java/org/apache/camel/CamelContext.java | 38 ++++++---
.../java/org/apache/camel/DelegateEndpoint.java | 3 +
.../main/java/org/apache/camel/Producer.java | 4 +-
.../src/main/java/org/apache/camel/Route.java | 3 +-
.../management/mbean/CamelOpenMBeanTypes.java | 6 +-
.../apache/camel/impl/DefaultCamelContext.java | 45 ++++++++--
.../org/apache/camel/impl/DefaultConsumer.java | 3 +-
.../camel/impl/DefaultEndpointRegistry.java | 10 +++
.../org/apache/camel/impl/RouteService.java | 69 +++++++++++-----
.../mbean/ManagedEndpointRegistry.java | 5 +-
.../apache/camel/processor/EndpointAware.java | 32 +++++++
.../org/apache/camel/processor/Enricher.java | 6 +-
.../apache/camel/processor/SendProcessor.java | 9 +-
.../camel/processor/UnitOfWorkProducer.java | 3 -
.../camel/processor/WireTapProcessor.java | 6 +-
.../org/apache/camel/spi/EndpointRegistry.java | 16 ++++
...eEndpointsRemoteRouteSharedEndpointTest.java | 87 ++++++++++++++++++++
...gistryKeepRouteEndpointsRemoteRouteTest.java | 9 +-
.../EndpointRegistryKeepRouteEndpointsTest.java | 3 +
19 files changed, 294 insertions(+), 63 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/CamelContext.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/CamelContext.java b/camel-core/src/main/java/org/apache/camel/CamelContext.java
index 9836321..06e2d9e 100644
--- a/camel-core/src/main/java/org/apache/camel/CamelContext.java
+++ b/camel-core/src/main/java/org/apache/camel/CamelContext.java
@@ -319,12 +319,15 @@ public interface CamelContext extends SuspendableService, RuntimeConfiguration {
// Endpoint Management Methods
//-----------------------------------------------------------------------
- EndpointRegistry getEndpointRegistry();
+ /**
+ * Gets the {@link org.apache.camel.spi.EndpointRegistry}
+ */
+ EndpointRegistry<String> getEndpointRegistry();
/**
* Resolves the given name to an {@link Endpoint} of the specified type.
* If the name has a singleton endpoint registered, then the singleton is returned.
- * Otherwise, a new {@link Endpoint} is created and registered.
+ * Otherwise, a new {@link Endpoint} is created and registered in the {@link org.apache.camel.spi.EndpointRegistry}.
*
* @param uri the URI of the endpoint
* @return the endpoint
@@ -334,7 +337,7 @@ public interface CamelContext extends SuspendableService, RuntimeConfiguration {
/**
* Resolves the given name to an {@link Endpoint} of the specified type.
* If the name has a singleton endpoint registered, then the singleton is returned.
- * Otherwise, a new {@link Endpoint} is created and registered.
+ * Otherwise, a new {@link Endpoint} is created and registered in the {@link org.apache.camel.spi.EndpointRegistry}.
*
* @param name the name of the endpoint
* @param endpointType the expected type
@@ -343,22 +346,21 @@ public interface CamelContext extends SuspendableService, RuntimeConfiguration {
<T extends Endpoint> T getEndpoint(String name, Class<T> endpointType);
/**
- * Returns the collection of all registered endpoints.
+ * Returns a new {@link Collection} of all of the endpoints from the {@link org.apache.camel.spi.EndpointRegistry}
*
* @return all endpoints
*/
Collection<Endpoint> getEndpoints();
/**
- * Returns a new Map containing all of the active endpoints with the key of the map being their
- * unique key.
+ * Returns a new {@link Map} containing all of the endpoints from the {@link org.apache.camel.spi.EndpointRegistry}
*
- * @return map of active endpoints
+ * @return map of endpoints
*/
Map<String, Endpoint> getEndpointMap();
/**
- * Is the given endpoint already registered?
+ * Is the given endpoint already registered in the {@link org.apache.camel.spi.EndpointRegistry}
*
* @param uri the URI of the endpoint
* @return the registered endpoint or <tt>null</tt> if not registered
@@ -366,30 +368,40 @@ public interface CamelContext extends SuspendableService, RuntimeConfiguration {
Endpoint hasEndpoint(String uri);
/**
- * Adds the endpoint to the context using the given URI.
+ * Adds the endpoint to the {@link org.apache.camel.spi.EndpointRegistry} using the given URI.
*
* @param uri the URI to be used to resolve this endpoint
- * @param endpoint the endpoint to be added to the context
+ * @param endpoint the endpoint to be added to the registry
* @return the old endpoint that was previously registered or <tt>null</tt> if none was registered
* @throws Exception if the new endpoint could not be started or the old endpoint could not be stopped
*/
Endpoint addEndpoint(String uri, Endpoint endpoint) throws Exception;
/**
- * Removes all endpoints with the given URI.
+ * Removes the endpoint from the {@link org.apache.camel.spi.EndpointRegistry}.
+ * <p/>
+ * The endpoint being removed will be stopped first.
+ *
+ * @param endpoint the endpoint
+ * @throws Exception if the endpoint could not be stopped
+ */
+ void removeEndpoint(Endpoint endpoint) throws Exception;
+
+ /**
+ * Removes all endpoints with the given URI from the {@link org.apache.camel.spi.EndpointRegistry}.
* <p/>
* The endpoints being removed will be stopped first.
*
* @param pattern an uri or pattern to match
* @return a collection of endpoints removed which could be empty if there are no endpoints found for the given <tt>pattern</tt>
* @throws Exception if at least one endpoint could not be stopped
- * @see org.apache.camel.util.EndpointHelper#matchEndpoint(CamelContext, String, String) for pattern
+ * @see org.apache.camel.util.EndpointHelper#matchEndpoint(CamelContext, String, String) for pattern
*/
Collection<Endpoint> removeEndpoints(String pattern) throws Exception;
/**
* Registers a {@link org.apache.camel.spi.EndpointStrategy callback} to allow you to do custom
- * logic when an {@link Endpoint} is about to be registered to the {@link CamelContext} endpoint registry.
+ * logic when an {@link Endpoint} is about to be registered to the {@link org.apache.camel.spi.EndpointRegistry}.
* <p/>
* When a callback is added it will be executed on the already registered endpoints allowing you to catch-up
*
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/DelegateEndpoint.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/DelegateEndpoint.java b/camel-core/src/main/java/org/apache/camel/DelegateEndpoint.java
index aa5ec93..f44a725 100644
--- a/camel-core/src/main/java/org/apache/camel/DelegateEndpoint.java
+++ b/camel-core/src/main/java/org/apache/camel/DelegateEndpoint.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel;
+/**
+ * An interface to represent an {@link org.apache.camel.Endpoint} which are delegated.
+ */
public interface DelegateEndpoint extends Endpoint {
/**
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/Producer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/Producer.java b/camel-core/src/main/java/org/apache/camel/Producer.java
index 6c2c466..21f22cd 100644
--- a/camel-core/src/main/java/org/apache/camel/Producer.java
+++ b/camel-core/src/main/java/org/apache/camel/Producer.java
@@ -16,13 +16,15 @@
*/
package org.apache.camel;
+import org.apache.camel.processor.EndpointAware;
+
/**
* Provides a channel on which clients can create and invoke message exchanges
* on an {@link Endpoint}
*
* @version
*/
-public interface Producer extends Processor, Service, IsSingleton {
+public interface Producer extends Processor, Service, IsSingleton, EndpointAware {
/**
* Gets the endpoint this producer sends to.
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/Route.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/Route.java b/camel-core/src/main/java/org/apache/camel/Route.java
index 3adb442..a04ca98 100644
--- a/camel-core/src/main/java/org/apache/camel/Route.java
+++ b/camel-core/src/main/java/org/apache/camel/Route.java
@@ -19,6 +19,7 @@ package org.apache.camel;
import java.util.List;
import java.util.Map;
+import org.apache.camel.processor.EndpointAware;
import org.apache.camel.spi.RouteContext;
/**
@@ -30,7 +31,7 @@ import org.apache.camel.spi.RouteContext;
* such as starting and stopping using the {@link org.apache.camel.CamelContext#startRoute(String)}
* and {@link org.apache.camel.CamelContext#stopRoute(String)} methods.
*/
-public interface Route {
+public interface Route extends EndpointAware {
String ID_PROPERTY = "id";
String PARENT_PROPERTY = "parent";
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java b/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java
index 40c8625..728e7ef 100644
--- a/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java
+++ b/camel-core/src/main/java/org/apache/camel/api/management/mbean/CamelOpenMBeanTypes.java
@@ -61,9 +61,9 @@ public final class CamelOpenMBeanTypes {
}
public static CompositeType listEndpointsCompositeType() throws OpenDataException {
- return new CompositeType("endpoints", "Endpoints", new String[]{"url"},
- new String[]{"Url"},
- new OpenType[]{SimpleType.STRING});
+ return new CompositeType("endpoints", "Endpoints", new String[]{"url", "static", "dynamic"},
+ new String[]{"Url", "Static", "Dynamic"},
+ new OpenType[]{SimpleType.STRING, SimpleType.BOOLEAN, SimpleType.BOOLEAN});
}
public static TabularType explainEndpointTabularType() throws OpenDataException {
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
index 7971252..f867f69 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
@@ -474,6 +474,10 @@ public class DefaultCamelContext extends ServiceSupport implements ModelCamelCon
return oldEndpoint;
}
+ public void removeEndpoint(Endpoint endpoint) throws Exception {
+ removeEndpoints(endpoint.getEndpointUri());
+ }
+
public Collection<Endpoint> removeEndpoints(String uri) throws Exception {
Collection<Endpoint> answer = new ArrayList<Endpoint>();
Endpoint oldEndpoint = endpoints.remove(getEndpointKey(uri));
@@ -945,6 +949,14 @@ public class DefaultCamelContext extends ServiceSupport implements ModelCamelCon
ErrorHandlerBuilderSupport builder = (ErrorHandlerBuilderSupport)getErrorHandlerBuilder();
builder.removeOnExceptionList(routeId);
}
+
+ // gather a map of all the endpoints in use by the routes, so we can known if a given endpoints is in use
+ // by one or more routes, when we remove the route
+ Map<String, Set<Endpoint>> endpointsInUse = new HashMap<String, Set<Endpoint>>();
+ for (Map.Entry<String, RouteService> entry : routeServices.entrySet()) {
+ endpointsInUse.put(entry.getKey(), entry.getValue().gatherEndpoints());
+ }
+
RouteService routeService = routeServices.get(routeId);
if (routeService != null) {
if (getRouteStatus(routeId).isStopped()) {
@@ -960,6 +972,27 @@ public class DefaultCamelContext extends ServiceSupport implements ModelCamelCon
it.remove();
}
}
+
+ // from the route which we have removed, then remove all its private endpoints
+ // (eg the endpoints which are not in use by other routes)
+ Set<Endpoint> toRemove = new LinkedHashSet<Endpoint>();
+ for (Endpoint endpoint : endpointsInUse.get(routeId)) {
+ // how many times is the endpoint in use
+ int count = 0;
+ for (Set<Endpoint> endpoints : endpointsInUse.values()) {
+ if (endpoints.contains(endpoint)) {
+ count++;
+ }
+ }
+ // notice we will count ourselves so if there is only 1 then its safe to remove
+ if (count <= 1) {
+ toRemove.add(endpoint);
+ }
+ }
+ for (Endpoint endpoint : toRemove) {
+ log.debug("Removing: {} which was only in use by route: {}", endpoint, routeId);
+ removeEndpoint(endpoint);
+ }
return true;
} else {
return false;
@@ -1056,16 +1089,14 @@ public class DefaultCamelContext extends ServiceSupport implements ModelCamelCon
}
public boolean removeService(Object object) throws Exception {
+ if (object instanceof Endpoint) {
+ removeEndpoint((Endpoint) object);
+ return true;
+ }
if (object instanceof Service) {
Service service = (Service) object;
-
for (LifecycleStrategy strategy : lifecycleStrategies) {
- if (service instanceof Endpoint) {
- // use specialized endpoint remove
- strategy.onEndpointRemove((Endpoint) service);
- } else {
- strategy.onServiceRemove(this, service, null);
- }
+ strategy.onServiceRemove(this, service, null);
}
return servicesToClose.remove(service);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/impl/DefaultConsumer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultConsumer.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultConsumer.java
index a141238..5a439cf 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultConsumer.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultConsumer.java
@@ -23,6 +23,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.Route;
import org.apache.camel.RouteAware;
+import org.apache.camel.processor.EndpointAware;
import org.apache.camel.spi.ExceptionHandler;
import org.apache.camel.spi.UnitOfWork;
import org.apache.camel.support.ServiceSupport;
@@ -38,7 +39,7 @@ import org.slf4j.LoggerFactory;
*
* @version
*/
-public class DefaultConsumer extends ServiceSupport implements Consumer, RouteAware {
+public class DefaultConsumer extends ServiceSupport implements Consumer, RouteAware, EndpointAware {
protected final Logger log = LoggerFactory.getLogger(getClass());
private final Endpoint endpoint;
private final Processor processor;
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpointRegistry.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpointRegistry.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpointRegistry.java
index 96565f6..87e595e 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpointRegistry.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultEndpointRegistry.java
@@ -187,6 +187,16 @@ public class DefaultEndpointRegistry extends LRUCache<EndpointKey, Endpoint> imp
}
@Override
+ public boolean isStatic(EndpointKey key) {
+ return staticMap.containsKey(key);
+ }
+
+ @Override
+ public boolean isDynamic(EndpointKey key) {
+ return super.containsKey(key);
+ }
+
+ @Override
public void stop() throws Exception {
ServiceHelper.stopServices(staticMap.values());
ServiceHelper.stopServices(values());
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/impl/RouteService.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/RouteService.java b/camel-core/src/main/java/org/apache/camel/impl/RouteService.java
index 0986a71..e93f48b 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/RouteService.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/RouteService.java
@@ -28,6 +28,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.camel.CamelContext;
import org.apache.camel.Channel;
import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
import org.apache.camel.Processor;
import org.apache.camel.Route;
import org.apache.camel.RouteAware;
@@ -36,6 +37,7 @@ import org.apache.camel.model.OnCompletionDefinition;
import org.apache.camel.model.OnExceptionDefinition;
import org.apache.camel.model.ProcessorDefinition;
import org.apache.camel.model.RouteDefinition;
+import org.apache.camel.processor.EndpointAware;
import org.apache.camel.processor.ErrorHandler;
import org.apache.camel.spi.LifecycleStrategy;
import org.apache.camel.spi.RouteContext;
@@ -95,6 +97,25 @@ public class RouteService extends ChildServiceSupport {
}
/**
+ * Gather all the endpoints this route service uses
+ */
+ public Set<Endpoint> gatherEndpoints() {
+ Set<Endpoint> answer = new LinkedHashSet<Endpoint>();
+ for (Route route : routes) {
+ Set<Service> services = gatherChildServices(route, true);
+ for (Service service : services) {
+ if (service instanceof EndpointAware) {
+ Endpoint endpoint = ((EndpointAware) service).getEndpoint();
+ if (endpoint != null) {
+ answer.add(endpoint);
+ }
+ }
+ }
+ }
+ return answer;
+ }
+
+ /**
* Gets the inputs to the routes.
*
* @return list of {@link Consumer} as inputs for the routes
@@ -207,19 +228,10 @@ public class RouteService extends ChildServiceSupport {
LOG.debug("Stopping services on route: {}", route.getId());
// gather list of services to stop as we need to start child services as well
- List<Service> services = new ArrayList<Service>();
- services.addAll(route.getServices());
- // also get route scoped services
- doGetRouteScopedServices(services, route);
- Set<Service> list = new LinkedHashSet<Service>();
- for (Service service : services) {
- list.addAll(ServiceHelper.getChildServices(service));
- }
- // also get route scoped error handler (which must be done last)
- doGetRouteScopedErrorHandler(list, route);
+ Set<Service> services = gatherChildServices(route, true);
// stop services
- stopChildService(route, list, isShutdownCamelContext);
+ stopChildService(route, services, isShutdownCamelContext);
// stop the route itself
if (isShutdownCamelContext) {
@@ -250,19 +262,10 @@ public class RouteService extends ChildServiceSupport {
LOG.debug("Shutting down services on route: {}", route.getId());
// gather list of services to stop as we need to start child services as well
- List<Service> services = new ArrayList<Service>();
- services.addAll(route.getServices());
- // also get route scoped services
- doGetRouteScopedServices(services, route);
- Set<Service> list = new LinkedHashSet<Service>();
- for (Service service : services) {
- list.addAll(ServiceHelper.getChildServices(service));
- }
- // also get route scoped error handler (which must be done last)
- doGetRouteScopedErrorHandler(list, route);
+ Set<Service> services = gatherChildServices(route, true);
// shutdown services
- stopChildService(route, list, true);
+ stopChildService(route, services, true);
// shutdown the route itself
ServiceHelper.stopAndShutdownServices(route);
@@ -359,6 +362,28 @@ public class RouteService extends ChildServiceSupport {
}
/**
+ * Gather all child services
+ */
+ private Set<Service> gatherChildServices(Route route, boolean includeErrorHandler) {
+ // gather list of services to stop as we need to start child services as well
+ List<Service> services = new ArrayList<Service>();
+ services.addAll(route.getServices());
+ // also get route scoped services
+ doGetRouteScopedServices(services, route);
+ Set<Service> list = new LinkedHashSet<Service>();
+ for (Service service : services) {
+ list.addAll(ServiceHelper.getChildServices(service));
+ }
+ if (includeErrorHandler) {
+ // also get route scoped error handler (which must be done last)
+ doGetRouteScopedErrorHandler(list, route);
+ }
+ Set<Service> answer = new LinkedHashSet<Service>();
+ answer.addAll(list);
+ return answer;
+ }
+
+ /**
* Gather the route scoped error handler from the given route
*/
private void doGetRouteScopedErrorHandler(Set<Service> services, Route route) {
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedEndpointRegistry.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedEndpointRegistry.java b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedEndpointRegistry.java
index c4489fb..98d6524 100644
--- a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedEndpointRegistry.java
+++ b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedEndpointRegistry.java
@@ -80,7 +80,10 @@ public class ManagedEndpointRegistry extends ManagedService implements ManagedEn
CompositeType ct = CamelOpenMBeanTypes.listEndpointsCompositeType();
String url = endpoint.getEndpointUri();
- CompositeData data = new CompositeDataSupport(ct, new String[]{"url"}, new Object[]{url});
+ boolean fromStatic = endpointRegistry.isStatic(url);
+ boolean fromDynamic = endpointRegistry.isDynamic(url);
+
+ CompositeData data = new CompositeDataSupport(ct, new String[]{"url", "static", "dynamic"}, new Object[]{url, fromStatic, fromDynamic});
answer.put(data);
}
return answer;
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/processor/EndpointAware.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/EndpointAware.java b/camel-core/src/main/java/org/apache/camel/processor/EndpointAware.java
new file mode 100644
index 0000000..6cc38a3
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/processor/EndpointAware.java
@@ -0,0 +1,32 @@
+/**
+ * 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
+ *
+ * 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 org.apache.camel.processor;
+
+import org.apache.camel.Endpoint;
+
+/**
+ * An interface to represent a {@link org.apache.camel.Processor} that uses an {@link Endpoint}
+ */
+public interface EndpointAware {
+
+ /**
+ * Gets the endpoint
+ *
+ * @return the endpoint
+ */
+ Endpoint getEndpoint();
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/processor/Enricher.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/Enricher.java b/camel-core/src/main/java/org/apache/camel/processor/Enricher.java
index 9ab0aa9..33a9af9 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/Enricher.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/Enricher.java
@@ -48,7 +48,7 @@ import static org.apache.camel.util.ExchangeHelper.copyResultsPreservePattern;
*
* @see PollEnricher
*/
-public class Enricher extends ServiceSupport implements AsyncProcessor {
+public class Enricher extends ServiceSupport implements AsyncProcessor, EndpointAware {
private static final Logger LOG = LoggerFactory.getLogger(Enricher.class);
private AggregationStrategy aggregationStrategy;
@@ -110,6 +110,10 @@ public class Enricher extends ServiceSupport implements AsyncProcessor {
this.aggregationStrategy = defaultAggregationStrategy();
}
+ public Endpoint getEndpoint() {
+ return producer.getEndpoint();
+ }
+
public void process(Exchange exchange) throws Exception {
AsyncProcessorHelper.process(this, exchange);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
index efd6a60..e3f75e5 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/SendProcessor.java
@@ -48,7 +48,7 @@ import org.slf4j.LoggerFactory;
*
* @version
*/
-public class SendProcessor extends ServiceSupport implements AsyncProcessor, Traceable {
+public class SendProcessor extends ServiceSupport implements AsyncProcessor, Traceable, EndpointAware {
protected static final Logger LOG = LoggerFactory.getLogger(SendProcessor.class);
protected final CamelContext camelContext;
protected final ExchangePattern pattern;
@@ -96,7 +96,12 @@ public class SendProcessor extends ServiceSupport implements AsyncProcessor, Tra
public String getTraceLabel() {
return URISupport.sanitizeUri(destination.getEndpointUri());
}
-
+
+ @Override
+ public Endpoint getEndpoint() {
+ return destination;
+ }
+
public void process(final Exchange exchange) throws Exception {
AsyncProcessorHelper.process(this, exchange);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/processor/UnitOfWorkProducer.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/UnitOfWorkProducer.java b/camel-core/src/main/java/org/apache/camel/processor/UnitOfWorkProducer.java
index 2b5f522..b3dc81b 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/UnitOfWorkProducer.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/UnitOfWorkProducer.java
@@ -23,8 +23,6 @@ import org.apache.camel.ExchangePattern;
import org.apache.camel.Producer;
import org.apache.camel.util.AsyncProcessorHelper;
import org.apache.camel.util.ServiceHelper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Ensures a {@link Producer} is executed within an {@link org.apache.camel.spi.UnitOfWork}.
@@ -33,7 +31,6 @@ import org.slf4j.LoggerFactory;
*/
public final class UnitOfWorkProducer implements Producer {
- private static final Logger LOG = LoggerFactory.getLogger(UnitOfWorkProducer.class);
private final Producer producer;
private final AsyncProcessor processor;
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java b/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
index 7f8d250..6614aa4 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
@@ -43,7 +43,7 @@ import org.slf4j.LoggerFactory;
*
* @version
*/
-public class WireTapProcessor extends ServiceSupport implements AsyncProcessor, Traceable {
+public class WireTapProcessor extends ServiceSupport implements AsyncProcessor, Traceable, EndpointAware {
private static final Logger LOG = LoggerFactory.getLogger(WireTapProcessor.class);
private final Endpoint destination;
private final Processor processor;
@@ -78,6 +78,10 @@ public class WireTapProcessor extends ServiceSupport implements AsyncProcessor,
return "wireTap(" + destination + ")";
}
+ public Endpoint getEndpoint() {
+ return destination;
+ }
+
public void process(Exchange exchange) throws Exception {
AsyncProcessorHelper.process(this, exchange);
}
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/main/java/org/apache/camel/spi/EndpointRegistry.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/spi/EndpointRegistry.java b/camel-core/src/main/java/org/apache/camel/spi/EndpointRegistry.java
index a3c3f57..1b7e69c 100644
--- a/camel-core/src/main/java/org/apache/camel/spi/EndpointRegistry.java
+++ b/camel-core/src/main/java/org/apache/camel/spi/EndpointRegistry.java
@@ -59,4 +59,20 @@ public interface EndpointRegistry<K> extends Map<K, Endpoint>, StaticService {
*/
void purge();
+ /**
+ * Whether the given endpoint is stored in the static cache
+ *
+ * @param key the endpoint key
+ * @return <tt>true</tt> if in static cache, <tt>false</tt> if not
+ */
+ boolean isStatic(K key);
+
+ /**
+ * Whether the given endpoint is stored in the dynamic cache
+ *
+ * @param key the endpoint key
+ * @return <tt>true</tt> if in dynamic cache, <tt>false</tt> if not
+ */
+ boolean isDynamic(K key);
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteSharedEndpointTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteSharedEndpointTest.java b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteSharedEndpointTest.java
new file mode 100644
index 0000000..922e35a
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteSharedEndpointTest.java
@@ -0,0 +1,87 @@
+/**
+ * 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
+ *
+ * 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 org.apache.camel.impl;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+
+public class EndpointRegistryKeepRouteEndpointsRemoteRouteSharedEndpointTest extends ContextTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getProperties().put(Exchange.MAXIMUM_ENDPOINT_CACHE_SIZE, "20");
+ return context;
+ }
+
+ public void testEndpointRegistryKeepRouteEndpointsRemoveRouteSharedEndpoint() throws Exception {
+ assertTrue(context.hasEndpoint("direct://start") != null);
+ assertTrue(context.hasEndpoint("log://start") != null);
+ assertTrue(context.hasEndpoint("log://foo") != null);
+ assertTrue(context.hasEndpoint("log://bar") != null);
+ assertTrue(context.hasEndpoint("mock://result") != null);
+ assertTrue(context.hasEndpoint("direct://bar") != null);
+ assertTrue(context.hasEndpoint("log://private") != null);
+
+ assertEquals(7, context.getEndpointRegistry().staticSize());
+
+ // we dont have this endpoint yet
+ assertFalse(context.hasEndpoint("mock://unknown0") != null);
+
+ for (int i = 0; i < 50; i++) {
+ template.sendBody("mock:unknown" + i, "Hello " + i);
+ }
+
+ assertEquals(7, context.getEndpointRegistry().staticSize());
+ assertTrue(context.hasEndpoint("direct://start") != null);
+ assertTrue(context.hasEndpoint("log://start") != null);
+ assertTrue(context.hasEndpoint("log://foo") != null);
+ assertTrue(context.hasEndpoint("log://bar") != null);
+ assertTrue(context.hasEndpoint("mock://result") != null);
+ assertTrue(context.hasEndpoint("direct://bar") != null);
+ assertTrue(context.hasEndpoint("log://private") != null);
+
+ // now stop and remove the bar route
+ context.stopRoute("bar");
+ context.removeRoute("bar");
+
+ assertEquals(5, context.getEndpointRegistry().staticSize());
+ assertTrue(context.hasEndpoint("direct://start") != null);
+ assertTrue(context.hasEndpoint("log://start") != null);
+ assertTrue(context.hasEndpoint("log://foo") != null);
+ // log:bar is still in use, so should still be here
+ assertTrue(context.hasEndpoint("log://bar") != null);
+ assertTrue(context.hasEndpoint("mock://result") != null);
+ assertFalse(context.hasEndpoint("direct://bar") != null);
+ assertFalse(context.hasEndpoint("log://private") != null);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start").routeId("foo").to("log:start").to("log:foo").to("log:bar").to("mock:result");
+
+ from("direct:bar").routeId("bar").to("log:private").to("log:bar");
+ }
+ };
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteTest.java b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteTest.java
index 29e2595..1b0acdb 100644
--- a/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteTest.java
+++ b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsRemoteRouteTest.java
@@ -47,6 +47,7 @@ public class EndpointRegistryKeepRouteEndpointsRemoteRouteTest extends ContextTe
template.sendBody("mock:unknown" + i, "Hello " + i);
}
+ assertEquals(6, context.getEndpointRegistry().staticSize());
assertTrue(context.hasEndpoint("direct://start") != null);
assertTrue(context.hasEndpoint("log://start") != null);
assertTrue(context.hasEndpoint("log://foo") != null);
@@ -54,23 +55,17 @@ public class EndpointRegistryKeepRouteEndpointsRemoteRouteTest extends ContextTe
assertTrue(context.hasEndpoint("direct://bar") != null);
assertTrue(context.hasEndpoint("log://bar") != null);
- assertEquals(6, context.getEndpointRegistry().staticSize());
-
// now stop and remove the bar route
context.stopRoute("bar");
context.removeRoute("bar");
- // TODO: need to remove endpoints when removing the route
- /*
+ assertEquals(4, context.getEndpointRegistry().staticSize());
assertTrue(context.hasEndpoint("direct://start") != null);
assertTrue(context.hasEndpoint("log://start") != null);
assertTrue(context.hasEndpoint("log://foo") != null);
assertTrue(context.hasEndpoint("mock://result") != null);
assertFalse(context.hasEndpoint("direct://bar") != null);
assertFalse(context.hasEndpoint("log://bar") != null);
-
- assertEquals(4, context.getEndpointRegistry().staticSize());
- */
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/017fd45b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsTest.java b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsTest.java
index e7ac30f..676ab2a 100644
--- a/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsTest.java
+++ b/camel-core/src/test/java/org/apache/camel/impl/EndpointRegistryKeepRouteEndpointsTest.java
@@ -58,6 +58,9 @@ public class EndpointRegistryKeepRouteEndpointsTest extends ContextTestSupport {
// we should have 4 static, 20 dynamic and 24 in total
assertEquals(4, context.getEndpointRegistry().staticSize());
+ assertTrue(context.getEndpointRegistry().isStatic("direct://start"));
+ assertTrue(context.getEndpointRegistry().isDynamic("mock://unknown49"));
+
assertEquals(20, context.getEndpointRegistry().dynamicSize());
assertEquals(24, context.getEndpointRegistry().size());