camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject svn commit: r1202222 - in /camel/branches/camel-2.8.x: ./ components/camel-jpa/src/main/java/org/apache/camel/component/jpa/ components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/
Date Tue, 15 Nov 2011 14:56:49 GMT
Author: davsclaus
Date: Tue Nov 15 14:56:49 2011
New Revision: 1202222

URL: http://svn.apache.org/viewvc?rev=1202222&view=rev
Log:
CAMEL-4683: Added consumer.transcted option to JpaConsumer to control TX behavior.

Added:
    camel/branches/camel-2.8.x/components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/JpaNonTXRollbackTest.java
      - copied unchanged from r1202215, camel/trunk/components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/JpaNonTXRollbackTest.java
Modified:
    camel/branches/camel-2.8.x/   (props changed)
    camel/branches/camel-2.8.x/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
    camel/branches/camel-2.8.x/components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/JpaTXRollbackTest.java

Propchange: camel/branches/camel-2.8.x/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Nov 15 14:56:49 2011
@@ -1 +1 @@
-/camel/trunk:1202148,1202167
+/camel/trunk:1202148,1202167,1202215

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

Modified: camel/branches/camel-2.8.x/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java?rev=1202222&r1=1202221&r2=1202222&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
(original)
+++ camel/branches/camel-2.8.x/components/camel-jpa/src/main/java/org/apache/camel/component/jpa/JpaConsumer.java
Tue Nov 15 14:56:49 2011
@@ -54,6 +54,7 @@ public class JpaConsumer extends Schedul
     private String nativeQuery;
     private Class resultClass;
     private int maxMessagesPerPoll;
+    private boolean transacted;
     private volatile ShutdownRunningTask shutdownRunningTask;
     private volatile int pendingExchanges;
 
@@ -96,17 +97,29 @@ public class JpaConsumer extends Schedul
                     answer.add(holder);
                 }
 
-                int messagePolled;
+                PersistenceException cause = null;
+                int messagePolled = 0;
                 try {
                     messagePolled = processBatch(CastUtils.cast(answer));
                 } catch (Exception e) {
                     if (e instanceof PersistenceException) {
-                        throw (PersistenceException) e;
+                        cause = (PersistenceException) e;
                     } else {
-                        throw new PersistenceException(e);
+                        cause = new PersistenceException(e);
                     }
                 }
 
+                if (cause != null) {
+                    if (!isTransacted()) {
+                        LOG.warn("Error processing last message due: {}. Will commit all
previous successful processed message, and ignore this last failure.", cause.getMessage(),
cause);
+                        entityManager.flush();
+                    } else {
+                        // rollback all by throwning exception
+                        throw cause;
+                    }
+                }
+
+                // commit
                 LOG.debug("Flushing EntityManager");
                 entityManager.flush();
                 return messagePolled;
@@ -270,7 +283,22 @@ public class JpaConsumer extends Schedul
 
     public void setResultClass(Class resultClass) {
         this.resultClass = resultClass;
-    }    
+    }
+
+    public boolean isTransacted() {
+        return transacted;
+    }
+
+    /**
+     * Sets whether to run in transacted mode or not.
+     * <p/>
+     * This option is default <tt>false</tt>. When <tt>false</tt>
then all the good messages
+     * will commit, and the first failed message will rollback.
+     * However when <tt>true</tt>, then all messages will rollback, if just one
message failed.
+     */
+    public void setTransacted(boolean transacted) {
+        this.transacted = transacted;
+    }
 
     // Implementation methods
     // -------------------------------------------------------------------------

Modified: camel/branches/camel-2.8.x/components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/JpaTXRollbackTest.java
URL: http://svn.apache.org/viewvc/camel/branches/camel-2.8.x/components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/JpaTXRollbackTest.java?rev=1202222&r1=1202221&r2=1202222&view=diff
==============================================================================
--- camel/branches/camel-2.8.x/components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/JpaTXRollbackTest.java
(original)
+++ camel/branches/camel-2.8.x/components/camel-jpa/src/test/java/org/apache/camel/processor/jpa/JpaTXRollbackTest.java
Tue Nov 15 14:56:49 2011
@@ -75,7 +75,7 @@ public class JpaTXRollbackTest extends C
         return new RouteBuilder() {
             @Override
             public void configure() throws Exception {
-                from("jpa://" + SendEmail.class.getName() + "?delay=2000").routeId("foo").noAutoStartup()
+                from("jpa://" + SendEmail.class.getName() + "?consumer.transacted=true&delay=1000").routeId("foo").noAutoStartup()
                         .process(new Processor() {
                             @Override
                             public void process(Exchange exchange) throws Exception {



Mime
View raw message