camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject svn commit: r1445148 - in /camel/branches/camel-2.10.x: ./ components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/ components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/ components/camel-hazelcast/src...
Date Tue, 12 Feb 2013 12:53:26 GMT
Author: davsclaus
Date: Tue Feb 12 12:53:26 2013
New Revision: 1445148

URL: http://svn.apache.org/r1445148
Log:
CAMEL-6054: Added transacted option to hazelcast seda consumer. Thanks to Thomas Vautrin for
the patch.

Added:
    camel/branches/camel-2.10.x/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaInOutTransactedTest.java
      - copied unchanged from r1445076, camel/trunk/components/camel-hazelcast/src/test/java/org/apache/camel/component/hazelcast/HazelcastSedaInOutTransactedTest.java
Modified:
    camel/branches/camel-2.10.x/   (props changed)
    camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastDefaultEndpoint.java
    camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java
    camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java
    camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaEndpoint.java

Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
  Merged /camel/trunk:r1445076

Propchange: camel/branches/camel-2.10.x/
------------------------------------------------------------------------------
Binary property 'svnmerge-integrated' - no diff available.

Modified: camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastDefaultEndpoint.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastDefaultEndpoint.java?rev=1445148&r1=1445147&r2=1445148&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastDefaultEndpoint.java
(original)
+++ camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/HazelcastDefaultEndpoint.java
Tue Feb 12 12:53:26 2013
@@ -45,4 +45,8 @@ public abstract class HazelcastDefaultEn
     public boolean isSingleton() {
         return true;
     }
+
+    public HazelcastInstance getHazelcastInstance() {
+        return hazelcastInstance;
+    }
 }

Modified: camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java?rev=1445148&r1=1445147&r2=1445148&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java
(original)
+++ camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConfiguration.java
Tue Feb 12 12:53:26 2013
@@ -25,6 +25,7 @@ public class HazelcastSedaConfiguration 
     private int pollInterval = 1000;
     private String queueName;
     private boolean transferExchange;
+    private boolean transacted;
 
     public HazelcastSedaConfiguration() {
     }
@@ -61,4 +62,12 @@ public class HazelcastSedaConfiguration 
         this.transferExchange = transferExchange;
     }
 
+    public boolean isTransacted() {
+        return transacted;
+    }
+
+    public void setTransacted(boolean transacted) {
+        this.transacted = transacted;
+    }
+
 }

Modified: camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java?rev=1445148&r1=1445147&r2=1445148&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java
(original)
+++ camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaConsumer.java
Tue Feb 12 12:53:26 2013
@@ -20,6 +20,7 @@ import java.util.concurrent.BlockingQueu
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import com.hazelcast.core.Transaction;
 import org.apache.camel.AsyncCallback;
 import org.apache.camel.AsyncProcessor;
 import org.apache.camel.Consumer;
@@ -74,6 +75,16 @@ public class HazelcastSedaConsumer exten
         while (queue != null && isRunAllowed()) {
             final Exchange exchange = this.getEndpoint().createExchange();
 
+            Transaction transaction = null;
+            if (endpoint.getConfiguration().isTransacted()) {
+                // Get and begin transaction if exist
+                transaction = endpoint.getHazelcastInstance().getTransaction();
+
+                if (transaction != null && transaction.getStatus() == Transaction.TXN_STATUS_NO_TXN)
{
+                    log.trace("Begin transaction: {}", transaction);
+                    transaction.begin();
+                }
+            }
             try {
                 final Object body = queue.poll(endpoint.getConfiguration().getPollInterval(),
TimeUnit.MILLISECONDS);
 
@@ -92,19 +103,38 @@ public class HazelcastSedaConsumer exten
                         });
 
                         if (exchange.getException() != null) {
+                            // Rollback
+                            if (transaction != null) {
+                                transaction.rollback();
+                            }
                             getExceptionHandler().handleException("Error processing exchange",
exchange, exchange.getException());
                         }
 
                     } catch (Exception e) {
                         LOG.error("Hzlq Exception caught: " + e, e);
+                        // Rollback
+                        if (transaction != null) {
+                            log.trace("Rollback transaction: {}", transaction);
+                            transaction.rollback();
+                        }
                     }
                 }
+                // It's OK, I commit
+                if (exchange.getException() == null && transaction != null &&
transaction.getStatus() == Transaction.TXN_STATUS_ACTIVE) {
+                    log.trace("Commit transaction: {}", transaction);
+                    transaction.commit();
+                }
             } catch (InterruptedException e) {
                 if (LOG.isDebugEnabled()) {
                     LOG.debug("Hzlq Consumer Interrupted: " + e, e);
                 }
                 continue;
             } catch (Throwable e) {
+                // Rollback
+                if (transaction != null) {
+                    log.trace("Rollback transaction: {}", transaction);
+                    transaction.rollback();
+                }
                 getExceptionHandler().handleException("Error processing exchange", exchange,
e);
             }
         }

Modified: camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaEndpoint.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaEndpoint.java?rev=1445148&r1=1445147&r2=1445148&view=diff
==============================================================================
--- camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaEndpoint.java
(original)
+++ camel/branches/camel-2.10.x/components/camel-hazelcast/src/main/java/org/apache/camel/component/hazelcast/seda/HazelcastSedaEndpoint.java
Tue Feb 12 12:53:26 2013
@@ -36,7 +36,7 @@ public class HazelcastSedaEndpoint exten
     private final HazelcastSedaConfiguration configuration;
 
     public HazelcastSedaEndpoint(final HazelcastInstance hazelcastInstance, final String
uri, final HazelcastComponent component, final HazelcastSedaConfiguration configuration) {
-        super(component.getHazelcastInstance(), uri, component);
+        super(hazelcastInstance, uri, component);
         this.queue = hazelcastInstance.getQueue(configuration.getQueueName());
         this.configuration = configuration;
         if (ObjectHelper.isEmpty(configuration.getQueueName())) {



Mime
View raw message