[hornetq-commits] JBoss hornetq SVN: r10583 - trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push.

[do-not-reply at jboss.org]

Author: clebert.suconic at jboss.com
Date: 2011-05-02 13:36:57 -0400 (Mon, 02 May 2011)
New Revision: 10583

Modified:
   trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/PushConsumer.java
Log:
changing commit with Bill

Modified: trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/PushConsumer.java
===================================================================
--- trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/PushConsumer.java	2011-04-29 17:04:14 UTC (rev 10582)
+++ trunk/hornetq-rest/hornetq-rest/src/main/java/org/hornetq/rest/queue/push/PushConsumer.java	2011-05-02 17:36:57 UTC (rev 10583)
@@ -70,7 +70,7 @@
       strategy.setRegistration(registration);
       strategy.start();
 
-      session = factory.createSession(true, true, 0);
+      session = factory.createSession(false, false, 0);
       if (registration.getSelector() != null)
       {
          consumer = session.createConsumer(destination, SelectorTranslator.convertToHornetQFilterString(registration.getSelector()));
@@ -127,6 +127,16 @@
    @Override
    public void onMessage(ClientMessage clientMessage)
    {
+
+      try
+      {
+           clientMessage.acknowledge();
+      }
+      catch (HornetQException e)
+      {
+           throw new RuntimeException(e.getMessage(), e);
+      }
+
       boolean acknowledge = strategy.push(clientMessage);
 
       if (acknowledge)
@@ -134,7 +144,7 @@
          try
          {
             log.debug("Acknowledging: " + clientMessage.getMessageID());
-            clientMessage.acknowledge();
+            session.commit();
             return;
          }
          catch (HornetQException e)
@@ -144,14 +154,20 @@
       }
       else
       {
-         if (registration.isDisableOnFailure())
+          try
+          {
+              session.rollback();
+          }
+          catch (HornetQException e)
+          {
+              throw new RuntimeException(e.getMessage(), e);
+          }
+          if (registration.isDisableOnFailure())
          {
             log.error("Failed to push message to " + registration.getTarget() + " disabling push registration...");
             disableFromFailure();
             return;
          }
-         // let hornetq decide what to do
-         throw new RuntimeException("Failed to push message to " + registration.getTarget());
       }
    }
 }