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 2016/02/19 16:40:23 UTC
[2/2] camel git commit: CAMEL-8736 - Create new JCache component
CAMEL-8736 - Create new JCache component
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/4496eea4
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/4496eea4
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/4496eea4
Branch: refs/heads/master
Commit: 4496eea4f08fc49edcd943d646749536a3e66d02
Parents: d7339b2
Author: lburgazzoli <lb...@gmail.com>
Authored: Fri Feb 5 13:04:44 2016 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Feb 19 16:37:12 2016 +0100
----------------------------------------------------------------------
components/camel-jcache/pom.xml | 77 +++++
.../camel-jcache/src/main/docs/jcache.adoc | 31 ++
.../camel/component/jcache/JCacheComponent.java | 44 +++
.../component/jcache/JCacheConfiguration.java | 302 +++++++++++++++++
.../camel/component/jcache/JCacheConstants.java | 28 ++
.../camel/component/jcache/JCacheConsumer.java | 116 +++++++
.../camel/component/jcache/JCacheEndpoint.java | 73 ++++
.../jcache/JCacheEntryEventFilters.java | 76 +++++
.../jcache/JCacheEntryEventListener.java | 53 +++
.../camel/component/jcache/JCacheManager.java | 157 +++++++++
.../camel/component/jcache/JCacheProducer.java | 334 +++++++++++++++++++
.../aggregate/JCacheAggregationRepository.java | 206 ++++++++++++
.../idempotent/JCacheIdempotentRepository.java | 125 +++++++
.../services/org/apache/camel/component/jcache | 1 +
.../jcache/JCacheComponentTestSupport.java | 50 +++
.../jcache/JCacheConfigurationTest.java | 118 +++++++
.../component/jcache/JCacheConsumerTest.java | 121 +++++++
.../jcache/JCacheProducerClearTest.java | 51 +++
.../component/jcache/JCacheProducerGetTest.java | 212 ++++++++++++
.../jcache/JCacheProducerInvokeTest.java | 134 ++++++++
.../component/jcache/JCacheProducerPutTest.java | 160 +++++++++
.../jcache/JCacheProducerRemoveTest.java | 165 +++++++++
.../jcache/JCacheProducerReplaceTest.java | 176 ++++++++++
.../JCacheAggregationRepositoryRoutesTest.java | 106 ++++++
.../JCacheAggregationRepositoryTest.java | 154 +++++++++
.../JCacheAggregationRepositoryTestSupport.java | 32 ++
.../JCacheIdempotentRepositoryTest.java | 104 ++++++
.../src/test/resources/log4j.properties | 35 ++
components/pom.xml | 1 +
parent/pom.xml | 1 +
30 files changed, 3243 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-jcache/pom.xml b/components/camel-jcache/pom.xml
new file mode 100644
index 0000000..33009f4
--- /dev/null
+++ b/components/camel-jcache/pom.xml
@@ -0,0 +1,77 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components</artifactId>
+ <version>2.17-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-jcache</artifactId>
+ <packaging>bundle</packaging>
+ <name>Camel :: JCache</name>
+ <description>Camel JCache support</description>
+
+ <properties>
+ <camel.osgi.export.pkg>
+ org.apache.camel.component.jcache.*,
+ org.apache.camel.component.jcache.processor.idempotent.*,
+ </camel.osgi.export.pkg>
+ <camel.osgi.export.service>
+ org.apache.camel.spi.ComponentResolver;component=jcache
+ </camel.osgi.export.service>
+ </properties>
+
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.servicemix.bundles</groupId>
+ <artifactId>org.apache.servicemix.bundles.javax-cache-api</artifactId>
+ <version>${jcache-bundle-version}</version>
+ </dependency>
+
+ <!-- testing -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.hazelcast</groupId>
+ <artifactId>hazelcast</artifactId>
+ <version>${hazelcast-version}</version>
+ <scope>test</scope>
+ </dependency>
+
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <version>${mockito-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>java-hamcrest</artifactId>
+ <version>${hamcrest-version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/docs/jcache.adoc
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/docs/jcache.adoc b/components/camel-jcache/src/main/docs/jcache.adoc
new file mode 100644
index 0000000..5defa57
--- /dev/null
+++ b/components/camel-jcache/src/main/docs/jcache.adoc
@@ -0,0 +1,31 @@
+[[JCache-JCache]]
+JCache
+~~~~~~
+
+
+// endpoint options: START
+The JCache component supports 5 endpoint options which are listed below:
+
+[width="100%",cols="2s,1,1m,1m,5",options="header"]
+|=======================================================================
+| Name | Group | Default | Java Type | Description
+| cacheName | common | | String | *Required* the cache name
+| bridgeErrorHandler | consumer | false | boolean | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN/ERROR level and ignored.
+| exceptionHandler | consumer (advanced) | | ExceptionHandler | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN/ERROR level and ignored.
+| exchangePattern | advanced | InOnly | ExchangePattern | Sets the default exchange pattern when creating an exchange
+| synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
+|=======================================================================
+// endpoint options: END
+
+
+
+// component options: START
+The JCache component has no options.
+
+
+[width="100%",cols="2s,1m,8",options="header"]
+|=======================================================================
+| Name | Java Type | Description
+|=======================================================================
+// component options: END
+
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheComponent.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheComponent.java
new file mode 100644
index 0000000..c94af86
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheComponent.java
@@ -0,0 +1,44 @@
+/**
+ * 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.component.jcache;
+
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.UriEndpointComponent;
+
+/**
+ * Represents the component that manages {@link JCacheEndpoint}.
+ */
+public class JCacheComponent extends UriEndpointComponent {
+ public JCacheComponent() {
+ super(JCacheEndpoint.class);
+ }
+
+ public JCacheComponent(CamelContext context) {
+ super(context, JCacheEndpoint.class);
+ }
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> parameters) throws Exception {
+ JCacheConfiguration configuration = new JCacheConfiguration(getCamelContext());
+ setProperties(configuration, parameters);
+
+ return new JCacheEndpoint(uri, this, configuration, remaining);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConfiguration.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConfiguration.java
new file mode 100644
index 0000000..af83153
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConfiguration.java
@@ -0,0 +1,302 @@
+/**
+ * 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.component.jcache;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+import javax.cache.Cache;
+import javax.cache.CacheManager;
+import javax.cache.configuration.Configuration;
+import javax.cache.configuration.Factory;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.EventType;
+import javax.cache.expiry.ExpiryPolicy;
+import javax.cache.integration.CacheLoader;
+import javax.cache.integration.CacheWriter;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.spi.UriParam;
+import org.apache.camel.spi.UriParams;
+import org.apache.camel.util.EndpointHelper;
+
+@UriParams
+public class JCacheConfiguration {
+ @UriParam(label = "advanced")
+ private String cachingProvider;
+
+ @UriParam(label = "advanced")
+ private Configuration cacheConfiguration;
+
+ @UriParam(label = "advanced")
+ private Properties cacheConfigurationProperties;
+
+ @UriParam(label = "advanced")
+ private String configurationUri;
+
+ @UriParam(label = "advanced")
+ private Factory<CacheLoader> cacheLoaderFactory;
+
+ @UriParam(label = "advanced")
+ private Factory<CacheWriter> cacheWriterFactory;
+
+ @UriParam(label = "advanced")
+ private Factory<ExpiryPolicy> expiryPolicyFactory;
+
+ @UriParam
+ private boolean readThrough;
+
+ @UriParam
+ private boolean writeThrough;
+
+ @UriParam(defaultValue = "true")
+ private boolean storeByValue = true;
+
+ @UriParam
+ private boolean statisticsEnabled;
+
+ @UriParam
+ private boolean managementEnabled;
+
+ @UriParam(label = "consumer", enums = "CREATED,UPDATED,REMOVED,EXPIRED")
+ private List<EventType> filteredEvents;
+
+ @UriParam(label = "consumer,advanced")
+ private List<CacheEntryEventFilter> eventFilters;
+
+ @UriParam(label = "consumer")
+ private boolean oldValueRequired;
+
+ @UriParam(label = "consumer")
+ private boolean synchronous;
+
+ @UriParam(label = "producer")
+ private String action;
+
+
+ private final CamelContext camelContext;
+
+ public JCacheConfiguration() {
+ this(null);
+ }
+
+ public JCacheConfiguration(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ /**
+ * The fully qualified class name of the {@link javax.cache.spi.CachingProvider}
+ */
+ public String getCachingProvider() {
+ return cachingProvider;
+ }
+
+ public void setCachingProvider(String cachingProvider) {
+ this.cachingProvider = cachingProvider;
+ }
+
+ /**
+ * A {@link Configuration} for the {@link Cache}
+ */
+ public Configuration getCacheConfiguration() {
+ return cacheConfiguration;
+ }
+
+ public void setCacheConfiguration(Configuration cacheConfiguration) {
+ this.cacheConfiguration = cacheConfiguration;
+ }
+
+ /**
+ * The {@link Properties} for the {@link javax.cache.spi.CachingProvider} to
+ * create the {@link CacheManager}
+ */
+ public Properties getCacheConfigurationProperties() {
+ return cacheConfigurationProperties;
+ }
+
+ public void setCacheConfigurationProperties(Properties cacheConfigurationProperties) {
+ this.cacheConfigurationProperties = cacheConfigurationProperties;
+ }
+
+ /**
+ * An implementation specific URI for the {@link CacheManager}
+ */
+ public String getConfigurationUri() {
+ return configurationUri;
+ }
+
+ public void setConfigurationUri(String configurationUri) {
+ this.configurationUri = configurationUri;
+ }
+
+ /**
+ * The {@link CacheLoader} factory
+ */
+ public Factory<CacheLoader> getCacheLoaderFactory() {
+ return cacheLoaderFactory;
+ }
+
+ public void setCacheLoaderFactory(Factory<CacheLoader> cacheLoaderFactory) {
+ this.cacheLoaderFactory = cacheLoaderFactory;
+ }
+
+ /**
+ * The {@link CacheWriter} factory
+ */
+ public Factory<CacheWriter> getCacheWriterFactory() {
+ return cacheWriterFactory;
+ }
+
+ public void setCacheWriterFactory(Factory<CacheWriter> cacheWriterFactory) {
+ this.cacheWriterFactory = cacheWriterFactory;
+ }
+
+ /**
+ * The {@link ExpiryPolicy} factory
+ */
+ public Factory<ExpiryPolicy> getExpiryPolicyFactory() {
+ return expiryPolicyFactory;
+ }
+
+ public void setExpiryPolicyFactory(Factory<ExpiryPolicy> expiryPolicyFactory) {
+ this.expiryPolicyFactory = expiryPolicyFactory;
+ }
+
+ /**
+ * If read-through caching should be used
+ */
+ public boolean isReadThrough() {
+ return readThrough;
+ }
+
+ public void setReadThrough(boolean readThrough) {
+ this.readThrough = readThrough;
+ }
+
+ /**
+ * If write-through caching should be used
+ */
+ public boolean isWriteThrough() {
+ return writeThrough;
+ }
+
+ public void setWriteThrough(boolean writeThrough) {
+ this.writeThrough = writeThrough;
+ }
+
+ /**
+ * If cache should use store-by-value or store-by-reference semantics
+ */
+ public boolean isStoreByValue() {
+ return storeByValue;
+ }
+
+ public void setStoreByValue(boolean storeByValue) {
+ this.storeByValue = storeByValue;
+ }
+
+ /**
+ * Whether statistics gathering is enabled
+ */
+ public boolean isStatisticsEnabled() {
+ return statisticsEnabled;
+ }
+
+ public void setStatisticsEnabled(boolean statisticsEnabled) {
+ this.statisticsEnabled = statisticsEnabled;
+ }
+
+ /**
+ * Whether management gathering is enabled
+ */
+ public boolean isManagementEnabled() {
+ return managementEnabled;
+ }
+
+ public void setManagementEnabled(boolean managementEnabled) {
+ this.managementEnabled = managementEnabled;
+ }
+
+ /**
+ * Events a consumer should filter
+ */
+ public List<EventType> getFilteredEvents() {
+ return filteredEvents;
+ }
+
+ public void setFilteredEvents(List<EventType> filteredEvents) {
+ this.filteredEvents = filteredEvents;
+ }
+
+ public void setFilteredEvents(String filteredEvents) {
+ this.filteredEvents = new ArrayList<>();
+
+ for (String event : filteredEvents.split(",")) {
+ this.filteredEvents.add(EventType.valueOf(event));
+ }
+ }
+
+ /**
+ * The CacheEntryEventFilter
+ */
+ public List<CacheEntryEventFilter> getEventFilters() {
+ return eventFilters;
+ }
+
+ public void setEventFilters(List<CacheEntryEventFilter> eventFilters) {
+ this.eventFilters = new LinkedList<>(eventFilters);
+ }
+
+ public void setEventFilters(String eventFilter) {
+ this.eventFilters = EndpointHelper.resolveReferenceListParameter(camelContext, eventFilter, CacheEntryEventFilter.class);
+ }
+
+ /**
+ * if the old value is required for events
+ */
+ public boolean isOldValueRequired() {
+ return oldValueRequired;
+ }
+
+ public void setOldValueRequired(boolean oldValueRequired) {
+ this.oldValueRequired = oldValueRequired;
+ }
+
+ /**
+ * if the the event listener should block the thread causing the event
+ */
+ public boolean isSynchronous() {
+ return synchronous;
+ }
+
+ public void setSynchronous(boolean synchronous) {
+ this.synchronous = synchronous;
+ }
+
+ public String getAction() {
+ return action;
+ }
+
+ /**
+ * To configure using a cache operation by default. If an operation in the
+ * message header, then the operation from the header takes precedence.
+ */
+ public void setAction(String action) {
+ this.action = action;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConstants.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConstants.java
new file mode 100644
index 0000000..42bcd79
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConstants.java
@@ -0,0 +1,28 @@
+/**
+ * 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.component.jcache;
+
+public interface JCacheConstants {
+ String ACTION = "CamelJCacheAction";
+ String RESULT = "CamelJCacheResult";
+ String EVENT_TYPE = "CamelJCacheEventType";
+ String KEY = "CamelJCacheKey";
+ String KEYS = "CamelJCacheKeys";
+ String OLD_VALUE = "CamelJCacheOldValue";
+ String ENTRY_PROCESSOR = "CamelJCacheEntryProcessor";
+ String ARGUMENTS = "CamelJCacheEntryArgs";
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConsumer.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConsumer.java
new file mode 100644
index 0000000..1b348ad
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheConsumer.java
@@ -0,0 +1,116 @@
+/**
+ * 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.component.jcache;
+
+import javax.cache.Cache;
+import javax.cache.configuration.CacheEntryListenerConfiguration;
+import javax.cache.configuration.Factory;
+import javax.cache.configuration.MutableCacheEntryListenerConfiguration;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.CacheEntryListener;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.Processor;
+import org.apache.camel.impl.DefaultConsumer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The JCache consumer.
+ */
+public class JCacheConsumer extends DefaultConsumer {
+ private static final Logger LOGGER = LoggerFactory.getLogger(JCacheConsumer.class);
+
+ private CacheEntryListenerConfiguration<Object, Object> entryListenerConfiguration;
+
+ public JCacheConsumer(final JCacheEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ this.entryListenerConfiguration = null;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (entryListenerConfiguration != null) {
+ getCache().deregisterCacheEntryListener(entryListenerConfiguration);
+ }
+
+ entryListenerConfiguration = createEntryListenerConfiguration();
+ getCache().registerCacheEntryListener(entryListenerConfiguration);
+
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (entryListenerConfiguration != null) {
+ getCache().deregisterCacheEntryListener(entryListenerConfiguration);
+ entryListenerConfiguration = null;
+ }
+
+ super.doStop();
+ }
+
+ private JCacheEndpoint getJCacheEndpoint() {
+ return (JCacheEndpoint)getEndpoint();
+ }
+
+ private Cache getCache() throws Exception {
+ return getJCacheEndpoint().getManager().getCache();
+ }
+
+ private CacheEntryListenerConfiguration<Object, Object> createEntryListenerConfiguration() {
+ return new MutableCacheEntryListenerConfiguration<>(
+ new Factory<CacheEntryListener<Object, Object>>() {
+ @Override
+ public CacheEntryListener<Object, Object> create() {
+ return new JCacheEntryEventListener() {
+ @Override
+ protected void onEvents(Iterable<CacheEntryEvent<?, ?>> events) {
+ for (CacheEntryEvent<?, ?> event : events) {
+ Exchange exchange = getEndpoint().createExchange();
+ Message message = exchange.getIn();
+ message.setHeader(JCacheConstants.EVENT_TYPE, event.getEventType().name());
+ message.setHeader(JCacheConstants.KEY, event.getKey());
+ message.setBody(event.getValue());
+
+ if (event.isOldValueAvailable()) {
+ message.setHeader(JCacheConstants.OLD_VALUE, event.getOldValue());
+ }
+
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ LOGGER.error("Error processing event ", e);
+ }
+ }
+ }
+ };
+ }
+ },
+ new Factory<CacheEntryEventFilter<Object, Object>>() {
+ @Override
+ public CacheEntryEventFilter<Object, Object> create() {
+ return getJCacheEndpoint().getManager().getEventFilter();
+ }
+ },
+ getJCacheEndpoint().getManager().getConfiguration().isOldValueRequired(),
+ getJCacheEndpoint().getManager().getConfiguration().isSynchronous()
+ );
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEndpoint.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEndpoint.java
new file mode 100644
index 0000000..b81ad00
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEndpoint.java
@@ -0,0 +1,73 @@
+/**
+ * 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.component.jcache;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriPath;
+
+/**
+ * Represents a JCache endpoint.
+ */
+@UriEndpoint(scheme = "jcache", title = "JCache", syntax = "jcache:cacheName", consumerClass = JCacheConsumer.class, label = "cache,datagrid,clustering")
+public class JCacheEndpoint extends DefaultEndpoint {
+ @UriPath(description = "the cache name")
+ @Metadata(required = "true")
+ private final String cacheName;
+ private final JCacheConfiguration cacheCnfiguration;
+ private final JCacheManager<Object, Object> cacheManager;
+
+ public JCacheEndpoint(String uri, JCacheComponent component, JCacheConfiguration configuration, String cacheName) {
+ super(uri, component);
+
+ this.cacheName = cacheName;
+ this.cacheCnfiguration = configuration;
+ this.cacheManager = new JCacheManager<>(
+ configuration,
+ cacheName,
+ getCamelContext().getApplicationContextClassLoader(),
+ super.getCamelContext());
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ return new JCacheProducer(this, cacheCnfiguration);
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new JCacheConsumer(this, processor);
+ }
+
+ @Override
+ public boolean isSingleton() {
+ return true;
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ cacheManager.close();
+ }
+
+ JCacheManager getManager() {
+ return cacheManager;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEntryEventFilters.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEntryEventFilters.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEntryEventFilters.java
new file mode 100644
index 0000000..08dbfd5
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEntryEventFilters.java
@@ -0,0 +1,76 @@
+/**
+ * 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.component.jcache;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.EventType;
+
+class JCacheEntryEventFilters {
+ public static class Named implements CacheEntryEventFilter<Object, Object> {
+ private List<EventType> filteredEvents;
+
+ public Named(Collection<EventType> filteredEventNames) {
+ if (filteredEventNames != null && !filteredEventNames.isEmpty()) {
+ this.filteredEvents = new ArrayList<>(filteredEventNames);
+ }
+ }
+
+ @Override
+ public boolean evaluate(CacheEntryEvent<?, ?> event) throws CacheEntryListenerException {
+ if (filteredEvents == null) {
+ return true;
+ }
+
+ return !filteredEvents.contains(event.getEventType());
+ }
+ }
+
+ public static class Chained implements CacheEntryEventFilter<Object, Object> {
+ private final List<CacheEntryEventFilter> filteredEvents;
+ private final int filteredEventsSize;
+
+ public Chained(List<CacheEntryEventFilter> filteredEvents) {
+ if (filteredEvents != null && !filteredEvents.isEmpty()) {
+ this.filteredEvents = new ArrayList<>(filteredEvents);
+ this.filteredEventsSize = this.filteredEvents.size();
+ } else {
+ this.filteredEvents = null;
+ this.filteredEventsSize = 0;
+ }
+ }
+
+ @Override
+ public boolean evaluate(CacheEntryEvent<?, ?> event) throws CacheEntryListenerException {
+ if (filteredEvents == null) {
+ return true;
+ }
+
+ for (int i = 0; i < filteredEventsSize; i++) {
+ if (!filteredEvents.get(i).evaluate(event)) {
+ return false;
+ }
+ }
+
+ return true;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEntryEventListener.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEntryEventListener.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEntryEventListener.java
new file mode 100644
index 0000000..697f4c0
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheEntryEventListener.java
@@ -0,0 +1,53 @@
+/**
+ * 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.component.jcache;
+
+import javax.cache.event.CacheEntryCreatedListener;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryExpiredListener;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.CacheEntryRemovedListener;
+import javax.cache.event.CacheEntryUpdatedListener;
+
+class JCacheEntryEventListener implements CacheEntryCreatedListener<Object, Object>,
+ CacheEntryUpdatedListener<Object, Object>,
+ CacheEntryRemovedListener<Object, Object>,
+ CacheEntryExpiredListener<Object, Object> {
+
+ @Override
+ public void onCreated(Iterable<CacheEntryEvent<?, ?>> events) throws CacheEntryListenerException {
+ onEvents(events);
+ }
+
+ @Override
+ public void onExpired(Iterable<CacheEntryEvent<?, ?>> events) throws CacheEntryListenerException {
+ onEvents(events);
+ }
+
+ @Override
+ public void onRemoved(Iterable<CacheEntryEvent<?, ?>> events) throws CacheEntryListenerException {
+ onEvents(events);
+ }
+
+ @Override
+ public void onUpdated(Iterable<CacheEntryEvent<?, ?>> events) throws CacheEntryListenerException {
+ onEvents(events);
+ }
+
+ protected void onEvents(Iterable<CacheEntryEvent<?, ?>> events) {
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheManager.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheManager.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheManager.java
new file mode 100644
index 0000000..5ec98b2
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheManager.java
@@ -0,0 +1,157 @@
+/**
+ * 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.component.jcache;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.net.URI;
+import javax.cache.Cache;
+import javax.cache.CacheManager;
+import javax.cache.Caching;
+import javax.cache.configuration.Configuration;
+import javax.cache.configuration.MutableConfiguration;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.spi.CachingProvider;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.util.ObjectHelper;
+
+public class JCacheManager<K, V> implements Closeable {
+ private JCacheConfiguration configuration;
+ private CachingProvider provider;
+ private CacheManager manager;
+ private ClassLoader classLoader;
+ private Cache<K, V> cache;
+ private String cacheName;
+ private CamelContext camelContext;
+
+ public JCacheManager(JCacheConfiguration configuration, String cacheName) {
+ this(configuration, cacheName, null, null);
+ }
+
+ public JCacheManager(JCacheConfiguration configuration, String cacheName, CamelContext camelContext) {
+ this(configuration, cacheName, null, camelContext);
+ }
+
+ public JCacheManager(JCacheConfiguration configuration, String cacheName, ClassLoader classLoader) {
+ this(configuration, cacheName, classLoader, null);
+ }
+
+ public JCacheManager(JCacheConfiguration configuration, String cacheName, ClassLoader classLoader, CamelContext camelContext) {
+ this.configuration = configuration;
+ this.provider = null;
+ this.manager = null;
+ this.classLoader = classLoader;
+ this.cache = null;
+ this.cacheName = cacheName;
+ this.camelContext = camelContext;
+ }
+
+ public JCacheManager(Cache<K, V> cache) {
+ this.cache = cache;
+ this.configuration = null;
+ this.cacheName = cache.getName();
+ this.provider = null;
+ this.manager = null;
+ this.camelContext = null;
+ }
+
+ public String getCacheName() {
+ return this.cacheName;
+ }
+
+ public JCacheConfiguration getConfiguration() {
+ return this.configuration;
+ }
+
+ public synchronized Cache<K, V> getCache() throws Exception {
+ if (cache == null) {
+ String uri = configuration.getConfigurationUri();
+ if (uri != null && camelContext != null) {
+ uri = camelContext.resolvePropertyPlaceholders(uri);
+ }
+
+ provider = configuration.getCachingProvider() != null
+ ? Caching.getCachingProvider(configuration.getCachingProvider())
+ : Caching.getCachingProvider();
+
+ manager = provider.getCacheManager(
+ ObjectHelper.isNotEmpty(uri) ? URI.create(uri) : null,
+ classLoader,
+ configuration.getCacheConfigurationProperties());
+
+ cache = manager.getCache(cacheName);
+ if (cache == null) {
+ cache = manager.createCache(
+ cacheName,
+ getOrCreateCacheConfiguration());
+ }
+ }
+
+ return cache;
+ }
+
+ @Override
+ public synchronized void close() throws IOException {
+ if (configuration != null) {
+ if (cache != null) {
+ cache.close();
+ }
+
+ if (manager != null) {
+ manager.close();
+ }
+
+ if (provider != null) {
+ provider.close();
+ }
+ }
+ }
+
+
+ Configuration getOrCreateCacheConfiguration() {
+ if (configuration.getCacheConfiguration() != null) {
+ return configuration.getCacheConfiguration();
+ }
+
+ MutableConfiguration mutableConfiguration = new MutableConfiguration();
+ if (configuration.getCacheLoaderFactory() != null) {
+ mutableConfiguration.setCacheLoaderFactory(configuration.getCacheLoaderFactory());
+ }
+ if (configuration.getCacheWriterFactory() != null) {
+ mutableConfiguration.setCacheWriterFactory(configuration.getCacheWriterFactory());
+ }
+ if (configuration.getExpiryPolicyFactory() != null) {
+ mutableConfiguration.setExpiryPolicyFactory(configuration.getExpiryPolicyFactory());
+ }
+
+ mutableConfiguration.setManagementEnabled(configuration.isManagementEnabled());
+ mutableConfiguration.setStatisticsEnabled(configuration.isStatisticsEnabled());
+ mutableConfiguration.setReadThrough(configuration.isReadThrough());
+ mutableConfiguration.setStoreByValue(configuration.isStoreByValue());
+
+ return mutableConfiguration;
+ }
+
+ CacheEntryEventFilter getEventFilter() {
+ if (configuration.getEventFilters() != null) {
+ return new JCacheEntryEventFilters.Chained(configuration.getEventFilters());
+ }
+
+ return new JCacheEntryEventFilters.Named(configuration.getFilteredEvents());
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheProducer.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheProducer.java
new file mode 100644
index 0000000..c3ff9fe
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/JCacheProducer.java
@@ -0,0 +1,334 @@
+/**
+ * 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.component.jcache;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import javax.cache.Cache;
+import javax.cache.processor.EntryProcessor;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.impl.DefaultProducer;
+import org.apache.camel.util.ObjectHelper;
+
+/**
+ * The JCache producer.
+ */
+public class JCacheProducer extends DefaultProducer {
+ private final JCacheConfiguration configuration;
+
+ public JCacheProducer(JCacheEndpoint endpoint, JCacheConfiguration configuration) {
+ super(endpoint);
+
+ this.configuration = configuration;
+ }
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ String actionName = exchange.getIn().getHeader(JCacheConstants.ACTION, String.class);
+ if (actionName == null) {
+ actionName = configuration.getAction();
+ }
+
+ ObjectHelper.notEmpty(actionName, JCacheConstants.ACTION);
+
+ Action action = Action.fromName(actionName);
+ if (action != null) {
+ Cache<Object, Object> cache = getJCacheEndpoint().getManager().getCache();
+ action.validate(cache, exchange);
+ action.execute(cache, exchange);
+ } else {
+ throw new IllegalArgumentException(
+ String.format("The value '%s' is not allowed for parameter '%s'", actionName, JCacheConstants.ACTION));
+ }
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ getCache();
+
+ super.doStart();
+ }
+
+ private JCacheEndpoint getJCacheEndpoint() {
+ return (JCacheEndpoint)getEndpoint();
+ }
+
+ private Cache getCache() throws Exception {
+ return getJCacheEndpoint().getManager().getCache();
+ }
+
+ private enum Action {
+ PUT {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEY);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ cache.put(
+ exchange.getIn().getHeader(JCacheConstants.KEY),
+ exchange.getIn().getBody()
+ );
+ }
+ },
+ PUTALL {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ cache.putAll(
+ exchange.getIn().getBody(Map.class)
+ );
+ }
+ },
+ PUTIFABSENT {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEY);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ exchange.getIn().setHeader(
+ JCacheConstants.RESULT,
+ cache.putIfAbsent(
+ exchange.getIn().getHeader(JCacheConstants.KEY),
+ exchange.getIn().getBody())
+ );
+ }
+ },
+ GET {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEY);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ exchange.getIn().setBody(
+ cache.get(exchange.getIn().getHeader(JCacheConstants.KEY))
+ );
+ }
+ },
+ GETALL {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEYS);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ exchange.getIn().setBody(
+ cache.getAll(
+ exchange.getIn().getHeader(JCacheConstants.KEYS, Set.class))
+ );
+ }
+ },
+ GETANDREMOVE {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEY);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ exchange.getIn().setBody(
+ cache.getAndRemove(
+ exchange.getIn().getHeader(JCacheConstants.KEY))
+ );
+ }
+ },
+ GETANDREPLACE {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEY);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ exchange.getIn().setBody(
+ cache.getAndReplace(
+ exchange.getIn().getHeader(JCacheConstants.KEY),
+ exchange.getIn().getBody())
+ );
+ }
+ },
+ GETANDPUT {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEY);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ exchange.getIn().setBody(
+ cache.getAndPut(
+ exchange.getIn().getHeader(JCacheConstants.KEY),
+ exchange.getIn().getBody())
+ );
+ }
+ },
+ REPLACE {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEY);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ Object oldValue = exchange.getIn().getHeader(JCacheConstants.OLD_VALUE);
+
+ exchange.getIn().setHeader(
+ JCacheConstants.RESULT,
+ oldValue != null
+ ? cache.replace(
+ exchange.getIn().getHeader(JCacheConstants.KEY),
+ oldValue,
+ exchange.getIn().getBody())
+ : cache.replace(
+ exchange.getIn().getHeader(JCacheConstants.KEY),
+ exchange.getIn().getBody())
+ );
+ }
+ },
+ REMOVE {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.KEY);
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ Object oldValue = exchange.getIn().getHeader(JCacheConstants.OLD_VALUE);
+
+ exchange.getIn().setHeader(
+ JCacheConstants.RESULT,
+ oldValue != null
+ ? cache.remove(
+ exchange.getIn().getHeader(JCacheConstants.KEY),
+ oldValue)
+ : cache.remove(
+ exchange.getIn().getHeader(JCacheConstants.KEY))
+ );
+ }
+ },
+ REMOVEALL {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ Set<Object> keys = exchange.getIn().getHeader(JCacheConstants.KEYS, Set.class);
+
+ if (keys != null) {
+ cache.removeAll(keys);
+ } else {
+ cache.removeAll();
+ }
+ }
+ },
+ INVOKE {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ headerIsNotNull(exchange, JCacheConstants.ENTRY_PROCESSOR);
+
+ if (exchange.getIn().getHeader(JCacheConstants.KEYS) == null
+ && exchange.getIn().getHeader(JCacheConstants.KEY) == null) {
+ throw new IllegalArgumentException(
+ String.format("Either %s or %s must be set for action %s",
+ JCacheConstants.KEYS,
+ JCacheConstants.KEY,
+ name())
+ );
+ }
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ Message message = exchange.getIn();
+ Set<Object> keys = message.getHeader(JCacheConstants.KEYS, Set.class);
+ EntryProcessor<Object, Object, Object> entryProcessor = message.getHeader(JCacheConstants.ENTRY_PROCESSOR, EntryProcessor.class);
+
+ Collection<Object> arguments = message.getHeader(JCacheConstants.ARGUMENTS, Collection.class);
+ if (arguments == null) {
+ arguments = Collections.emptyList();
+ }
+
+ message.setBody(
+ keys != null
+ ? cache.invokeAll(
+ keys,
+ entryProcessor,
+ arguments)
+ : cache.invoke(
+ exchange.getIn().getHeader(JCacheConstants.KEY),
+ entryProcessor,
+ arguments)
+ );
+ }
+ },
+ CLEAR {
+ @Override
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ }
+
+ @Override
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ cache.clear();
+ }
+ };
+
+
+ static final Action[] VALUES = values();
+
+ static Action fromName(String name) {
+ if (ObjectHelper.isNotEmpty(name)) {
+ for (int i = 0; i < VALUES.length; i++) {
+ if (VALUES[i].name().equalsIgnoreCase(name)) {
+ return VALUES[i];
+ }
+ }
+ }
+
+ return null;
+ }
+
+ void validate(Cache<Object, Object> cache, Exchange exchange) throws Exception {
+ }
+
+ void execute(Cache<Object, Object> cache, Exchange exchange) {
+ }
+
+ protected void headerIsNotNull(Exchange exchange, String... keys) throws Exception {
+ for (int i = keys.length - 1; i >= 0; i--) {
+ if (exchange.getIn().getHeader(keys[i]) == null) {
+ throw new IllegalArgumentException(
+ String.format("Header %s must be set for action %s", keys[i], name())
+ );
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/aggregate/JCacheAggregationRepository.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/aggregate/JCacheAggregationRepository.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/aggregate/JCacheAggregationRepository.java
new file mode 100644
index 0000000..341c9ca
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/aggregate/JCacheAggregationRepository.java
@@ -0,0 +1,206 @@
+/**
+ * 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.component.jcache.processor.aggregate;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+import javax.cache.Cache;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.component.jcache.JCacheConfiguration;
+import org.apache.camel.component.jcache.JCacheManager;
+import org.apache.camel.impl.DefaultExchange;
+import org.apache.camel.impl.DefaultExchangeHolder;
+import org.apache.camel.spi.OptimisticLockingAggregationRepository;
+import org.apache.camel.support.ServiceSupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class JCacheAggregationRepository extends ServiceSupport implements OptimisticLockingAggregationRepository {
+ private static final Logger LOG = LoggerFactory.getLogger(JCacheAggregationRepository.class);
+
+ private JCacheConfiguration configuration = new JCacheConfiguration();
+ private String cacheName;
+ private ClassLoader classLoader;
+ private CamelContext camelContext;
+ private Cache<String, DefaultExchangeHolder> cache;
+ private boolean optimistic;
+ private boolean allowSerializedHeaders;
+
+ private JCacheManager<String, DefaultExchangeHolder> cacheManager;
+
+ public JCacheConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(JCacheConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ public String getCacheName() {
+ return cacheName;
+ }
+
+ public void setCacheName(String cacheName) {
+ this.cacheName = cacheName;
+ }
+
+ public ClassLoader getClassLoader() {
+ return classLoader;
+ }
+
+ public void setClassLoader(ClassLoader classLoader) {
+ this.classLoader = classLoader;
+ }
+
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ public Cache<String, DefaultExchangeHolder> getCache() {
+ return cache;
+ }
+
+ public void setCache(Cache<String, DefaultExchangeHolder> cache) {
+ this.cache = cache;
+ }
+
+ public boolean isOptimistic() {
+ return optimistic;
+ }
+
+ public void setOptimistic(boolean optimistic) {
+ this.optimistic = optimistic;
+ }
+
+ public boolean isAllowSerializedHeaders() {
+ return allowSerializedHeaders;
+ }
+
+ public void setAllowSerializedHeaders(boolean allowSerializedHeaders) {
+ this.allowSerializedHeaders = allowSerializedHeaders;
+ }
+
+ @Override
+ public Exchange add(CamelContext camelContext, String key, Exchange oldExchange, Exchange newExchange) throws OptimisticLockingException {
+ if (!optimistic) {
+ throw new UnsupportedOperationException();
+ }
+
+ LOG.trace("Adding an Exchange with ID {} for key {} in an optimistic manner.", newExchange.getExchangeId(), key);
+ if (oldExchange == null) {
+ DefaultExchangeHolder newHolder = DefaultExchangeHolder.marshal(newExchange, true, allowSerializedHeaders);
+ DefaultExchangeHolder oldHolder = cache.getAndPut(key, newHolder);
+ if (oldHolder != null) {
+ Exchange exchange = unmarshallExchange(camelContext, oldHolder);
+ LOG.error("Optimistic locking failed for exchange with key {}: IMap#putIfAbsend returned Exchange with ID {}, while it's expected no exchanges to be returned",
+ key,
+ exchange != null ? exchange.getExchangeId() : "<null>");
+
+ throw new OptimisticLockingException();
+ }
+ } else {
+ DefaultExchangeHolder oldHolder = DefaultExchangeHolder.marshal(oldExchange, true, allowSerializedHeaders);
+ DefaultExchangeHolder newHolder = DefaultExchangeHolder.marshal(newExchange, true, allowSerializedHeaders);
+ if (!cache.replace(key, oldHolder, newHolder)) {
+ LOG.error("Optimistic locking failed for exchange with key {}: IMap#replace returned no Exchanges, while it's expected to replace one", key);
+ throw new OptimisticLockingException();
+ }
+ }
+ LOG.trace("Added an Exchange with ID {} for key {} in optimistic manner.", newExchange.getExchangeId(), key);
+ return oldExchange;
+ }
+
+ @Override
+ public Exchange add(CamelContext camelContext, String key, Exchange exchange) {
+ if (optimistic) {
+ throw new UnsupportedOperationException();
+ }
+ LOG.trace("Adding an Exchange with ID {} for key {} in a thread-safe manner.", exchange.getExchangeId(), key);
+ DefaultExchangeHolder newHolder = DefaultExchangeHolder.marshal(exchange, true, allowSerializedHeaders);
+ DefaultExchangeHolder oldHolder = cache.getAndPut(key, newHolder);
+ return unmarshallExchange(camelContext, oldHolder);
+ }
+
+ @Override
+ public Exchange get(CamelContext camelContext, String key) {
+ return unmarshallExchange(camelContext, cache.get(key));
+ }
+
+ @Override
+ public void remove(CamelContext camelContext, String key, Exchange exchange) {
+ DefaultExchangeHolder holder = DefaultExchangeHolder.marshal(exchange, true, allowSerializedHeaders);
+ if (optimistic) {
+ LOG.trace("Removing an exchange with ID {} for key {} in an optimistic manner.", exchange.getExchangeId(), key);
+ if (!cache.remove(key, holder)) {
+ LOG.error("Optimistic locking failed for exchange with key {}: IMap#remove removed no Exchanges, while it's expected to remove one.", key);
+ throw new OptimisticLockingException();
+ }
+ LOG.trace("Removed an exchange with ID {} for key {} in an optimistic manner.", exchange.getExchangeId(), key);
+ } else {
+ cache.remove(key);
+ }
+ }
+
+ @Override
+ public void confirm(CamelContext camelContext, String exchangeId) {
+ LOG.trace("Confirming an exchange with ID {}.", exchangeId);
+ }
+
+ @Override
+ public Set<String> getKeys() {
+ Set<String> keys = new HashSet<>();
+
+ Iterator<Cache.Entry<String, DefaultExchangeHolder>> entries = cache.iterator();
+ while (entries.hasNext()) {
+ keys.add(entries.next().getKey());
+ }
+
+ return Collections.unmodifiableSet(keys);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (cache != null) {
+ cacheManager = new JCacheManager<>(cache);
+ } else {
+ cacheManager = new JCacheManager(configuration, cacheName, classLoader, camelContext);
+ cache = cacheManager.getCache();
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ cacheManager.close();
+ }
+
+ protected Exchange unmarshallExchange(CamelContext camelContext, DefaultExchangeHolder holder) {
+ Exchange exchange = null;
+ if (holder != null) {
+ exchange = new DefaultExchange(camelContext);
+ DefaultExchangeHolder.unmarshal(exchange, holder);
+ }
+ return exchange;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/idempotent/JCacheIdempotentRepository.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/idempotent/JCacheIdempotentRepository.java b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/idempotent/JCacheIdempotentRepository.java
new file mode 100644
index 0000000..7564ce5
--- /dev/null
+++ b/components/camel-jcache/src/main/java/org/apache/camel/component/jcache/processor/idempotent/JCacheIdempotentRepository.java
@@ -0,0 +1,125 @@
+/**
+ * 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.component.jcache.processor.idempotent;
+
+import javax.cache.Cache;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.api.management.ManagedAttribute;
+import org.apache.camel.api.management.ManagedOperation;
+import org.apache.camel.api.management.ManagedResource;
+import org.apache.camel.component.jcache.JCacheConfiguration;
+import org.apache.camel.component.jcache.JCacheManager;
+import org.apache.camel.spi.IdempotentRepository;
+import org.apache.camel.support.ServiceSupport;
+
+@ManagedResource(description = "JCache based message id repository")
+public class JCacheIdempotentRepository extends ServiceSupport implements IdempotentRepository<Object> {
+ private JCacheConfiguration configuration = new JCacheConfiguration();
+ private String cacheName;
+ private ClassLoader classLoader;
+ private CamelContext camelContext;
+ private Cache<Object, Boolean> cache;
+
+ private JCacheManager<Object, Boolean> cacheManager;
+
+ public JCacheConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(JCacheConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ public ClassLoader getClassLoader() {
+ return classLoader;
+ }
+
+ public void setClassLoader(ClassLoader classLoader) {
+ this.classLoader = classLoader;
+ }
+
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ public Cache<Object, Boolean> getCache() {
+ return cache;
+ }
+
+ public void setCache(Cache<Object, Boolean> cache) {
+ this.cache = cache;
+ }
+
+ @Override
+ @ManagedOperation(description = "Adds the key to the store")
+ public boolean add(Object key) {
+ return cache.putIfAbsent(key, true);
+ }
+
+ @Override
+ @ManagedOperation(description = "Does the store contain the given key")
+ public boolean contains(Object key) {
+ return cache.containsKey(key);
+ }
+
+ @Override
+ @ManagedOperation(description = "Remove the key from the store")
+ public boolean remove(Object key) {
+ return cache.remove(key);
+ }
+
+ @Override
+ @ManagedOperation(description = "Clear the store")
+ public void clear() {
+ cache.clear();
+ }
+
+ public void setCacheName(String cacheName) {
+ this.cacheName = cacheName;
+ }
+
+ @ManagedAttribute(description = "The processor name")
+ public String getCacheName() {
+ return cacheName;
+ }
+
+ @Override
+ public boolean confirm(Object key) {
+ return cache.replace(key, false, true);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (cache != null) {
+ cacheManager = new JCacheManager<>(cache);
+ } else {
+ cacheManager = new JCacheManager(configuration, cacheName, classLoader, camelContext);
+ cache = cacheManager.getCache();
+ }
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ cacheManager.close();
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/main/resources/META-INF/services/org/apache/camel/component/jcache
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/main/resources/META-INF/services/org/apache/camel/component/jcache b/components/camel-jcache/src/main/resources/META-INF/services/org/apache/camel/component/jcache
new file mode 100644
index 0000000..bbf037c
--- /dev/null
+++ b/components/camel-jcache/src/main/resources/META-INF/services/org/apache/camel/component/jcache
@@ -0,0 +1 @@
+class=org.apache.camel.component.jcache.JCacheComponent
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheComponentTestSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheComponentTestSupport.java b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheComponentTestSupport.java
new file mode 100644
index 0000000..6f93dbe
--- /dev/null
+++ b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheComponentTestSupport.java
@@ -0,0 +1,50 @@
+/**
+ * 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.component.jcache;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import javax.cache.Cache;
+
+import org.apache.camel.test.junit4.CamelTestSupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+class JCacheComponentTestSupport extends CamelTestSupport {
+ protected static final Logger LOGGER = LoggerFactory.getLogger(JCacheComponentTestSupport.class);
+
+ protected Cache<Object, Object> getCacheFromEndpoint(String endpoint) throws Exception {
+ return resolveMandatoryEndpoint(endpoint, JCacheEndpoint.class)
+ .getManager()
+ .getCache();
+ }
+
+ protected String randomString() {
+ return UUID.randomUUID().toString();
+ }
+
+ protected Map<Object, Object> generateRandomMap(int size) {
+ Map<Object, Object> map = new HashMap<>();
+ for (int i = 0; i < size; i++) {
+ map.put(randomString(), randomString());
+ }
+
+ return map;
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheConfigurationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheConfigurationTest.java b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheConfigurationTest.java
new file mode 100644
index 0000000..26b4aa7
--- /dev/null
+++ b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheConfigurationTest.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.component.jcache;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Map;
+import javax.cache.Cache;
+import javax.cache.configuration.CompleteConfiguration;
+import javax.cache.configuration.Factory;
+import javax.cache.configuration.FactoryBuilder;
+import javax.cache.expiry.AccessedExpiryPolicy;
+import javax.cache.expiry.Duration;
+import javax.cache.expiry.ExpiryPolicy;
+import javax.cache.integration.CacheLoader;
+import javax.cache.integration.CacheLoaderException;
+import javax.cache.integration.CacheWriter;
+import javax.cache.integration.CacheWriterException;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.junit.Test;
+
+public class JCacheConfigurationTest extends JCacheComponentTestSupport {
+ private static final Factory<ExpiryPolicy> EXPIRY_POLICY_FACTORY = AccessedExpiryPolicy.factoryOf(Duration.ONE_MINUTE);
+ private static final Factory<CacheWriter<Object, Object>> CACHE_WRITER_FACTORY = MyCacheWriter.factory();
+ private static final Factory<CacheLoader<Object, Object>> CACHE_LOADER_FACTORY = MyCacheLoader.factory();
+
+ @EndpointInject(
+ uri = "jcache://test-cache"
+ + "?expiryPolicyFactory=#myExpiryPolicyFactory"
+ + "&cacheWriterFactory=#myCacheWriterFactory"
+ + "&cacheLoaderFactory=#myCacheLoaderFactory"
+ )
+ JCacheEndpoint from;
+
+ @EndpointInject(uri = "mock:to")
+ MockEndpoint to;
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry registry = super.createRegistry();
+ registry.bind("myExpiryPolicyFactory", EXPIRY_POLICY_FACTORY);
+ registry.bind("myCacheWriterFactory", CACHE_WRITER_FACTORY);
+ registry.bind("myCacheLoaderFactory", CACHE_LOADER_FACTORY);
+
+ return registry;
+ }
+
+ @Test
+ public void testConfigurations() throws Exception {
+ final Cache<Object, Object> cache = from.getManager().getCache();
+ final CompleteConfiguration<Object, Object> conf = cache.getConfiguration(CompleteConfiguration.class);
+
+ assertEquals(EXPIRY_POLICY_FACTORY, conf.getExpiryPolicyFactory());
+ assertEquals(CACHE_WRITER_FACTORY, conf.getCacheWriterFactory());
+ assertEquals(CACHE_LOADER_FACTORY, conf.getCacheLoaderFactory());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ from(from).to(to);
+ }
+ };
+ }
+
+ private static final class MyCacheLoader implements CacheLoader<Object, Object>, Serializable {
+ @Override
+ public Object load(Object key) throws CacheLoaderException {
+ return null;
+ }
+ @Override
+ public Map<Object, Object> loadAll(Iterable<?> keys) throws CacheLoaderException {
+ return null;
+ }
+
+ public static Factory<CacheLoader<Object, Object>> factory() {
+ return new FactoryBuilder.SingletonFactory(new MyCacheLoader());
+ }
+ }
+
+ private static final class MyCacheWriter implements CacheWriter<Object, Object>, Serializable {
+ @Override
+ public void write(Cache.Entry<?, ?> entry) throws CacheWriterException {
+ }
+ @Override
+ public void writeAll(Collection<Cache.Entry<?, ?>> entries) throws CacheWriterException {
+ }
+ @Override
+ public void delete(Object key) throws CacheWriterException {
+ }
+ @Override
+ public void deleteAll(Collection<?> keys) throws CacheWriterException {
+ }
+
+ public static Factory<CacheWriter<Object, Object>> factory() {
+ return new FactoryBuilder.SingletonFactory(new MyCacheWriter());
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheConsumerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheConsumerTest.java b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheConsumerTest.java
new file mode 100644
index 0000000..975f721
--- /dev/null
+++ b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheConsumerTest.java
@@ -0,0 +1,121 @@
+/**
+ * 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.component.jcache;
+
+import javax.cache.Cache;
+import javax.cache.event.CacheEntryEvent;
+import javax.cache.event.CacheEntryEventFilter;
+import javax.cache.event.CacheEntryListenerException;
+import javax.cache.event.EventType;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Predicate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.JndiRegistry;
+import org.junit.Test;
+
+public class JCacheConsumerTest extends JCacheComponentTestSupport {
+
+ @Override
+ protected JndiRegistry createRegistry() throws Exception {
+ JndiRegistry registry = super.createRegistry();
+ registry.bind("myFilter", new CacheEntryEventFilter<Object, Object>() {
+ @Override
+ public boolean evaluate(CacheEntryEvent<?, ?> event) throws CacheEntryListenerException {
+ if (event.getEventType() == EventType.REMOVED) {
+ return false;
+ }
+
+ return !event.getValue().toString().startsWith("to-filter-");
+ }
+ });
+
+ return registry;
+ }
+
+ @Test
+ public void testFilters() throws Exception {
+ final Cache<Object, Object> cache = getCacheFromEndpoint("jcache://test-cache");
+
+ final String key = randomString();
+ final String val1 = "to-filter-" + randomString();
+ final String val2 = randomString();
+
+ cache.put(key, val1);
+ cache.put(key, val2);
+ cache.remove(key);
+
+ MockEndpoint mockCreated = getMockEndpoint("mock:created");
+ mockCreated.expectedMinimumMessageCount(1);
+ mockCreated.expectedHeaderReceived(JCacheConstants.KEY, key);
+ mockCreated.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ return exchange.getIn().getBody(String.class).equals(val1);
+ }
+ });
+
+ MockEndpoint mockUpdated = getMockEndpoint("mock:updated");
+ mockUpdated.expectedMinimumMessageCount(1);
+ mockUpdated.expectedHeaderReceived(JCacheConstants.KEY, key);
+ mockUpdated.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ return exchange.getIn().getBody(String.class).equals(val2);
+ }
+ });
+
+ MockEndpoint mockRemoved = getMockEndpoint("mock:removed");
+ mockRemoved.expectedMinimumMessageCount(1);
+ mockRemoved.expectedHeaderReceived(JCacheConstants.KEY, key);
+ mockRemoved.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ return exchange.getIn().getBody(String.class).equals(val2);
+ }
+ });
+
+ MockEndpoint mockMyFilter = getMockEndpoint("mock:my-filter");
+ mockMyFilter.expectedMinimumMessageCount(1);
+ mockMyFilter.expectedHeaderReceived(JCacheConstants.KEY, key);
+ mockMyFilter.expectedMessagesMatches(new Predicate() {
+ @Override
+ public boolean matches(Exchange exchange) {
+ return exchange.getIn().getBody(String.class).equals(val2);
+ }
+ });
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ from("jcache://test-cache?filteredEvents=UPDATED,REMOVED,EXPIRED")
+ .to("mock:created");
+ from("jcache://test-cache?filteredEvents=CREATED,REMOVED,EXPIRED")
+ .to("mock:updated");
+ from("jcache://test-cache?filteredEvents=CREATED,UPDATED,EXPIRED")
+ .to("mock:removed");
+ from("jcache://test-cache?eventFilters=#myFilter")
+ .to("mock:my-filter");
+ }
+ };
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/4496eea4/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheProducerClearTest.java
----------------------------------------------------------------------
diff --git a/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheProducerClearTest.java b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheProducerClearTest.java
new file mode 100644
index 0000000..d571734
--- /dev/null
+++ b/components/camel-jcache/src/test/java/org/apache/camel/component/jcache/JCacheProducerClearTest.java
@@ -0,0 +1,51 @@
+/**
+ * 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.component.jcache;
+
+import java.util.HashMap;
+import java.util.Map;
+import javax.cache.Cache;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.Test;
+
+public class JCacheProducerClearTest extends JCacheComponentTestSupport {
+
+ @Test
+ public void testClear() throws Exception {
+ final Map<String, Object> headers = new HashMap<>();
+ final Cache<Object, Object> cache = getCacheFromEndpoint("jcache://test-cache");
+
+ cache.putAll(generateRandomMap(4));
+
+ headers.clear();
+ headers.put(JCacheConstants.ACTION, "CLEAR");
+ sendBody("direct:clear", null, headers);
+
+ assertFalse(cache.iterator().hasNext());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:clear")
+ .to("jcache://test-cache");
+ }
+ };
+ }
+}