Fix for https://issues.apache.org/activemq/browse/AMQ-1656

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@646880 13f79535-47bb-0310-9956-ffa450edef68

activemq-core/src/main/java/org/apache/activemq/broker/Broker.java

@@ -313,4 +313,9 @@ public interface Broker extends Region, Service {
      */
     void sendToDeadLetterQueue(ConnectionContext context, MessageReference messageReference);
     
+    /**
+     * @return the broker sequence id
+     */
+    long getBrokerSequenceId();
+
 }

activemq-core/src/main/java/org/apache/activemq/broker/BrokerFilter.java

@@ -260,4 +260,8 @@ public class BrokerFilter implements Broker {
     public Broker getRoot() {
         return next.getRoot();
     }
+
+    public long getBrokerSequenceId() {
+        return next.getBrokerSequenceId();
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/EmptyBroker.java

@@ -252,4 +252,8 @@ public class EmptyBroker implements Broker {
     public Broker getRoot() {
         return null;
     }
+    
+    public long getBrokerSequenceId() {
+        return -1l;
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/ErrorBroker.java

@@ -263,4 +263,8 @@ public class ErrorBroker implements Broker {
     public Broker getRoot() {
         throw new BrokerStoppedException(this.message);
     }
+    
+    public long getBrokerSequenceId() {
+        throw new BrokerStoppedException(this.message);
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java

@@ -274,4 +274,8 @@ public class MutableBrokerFilter implements Broker {
         return getNext().getRoot();
     }
     
+    public long getBrokerSequenceId() {
+        return getNext().getBrokerSequenceId();
+    }
+
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/BaseDestination.java

@@ -46,6 +46,7 @@ public abstract class BaseDestination implements Destination {
     private boolean lazyDispatch=false;
     protected final DestinationStatistics destinationStatistics = new DestinationStatistics();
     protected final BrokerService brokerService;
+    protected final Broker regionBroker;
     
     /**
      * @param broker 
@@ -66,6 +67,7 @@ public abstract class BaseDestination implements Destination {
         this.systemUsage = brokerService.getProducerSystemUsage();
         this.memoryUsage = new MemoryUsage(systemUsage.getMemoryUsage(), destination.toString());
         this.memoryUsage.setUsagePortion(1.0f);
+        this.regionBroker = brokerService.getRegionBroker();
     }
     
     /**
@@ -194,4 +196,8 @@ public abstract class BaseDestination implements Destination {
     public void setLazyDispatch(boolean lazyDispatch) {
         this.lazyDispatch = lazyDispatch;
     } 
+    
+    protected long getDestinationSequenceId() {
+        return regionBroker.getBrokerSequenceId();
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java

@@ -430,7 +430,7 @@ public class Queue extends BaseDestination implements Task {
                                 "Connection closed, send aborted.");
                     }
                 }
-
+                message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
                 store.addMessage(context, message);
 
             }

activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java

@@ -418,8 +418,6 @@ public class RegionBroker implements Broker {
     }
 
     public void send(ProducerBrokerExchange producerExchange, Message message) throws Exception {
-        long si = sequenceGenerator.getNextSequenceId();
-        message.getMessageId().setBrokerSequenceId(si);
         message.setBrokerInTime(System.currentTimeMillis());
         if (producerExchange.isMutable() || producerExchange.getRegion() == null) {
             ActiveMQDestination destination = message.getDestination();
@@ -730,4 +728,13 @@ public class RegionBroker implements Broker {
             throw new RuntimeException("The broker from the BrokerService should not throw an exception");
         }
     }
+    
+    /**
+     * @return the broker sequence id
+     */
+    public long getBrokerSequenceId() {
+        synchronized(sequenceGenerator) {
+            return sequenceGenerator.getNextSequenceId();
+        }
+    }
 }

activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java

@@ -382,6 +382,7 @@ public class Topic  extends BaseDestination  implements Task{
         final ConnectionContext context = producerExchange
                 .getConnectionContext();
         message.setRegionDestination(this);
+        message.getMessageId().setBrokerSequenceId(getDestinationSequenceId());
 
         if (topicStore != null && message.isPersistent()
                 && !canOptimizeOutPersistence()) {

activemq-core/src/test/java/org/apache/activemq/broker/StubBroker.java

@@ -258,4 +258,8 @@ public class StubBroker implements Broker {
     public Broker getRoot() {
         return this;
     }
+    
+    public long getBrokerSequenceId() {
+        return -1l;
+    }
 }