Author: davsclaus Date: Wed Apr 28 08:44:19 2010 New Revision: 938819 URL: http://svn.apache.org/viewvc?rev=938819&view=rev Log: CAMEL-2603: camel-cache now supports GET operation. Refactored code to properly shutdown cache manager. Applied patch with thanks to Volodymyr Buell.
Added: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConstants.java (with props) camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheManagerFactory.java (contents, props changed) - copied, changed from r938776, camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/factory/CacheManagerFactory.java Removed: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/factory/ Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheComponent.java camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConfiguration.java camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConsumer.java camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheEndpoint.java camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheProducer.java camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedMessageBodyReplacer.java camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedTokenReplacer.java camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedXPathReplacer.java camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheConsumerTest.java camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheProducerTest.java Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheComponent.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheComponent.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheComponent.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheComponent.java Wed Apr 28 08:44:19 2010 @@ -22,9 +22,11 @@ import java.util.Map; import org.apache.camel.CamelContext; import org.apache.camel.Endpoint; import org.apache.camel.impl.DefaultComponent; +import org.apache.camel.util.ServiceHelper; public class CacheComponent extends DefaultComponent { - CacheConfiguration config; + private CacheConfiguration config; + private CacheManagerFactory cacheManagerFactory = new CacheManagerFactory(); public CacheComponent() { config = new CacheConfiguration(); @@ -40,9 +42,28 @@ public class CacheComponent extends Defa protected Endpoint createEndpoint(String uri, String remaining, Map parameters) throws Exception { config.parseURI(new URI(uri)); - CacheEndpoint cacheEndpoint = new CacheEndpoint(uri, this, config); + CacheEndpoint cacheEndpoint = new CacheEndpoint(uri, this, config, cacheManagerFactory); setProperties(cacheEndpoint.getConfig(), parameters); return cacheEndpoint; } + public CacheManagerFactory getCacheManagerFactory() { + return cacheManagerFactory; + } + + public void setCacheManagerFactory(CacheManagerFactory cacheManagerFactory) { + this.cacheManagerFactory = cacheManagerFactory; + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + ServiceHelper.startService(cacheManagerFactory); + } + + @Override + protected void doStop() throws Exception { + ServiceHelper.stopService(cacheManagerFactory); + super.doStop(); + } } Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConfiguration.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConfiguration.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConfiguration.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConfiguration.java Wed Apr 28 08:44:19 2010 @@ -44,7 +44,6 @@ public class CacheConfiguration { public CacheConfiguration(URI uri) throws Exception { parseURI(uri); - } public void parseURI(URI uri) throws Exception { Added: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConstants.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConstants.java?rev=938819&view=auto ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConstants.java (added) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConstants.java Wed Apr 28 08:44:19 2010 @@ -0,0 +1,34 @@ +/** + * 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.cache; + +/** + * Constants used in this module + */ +public interface CacheConstants { + + String CACHE_OPERATION = "CACHE_OPERATION"; + String CACHE_KEY = "CACHE_KEY"; + String CACHE_ELEMENT_WAS_FOUND = "CACHE_ELEMENT_WAS_FOUND"; + + String CACHE_OPERATION_ADD = "ADD"; + String CACHE_OPERATION_UPDATE = "UPDATE"; + String CACHE_OPERATION_DELETE = "DELETE"; + String CACHE_OPERATION_DELETEALL = "DELETEALL"; + String CACHE_OPERATION_GET = "GET"; + String CACHE_OPERATION_CHECK = "CHECK"; +} Propchange: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConstants.java ------------------------------------------------------------------------------ svn:eol-style = native Propchange: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConstants.java ------------------------------------------------------------------------------ svn:keywords = Rev Date Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConsumer.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConsumer.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConsumer.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheConsumer.java Wed Apr 28 08:44:19 2010 @@ -22,7 +22,6 @@ import net.sf.ehcache.Ehcache; import org.apache.camel.Endpoint; import org.apache.camel.Processor; -import org.apache.camel.component.cache.factory.CacheManagerFactory; import org.apache.camel.impl.DefaultConsumer; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -30,14 +29,12 @@ import org.apache.commons.logging.LogFac public class CacheConsumer extends DefaultConsumer { private static final transient Log LOG = LogFactory.getLog(CacheConsumer.class); - CacheEndpoint endpoint; - CacheConfiguration config; - Ehcache cache; - CacheManager cacheManager; - + private CacheConfiguration config; + private Ehcache cache; + private CacheManager cacheManager; + public CacheConsumer(Endpoint endpoint, Processor processor, CacheConfiguration config) { super(endpoint, processor); - this.endpoint = (CacheEndpoint) endpoint; this.config = config; } @@ -55,11 +52,11 @@ public class CacheConsumer extends Defau @Override public CacheEndpoint getEndpoint() { - return endpoint; + return (CacheEndpoint) super.getEndpoint(); } protected void createConsumerCacheConnection() { - cacheManager = new CacheManagerFactory().instantiateCacheManager(); + cacheManager = getEndpoint().getCacheManagerFactory().instantiateCacheManager(); CacheEventListener cacheEventListener = new CacheEventListenerFactory().createCacheEventListener(null); cacheEventListener.setCacheConsumer(this); @@ -89,9 +86,6 @@ public class CacheConsumer extends Defau protected void removeConsumerCacheConnection() { cacheManager.removeCache(config.getCacheName()); - if (cacheManager.getCacheNames().length == 0) { - cacheManager.shutdown(); - } } } Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheEndpoint.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheEndpoint.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheEndpoint.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheEndpoint.java Wed Apr 28 08:44:19 2010 @@ -25,28 +25,36 @@ import org.apache.camel.Producer; import org.apache.camel.impl.DefaultEndpoint; import org.apache.camel.impl.DefaultExchange; import org.apache.camel.impl.DefaultMessage; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; +import org.apache.camel.util.ObjectHelper; public class CacheEndpoint extends DefaultEndpoint { - private static final transient Log LOG = LogFactory.getLog(CacheEndpoint.class); - CacheConfiguration config; - - public CacheEndpoint(String endpointUri, Component component, CacheConfiguration config) { + private CacheConfiguration config; + private CacheManagerFactory cacheManagerFactory; + + public CacheEndpoint() { + } + + public CacheEndpoint(String endpointUri, Component component, CacheConfiguration config, + CacheManagerFactory cacheManagerFactory) { super(endpointUri, component); this.config = config; + this.cacheManagerFactory = cacheManagerFactory; } public Consumer createConsumer(Processor processor) throws Exception { + ObjectHelper.notNull(config, "config"); + ObjectHelper.notNull(cacheManagerFactory, "cacheManagerFactory"); return new CacheConsumer(this, processor, config); } public Producer createProducer() throws Exception { + ObjectHelper.notNull(config, "config"); + ObjectHelper.notNull(cacheManagerFactory, "cacheManagerFactory"); return new CacheProducer(this, config); } public boolean isSingleton() { - return false; + return true; } public CacheConfiguration getConfig() { @@ -57,6 +65,14 @@ public class CacheEndpoint extends Defau this.config = config; } + public CacheManagerFactory getCacheManagerFactory() { + return cacheManagerFactory; + } + + public void setCacheManagerFactory(CacheManagerFactory cacheManagerFactory) { + this.cacheManagerFactory = cacheManagerFactory; + } + public Exchange createCacheExchange(String operation, String key, Object value) { Exchange exchange = new DefaultExchange(this.getCamelContext(), getExchangePattern()); Message message = new DefaultMessage(); Copied: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheManagerFactory.java (from r938776, camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/factory/CacheManagerFactory.java) URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheManagerFactory.java?p2=camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheManagerFactory.java&p1=camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/factory/CacheManagerFactory.java&r1=938776&r2=938819&rev=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/factory/CacheManagerFactory.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheManagerFactory.java Wed Apr 28 08:44:19 2010 @@ -14,12 +14,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.camel.component.cache.factory; +package org.apache.camel.component.cache; import net.sf.ehcache.CacheManager; +import org.apache.camel.impl.ServiceSupport; -public class CacheManagerFactory { - +public class CacheManagerFactory extends ServiceSupport { private CacheManager cacheManager; public synchronized CacheManager instantiateCacheManager() { @@ -30,4 +30,15 @@ public class CacheManagerFactory { return cacheManager; } + @Override + protected void doStart() throws Exception { + } + + @Override + protected void doStop() throws Exception { + // shutdown cache manager when stopping + if (cacheManager != null) { + cacheManager.shutdown(); + } + } } Propchange: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheManagerFactory.java ------------------------------------------------------------------------------ svn:eol-style = native Propchange: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheManagerFactory.java ------------------------------------------------------------------------------ svn:executable = * Propchange: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheManagerFactory.java ------------------------------------------------------------------------------ svn:keywords = Rev Date Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheProducer.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheProducer.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheProducer.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/component/cache/CacheProducer.java Wed Apr 28 08:44:19 2010 @@ -25,72 +25,77 @@ import net.sf.ehcache.CacheException; import net.sf.ehcache.CacheManager; import net.sf.ehcache.Ehcache; import net.sf.ehcache.Element; - import org.apache.camel.Endpoint; import org.apache.camel.Exchange; -import org.apache.camel.component.cache.factory.CacheManagerFactory; import org.apache.camel.impl.DefaultProducer; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; public class CacheProducer extends DefaultProducer { private static final transient Log LOG = LogFactory.getLog(CacheProducer.class); - Endpoint endpoint; - CacheConfiguration config; - CacheManager cacheManager; - Ehcache cache; - + private CacheConfiguration config; + private CacheManager cacheManager; + private Ehcache cache; + public CacheProducer(Endpoint endpoint, CacheConfiguration config) throws Exception { super(endpoint); - this.endpoint = endpoint; this.config = config; } + @Override + protected void doStart() throws Exception { + cacheManager = getEndpoint().getCacheManagerFactory().instantiateCacheManager(); + super.doStart(); + } + + @Override + public CacheEndpoint getEndpoint() { + return (CacheEndpoint) super.getEndpoint(); + } + public void process(Exchange exchange) throws Exception { - - cacheManager = new CacheManagerFactory().instantiateCacheManager(); - if (LOG.isTraceEnabled()) { LOG.trace("Cache Name: " + config.getCacheName()); } + if (cacheManager.cacheExists(config.getCacheName())) { if (LOG.isTraceEnabled()) { LOG.trace("Found an existing cache: " + config.getCacheName()); - LOG.trace("Cache " + config.getCacheName() + " currently contains " + cacheManager.getCache(config.getCacheName()).getSize() + " elements"); + LOG.trace("Cache " + config.getCacheName() + " currently contains " + + cacheManager.getCache(config.getCacheName()).getSize() + " elements"); } cache = cacheManager.getCache(config.getCacheName()); } else { - cache = new Cache(config.getCacheName(), + cache = new Cache(config.getCacheName(), config.getMaxElementsInMemory(), - config.getMemoryStoreEvictionPolicy(), - config.isOverflowToDisk(), - config.getDiskStorePath(), - config.isEternal(), - config.getTimeToLiveSeconds(), - config.getTimeToIdleSeconds(), - config.isDiskPersistent(), - config.getDiskExpiryThreadIntervalSeconds(), + config.getMemoryStoreEvictionPolicy(), + config.isOverflowToDisk(), + config.getDiskStorePath(), + config.isEternal(), + config.getTimeToLiveSeconds(), + config.getTimeToIdleSeconds(), + config.isDiskPersistent(), + config.getDiskExpiryThreadIntervalSeconds(), null); cacheManager.addCache(cache); if (LOG.isDebugEnabled()) { LOG.debug("Added a new cache: " + cache.getName()); } } - - - String key = (String) exchange.getIn().getHeader("CACHE_KEY"); - String operation = (String) exchange.getIn().getHeader("CACHE_OPERATION"); + + String key = exchange.getIn().getHeader(CacheConstants.CACHE_KEY, String.class); + String operation = exchange.getIn().getHeader(CacheConstants.CACHE_OPERATION, String.class); + if (operation == null) { - throw new CacheException("Operation property is not specified in the incoming exchange header." - + "A valid Operation property must be set to ADD, UPDATE, DELETE, DELETEALL"); + throw new CacheException("Operation not specified in the message header [" + CacheConstants.CACHE_KEY + "]"); } - if ((key == null) && (!operation.equalsIgnoreCase("DELETEALL"))) { - throw new CacheException("Cache Key is not specified in exchange either header or URL. Unable to add objects to the cache without a Key"); + if ((key == null) && (!operation.equalsIgnoreCase(CacheConstants.CACHE_OPERATION_DELETEALL))) { + throw new CacheException("Cache Key is not specified in message header header or endpoint URL."); } - + performCacheOperation(exchange, operation, key); } - + private void performCacheOperation(Exchange exchange, String operation, String key) throws Exception { Object element; @@ -99,37 +104,51 @@ public class CacheProducer extends Defau element = body; } else { InputStream is = exchange.getContext().getTypeConverter().mandatoryConvertTo(InputStream.class, body); - // Read InputStream into a byte[] buffer - byte[] buffer = new byte[is.available()]; - int n = is.available(); - for (int j = 0; j < n; j++) { - buffer[j] = (byte)is.read(); - } - - element = buffer; + element = exchange.getContext().getTypeConverter().mandatoryConvertTo(byte[].class, is); } - if (operation.equalsIgnoreCase("ADD")) { + if (operation.equalsIgnoreCase(CacheConstants.CACHE_OPERATION_ADD)) { if (LOG.isDebugEnabled()) { LOG.debug("Adding an element with key " + key + " into the Cache"); } cache.put(new Element(key, element), true); - } else if (operation.equalsIgnoreCase("UPDATE")) { + } else if (operation.equalsIgnoreCase(CacheConstants.CACHE_OPERATION_UPDATE)) { if (LOG.isDebugEnabled()) { LOG.debug("Updating an element with key " + key + " into the Cache"); } cache.put(new Element(key, element), true); - } else if (operation.equalsIgnoreCase("DELETEALL")) { + } else if (operation.equalsIgnoreCase(CacheConstants.CACHE_OPERATION_DELETEALL)) { if (LOG.isDebugEnabled()) { LOG.debug("Deleting All elements from the Cache"); } cache.removeAll(); - } else if (operation.equalsIgnoreCase("DELETE")) { + } else if (operation.equalsIgnoreCase(CacheConstants.CACHE_OPERATION_DELETE)) { if (LOG.isDebugEnabled()) { LOG.debug("Deleting an element with key " + key + " into the Cache"); } cache.remove(key, true); + } else if (operation.equalsIgnoreCase(CacheConstants.CACHE_OPERATION_GET)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Quering an element with key " + key + " from the Cache"); + } + if (cache.isKeyInCache(key)) { + exchange.getIn().setHeader(CacheConstants.CACHE_ELEMENT_WAS_FOUND, true); + exchange.getIn().setBody(cache.get(key).getValue()); + } else { + exchange.getIn().removeHeader(CacheConstants.CACHE_ELEMENT_WAS_FOUND); + } + } else if (operation.equalsIgnoreCase(CacheConstants.CACHE_OPERATION_CHECK)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Querying an element with key " + key + " from the Cache"); + } + if (cache.isKeyInCache(key)) { + exchange.getIn().setHeader(CacheConstants.CACHE_ELEMENT_WAS_FOUND, true); + } else { + exchange.getIn().removeHeader(CacheConstants.CACHE_ELEMENT_WAS_FOUND); + } + } else { + throw new CacheException("Operation " + operation + " is not supported."); } } Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedMessageBodyReplacer.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedMessageBodyReplacer.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedMessageBodyReplacer.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedMessageBodyReplacer.java Wed Apr 28 08:44:19 2010 @@ -21,7 +21,7 @@ import net.sf.ehcache.Ehcache; import org.apache.camel.Exchange; import org.apache.camel.Processor; -import org.apache.camel.component.cache.factory.CacheManagerFactory; +import org.apache.camel.component.cache.CacheManagerFactory; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedTokenReplacer.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedTokenReplacer.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedTokenReplacer.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedTokenReplacer.java Wed Apr 28 08:44:19 2010 @@ -23,7 +23,7 @@ import net.sf.ehcache.Ehcache; import org.apache.camel.Exchange; import org.apache.camel.Processor; -import org.apache.camel.component.cache.factory.CacheManagerFactory; +import org.apache.camel.component.cache.CacheManagerFactory; import org.apache.camel.converter.IOConverter; import org.apache.camel.util.ObjectHelper; import org.apache.commons.logging.Log; Modified: camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedXPathReplacer.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedXPathReplacer.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedXPathReplacer.java (original) +++ camel/trunk/components/camel-cache/src/main/java/org/apache/camel/processor/cache/CacheBasedXPathReplacer.java Wed Apr 28 08:44:19 2010 @@ -30,9 +30,10 @@ import org.w3c.dom.Document; import net.sf.ehcache.CacheManager; import net.sf.ehcache.Ehcache; + import org.apache.camel.Exchange; import org.apache.camel.Processor; -import org.apache.camel.component.cache.factory.CacheManagerFactory; +import org.apache.camel.component.cache.CacheManagerFactory; import org.apache.camel.converter.IOConverter; import org.apache.camel.converter.jaxp.XmlConverter; import org.apache.camel.util.ObjectHelper; Modified: camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheConsumerTest.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheConsumerTest.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheConsumerTest.java (original) +++ camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheConsumerTest.java Wed Apr 28 08:44:19 2010 @@ -48,16 +48,16 @@ public class CacheConsumerTest extends C resultEndpoint.expectedMessageCount(3); List<String> operations = new ArrayList<String>(); - operations.add("ADD"); - operations.add("UPDATE"); - operations.add("DELETE"); + operations.add(CacheConstants.CACHE_OPERATION_ADD); + operations.add(CacheConstants.CACHE_OPERATION_UPDATE); + operations.add(CacheConstants.CACHE_OPERATION_DELETE); for (final String operation : operations) { producerTemplate.send(new Processor() { public void process(Exchange exchange) throws Exception { exchange.setProperty(Exchange.CHARSET_NAME, "UTF-8"); Message in = exchange.getIn(); - in.setHeader("CACHE_OPERATION", operation); - in.setHeader("CACHE_KEY", "greeting"); + in.setHeader(CacheConstants.CACHE_OPERATION, operation); + in.setHeader(CacheConstants.CACHE_KEY, "greeting"); in.setBody("Hello World"); } }); @@ -74,20 +74,20 @@ public class CacheConsumerTest extends C resultEndpoint.expectedMessageCount(3); List<String> operations = new ArrayList<String>(); - operations.add("ADD"); - operations.add("UPDATE"); - operations.add("DELETE"); + operations.add(CacheConstants.CACHE_OPERATION_ADD); + operations.add(CacheConstants.CACHE_OPERATION_UPDATE); + operations.add(CacheConstants.CACHE_OPERATION_DELETE); for (final String operation : operations) { producerTemplate.send(new Processor() { public void process(Exchange exchange) throws Exception { Poetry p = new Poetry(); p.setPoet("Ralph Waldo Emerson"); p.setPoem("Brahma"); - + exchange.setProperty(Exchange.CHARSET_NAME, "UTF-8"); Message in = exchange.getIn(); - in.setHeader("CACHE_OPERATION", operation); - in.setHeader("CACHE_KEY", "poetry"); + in.setHeader(CacheConstants.CACHE_OPERATION, operation); + in.setHeader(CacheConstants.CACHE_KEY, "poetry"); in.setBody(p); } }); @@ -96,15 +96,15 @@ public class CacheConsumerTest extends C resultEndpoint.assertIsSatisfied(); LOG.debug("Completed Test ---> testReceivingFileFromCache()"); } - + @Override protected RouteBuilder createRouteBuilder() throws Exception { return new RouteBuilder() { public void configure() { from("cache://TestCache1").process(new Processor() { public void process(Exchange exchange) throws Exception { - String operation = (String)exchange.getIn().getHeader("CACHE_OPERATION"); - String key = (String)exchange.getIn().getHeader("CACHE_KEY"); + String operation = (String) exchange.getIn().getHeader(CacheConstants.CACHE_OPERATION); + String key = (String) exchange.getIn().getHeader(CacheConstants.CACHE_KEY); Object body = exchange.getIn().getBody(); String data = exchange.getContext().getTypeConverter().convertTo(String.class, body); Modified: camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheProducerTest.java URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheProducerTest.java?rev=938819&r1=938818&r2=938819&view=diff ============================================================================== --- camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheProducerTest.java (original) +++ camel/trunk/components/camel-cache/src/test/java/org/apache/camel/component/cache/CacheProducerTest.java Wed Apr 28 08:44:19 2010 @@ -21,10 +21,13 @@ import java.io.BufferedInputStream; import java.io.FileInputStream; import java.io.InputStream; +import net.sf.ehcache.CacheException; +import org.apache.camel.EndpointInject; import org.apache.camel.Exchange; import org.apache.camel.Message; import org.apache.camel.Processor; import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; import org.apache.camel.converter.IOConverter; import org.apache.camel.test.junit4.CamelTestSupport; import org.apache.commons.logging.Log; @@ -32,48 +35,59 @@ import org.apache.commons.logging.LogFac import org.junit.Test; public class CacheProducerTest extends CamelTestSupport { + + private static final String FILEPATH_UPDATEDTEST_TXT = "./src/test/resources/updatedtest.txt"; + + private static final String FILEPATH_TEST_TXT = "./src/test/resources/test.txt"; + private static final transient Log LOG = LogFactory.getLog(CacheProducerTest.class); - + + @EndpointInject(uri = "mock:CacheProducerTest.result") + protected MockEndpoint resultEndpoint; + + @EndpointInject(uri = "mock:CacheProducerTest.exception") + protected MockEndpoint exceptionEndpoint; + @Override public boolean isUseRouteBuilder() { return false; } - private void sendFile() throws Exception { + private void sendFile(final String path) throws Exception { template.send("direct:a", new Processor() { public void process(Exchange exchange) throws Exception { - // Read from an input stream + // Read from an input stream InputStream is = new BufferedInputStream( - new FileInputStream("./src/test/resources/test.txt")); + new FileInputStream(path)); // "./src/test/resources/test.txt")); byte buffer[] = IOConverter.toBytes(is); is.close(); - + // Set the property of the charset encoding exchange.setProperty(Exchange.CHARSET_NAME, "UTF-8"); Message in = exchange.getIn(); in.setBody(buffer); - } + } }); } - - private void sendUpdatedFile() throws Exception { - template.send("direct:a", new Processor() { - public void process(Exchange exchange) throws Exception { - // Read from an input stream - InputStream is = new BufferedInputStream( - new FileInputStream("./src/test/resources/updatedtest.txt")); - byte buffer[] = IOConverter.toBytes(is); - is.close(); - - // Set the property of the charset encoding - exchange.setProperty(Exchange.CHARSET_NAME, "UTF-8"); - Message in = exchange.getIn(); - in.setBody(buffer); - } - }); - } + private byte[] getFileAsByteArray(String path) throws Exception { + // Read from an input stream + InputStream is = new BufferedInputStream(new FileInputStream(path)); + + byte[] buffer = IOConverter.toBytes(is); + is.close(); + + return buffer; + } + + private void sendOriginalFile() throws Exception { + sendFile(FILEPATH_TEST_TXT); + } + + private void sendUpdatedFile() throws Exception { + sendFile(FILEPATH_UPDATEDTEST_TXT); + } private void sendSerializedData() throws Exception { template.send("direct:a", new Processor() { @@ -81,94 +95,173 @@ public class CacheProducerTest extends C Poetry p = new Poetry(); p.setPoet("Ralph Waldo Emerson"); p.setPoem("Brahma"); - + // Set the property of the charset encoding exchange.setProperty(Exchange.CHARSET_NAME, "UTF-8"); Message in = exchange.getIn(); in.setBody(p); - } + } }); } - + @Test public void testAddingDataToCache() throws Exception { context.addRoutes(new RouteBuilder() { public void configure() { - from("direct:b"). - setHeader("CACHE_OPERATION", constant("ADD")). - setHeader("CACHE_KEY", constant("Ralph_Waldo_Emerson")). - to("cache://TestCache1"); + from("direct:a"). + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_ADD)). + setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")). + to("cache://TestCache1"); } }); context.start(); LOG.debug("------------Beginning CacheProducer Add Test---------------"); - sendFile(); + sendOriginalFile(); } @Test public void testAddingSerializableDataToCache() throws Exception { context.addRoutes(new RouteBuilder() { public void configure() { - from("direct:b"). - setHeader("CACHE_OPERATION", constant("ADD")). - setHeader("CACHE_KEY", constant("Ralph_Waldo_Emerson")). - to("cache://TestCache1"); + from("direct:a"). + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_ADD)). + setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")). + to("cache://TestCache1"); } }); context.start(); LOG.debug("------------Beginning CacheProducer Add Test---------------"); - sendFile(); + sendOriginalFile(); } + @Test public void testUpdatingDataInCache() throws Exception { context.addRoutes(new RouteBuilder() { public void configure() { - from("direct:b"). - setHeader("CACHE_OPERATION", constant("UPDATE")). - setHeader("CACHE_KEY", constant("Ralph_Waldo_Emerson")). - to("cache://TestCache1"); + from("direct:a"). + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_UPDATE)). + setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")). + to("cache://TestCache1"); } }); context.start(); LOG.debug("------------Beginning CacheProducer Update Test---------------"); sendSerializedData(); } - + @Test public void testDeletingDataFromCache() throws Exception { context.addRoutes(new RouteBuilder() { public void configure() { - from("direct:c"). - setHeader("CACHE_OPERATION", constant("DELETE")). - setHeader("CACHE_KEY", constant("Ralph_Waldo_Emerson")). - to("cache://TestCache1"); + from("direct:a"). + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_DELETE)). + setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")). + to("cache://TestCache1"); } }); context.start(); LOG.debug("------------Beginning CacheProducer Delete Test---------------"); sendUpdatedFile(); } - + @Test public void testDeletingAllDataFromCache() throws Exception { context.addRoutes(new RouteBuilder() { public void configure() { from("direct:a"). - setHeader("CACHE_OPERATION", constant("ADD")). - setHeader("CACHE_KEY", constant("Ralph_Waldo_Emerson")). - to("cache://TestCache1"); - from("direct:b"). - setHeader("CACHE_OPERATION", constant("ADD")). - setHeader("CACHE_KEY", constant("Ralph_Waldo_Emerson2")). - to("cache://TestCache1"); - from("direct:c"). - setHeader("CACHE_OPERATION", constant("DELETEALL")). - to("cache://TestCache1"); + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_DELETEALL)). + to("cache://TestCache1"); } }); context.start(); LOG.debug("------------Beginning CacheProducer Delete All Elements Test---------------"); sendUpdatedFile(); - } - + } + + @Test + public void testUnknownOperation() throws Exception { + context.addRoutes(new RouteBuilder() { + public void configure() { + onException(CacheException.class). + handled(true). + to("log:*** LOGGER"). + to("mock:CacheProducerTest.exception"); + + from("direct:a"). + setHeader(CacheConstants.CACHE_OPERATION, constant("UNKNOWN")). + setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")). + to("cache://TestCache1"). + to("mock:CacheProducerTest.result"); + } + }); + resultEndpoint.expectedMessageCount(0); + exceptionEndpoint.expectedMessageCount(1); + context.start(); + LOG.debug("------------Beginning CacheProducer Query An Elements Test---------------"); + sendUpdatedFile(); + resultEndpoint.assertIsSatisfied(); + exceptionEndpoint.assertIsSatisfied(); + } + + @Test + public void testQueringNonExistingDataFromCache() throws Exception { + context.addRoutes(new RouteBuilder() { + public void configure() { + onException(CacheException.class). + handled(true). + to("log:*** LOGGER"). + to("mock:CacheProducerTest.exception"); + + from("direct:a"). + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_DELETEALL)). + to("cache://TestCache1"). + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_GET)). + setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")). + to("cache://TestCache1"). + choice().when(header(CacheConstants.CACHE_ELEMENT_WAS_FOUND).isNotNull()). + to("mock:CacheProducerTest.result").end(); + } + }); + resultEndpoint.expectedMessageCount(0); + exceptionEndpoint.expectedMessageCount(0); + context.start(); + LOG.debug("------------Beginning CacheProducer Query An Elements Test---------------"); + sendUpdatedFile(); + resultEndpoint.assertIsSatisfied(); + exceptionEndpoint.assertIsSatisfied(); + } + + @Test + public void testQueringDataFromCache() throws Exception { + context.addRoutes(new RouteBuilder() { + public void configure() { + onException(CacheException.class). + handled(true). + to("log:*** LOGGER"). + to("mock:CacheProducerTest.exception"); + + from("direct:a"). + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_ADD)). + setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")). + to("cache://TestCache1"). + setBody(constant("Don't care. This body will be overridden.")). + setHeader(CacheConstants.CACHE_OPERATION, constant(CacheConstants.CACHE_OPERATION_GET)). + setHeader(CacheConstants.CACHE_KEY, constant("Ralph_Waldo_Emerson")). + to("cache://TestCache1"). + choice().when(header(CacheConstants.CACHE_ELEMENT_WAS_FOUND).isNotNull()). + to("mock:CacheProducerTest.result").end(); + } + }); + + resultEndpoint.expectedMessageCount(1); + exceptionEndpoint.expectedMessageCount(0); + String body = new String(getFileAsByteArray(FILEPATH_UPDATEDTEST_TXT), "UTF-8"); + resultEndpoint.expectedBodiesReceived(body); + context.start(); + LOG.debug("------------Beginning CacheProducer Query An Elements Test---------------"); + sendUpdatedFile(); + resultEndpoint.assertIsSatisfied(); + exceptionEndpoint.assertIsSatisfied(); + } + }