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 2020/02/21 12:04:30 UTC
[camel] 01/11: CAMEL-14596: camel-core - Optimize toD when
cacheSize = -1 to avoid creating endpoints. Introduced getPrototypeEndpoint
API.
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 0e5dde03039a6c85298acbd13330072875949a75
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Thu Feb 20 16:06:03 2020 +0100
CAMEL-14596: camel-core - Optimize toD when cacheSize = -1 to avoid creating endpoints. Introduced getPrototypeEndpoint API.
---
.../main/java/org/apache/camel/CamelContext.java | 28 +++++
.../camel/impl/engine/AbstractCamelContext.java | 60 ++++++++---
.../camel/processor/SendDynamicProcessor.java | 35 +++++-
.../apache/camel/impl/EndpointPrototypeTest.java | 118 +++++++++++++++++++++
.../camel/processor/ToDynamicNoCacheTest.java | 91 ++++++++++++++++
.../apache/camel/support/CamelContextHelper.java | 14 +++
.../org/apache/camel/support/ExchangeHelper.java | 21 ++++
7 files changed, 346 insertions(+), 21 deletions(-)
diff --git a/core/camel-api/src/main/java/org/apache/camel/CamelContext.java b/core/camel-api/src/main/java/org/apache/camel/CamelContext.java
index bb122b3..e834261 100644
--- a/core/camel-api/src/main/java/org/apache/camel/CamelContext.java
+++ b/core/camel-api/src/main/java/org/apache/camel/CamelContext.java
@@ -268,6 +268,17 @@ public interface CamelContext extends StatefulService, RuntimeConfiguration {
void addService(Object object, boolean stopOnShutdown, boolean forceStart) throws Exception;
/**
+ * Adds a service to this CamelContext (prototype scope).
+ * <p/>
+ * The service will also have {@link CamelContext} injected if its {@link CamelContextAware}.
+ * The service will be started, if its not already started.
+ *
+ * @param object the service
+ * @throws Exception can be thrown when starting the service
+ */
+ void addPrototypeService(Object object) throws Exception;
+
+ /**
* Removes a service from this CamelContext.
* <p/>
* The service is assumed to have been previously added using {@link #addService(Object)} method.
@@ -423,10 +434,27 @@ public interface CamelContext extends StatefulService, RuntimeConfiguration {
*
* @param uri the URI of the endpoint
* @return the endpoint
+ *
+ * @see #getPrototypeEndpoint(String)
*/
Endpoint getEndpoint(String uri);
/**
+ * Resolves the given name to an {@link Endpoint} of the specified type (scope is prototype).
+ * If the name has a singleton endpoint registered, then the singleton is returned.
+ * Otherwise, a new {@link Endpoint} is created.
+ *
+ * The endpoint is NOT registered in the {@link org.apache.camel.spi.EndpointRegistry} as its prototype scoped,
+ * and therefore expected to be short lived and discarded after use.
+ *
+ * @param uri the URI of the endpoint
+ * @return the endpoint
+ *
+ * @see #getEndpoint(String)
+ */
+ Endpoint getPrototypeEndpoint(String uri);
+
+ /**
* 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 in the {@link org.apache.camel.spi.EndpointRegistry}.
diff --git a/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java b/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
index bfbeec5..7e5592c 100644
--- a/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
+++ b/core/camel-base/src/main/java/org/apache/camel/impl/engine/AbstractCamelContext.java
@@ -731,6 +731,15 @@ public abstract class AbstractCamelContext extends ServiceSupport implements Ext
@Override
public Endpoint getEndpoint(String uri) {
+ return doGetEndpoint(uri, false);
+ }
+
+ @Override
+ public Endpoint getPrototypeEndpoint(String uri) {
+ return doGetEndpoint(uri, true);
+ }
+
+ protected Endpoint doGetEndpoint(String uri, boolean prototype) {
// ensure CamelContext are initialized before we can get an endpoint
init();
@@ -758,7 +767,8 @@ public abstract class AbstractCamelContext extends ServiceSupport implements Ext
String scheme = null;
// use optimized method to get the endpoint uri
EndpointKey key = getEndpointKeyPreNormalized(uri);
- answer = endpoints.get(key);
+ // only lookup and reuse existing endpoints if not prototype scoped
+ answer = prototype ? null : endpoints.get(key);
if (answer == null) {
try {
// Use the URI prefix to find the component.
@@ -820,8 +830,18 @@ public abstract class AbstractCamelContext extends ServiceSupport implements Ext
}
if (answer != null) {
- addService(answer);
- answer = addEndpointToRegistry(uri, answer);
+ if (!prototype) {
+ addService(answer);
+ // register in registry
+ answer = addEndpointToRegistry(uri, answer);
+ } else {
+ addPrototypeService(answer);
+ // if there is endpoint strategies, then use the endpoints they return
+ // as this allows to intercept endpoints etc.
+ for (EndpointStrategy strategy : endpointStrategies) {
+ answer = strategy.registerEndpoint(uri, answer);
+ }
+ }
}
} catch (Exception e) {
throw new ResolveEndpointFailedException(uri, e);
@@ -1387,7 +1407,12 @@ public abstract class AbstractCamelContext extends ServiceSupport implements Ext
@Override
public void addService(Object object, boolean stopOnShutdown, boolean forceStart) throws Exception {
- internalAddService(object, stopOnShutdown, forceStart);
+ internalAddService(object, stopOnShutdown, forceStart, true);
+ }
+
+ @Override
+ public void addPrototypeService(Object object) throws Exception {
+ doAddService(object, false, true, false);
}
protected <T> T doAddService(T object) {
@@ -1395,19 +1420,20 @@ public abstract class AbstractCamelContext extends ServiceSupport implements Ext
}
protected <T> T doAddService(T object, boolean stopOnShutdown) {
- return doAddService(object, stopOnShutdown, true);
+ return doAddService(object, stopOnShutdown, true, true);
}
- protected <T> T doAddService(T object, boolean stopOnShutdown, boolean forceStart) {
+ protected <T> T doAddService(T object, boolean stopOnShutdown, boolean forceStart, boolean useLifecycleStrategies) {
try {
- internalAddService(object, stopOnShutdown, forceStart);
+ internalAddService(object, stopOnShutdown, forceStart, useLifecycleStrategies);
} catch (Exception e) {
throw RuntimeCamelException.wrapRuntimeCamelException(e);
}
return object;
}
- private void internalAddService(Object object, boolean stopOnShutdown, boolean forceStart) throws Exception {
+ private void internalAddService(Object object, boolean stopOnShutdown,
+ boolean forceStart, boolean useLifecycleStrategies) throws Exception {
// inject CamelContext
if (object instanceof CamelContextAware) {
@@ -1418,13 +1444,15 @@ public abstract class AbstractCamelContext extends ServiceSupport implements Ext
if (object instanceof Service) {
Service service = (Service)object;
- for (LifecycleStrategy strategy : lifecycleStrategies) {
- if (service instanceof Endpoint) {
- // use specialized endpoint add
- strategy.onEndpointAdd((Endpoint)service);
- } else {
- Route route = setupRoute.get();
- strategy.onServiceAdd(this, service, route);
+ if (useLifecycleStrategies) {
+ for (LifecycleStrategy strategy : lifecycleStrategies) {
+ if (service instanceof Endpoint) {
+ // use specialized endpoint add
+ strategy.onEndpointAdd((Endpoint) service);
+ } else {
+ Route route = setupRoute.get();
+ strategy.onServiceAdd(this, service, route);
+ }
}
}
@@ -4098,7 +4126,7 @@ public abstract class AbstractCamelContext extends ServiceSupport implements Ext
@Override
public void setStreamCachingStrategy(StreamCachingStrategy streamCachingStrategy) {
- this.streamCachingStrategy = doAddService(streamCachingStrategy, true, false);
+ this.streamCachingStrategy = doAddService(streamCachingStrategy, true, false, true);
}
@Override
diff --git a/core/camel-base/src/main/java/org/apache/camel/processor/SendDynamicProcessor.java b/core/camel-base/src/main/java/org/apache/camel/processor/SendDynamicProcessor.java
index dabc2f9..11fa8ef 100644
--- a/core/camel-base/src/main/java/org/apache/camel/processor/SendDynamicProcessor.java
+++ b/core/camel-base/src/main/java/org/apache/camel/processor/SendDynamicProcessor.java
@@ -113,6 +113,7 @@ public class SendDynamicProcessor extends AsyncProcessorSupport implements IdAwa
Processor preAwareProcessor = null;
Processor postAwareProcessor = null;
String staticUri = null;
+ boolean prototype = cacheSize < 0;
try {
recipient = expression.evaluate(exchange, Object.class);
if (dynamicAware != null) {
@@ -134,10 +135,14 @@ public class SendDynamicProcessor extends AsyncProcessorSupport implements IdAwa
}
}
}
- if (staticUri != null) {
- endpoint = resolveEndpoint(exchange, staticUri);
+ Object targetRecipient = staticUri != null ? staticUri : recipient;
+ Endpoint existing = getExistingEndpoint(exchange, targetRecipient);
+ if (existing == null) {
+ endpoint = resolveEndpoint(exchange, targetRecipient, prototype);
} else {
- endpoint = resolveEndpoint(exchange, recipient);
+ endpoint = existing;
+ // we have an existing endpoint then its not a prototype scope
+ prototype = false;
}
if (endpoint == null) {
if (LOG.isDebugEnabled()) {
@@ -165,6 +170,7 @@ public class SendDynamicProcessor extends AsyncProcessorSupport implements IdAwa
final Processor postProcessor = postAwareProcessor;
// destination exchange pattern overrides pattern
final ExchangePattern pattern = destinationExchangePattern != null ? destinationExchangePattern : this.pattern;
+ final boolean stopEndpoint = prototype;
return producerCache.doInAsyncProducer(endpoint, exchange, callback, (p, e, c) -> {
final Exchange target = configureExchange(e, pattern, endpoint);
try {
@@ -191,6 +197,10 @@ public class SendDynamicProcessor extends AsyncProcessorSupport implements IdAwa
} catch (Throwable e) {
target.setException(e);
}
+ // stop endpoint if prototype as it was only used once
+ if (stopEndpoint) {
+ ServiceHelper.stopAndShutdownService(endpoint);
+ }
// signal we are done
c.done(doneSync);
}
@@ -228,7 +238,22 @@ public class SendDynamicProcessor extends AsyncProcessorSupport implements IdAwa
return ExchangeHelper.resolveScheme(uri);
}
- protected static Endpoint resolveEndpoint(Exchange exchange, Object recipient) throws NoTypeConversionAvailableException {
+ protected static Endpoint getExistingEndpoint(Exchange exchange, Object recipient) throws NoTypeConversionAvailableException {
+ // trim strings as end users might have added spaces between separators
+ if (recipient instanceof Endpoint) {
+ return (Endpoint) recipient;
+ } else if (recipient instanceof String) {
+ recipient = ((String) recipient).trim();
+ }
+ if (recipient != null) {
+ // convert to a string type we can work with
+ String uri = exchange.getContext().getTypeConverter().mandatoryConvertTo(String.class, exchange, recipient);
+ return exchange.getContext().hasEndpoint(uri);
+ }
+ return null;
+ }
+
+ protected static Endpoint resolveEndpoint(Exchange exchange, Object recipient, boolean prototype) throws NoTypeConversionAvailableException {
// trim strings as end users might have added spaces between separators
if (recipient instanceof String) {
recipient = ((String) recipient).trim();
@@ -240,7 +265,7 @@ public class SendDynamicProcessor extends AsyncProcessorSupport implements IdAwa
}
if (recipient != null) {
- return ExchangeHelper.resolveEndpoint(exchange, recipient);
+ return prototype ? ExchangeHelper.resolvePrototypeEndpoint(exchange, recipient) : ExchangeHelper.resolveEndpoint(exchange, recipient);
} else {
return null;
}
diff --git a/core/camel-core/src/test/java/org/apache/camel/impl/EndpointPrototypeTest.java b/core/camel-core/src/test/java/org/apache/camel/impl/EndpointPrototypeTest.java
new file mode 100644
index 0000000..c4523f7
--- /dev/null
+++ b/core/camel-core/src/test/java/org/apache/camel/impl/EndpointPrototypeTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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 java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Endpoint;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.support.LifecycleStrategySupport;
+import org.junit.Test;
+
+public class EndpointPrototypeTest extends ContextTestSupport {
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ public void testGetPrototype() throws Exception {
+ context.start();
+
+ assertEquals(0, context.getEndpointRegistry().size());
+
+ context.getEndpoint("mock:foo");
+
+ assertEquals(1, context.getEndpointRegistry().size());
+
+ // now get a prototype which should not be added
+
+ Endpoint prototype = context.getPrototypeEndpoint("mock:bar");
+ assertNotNull(prototype);
+
+ // and should be started
+ MockEndpoint bar = assertIsInstanceOf(MockEndpoint.class, prototype);
+ assertTrue(bar.getStatus().isStarted());
+
+ // but registry is still at 1
+ assertEquals(1, context.getEndpointRegistry().size());
+
+ // and now at 2
+ context.getEndpoint("mock:foo2");
+ assertEquals(2, context.getEndpointRegistry().size());
+
+ context.stop();
+
+ // should not be stopped as we need to handle that ourselves due to prototype scoped
+ assertFalse(bar.getStatus().isStopped());
+ bar.stop();
+ assertTrue(bar.getStatus().isStopped());
+ }
+
+ @Test
+ public void testGetPrototypeNoLifecycleStrategy() throws Exception {
+ final List<Endpoint> endpoints = new ArrayList<>();
+
+ LifecycleStrategySupport dummy = new LifecycleStrategySupport() {
+ @Override
+ public void onEndpointAdd(Endpoint endpoint) {
+ endpoints.add(endpoint);
+ }
+ };
+
+ context.addLifecycleStrategy(dummy);
+ context.start();
+
+ assertEquals(0, context.getEndpointRegistry().size());
+
+ context.getEndpoint("mock:foo");
+
+ assertEquals(1, context.getEndpointRegistry().size());
+
+ // now get a prototype which should not be added
+
+ Endpoint prototype = context.getPrototypeEndpoint("mock:bar");
+ assertNotNull(prototype);
+
+ // and should be started
+ MockEndpoint bar = assertIsInstanceOf(MockEndpoint.class, prototype);
+ assertTrue(bar.getStatus().isStarted());
+
+ // but registry is still at 1
+ assertEquals(1, context.getEndpointRegistry().size());
+
+ // and now at 2
+ context.getEndpoint("mock:foo2");
+ assertEquals(2, context.getEndpointRegistry().size());
+
+ context.stop();
+
+ // should not be stopped as we need to handle that ourselves due to prototype scoped
+ assertFalse(bar.getStatus().isStopped());
+ bar.stop();
+ assertTrue(bar.getStatus().isStopped());
+
+ // should only be mock:foo, mock:foo2, and no mock:bar
+ assertEquals(2, endpoints.size());
+ assertEquals("mock://foo", endpoints.get(0).getEndpointUri());
+ assertEquals("mock://foo2", endpoints.get(1).getEndpointUri());
+ }
+
+}
diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/ToDynamicNoCacheTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/ToDynamicNoCacheTest.java
new file mode 100644
index 0000000..29f1f88
--- /dev/null
+++ b/core/camel-core/src/test/java/org/apache/camel/processor/ToDynamicNoCacheTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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 java.util.List;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Test;
+
+public class ToDynamicNoCacheTest extends ContextTestSupport {
+
+ @Test
+ public void testNoCache() throws Exception {
+ assertEquals(1, context.getEndpointRegistry().size());
+
+ sendBody("foo", "mock:x");
+ sendBody("foo", "mock:y");
+ sendBody("foo", "mock:z");
+ sendBody("bar", "mock:x");
+ sendBody("bar", "mock:y");
+ sendBody("bar", "mock:z");
+
+ // make sure its using an empty producer cache as the cache is disabled
+ List<Processor> list = context.getRoute("route1").filter("foo");
+ SendDynamicProcessor sdp = (SendDynamicProcessor) list.get(0);
+ assertNotNull(sdp);
+ assertEquals(-1, sdp.getCacheSize());
+
+ // check no additional endpoints added as cache was disabled
+ assertEquals(1, context.getEndpointRegistry().size());
+
+ // now send again with mocks which then add endpoints
+
+ MockEndpoint x = getMockEndpoint("mock:x");
+ MockEndpoint y = getMockEndpoint("mock:y");
+ MockEndpoint z = getMockEndpoint("mock:z");
+
+ x.expectedBodiesReceived("foo", "bar");
+ y.expectedBodiesReceived("foo", "bar");
+ z.expectedBodiesReceived("foo", "bar");
+
+ assertEquals(4, context.getEndpointRegistry().size());
+
+ sendBody("foo", "mock:x");
+ sendBody("foo", "mock:y");
+ sendBody("foo", "mock:z");
+ sendBody("bar", "mock:x");
+ sendBody("bar", "mock:y");
+ sendBody("bar", "mock:z");
+
+ // should not register as new endpoint so we keep at 4
+ sendBody("dummy", "mock:dummy");
+
+ assertMockEndpointsSatisfied();
+
+ assertEquals(4, context.getEndpointRegistry().size());
+ }
+
+ protected void sendBody(String body, String uri) {
+ template.sendBodyAndHeader("direct:a", body, "myHeader", uri);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:a")
+ .toD("${header.myHeader}", -1).id("foo");
+ }
+ };
+
+ }
+
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/CamelContextHelper.java b/core/camel-support/src/main/java/org/apache/camel/support/CamelContextHelper.java
index c4eab76..48a4510 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/CamelContextHelper.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/CamelContextHelper.java
@@ -63,6 +63,20 @@ public final class CamelContextHelper {
}
/**
+ * Returns the mandatory endpoint (prototype scope) for the given URI or the
+ * {@link org.apache.camel.NoSuchEndpointException} is thrown
+ */
+ public static Endpoint getMandatoryPrototypeEndpoint(CamelContext camelContext, String uri)
+ throws NoSuchEndpointException {
+ Endpoint endpoint = camelContext.getPrototypeEndpoint(uri);
+ if (endpoint == null) {
+ throw new NoSuchEndpointException(uri);
+ } else {
+ return endpoint;
+ }
+ }
+
+ /**
* Returns the mandatory endpoint for the given URI and type or the
* {@link org.apache.camel.NoSuchEndpointException} is thrown
*/
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
index f66b576..bfac356 100644
--- a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
+++ b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
@@ -100,6 +100,27 @@ public final class ExchangeHelper {
}
/**
+ * Attempts to resolve the endpoint (prototype scope) for the given value
+ *
+ * @param exchange the message exchange being processed
+ * @param value the value which can be an {@link Endpoint} or an object
+ * which provides a String representation of an endpoint via
+ * {@link #toString()}
+ * @return the endpoint
+ * @throws NoSuchEndpointException if the endpoint cannot be resolved
+ */
+ public static Endpoint resolvePrototypeEndpoint(Exchange exchange, Object value) throws NoSuchEndpointException {
+ Endpoint endpoint;
+ if (value instanceof Endpoint) {
+ endpoint = (Endpoint) value;
+ } else {
+ String uri = value.toString().trim();
+ endpoint = CamelContextHelper.getMandatoryPrototypeEndpoint(exchange.getContext(), uri);
+ }
+ return endpoint;
+ }
+
+ /**
* Gets the mandatory property of the exchange of the correct type
*
* @param exchange the exchange