camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r566207 - in /activemq/camel/trunk: ./ components/camel-activemq/src/main/java/org/apache/camel/component/activemq/ components/camel-activemq/src/main/resources/META-INF/services/org/apache/camel/component/ components/camel-activemq/src/tes...
Date Wed, 15 Aug 2007 14:53:22 GMT
Author: chirino
Date: Wed Aug 15 07:53:21 2007
New Revision: 566207

URL: http://svn.apache.org/viewvc?view=rev&rev=566207
Log:
Adding support for a activemq.journal component that can be used to persist exchanges in file
based journal

Added:
    activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalComponent.java
  (with props)
    activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalEndpoint.java
  (with props)
    activemq/camel/trunk/components/camel-activemq/src/main/resources/META-INF/services/org/apache/camel/component/activemq.journal
    activemq/camel/trunk/components/camel-activemq/src/test/java/org/apache/camel/component/activemq/JournalRouteTest.java
  (with props)
Modified:
    activemq/camel/trunk/pom.xml

Added: activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalComponent.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalComponent.java?view=auto&rev=566207
==============================================================================
--- activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalComponent.java
(added)
+++ activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalComponent.java
Wed Aug 15 07:53:21 2007
@@ -0,0 +1,40 @@
+/**
+ * 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.activemq;
+
+import java.io.File;
+import java.util.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.DefaultComponent;
+import org.apache.camel.impl.DefaultExchange;
+import org.apache.camel.util.IntrospectionSupport;
+
+/**
+ * The <a href="http://activemq.apache.org/camel/activemq.html">ActiveMQ Component</a>
+ *
+ * @version $Revision$
+ */
+public class JournalComponent extends DefaultComponent<DefaultExchange> {
+
+    @Override
+    protected Endpoint<DefaultExchange> createEndpoint(String uri, String remaining,
Map parameters) throws Exception {        
+        JournalEndpoint endpoint = new JournalEndpoint(uri, this, new File(remaining));
+        IntrospectionSupport.setProperties(endpoint, parameters);
+        return endpoint;
+    }
+}

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

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

Added: activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalEndpoint.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalEndpoint.java?view=auto&rev=566207
==============================================================================
--- activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalEndpoint.java
(added)
+++ activemq/camel/trunk/components/camel-activemq/src/main/java/org/apache/camel/component/activemq/JournalEndpoint.java
Wed Aug 15 07:53:21 2007
@@ -0,0 +1,233 @@
+/**
+ * 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.activemq;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.activemq.kaha.impl.async.AsyncDataManager;
+import org.apache.activemq.kaha.impl.async.Location;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.camel.CamelExchangeException;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.impl.DefaultExchange;
+import org.apache.camel.impl.DefaultProducer;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+public class JournalEndpoint extends DefaultEndpoint<DefaultExchange> {
+
+    private static final transient Log LOG = LogFactory.getLog(JournalEndpoint.class);
+
+    public final class JournalProducer extends DefaultProducer<DefaultExchange> {
+        private boolean syncWrite = JournalEndpoint.this.syncWrite;
+
+        public JournalProducer(Endpoint<DefaultExchange> endpoint) {
+            super(endpoint);
+        }
+
+        public void process(Exchange exchange) throws Exception {
+            incrementReference();
+            try {
+
+                ByteSequence body = exchange.getIn().getBody(ByteSequence.class);
+                if (body == null) {
+                    byte[] bytes = exchange.getIn().getBody(byte[].class);
+                    if (bytes != null) {
+                        body = new ByteSequence(bytes);
+                    }
+                }
+                if (body == null) {
+                    throw new CamelExchangeException("In body message could not be converted
to a ByteSequence or a byte array.", exchange);
+                }
+                dataManager.write(body, true);
+
+            } finally {
+                decrementReference();
+            }
+        }
+
+        public boolean isSyncWrite() {
+            return syncWrite;
+        }
+
+        public void setSyncWrite(boolean syncWrite) {
+            this.syncWrite = syncWrite;
+        }
+    }
+
+    private final File directory;
+    private final AtomicReference<DefaultConsumer<DefaultExchange>> consumer
= new AtomicReference<DefaultConsumer<DefaultExchange>>();
+    private final Object activationMutex = new Object();
+    private int referenceCount;
+    private AsyncDataManager dataManager;
+    private Thread thread;
+    private Location lastReadLocation;
+    private boolean syncWrite = true;
+    private long idleDelay = 1000;
+
+    public JournalEndpoint(String uri, JournalComponent journalComponent, File directory)
{
+        super(uri, journalComponent.getCamelContext());
+        this.directory = directory;
+    }
+
+    public DefaultExchange createExchange() {
+        return new DefaultExchange(getContext());
+    }
+
+    public boolean isSingleton() {
+        return true;
+    }
+
+    public File getDirectory() {
+        return directory;
+    }
+
+    public Consumer<DefaultExchange> createConsumer(Processor processor) throws Exception
{
+        return new DefaultConsumer<DefaultExchange>(this, processor) {
+            @Override
+            public void start() throws Exception {
+                super.start();
+                activateConsumer(this);
+            }
+
+            @Override
+            public void stop() throws Exception {
+                deactivateConsumer(this);
+                super.stop();
+            }
+        };
+    }
+
+    protected void decrementReference() throws IOException {
+        synchronized (activationMutex) {
+            referenceCount--;
+            if (referenceCount == 0) {
+                LOG.debug("Closing data manager: " + directory);
+                LOG.debug("Last mark at: " + lastReadLocation);
+                dataManager.close();
+                dataManager = null;
+            }
+        }
+    }
+
+    protected void incrementReference() throws IOException {
+        synchronized (activationMutex) {
+            referenceCount++;
+            if (referenceCount == 1) {
+                LOG.debug("Opening data manager: " + directory);
+                dataManager = new AsyncDataManager();
+                dataManager.setDirectory(directory);
+                dataManager.start();
+
+                lastReadLocation = dataManager.getMark();
+                LOG.debug("Last mark at: " + lastReadLocation);
+            }
+        }
+    }
+
+    protected void deactivateConsumer(DefaultConsumer<DefaultExchange> consumer) throws
IOException {
+        synchronized (activationMutex) {
+            if (this.consumer.get() != consumer) {
+                throw new RuntimeCamelException("Consumer was not active.");
+            }
+            this.consumer.set(null);
+            try {
+                thread.join();
+            } catch (InterruptedException e) {
+                throw new InterruptedIOException();
+            }
+            decrementReference();
+        }
+    }
+
+    protected void activateConsumer(DefaultConsumer<DefaultExchange> consumer) throws
IOException {
+        synchronized (activationMutex) {
+            if (this.consumer.get() != null) {
+                throw new RuntimeCamelException("Consumer already active: journal endpoints
only support 1 active consumer");
+            }
+            incrementReference();
+            this.consumer.set(consumer);
+            thread = new Thread() {
+                @Override
+                public void run() {
+                    dispatchToConsumer();
+                }
+            };
+            thread.setName("Dipatch thread: " + getEndpointUri());
+            thread.setDaemon(true);
+            thread.start();
+        }
+    }
+
+    protected void dispatchToConsumer() {
+        try {
+            DefaultConsumer<DefaultExchange> consumer;
+            while ((consumer = this.consumer.get()) != null) {
+                // See if there is a new record to process
+                Location location = dataManager.getNextLocation(lastReadLocation);
+                if (location != null) {
+
+                    // Send it on.
+                    ByteSequence read = dataManager.read(location);
+                    DefaultExchange exchange = createExchange();
+                    exchange.getIn().setBody(read);
+                    exchange.getIn().setHeader("journal", getEndpointUri());
+                    exchange.getIn().setHeader("location", location);
+                    consumer.getProcessor().process(exchange);
+
+                    // Setting the mark makes the data manager forget about
+                    // everything
+                    // before that record.
+                    if (LOG.isDebugEnabled()) {
+                        LOG.debug("Consumed record at: " + location);
+                    }
+                    dataManager.setMark(location, true);
+                    lastReadLocation = location;
+                } else {
+                    // Avoid a tight CPU loop if there is no new record to read.
+                    LOG.debug("Sleeping due to no records being available.");
+                    Thread.sleep(idleDelay);
+                }
+            }
+        } catch (Throwable e) {
+            e.printStackTrace();
+        }
+    }
+
+    public Producer<DefaultExchange> createProducer() throws Exception {
+        return new JournalProducer(this);
+    }
+
+    public boolean isSyncWrite() {
+        return syncWrite;
+    }
+
+    public void setSyncWrite(boolean syncWrite) {
+        this.syncWrite = syncWrite;
+    }
+
+}

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

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

Added: activemq/camel/trunk/components/camel-activemq/src/main/resources/META-INF/services/org/apache/camel/component/activemq.journal
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-activemq/src/main/resources/META-INF/services/org/apache/camel/component/activemq.journal?view=auto&rev=566207
==============================================================================
--- activemq/camel/trunk/components/camel-activemq/src/main/resources/META-INF/services/org/apache/camel/component/activemq.journal
(added)
+++ activemq/camel/trunk/components/camel-activemq/src/main/resources/META-INF/services/org/apache/camel/component/activemq.journal
Wed Aug 15 07:53:21 2007
@@ -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.activemq.JournalComponent

Added: activemq/camel/trunk/components/camel-activemq/src/test/java/org/apache/camel/component/activemq/JournalRouteTest.java
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/components/camel-activemq/src/test/java/org/apache/camel/component/activemq/JournalRouteTest.java?view=auto&rev=566207
==============================================================================
--- activemq/camel/trunk/components/camel-activemq/src/test/java/org/apache/camel/component/activemq/JournalRouteTest.java
(added)
+++ activemq/camel/trunk/components/camel-activemq/src/test/java/org/apache/camel/component/activemq/JournalRouteTest.java
Wed Aug 15 07:53:21 2007
@@ -0,0 +1,65 @@
+/**
+ * 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.activemq;
+
+import java.util.List;
+
+import org.apache.activemq.util.ByteSequence;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.AssertionClause;
+import org.apache.camel.component.mock.MockEndpoint;
+
+/**
+ * @version $Revision$
+ */
+public class JournalRouteTest extends ContextTestSupport {
+
+    public void testForwardingAMessageAcrossJMSKeepingCustomJMSHeaders() throws Exception
{
+
+        byte[] payload = "Hello World".getBytes();
+        
+        
+        MockEndpoint resultEndpoint = resolveMandatoryEndpoint("mock:result", MockEndpoint.class);
+        resultEndpoint.expectedMessageCount(1);
+        
+        AssertionClause firstMessageExpectations = resultEndpoint.message(0);
+        firstMessageExpectations.header("journal").isEqualTo("activemq.journal:test.a");
+        firstMessageExpectations.header("location").isNotNull();
+        firstMessageExpectations.body().isInstanceOf(ByteSequence.class);
+
+        template.sendBody("direct:test.a", payload);
+
+        resultEndpoint.assertIsSatisfied();
+
+        List<Exchange> list = resultEndpoint.getReceivedExchanges();
+        Exchange exchange = list.get(0);
+        ByteSequence body = (ByteSequence)exchange.getIn().getBody();
+        body.compact(); // trims the byte array to the actual size.
+        assertEquals("body", new String(payload), new String(body.data));
+    }
+
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            public void configure() throws Exception {
+                from("direct:test.a").to("activemq.journal:test.a");
+                from("activemq.journal:test.a").to("mock:result");
+            }
+        };
+    }
+}
\ No newline at end of file

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

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

Modified: activemq/camel/trunk/pom.xml
URL: http://svn.apache.org/viewvc/activemq/camel/trunk/pom.xml?view=diff&rev=566207&r1=566206&r2=566207
==============================================================================
--- activemq/camel/trunk/pom.xml (original)
+++ activemq/camel/trunk/pom.xml Wed Aug 15 07:53:21 2007
@@ -395,12 +395,12 @@
       <dependency>
         <groupId>org.apache.activemq</groupId>
         <artifactId>apache-activemq</artifactId>
-        <version>4.1.1</version>
+        <version>5.0-SNAPSHOT</version>
       </dependency>
       <dependency>
         <groupId>org.apache.activemq</groupId>
         <artifactId>activemq-core</artifactId>
-        <version>4.1.1</version>
+        <version>5.0-SNAPSHOT</version>
       </dependency>
       <dependency>
         <groupId>org.apache.xbean</groupId>



Mime
View raw message