svn commit: r1057218 [1/2] - in /camel/trunk: apache-camel/src/main/descriptors/ components/ components/camel-kestrel/ components/camel-kestrel/src/ components/camel-kestrel/src/main/ components/camel-kestrel/src/main/java/ components/camel-kestrel/src...

classic Classic list List threaded Threaded
1 message Options
Reply | Threaded
Open this post in threaded view
|

svn commit: r1057218 [1/2] - in /camel/trunk: apache-camel/src/main/descriptors/ components/ components/camel-kestrel/ components/camel-kestrel/src/ components/camel-kestrel/src/main/ components/camel-kestrel/src/main/java/ components/camel-kestrel/src...

davsclaus-2
Author: davsclaus
Date: Mon Jan 10 14:46:56 2011
New Revision: 1057218

URL: http://svn.apache.org/viewvc?rev=1057218&view=rev
Log:
CAMEL-3480: Added new camel-kestrel component. Thanks to Dan Checkoway for the contribution.

Added:
    camel/trunk/components/camel-kestrel/   (with props)
    camel/trunk/components/camel-kestrel/pom.xml   (with props)
    camel/trunk/components/camel-kestrel/src/
    camel/trunk/components/camel-kestrel/src/main/
    camel/trunk/components/camel-kestrel/src/main/java/
    camel/trunk/components/camel-kestrel/src/main/java/org/
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelComponent.java   (with props)
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConfiguration.java   (with props)
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConsumer.java   (with props)
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelEndpoint.java   (with props)
    camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelProducer.java   (with props)
    camel/trunk/components/camel-kestrel/src/main/resources/
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/LICENSE.txt   (with props)
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/NOTICE.txt   (with props)
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/apache/
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/apache/camel/
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/apache/camel/component/
    camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/apache/camel/component/kestrel
    camel/trunk/components/camel-kestrel/src/test/
    camel/trunk/components/camel-kestrel/src/test/java/
    camel/trunk/components/camel-kestrel/src/test/java/org/
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelEndpointTest.java   (with props)
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProduceAndConsumeTest.java   (with props)
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProducerTest.java   (with props)
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSingleThreadedConsumerTest.java   (with props)
    camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSpringXmlTest.java   (with props)
    camel/trunk/components/camel-kestrel/src/test/resources/
    camel/trunk/components/camel-kestrel/src/test/resources/log4j.properties   (with props)
    camel/trunk/components/camel-kestrel/src/test/resources/org/
    camel/trunk/components/camel-kestrel/src/test/resources/org/apache/
    camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/
    camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/
    camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/
    camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/KestrelSpringXmlTest-context.xml   (with props)
Modified:
    camel/trunk/apache-camel/src/main/descriptors/common-bin.xml
    camel/trunk/components/pom.xml
    camel/trunk/parent/pom.xml
    camel/trunk/platforms/karaf/features/pom.xml
    camel/trunk/platforms/karaf/features/src/main/resources/bundles.properties
    camel/trunk/platforms/karaf/features/src/main/resources/features-spring2.xml
    camel/trunk/platforms/karaf/features/src/main/resources/features.xml

Modified: camel/trunk/apache-camel/src/main/descriptors/common-bin.xml
URL: http://svn.apache.org/viewvc/camel/trunk/apache-camel/src/main/descriptors/common-bin.xml?rev=1057218&r1=1057217&r2=1057218&view=diff
==============================================================================
--- camel/trunk/apache-camel/src/main/descriptors/common-bin.xml (original)
+++ camel/trunk/apache-camel/src/main/descriptors/common-bin.xml Mon Jan 10 14:46:56 2011
@@ -81,6 +81,7 @@
         <include>org.apache.camel:camel-jt400</include>
         <include>org.apache.camel:camel-juel</include>
         <include>org.apache.camel:camel-jxpath</include>
+        <include>org.apache.camel:camel-kestrel</include>
         <include>org.apache.camel:camel-ldap</include>
         <include>org.apache.camel:camel-lucene</include>
         <include>org.apache.camel:camel-mail</include>

Propchange: camel/trunk/components/camel-kestrel/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Mon Jan 10 14:46:56 2011
@@ -0,0 +1,9 @@
+.project
+.checkstyle
+.pmd
+.classpath
+target
+.settings
+eclipse-classes
+*.i??
+classes

Added: camel/trunk/components/camel-kestrel/pom.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/pom.xml?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/pom.xml (added)
+++ camel/trunk/components/camel-kestrel/pom.xml Mon Jan 10 14:46:56 2011
@@ -0,0 +1,84 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  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.
+-->
+<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>camel-parent</artifactId>
+        <version>2.6-SNAPSHOT</version>
+        <relativePath>../../parent</relativePath>
+    </parent>
+
+    <artifactId>camel-kestrel</artifactId>
+    <packaging>bundle</packaging>
+    <name>Camel :: Kestrel</name>
+    <description>Camel Kestrel support</description>
+
+    <properties>
+        <camel.osgi.export.pkg>org.apache.camel.component.kestrel.*</camel.osgi.export.pkg>
+    </properties>
+
+    <!-- TODO: we should use SMX bundle JAR -->
+    <repositories>
+        <repository>
+            <id>spy</id>
+            <name>Spy Repository</name>
+            <layout>default</layout>
+            <url>http://bleu.west.spy.net/~dustin/m2repo/</url>
+            <snapshots>
+                <enabled>false</enabled>
+            </snapshots>
+        </repository>
+    </repositories>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>spy</groupId>
+            <artifactId>memcached</artifactId>
+            <version>${spymemcached-version}</version>
+        </dependency>
+
+        <!-- testing -->
+        <dependency>
+            <groupId>org.apache.camel</groupId>
+            <artifactId>camel-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>log4j</groupId>
+            <artifactId>log4j</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+
+</project>

Propchange: camel/trunk/components/camel-kestrel/pom.xml
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/pom.xml
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Propchange: camel/trunk/components/camel-kestrel/pom.xml
------------------------------------------------------------------------------
    svn:mime-type = text/xml

Added: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelComponent.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelComponent.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelComponent.java (added)
+++ camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelComponent.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,204 @@
+/**
+ * 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.kestrel;
+
+import java.net.URI;
+import java.util.HashMap;
+import java.util.Map;
+
+import net.spy.memcached.ConnectionFactory;
+import net.spy.memcached.ConnectionFactoryBuilder;
+import net.spy.memcached.FailureMode;
+import net.spy.memcached.MemcachedClient;
+import org.apache.camel.CamelContext;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.impl.DefaultComponent;
+import org.apache.camel.util.ServiceHelper;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Camel component which offers queueing over the Memcached protocol
+ * as supported by Kestrel.
+ */
+public class KestrelComponent extends DefaultComponent {
+    private static final transient Log LOG = LogFactory.getLog(KestrelComponent.class);
+
+    private KestrelConfiguration configuration;
+    private ConnectionFactory memcachedConnectionFactory;
+
+    /**
+     * We cache the memcached clients by queue for reuse
+     */
+    private final Map<String, MemcachedClient> memcachedClientCache = new HashMap<String, MemcachedClient>();
+
+    public KestrelComponent() {
+        configuration = new KestrelConfiguration();
+    }
+
+    public KestrelComponent(KestrelConfiguration configuration) {
+        this.configuration = configuration;
+    }
+
+    public KestrelComponent(CamelContext context) {
+        super(context);
+        configuration = new KestrelConfiguration();
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+
+        ConnectionFactoryBuilder builder = new ConnectionFactoryBuilder();
+        // We never want spymemcached to time out
+        builder.setOpTimeout(9999999);
+        // Retry upon failure
+        builder.setFailureMode(FailureMode.Retry);
+        memcachedConnectionFactory = builder.build();
+    }
+
+    public KestrelConfiguration getConfiguration() {
+        return configuration;
+    }
+
+    public void setConfiguration(KestrelConfiguration configuration) {
+        this.configuration = configuration;
+    }
+
+    @SuppressWarnings("unchecked")
+    protected KestrelEndpoint createEndpoint(String uri, String remaining, Map parameters) throws Exception {
+        // Copy the configuration as each endpoint can override defaults
+        KestrelConfiguration config = getConfiguration().copy();
+
+        // Parse the URI, expected to be in one of the following formats:
+        // 1. Use the base KestrelConfiguration for host addresses:
+        //      kestrel://queue[?parameters]
+        //      kestrel:///queue[?parameters]
+        // 2. Override the host, but use the default port:
+        //      kestrel://host/queue[?parameters]
+        // 3. Override the host and port:
+        //      kestrel://host:port/queue[?parameters]
+        // 4. Supply a list of host addresses:
+        //      kestrel://host[:port],host[:port]/queue[?parameters]
+        URI u = new URI(uri);
+        String queue;
+        String[] addresses = null;
+        if (u.getPath() == null || "".equals(u.getPath())) {
+            // This would be the case when they haven't specified any explicit
+            // address(es), and the queue ends up in the "authority" portion of
+            // the URI.  For example:
+            //      kestrel://queue[?parameters]
+            queue = u.getAuthority();
+        } else if (u.getAuthority() == null || "".equals(u.getAuthority())) {
+            // The "path" was present without an authority, such as:
+            //      kestrel:///queue[?parameters]
+            queue = u.getPath();
+        } else {
+            // Both "path" and "authority" were present in the URI, which
+            // means both address(es) and the queue were specified, i.e.:
+            //      kestrel://host/queue[?parameters]
+            //      kestrel://host:port/queue[?parameters]
+            //      kestrel://host[:port],host[:port]/queue[?parameters]
+            addresses = u.getAuthority().split(",");
+            queue = u.getPath();
+        }
+
+        // Trim off any slash(es), i.e. "/queue/" -> "queue"
+        while (queue.startsWith("/")) {
+            queue = queue.substring(1);
+        }
+        while (queue.endsWith("/")) {
+            queue = queue.substring(0, queue.length() - 1);
+        }
+
+        if ("".equals(queue)) {
+            // This would be the case if the URI didn't include a path, or if
+            // the path was just "/" or something...throw an exception.
+            throw new IllegalArgumentException("Queue not specified in endpoint URI: " + uri);
+        }
+
+        if (addresses != null && addresses.length > 0) {
+            // Override the addresses on the copied config
+            config.setAddresses(addresses);
+        } else {
+            // Explicit address(es) weren't specified on the URI, which is
+            // no problem...just default the addresses to whatever was set on
+            // the base KestrelConfiguration.  And since we've already copied
+            // the config, there's nothing else we need to do there.  But let's
+            // make sure the addresses field was indeed set on the base config.
+            if (config.getAddresses() == null) {
+                throw new IllegalArgumentException("Addresses not set in base configuration or endpoint: " + uri);
+            }
+        }
+
+        LOG.info("Creating endpoint for queue \"" + queue + "\" on " + config.getAddressesAsString() + ", parameters=" + parameters);
+
+        // Finally, override config with any supplied URI parameters
+        setProperties(config, parameters);
+
+        // Create the endpoint for the given queue with the config we built
+        return new KestrelEndpoint(uri, this, config, queue);
+    }
+
+    public MemcachedClient getMemcachedClient(KestrelConfiguration config, String queue) {
+        String key = config.getAddressesAsString() + "/" + queue;
+        MemcachedClient memcachedClient = memcachedClientCache.get(key);
+        if (memcachedClient != null) {
+            return memcachedClient;
+        }
+        synchronized (memcachedClientCache) {
+            if ((memcachedClient = memcachedClientCache.get(key)) == null) {
+                LOG.info("Creating MemcachedClient for " + key);
+                try {
+                    memcachedClient = new MemcachedClient(memcachedConnectionFactory, config.getInetSocketAddresses());
+                } catch (Exception e) {
+                    throw new RuntimeCamelException("Failed to connect to " + key, e);
+                }
+                memcachedClientCache.put(key, memcachedClient);
+            }
+        }
+        return memcachedClient;
+    }
+
+    public void closeMemcachedClient(String key, MemcachedClient memcachedClient) {
+        try {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Closing client connection to " + key);
+            }
+            memcachedClient.shutdown();
+            memcachedClientCache.remove(key);
+        } catch (Exception e) {
+            LOG.warn("Failed to close client connection to " + key, e);
+        }
+    }
+
+    @Override
+    protected synchronized void doStop() throws Exception {
+        // Use a copy so we can clear the memcached client cache eagerly
+        Map<String, MemcachedClient> copy;
+        synchronized (memcachedClientCache) {
+            copy = new HashMap<String, MemcachedClient>(memcachedClientCache);
+            memcachedClientCache.clear();
+        }
+
+        for (Map.Entry<String, MemcachedClient> entry : copy.entrySet()) {
+            closeMemcachedClient(entry.getKey(), entry.getValue());
+        }
+
+        super.doStop();
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelComponent.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelComponent.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConfiguration.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConfiguration.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConfiguration.java (added)
+++ camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConfiguration.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,114 @@
+/**
+ * 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.kestrel;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.RuntimeCamelException;
+
+/**
+ * Represents the configuration of the Kestrel component and/or endpoint.
+ */
+public class KestrelConfiguration implements Cloneable {
+    /**
+     * The default port on which kestrel runs
+     */
+    public static final int DEFAULT_KESTREL_PORT = 22133;
+
+    /**
+     * The address(es) on which kestrel is running
+     */
+    private String[] addresses = new String[]{"localhost:" + DEFAULT_KESTREL_PORT};
+
+    /**
+     * How long a given wait should block (server side), in milliseconds
+     */
+    private int waitTimeMs = 100;
+
+    /**
+     * How many concurrent listeners to schedule for the thread pool
+     */
+    private int concurrentConsumers = 1;
+
+    public String[] getAddresses() {
+        return addresses;
+    }
+
+    public void setAddresses(String[] addresses) {
+        this.addresses = addresses;
+    }
+
+    public int getWaitTimeMs() {
+        return waitTimeMs;
+    }
+
+    public void setWaitTimeMs(int waitTimeMs) {
+        this.waitTimeMs = waitTimeMs;
+    }
+
+    public int getConcurrentConsumers() {
+        return concurrentConsumers;
+    }
+
+    public void setConcurrentConsumers(int concurrentConsumers) {
+        if (concurrentConsumers <= 0) {
+            throw new IllegalArgumentException("Invalid value for concurrentConsumers: " + concurrentConsumers);
+        }
+        this.concurrentConsumers = concurrentConsumers;
+    }
+
+    public String getAddressesAsString() {
+        StringBuilder bld = new StringBuilder();
+        for (String address : addresses) {
+            if (bld.length() > 0) {
+                bld.append(',');
+            }
+            bld.append(address);
+        }
+        return bld.toString();
+    }
+
+    public List<InetSocketAddress> getInetSocketAddresses() {
+        List<InetSocketAddress> list = new ArrayList<InetSocketAddress>();
+        for (String address : addresses) {
+            String[] tok = address.split(":");
+            String host;
+            int port;
+            if (tok.length == 2) {
+                host = tok[0];
+                port = Integer.parseInt(tok[1]);
+            } else if (tok.length == 1) {
+                host = tok[0];
+                port = DEFAULT_KESTREL_PORT;
+            } else {
+                throw new IllegalArgumentException("Invalid address: " + address);
+            }
+            list.add(new InetSocketAddress(host, port));
+        }
+        return list;
+    }
+
+    public KestrelConfiguration copy() {
+        try {
+            return (KestrelConfiguration) clone();
+        } catch (CloneNotSupportedException e) {
+            throw new RuntimeCamelException(e);
+        }
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConfiguration.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConfiguration.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConsumer.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConsumer.java (added)
+++ camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConsumer.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,326 @@
+/**
+ * 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.kestrel;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Exchanger;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import net.spy.memcached.MemcachedClient;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.ShutdownRunningTask;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.spi.ShutdownAware;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * A Camel consumer that polls a kestrel queue.
+ */
+public class KestrelConsumer extends DefaultConsumer implements ShutdownAware {
+    private final KestrelEndpoint endpoint;
+    private final MemcachedClient memcachedClient;
+    private final BlockingQueue<Exchanger> exchangerQueue = new LinkedBlockingQueue<Exchanger>();
+    private ExecutorService pollerExecutor;
+    private ExecutorService handlerExecutor;
+    private volatile boolean shutdownPending;
+    private CountDownLatch shutdownLatch;
+    private AtomicInteger pendingExchangeCount = new AtomicInteger(0);
+
+    public KestrelConsumer(final KestrelEndpoint endpoint, Processor processor, final MemcachedClient memcachedClient) {
+        super(endpoint, processor);
+        this.endpoint = endpoint;
+        this.memcachedClient = memcachedClient;
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        log.info("Starting consumer for " + endpoint.getEndpointUri());
+
+        int poolSize = endpoint.getConfiguration().getConcurrentConsumers();
+
+        shutdownPending = false;
+
+        if (poolSize > 1) {
+            // We'll set the shutdown latch to poolSize + 1, since we'll also
+            // wait for the poller thread when shutting down.
+            shutdownLatch = new CountDownLatch(poolSize + 1);
+
+            // Fire up the handler thread pool
+            handlerExecutor = endpoint.getCamelContext().getExecutorServiceStrategy().newFixedThreadPool(this, "Handlers-" + endpoint.getEndpointUri(), poolSize);
+            for (int k = 0; k < poolSize; ++k) {
+                handlerExecutor.execute(new Handler());
+            }
+        } else {
+            // Since we only have concurrentConsumers=1, we'll do the handling
+            // inside the poller thread, so there will only be one thread to
+            // wait for on this latch.
+            shutdownLatch = new CountDownLatch(1);
+        }
+
+        // Fire up the single poller thread
+        pollerExecutor = endpoint.getCamelContext().getExecutorServiceStrategy().newSingleThreadExecutor(this, "Poller-" + endpoint.getEndpointUri());
+        pollerExecutor.submit(new Poller(poolSize > 1));
+
+        super.doStart();
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        log.info("Stopping consumer for " + endpoint.getEndpointUri());
+
+        if (pollerExecutor != null) {
+            endpoint.getCamelContext().getExecutorServiceStrategy().shutdownNow(pollerExecutor);
+        }
+        if (handlerExecutor != null) {
+            endpoint.getCamelContext().getExecutorServiceStrategy().shutdownNow(handlerExecutor);
+        }
+
+        super.doStop();
+    }
+
+    public boolean deferShutdown(ShutdownRunningTask shutdownRunningTask) {
+        return false;
+    }
+
+    public int getPendingExchangesSize() {
+        return pendingExchangeCount.get();
+    }
+
+    public void prepareShutdown() {
+        // Signal to our threads that shutdown is happening
+        shutdownPending = true;
+
+        if (log.isDebugEnabled()) {
+            log.debug("Preparing to shutdown, waiting for " + shutdownLatch.getCount() + " threads to complete.");
+        }
+
+        // Wait for all threads to end
+        try {
+            shutdownLatch.await();
+        } catch (InterruptedException e) {
+            // ignore
+        }
+    }
+
+    /**
+     * This single thread is responsible for reading objects from kestrel and
+     * dispatching them to the handler threads.  The catch is that we don't
+     * want to poll kestrel until we know we have a handler thread available
+     * and waiting to handle whatever comes up.  So the way we deal with that
+     * is...each handler thread has an exchanger used to "receive" objects
+     * from the kestrel reader thread.  When a handler thread is ready for
+     * work, it simply puts its exchanger in the queue.  The kestrel reader
+     * thread takes an exchanger from the queue (which will block until one
+     * is there), and *then* it can poll kestrel.  Once an object is received
+     * from kestrel, it gets exchanged with the handler thread, which can
+     * take the object and process it.  Repeat...
+     */
+    @SuppressWarnings("unchecked")
+    private final class Poller implements Runnable {
+        private boolean concurrent;
+
+        private Poller(boolean concurrent) {
+            this.concurrent = concurrent;
+        }
+
+        public void run() {
+            if (log.isTraceEnabled()) {
+                log.trace("Kestrel poller is running");
+            }
+
+            // Construct the target key that we'll be requesting from kestrel.
+            // Include the /t=... wait time as applicable.
+            String target;
+            if (endpoint.getConfiguration().getWaitTimeMs() > 0) {
+                target = endpoint.getQueue() + "/t=" + endpoint.getConfiguration().getWaitTimeMs();
+            } else {
+                target = endpoint.getQueue();
+            }
+
+            Exchanger exchanger = null;
+            while (isRunAllowed() && !shutdownPending) {
+                if (concurrent) {
+                    // Wait until an exchanger is available, indicating that a
+                    // handler thread is ready to handle the next request.
+                    // Don't read from kestrel until we know a handler is ready.
+                    try {
+                        exchanger = exchangerQueue.take();
+                    } catch (InterruptedException e) {
+                        if (log.isDebugEnabled()) {
+                            log.debug("Interrupted, are we stopping? " + (isStopping() || isStopped()));
+                        }
+                        continue;
+                    }
+
+                    // We have the exchanger, so there's a handler thread ready
+                    // to handle whatever we may read...so read the next object
+                    // from the queue.
+                }
+
+                // Poll kestrel until we get an object back
+                Object value = null;
+                while (isRunAllowed() && !shutdownPending) {
+                    if (log.isTraceEnabled()) {
+                        log.trace("Polling " + target);
+                    }
+                    try {
+                        value = memcachedClient.get(target);
+                        if (value != null) {
+                            break;
+                        }
+                    } catch (Exception e) {
+                        if (isRunAllowed() && !shutdownPending) {
+                            getExceptionHandler().handleException("Failed to get object from kestrel", e);
+                        }
+                    }
+
+                    // We didn't get a value back from kestrel
+                    if (isRunAllowed() && !shutdownPending) {
+                        if (endpoint.getConfiguration().getWaitTimeMs() > 0) {
+                            // Kestrel did the blocking for us
+                        } else {
+                            // We're doing non-blocking get, so in between we
+                            // should at least sleep some short period of time
+                            // so this loop doesn't go nuts so tightly.
+                            try {
+                                Thread.sleep(100);
+                            } catch (InterruptedException ignored) {
+                            }
+                        }
+                    }
+                }
+
+                if (log.isTraceEnabled()) {
+                    log.trace("Got object from " + target);
+                }
+
+                if (concurrent) {
+                    // Pass the object to the handler thread via the exchanger.
+                    // The handler will take it from there.
+                    try {
+                        exchanger.exchange(value);
+                    } catch (InterruptedException e) {
+                        if (log.isDebugEnabled()) {
+                            log.debug("Interrupted, are we stopping? " + (isStopping() || isStopped()));
+                        }
+                        continue;
+                    }
+                } else {
+                    // We're non-concurrent, so handle it right here
+                    pendingExchangeCount.incrementAndGet();
+                    try {
+                        // Create the exchange and let camel process/route it
+                        Exchange exchange = null;
+                        try {
+                            exchange = endpoint.createExchange();
+                            exchange.getIn().setBody(value);
+                            getProcessor().process(exchange);
+                        } catch (Exception e) {
+                            if (exchange != null) {
+                                getExceptionHandler().handleException("Error processing exchange", exchange, e);
+                            } else {
+                                getExceptionHandler().handleException(e);
+                            }
+                        }
+                    } finally {
+                        // Decrement our pending exchange counter
+                        pendingExchangeCount.decrementAndGet();
+                    }
+                }
+            }
+            if (log.isTraceEnabled()) {
+                log.trace("Finished polling " + target);
+            }
+
+            // Decrement the shutdown countdown latch
+            shutdownLatch.countDown();
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private final class Handler implements Runnable {
+        private Exchanger exchanger = new Exchanger();
+
+        public void run() {
+            if (log.isTraceEnabled()) {
+                log.trace(Thread.currentThread().getName() + " is starting");
+            }
+
+            while (isRunAllowed() && !shutdownPending) {
+                // First things first, add our exchanger to the queue,
+                // indicating that we're ready for a hand-off of work
+                try {
+                    exchangerQueue.put(exchanger);
+                } catch (InterruptedException e) {
+                    if (log.isDebugEnabled()) {
+                        log.debug("Interrupted, are we stopping? " + (isStopping() || isStopped()));
+                    }
+                    continue;
+                }
+
+                // Optimistically increment our internal pending exchange
+                // counter, anticipating getting a value back from the exchanger
+                pendingExchangeCount.incrementAndGet();
+                try {
+                    // Now wait for an object to come through the exchanger
+                    Object value;
+                    try {
+                        value = exchanger.exchange(this);
+                    } catch (InterruptedException e) {
+                        if (log.isDebugEnabled()) {
+                            log.debug("Interrupted, are we stopping? " + (isStopping() || isStopped()));
+                        }
+                        continue;
+                    }
+
+                    if (log.isTraceEnabled()) {
+                        log.trace("Got a value from the exchanger");
+                    }
+
+                    // Create the exchange and let camel process/route it
+                    Exchange exchange = null;
+                    try {
+                        exchange = endpoint.createExchange();
+                        exchange.getIn().setBody(value);
+                        getProcessor().process(exchange);
+                    } catch (Exception e) {
+                        if (exchange != null) {
+                            getExceptionHandler().handleException("Error processing exchange", exchange, e);
+                        } else {
+                            getExceptionHandler().handleException(e);
+                        }
+                    }
+                } finally {
+                    // Decrement our pending exchange counter
+                    pendingExchangeCount.decrementAndGet();
+                }
+            }
+
+            // Decrement the shutdown countdown latch
+            shutdownLatch.countDown();
+
+            if (log.isTraceEnabled()) {
+                log.trace(Thread.currentThread().getName() + " is finished");
+            }
+        }
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConsumer.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelConsumer.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelEndpoint.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelEndpoint.java (added)
+++ camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelEndpoint.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,92 @@
+/**
+ * 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.kestrel;
+
+import net.spy.memcached.MemcachedClient;
+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.util.ServiceHelper;
+
+/**
+ * Camel endpoint for communication with a kestrel based queue.
+ */
+public class KestrelEndpoint extends DefaultEndpoint {
+
+    /**
+     * The configuration of this endpoint
+     */
+    private KestrelConfiguration configuration;
+
+    /**
+     * The queue we are polling
+     */
+    private String queue;
+
+    /**
+     * The kestrel component itself
+     */
+    private KestrelComponent component;
+
+    public KestrelEndpoint(String endPointURI, KestrelComponent component, KestrelConfiguration configuration, String queue) {
+        super(endPointURI, component);
+        this.component = component;
+        this.configuration = configuration;
+        this.queue = queue;
+    }
+
+    public KestrelConfiguration getConfiguration() {
+        return configuration;
+    }
+
+    public void setConfiguration(KestrelConfiguration configuration) {
+        this.configuration = configuration;
+    }
+
+    public String getQueue() {
+        return queue;
+    }
+
+    public void setQueue(String queue) {
+        this.queue = queue;
+    }
+
+    public Producer createProducer() throws Exception {
+        return new KestrelProducer(this, getMemcachedClient());
+    }
+
+    public Consumer createConsumer(Processor processor) throws Exception {
+        return new KestrelConsumer(this, processor, getMemcachedClient());
+    }
+
+    /**
+     * @return a client to kestrel using the memcached client as configured by this endpoint
+     */
+    private MemcachedClient getMemcachedClient() {
+        return component.getMemcachedClient(configuration, queue);
+    }
+
+    @Override
+    public boolean isLenientProperties() {
+        return false;
+    }
+
+    public boolean isSingleton() {
+        return false;
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelEndpoint.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelEndpoint.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelProducer.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelProducer.java (added)
+++ camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelProducer.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,55 @@
+/**
+ * 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.kestrel;
+
+import net.spy.memcached.MemcachedClient;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultProducer;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+/**
+ * Camel producer for communication with a kestrel based queue.
+ */
+public class KestrelProducer extends DefaultProducer {
+    private final KestrelEndpoint endpoint;
+    private final MemcachedClient memcachedClient;
+
+    public KestrelProducer(final KestrelEndpoint endpoint, final MemcachedClient memcachedClient) {
+        super(endpoint);
+        this.endpoint = endpoint;
+        this.memcachedClient = memcachedClient;
+    }
+
+    public void process(Exchange exchange) throws Exception {
+        String msg = exchange.getIn().getBody(String.class);
+        String queue = endpoint.getQueue();
+        if (msg != null) {
+            try {
+                if (log.isDebugEnabled()) {
+                    log.debug("Sending to: " + queue + " message: " + msg);
+                }
+                memcachedClient.set(queue, 0, msg);
+            } catch (Exception e) {
+                throw new CamelExchangeException("Error sending to: " + queue, exchange, e);
+            }
+        } else {
+            log.debug("No message body to send to: " + queue);
+        }
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelProducer.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/main/java/org/apache/camel/component/kestrel/KestrelProducer.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/main/resources/META-INF/LICENSE.txt
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/main/resources/META-INF/LICENSE.txt?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/main/resources/META-INF/LICENSE.txt (added)
+++ camel/trunk/components/camel-kestrel/src/main/resources/META-INF/LICENSE.txt Mon Jan 10 14:46:56 2011
@@ -0,0 +1,203 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright [yyyy] [name of copyright owner]
+
+   Licensed 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.
+

Propchange: camel/trunk/components/camel-kestrel/src/main/resources/META-INF/LICENSE.txt
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/main/resources/META-INF/LICENSE.txt
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Added: camel/trunk/components/camel-kestrel/src/main/resources/META-INF/NOTICE.txt
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/main/resources/META-INF/NOTICE.txt?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/main/resources/META-INF/NOTICE.txt (added)
+++ camel/trunk/components/camel-kestrel/src/main/resources/META-INF/NOTICE.txt Mon Jan 10 14:46:56 2011
@@ -0,0 +1,11 @@
+   =========================================================================
+   ==  NOTICE file corresponding to the section 4 d of                    ==
+   ==  the Apache License, Version 2.0,                                   ==
+   ==  in this case for the Apache Camel distribution.                    ==
+   =========================================================================
+
+   This product includes software developed by
+   The Apache Software Foundation (http://www.apache.org/).
+
+   Please read the different LICENSE files present in the licenses directory of
+   this distribution.

Propchange: camel/trunk/components/camel-kestrel/src/main/resources/META-INF/NOTICE.txt
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/main/resources/META-INF/NOTICE.txt
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Added: camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/apache/camel/component/kestrel
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/apache/camel/component/kestrel?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/apache/camel/component/kestrel (added)
+++ camel/trunk/components/camel-kestrel/src/main/resources/META-INF/services/org/apache/camel/component/kestrel Mon Jan 10 14:46:56 2011
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+class=org.apache.camel.component.kestrel.KestrelComponent

Added: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelEndpointTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelEndpointTest.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelEndpointTest.java (added)
+++ camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelEndpointTest.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,140 @@
+/**
+ * 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.kestrel;
+
+import java.util.Arrays;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.test.CamelTestSupport;
+
+public class KestrelEndpointTest extends CamelTestSupport {
+
+    private static final class TestCase {
+        String uri;
+        String[] addresses;
+        String queue;
+        Integer waitTimeMs;
+        Integer concurrentConsumers;
+        TestCase(String uri, String[] addresses, String queue, Integer waitTimeMs, Integer concurrentConsumers) {
+            this.uri = uri;
+            this.addresses = addresses;
+            this.queue = queue;
+            this.waitTimeMs = waitTimeMs;
+            this.concurrentConsumers = concurrentConsumers;
+        }
+    }
+    
+    private static final TestCase[] TEST_CASES = new TestCase[] {
+        new TestCase("kestrel:///queuename",
+                     null,
+                     "queuename",
+                     null,
+                     null),
+        new TestCase("kestrel://queuename?concurrentConsumers=44",
+                     null,
+                     "queuename",
+                     null,
+                     44),
+        new TestCase("kestrel://queuename?waitTimeMs=4567",
+                     null,
+                     "queuename",
+                     4567,
+                     null),
+        new TestCase("kestrel://localhost/queuename",
+                     new String[] {"localhost"},
+                     "queuename",
+                     null,
+                     null),
+        new TestCase("kestrel://127.0.0.1:22133,localhost:22134/queuename?waitTimeMs=4567&concurrentConsumers=99",
+                     new String[] {"127.0.0.1:22133", "localhost:22134"},
+                     "queuename",
+                     4567,
+                     99),
+        new TestCase("kestrel://127.0.0.1:22133/queuename?concurrentConsumers=10&waitTimeMs=4567",
+                     new String[] {"127.0.0.1:22133"},
+                     "queuename",
+                     4567,
+                     10),
+        new TestCase("kestrel://localhost/queuename?concurrentConsumers=20",
+                     new String[] {"localhost"},
+                     "queuename",
+                     null,
+                     20),
+        new TestCase("kestrel://localhost,otherhost/queuename?waitTimeMs=4567",
+                     new String[] {"localhost", "otherhost"},
+                     "queuename",
+                     4567,
+                     null),
+        new TestCase("kestrel://localhost:22133,otherhost/queuename?waitTimeMs=4567&concurrentConsumers=5",
+                     new String[] {"localhost:22133", "otherhost"},
+                     "queuename",
+                     4567,
+                     5),
+        new TestCase("kestrel://localhost,otherhost:22133/queuename?waitTimeMs=4567",
+                     new String[] {"localhost", "otherhost:22133"},
+                     "queuename",
+                     4567,
+                     null),
+    };
+
+    private KestrelConfiguration baseConfiguration;
+    private KestrelComponent kestrelComponent;
+    
+    public void testEndpoints() throws Exception {
+        for (TestCase testCase : TEST_CASES) {
+            KestrelEndpoint endpoint = (KestrelEndpoint)
+                resolveMandatoryEndpoint(testCase.uri);
+            assertEquals("getQueue(" + testCase.uri + ")", testCase.queue, endpoint.getQueue());
+            
+            KestrelConfiguration configuration = endpoint.getConfiguration();
+            
+            if (testCase.addresses != null) {
+                assertEquals("getAddresses(" + testCase.uri + ")", Arrays.asList(testCase.addresses), Arrays.asList(configuration.getAddresses()));
+            } else {
+                assertEquals("getAddresses(" + testCase.uri + ")", Arrays.asList(baseConfiguration.getAddresses()), Arrays.asList(configuration.getAddresses()));
+            }
+            
+            if (testCase.waitTimeMs != null) {
+                assertEquals("getWaitTimeMs(" + testCase.uri + ")", (Object)testCase.waitTimeMs, configuration.getWaitTimeMs());
+            } else {
+                assertEquals("getWaitTimeMs(" + testCase.uri + ")", baseConfiguration.getWaitTimeMs(), configuration.getWaitTimeMs());
+            }
+            
+            if (testCase.concurrentConsumers != null) {
+                assertEquals("getConcurrentConsumers(" + testCase.uri + ")", (Object)testCase.concurrentConsumers, configuration.getConcurrentConsumers());
+            } else {
+                assertEquals("getConcurrentConsumers(" + testCase.uri + ")", baseConfiguration.getConcurrentConsumers(), configuration.getConcurrentConsumers());
+            }
+        }
+    }
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        baseConfiguration = new KestrelConfiguration();
+        baseConfiguration.setAddresses(new String[] {"base:12345"});
+        baseConfiguration.setWaitTimeMs(9999);
+        baseConfiguration.setConcurrentConsumers(11);
+        
+        kestrelComponent = new KestrelComponent();
+        kestrelComponent.setConfiguration(baseConfiguration);
+
+        CamelContext camelContext = super.createCamelContext();
+        camelContext.addComponent("kestrel", kestrelComponent);
+
+        return camelContext;
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelEndpointTest.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelEndpointTest.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProduceAndConsumeTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProduceAndConsumeTest.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProduceAndConsumeTest.java (added)
+++ camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProduceAndConsumeTest.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,60 @@
+/**
+ * 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.kestrel;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.CamelTestSupport;
+import org.junit.Ignore;
+
+@Ignore("Manual test as you need to start a Kestrel broker")
+public class KestrelProduceAndConsumeTest extends CamelTestSupport {
+
+    @EndpointInject(uri = "kestrel://cameltest2")
+    private ProducerTemplate producerTemplate;
+
+    public void testProduceAndConsume() throws Exception {
+        List<String> bodies = new ArrayList<String>();
+        for (int k = 0; k < 10; ++k) {
+            bodies.add("this is body #" + k);
+        }
+        
+        MockEndpoint mock = getMockEndpoint("mock:result");
+        mock.expectedMinimumMessageCount(bodies.size());
+        mock.expectedBodiesReceivedInAnyOrder(bodies);
+
+        for (String body : bodies) {
+            producerTemplate.sendBody(body);
+        }
+
+        mock.assertIsSatisfied();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            public void configure() {
+                from("kestrel://cameltest2?concurrentConsumers=3").to("log:incoming").to("mock:result");
+            }
+        };
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProduceAndConsumeTest.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProduceAndConsumeTest.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProducerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProducerTest.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProducerTest.java (added)
+++ camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProducerTest.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,35 @@
+/**
+ * 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.kestrel;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.test.CamelTestSupport;
+import org.junit.Ignore;
+
+@Ignore("Manual test as you need to start a Kestrel broker")
+public class KestrelProducerTest extends CamelTestSupport {
+
+    @EndpointInject(uri = "kestrel://cameltest1")
+    ProducerTemplate producerTemplate;
+
+    public void testKestrelProducer() throws Exception {
+        for (int k = 0; k < 100; ++k) {
+            producerTemplate.sendBody("test body " + k);
+        }
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProducerTest.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelProducerTest.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSingleThreadedConsumerTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSingleThreadedConsumerTest.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSingleThreadedConsumerTest.java (added)
+++ camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSingleThreadedConsumerTest.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,60 @@
+/**
+ * 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.kestrel;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.CamelTestSupport;
+import org.junit.Ignore;
+
+@Ignore("Manual test as you need to start a Kestrel broker")
+public class KestrelSingleThreadedConsumerTest extends CamelTestSupport {
+
+    @EndpointInject(uri = "kestrel://cameltest4")
+    private ProducerTemplate producerTemplate;
+
+    public void testProduceAndConsume() throws Exception {
+        List<String> bodies = new ArrayList<String>();
+        for (int k = 0; k < 10; ++k) {
+            bodies.add("this is body #" + k);
+        }
+        
+        MockEndpoint mock = getMockEndpoint("mock:result");
+        mock.expectedMinimumMessageCount(bodies.size());
+        mock.expectedBodiesReceivedInAnyOrder(bodies);
+
+        for (String body : bodies) {
+            producerTemplate.sendBody(body);
+        }
+
+        mock.assertIsSatisfied();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            public void configure() {
+                from("kestrel://cameltest4?concurrentConsumers=1").to("log:incoming").to("mock:result");
+            }
+        };
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSingleThreadedConsumerTest.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSingleThreadedConsumerTest.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSpringXmlTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSpringXmlTest.java?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSpringXmlTest.java (added)
+++ camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSpringXmlTest.java Mon Jan 10 14:46:56 2011
@@ -0,0 +1,56 @@
+/**
+ * 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.kestrel;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Ignore;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit38.AbstractJUnit38SpringContextTests;
+
+@ContextConfiguration
+@Ignore("Manual test as you need to start a Kestrel broker")
+public class KestrelSpringXmlTest extends AbstractJUnit38SpringContextTests {
+
+    @EndpointInject(uri = "mock:result")
+    protected MockEndpoint mock;
+    
+    @EndpointInject(uri = "kestrel://cameltest3")
+    private ProducerTemplate producerTemplate;
+
+    @DirtiesContext
+    public void testProduceAndConsume() throws Exception {
+        List<String> bodies = new ArrayList<String>();
+        for (int k = 0; k < 10; ++k) {
+            bodies.add("this is body #" + k);
+        }
+        
+        mock.expectedMinimumMessageCount(bodies.size());
+        mock.expectedBodiesReceivedInAnyOrder(bodies);
+
+        for (String body : bodies) {
+            producerTemplate.sendBody(body);
+        }
+
+        mock.assertIsSatisfied();
+    }
+}

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSpringXmlTest.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/test/java/org/apache/camel/component/kestrel/KestrelSpringXmlTest.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: camel/trunk/components/camel-kestrel/src/test/resources/log4j.properties
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/test/resources/log4j.properties?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/test/resources/log4j.properties (added)
+++ camel/trunk/components/camel-kestrel/src/test/resources/log4j.properties Mon Jan 10 14:46:56 2011
@@ -0,0 +1,35 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+#
+# The logging properties used during tests..
+#
+log4j.rootLogger=INFO, out
+
+#log4j.logger.org.apache.camel.component.kestrel=DEBUG
+
+# CONSOLE appender not used by default
+log4j.appender.stdout=org.apache.log4j.ConsoleAppender
+log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
+log4j.appender.stdout.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+
+# File appender
+log4j.appender.out=org.apache.log4j.FileAppender
+log4j.appender.out.layout=org.apache.log4j.PatternLayout
+log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
+log4j.appender.out.file=target/camel-kestrel-test.log
+log4j.appender.out.append=true

Propchange: camel/trunk/components/camel-kestrel/src/test/resources/log4j.properties
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/test/resources/log4j.properties
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Propchange: camel/trunk/components/camel-kestrel/src/test/resources/log4j.properties
------------------------------------------------------------------------------
    svn:mime-type = text/plain

Added: camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/KestrelSpringXmlTest-context.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/KestrelSpringXmlTest-context.xml?rev=1057218&view=auto
==============================================================================
--- camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/KestrelSpringXmlTest-context.xml (added)
+++ camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/KestrelSpringXmlTest-context.xml Mon Jan 10 14:46:56 2011
@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  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.
+-->
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xsi:schemaLocation="http://www.springframework.org/schema/beans
+                           http://www.springframework.org/schema/beans/spring-beans.xsd
+                           http://camel.apache.org/schema/spring
+                           http://camel.apache.org/schema/spring/camel-spring.xsd">
+
+    <camelContext xmlns="http://camel.apache.org/schema/spring">
+        <route>
+            <from uri="kestrel://cameltest3?concurrentConsumers=3"/>
+            <to uri="mock:result"/>
+        </route>
+    </camelContext>
+
+</beans>

Propchange: camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/KestrelSpringXmlTest-context.xml
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/KestrelSpringXmlTest-context.xml
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Propchange: camel/trunk/components/camel-kestrel/src/test/resources/org/apache/camel/component/kestrel/KestrelSpringXmlTest-context.xml
------------------------------------------------------------------------------
    svn:mime-type = text/xml

Modified: camel/trunk/components/pom.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/pom.xml?rev=1057218&r1=1057217&r2=1057218&view=diff
==============================================================================
--- camel/trunk/components/pom.xml (original)
+++ camel/trunk/components/pom.xml Mon Jan 10 14:46:56 2011
@@ -80,6 +80,7 @@
     <module>camel-jt400</module>
     <module>camel-juel</module>
     <module>camel-jxpath</module>
+    <module>camel-kestrel</module>
     <module>camel-ldap</module>
     <module>camel-lucene</module>
     <module>camel-mail</module>

Modified: camel/trunk/parent/pom.xml
URL: http://svn.apache.org/viewvc/camel/trunk/parent/pom.xml?rev=1057218&r1=1057217&r2=1057218&view=diff
==============================================================================
--- camel/trunk/parent/pom.xml (original)
+++ camel/trunk/parent/pom.xml Mon Jan 10 14:46:56 2011
@@ -133,6 +133,7 @@
     <spring-osgi-version>1.2.0</spring-osgi-version>
     <spring-security-version>3.0.5.RELEASE</spring-security-version>
     <spring-ws-version>1.5.9</spring-ws-version>
+    <spymemcached-version>2.5</spymemcached-version>
     <stax-api-version>1.0.1</stax-api-version>
     <stringtemplate-version>3.0</stringtemplate-version>
     <tagsoup-version>1.2</tagsoup-version>
@@ -459,6 +460,11 @@
       </dependency>
       <dependency>
         <groupId>org.apache.camel</groupId>
+        <artifactId>camel-kestrel</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.camel</groupId>
         <artifactId>camel-lucene</artifactId>
         <version>${project.version}</version>
       </dependency>