activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cmacn...@apache.org
Subject svn commit: r786344 - /activemq/sandbox/activemq-flow/activemq-openwire/src/main/java/org/apache/activemq/broker/openwire/OpenwireProtocolHandler.java
Date Fri, 19 Jun 2009 00:49:20 GMT
Author: cmacnaug
Date: Fri Jun 19 00:49:19 2009
New Revision: 786344

URL: http://svn.apache.org/viewvc?rev=786344&view=rev
Log:
Changing isRemoveOnDispatch to return false for qeueue receivers. Otherwise messages are removed
from the queue too soon.

Modified:
    activemq/sandbox/activemq-flow/activemq-openwire/src/main/java/org/apache/activemq/broker/openwire/OpenwireProtocolHandler.java

Modified: activemq/sandbox/activemq-flow/activemq-openwire/src/main/java/org/apache/activemq/broker/openwire/OpenwireProtocolHandler.java
URL: http://svn.apache.org/viewvc/activemq/sandbox/activemq-flow/activemq-openwire/src/main/java/org/apache/activemq/broker/openwire/OpenwireProtocolHandler.java?rev=786344&r1=786343&r2=786344&view=diff
==============================================================================
--- activemq/sandbox/activemq-flow/activemq-openwire/src/main/java/org/apache/activemq/broker/openwire/OpenwireProtocolHandler.java
(original)
+++ activemq/sandbox/activemq-flow/activemq-openwire/src/main/java/org/apache/activemq/broker/openwire/OpenwireProtocolHandler.java
Fri Jun 19 00:49:19 2009
@@ -100,12 +100,12 @@
     private Router router;
     private VirtualHost host;
     private final CommandVisitor visitor;
-    
-    ArrayList<ActiveMQDestination> temporaryDestinations = new ArrayList<ActiveMQDestination>();

+
+    ArrayList<ActiveMQDestination> temporaryDestinations = new ArrayList<ActiveMQDestination>();
 
     public OpenwireProtocolHandler() {
         setStoreWireFormat(new OpenWireFormat());
-        
+
         visitor = new CommandVisitor() {
 
             // /////////////////////////////////////////////////////////////////
@@ -152,7 +152,7 @@
             public Response processRemoveConsumer(RemoveInfo remove, ConsumerId info, long
arg1) throws Exception {
                 ConsumerContext ctx = consumers.remove(info);
                 if (ctx != null) {
-                	ctx.stop();
+                    ctx.stop();
                 }
                 ack(remove);
                 return null;
@@ -162,10 +162,10 @@
             // Message Processing Methods.
             // /////////////////////////////////////////////////////////////////
             public Response processMessage(Message info) throws Exception {
-            	if( info.getOriginalDestination() == null ) {
-            		info.setOriginalDestination(info.getDestination());
-            	}
-            	
+                if (info.getOriginalDestination() == null) {
+                    info.setOriginalDestination(info.getDestination());
+                }
+
                 ProducerId producerId = info.getProducerId();
                 ProducerContext producerContext = producers.get(producerId);
 
@@ -214,10 +214,10 @@
                 // broker.
                 BrokerInfo brokerInfo = new BrokerInfo();
                 Broker broker = connection.getBroker();
-				brokerInfo.setBrokerId(new BrokerId(broker.getName()));
+                brokerInfo.setBrokerId(new BrokerId(broker.getName()));
                 brokerInfo.setBrokerName(broker.getName());
-                if( !broker.getConnectUris().isEmpty() ) {
-                	brokerInfo.setBrokerURL(broker.getConnectUris().get(0));
+                if (!broker.getConnectUris().isEmpty()) {
+                    brokerInfo.setBrokerURL(broker.getConnectUris().get(0));
                 }
                 connection.write(brokerInfo);
                 return ack(info);
@@ -256,13 +256,13 @@
             // Methods for server management
             // /////////////////////////////////////////////////////////////////
             public Response processAddDestination(DestinationInfo info) throws Exception
{
-            	ActiveMQDestination destination = info.getDestination();
-				if( destination.isTemporary() ) {
-					// Keep track of it so that we can remove them this connection 
-					// shuts down.
-            		temporaryDestinations.add(destination);
-            	}
-            	host.createQueue(destination);
+                ActiveMQDestination destination = info.getDestination();
+                if (destination.isTemporary()) {
+                    // Keep track of it so that we can remove them this connection 
+                    // shuts down.
+                    temporaryDestinations.add(destination);
+                }
+                host.createQueue(destination);
                 return ack(info);
             }
 
@@ -348,6 +348,7 @@
         Command command = (Command) o;
         boolean responseRequired = command.isResponseRequired();
         try {
+            //System.out.println(o);
             command.visit(visitor);
         } catch (Exception e) {
             if (responseRequired) {
@@ -449,7 +450,7 @@
 
         private HashMap<MessageId, SubscriptionDeliveryCallback> pendingMessages =
new HashMap<MessageId, SubscriptionDeliveryCallback>();
         private LinkedList<MessageId> pendingMessageIds = new LinkedList<MessageId>();
-		private BrokerSubscription brokerSubscription;
+        private BrokerSubscription brokerSubscription;
 
         public ConsumerContext(final ConsumerInfo info) throws Exception {
             this.info = info;
@@ -464,22 +465,22 @@
                 }
             };
 
+            isQueueReceiver = info.getDestination().isQueue();
+
             controller = new FlowController<MessageDelivery>(null, flow, limiter, this);
             controller.useOverFlowQueue(false);
             controller.setExecutor(connection.getDispatcher().createPriorityExecutor(connection.getDispatcher().getDispatchPriorities()
- 1));
             super.onFlowOpened(controller);
         }
 
-        
-		public void start() throws Exception {
+        public void start() throws Exception {
             brokerSubscription = host.createSubscription(this);
             brokerSubscription.connect(this);
-		}
-
-		public void stop() throws Exception {
-			brokerSubscription.disconnect(this);
-		}
+        }
 
+        public void stop() throws Exception {
+            brokerSubscription.disconnect(this);
+        }
 
         public boolean offer(final MessageDelivery message, ISourceController<?> source,
SubscriptionDeliveryCallback callback) {
             if (!controller.offer(message, source)) {
@@ -587,7 +588,10 @@
          * .Object)
          */
         public boolean isRemoveOnDispatch(MessageDelivery elem) {
-            return !elem.isPersistent() || !(isDurable || isQueueReceiver);
+            if (isQueueReceiver()) {
+                return false;
+            }
+            return !elem.isPersistent() || !isDurable;
         }
 
         /*
@@ -696,9 +700,9 @@
             return offer(message, source, null);
         }
 
-		public boolean autoCreateDestination() {
-			return true;
-		}
+        public boolean autoCreateDestination() {
+            return true;
+        }
 
     }
 



Mime
View raw message