qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lqu...@apache.org
Subject [2/3] qpid-broker-j git commit: QPID-7958: [Java Broker] Ensure a message reference is always taken for SystemMessageSource messages
Date Mon, 09 Oct 2017 13:27:16 GMT
QPID-7958: [Java Broker] Ensure a message reference is always taken for SystemMessageSource
messages


Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/2967b682
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/2967b682
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/2967b682

Branch: refs/heads/master
Commit: 2967b682326ae7f0f862a0086ae54ea733c5cc23
Parents: 3eb2426
Author: Lorenz Quack <lquack@apache.org>
Authored: Mon Oct 9 12:08:21 2017 +0100
Committer: Lorenz Quack <lquack@apache.org>
Committed: Mon Oct 9 13:32:31 2017 +0100

----------------------------------------------------------------------
 .../AbstractSystemMessageSource.java            | 30 ++++++++++++++++----
 1 file changed, 25 insertions(+), 5 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/2967b682/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
index 24e19f5..3d6d009 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
@@ -34,13 +34,14 @@ import org.apache.qpid.server.consumer.ConsumerTarget;
 import org.apache.qpid.server.filter.FilterManager;
 import org.apache.qpid.server.filter.Filterable;
 import org.apache.qpid.server.message.InstanceProperties;
+import org.apache.qpid.server.message.MessageContainer;
 import org.apache.qpid.server.message.MessageInstance;
 import org.apache.qpid.server.message.MessageInstanceConsumer;
+import org.apache.qpid.server.message.MessageReference;
 import org.apache.qpid.server.message.MessageSource;
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.message.internal.InternalMessage;
 import org.apache.qpid.server.model.NamedAddressSpace;
-import org.apache.qpid.server.message.MessageContainer;
 import org.apache.qpid.server.session.AMQPSession;
 import org.apache.qpid.server.store.MessageDurability;
 import org.apache.qpid.server.store.MessageEnqueueRecord;
@@ -115,15 +116,14 @@ public abstract class AbstractSystemMessageSource implements MessageSource
         return MessageConversionExceptionHandlingPolicy.CLOSE;
     }
 
-    protected class Consumer<T extends ConsumerTarget> implements MessageInstanceConsumer<T>
+    protected class Consumer<T extends ConsumerTarget> implements MessageInstanceConsumer<T>,
TransactionLogResource
     {
 
         private final List<PropertiesMessageInstance> _queue =
                 Collections.synchronizedList(new ArrayList<PropertiesMessageInstance>());
         private final T _target;
         private final String _name;
-        private final Object _identifier = new Object();
-
+        private final UUID _identifier = UUID.randomUUID();
 
         public Consumer(final String consumerName, T target)
         {
@@ -161,7 +161,9 @@ public abstract class AbstractSystemMessageSource implements MessageSource
                 if (!_target.isSuspended() && _target.allocateCredit(propertiesMessageInstance.getMessage()))
                 {
                     _queue.remove(0);
-                    return new MessageContainer(propertiesMessageInstance, null, false);
+                    return new MessageContainer(propertiesMessageInstance,
+                                                propertiesMessageInstance.getMessageReference(),
+                                                false);
                 }
             }
             return null;
@@ -200,6 +202,18 @@ public abstract class AbstractSystemMessageSource implements MessageSource
             return _name;
         }
 
+        @Override
+        public UUID getId()
+        {
+            return _identifier;
+        }
+
+        @Override
+        public MessageDurability getMessageDurability()
+        {
+            return MessageDurability.NEVER;
+        }
+
         public void send(final InternalMessage response)
         {
             _queue.add(new PropertiesMessageInstance(this, response));
@@ -210,6 +224,7 @@ public abstract class AbstractSystemMessageSource implements MessageSource
     class PropertiesMessageInstance implements MessageInstance
     {
         private final Consumer _consumer;
+        private final MessageReference _messageReference;
         private int _deliveryCount;
         private boolean _isRedelivered;
         private boolean _isDelivered;
@@ -220,6 +235,7 @@ public abstract class AbstractSystemMessageSource implements MessageSource
         {
             _consumer = consumer;
             _message = message;
+            _messageReference = message.newReference(consumer);
         }
 
         @Override
@@ -429,5 +445,9 @@ public abstract class AbstractSystemMessageSource implements MessageSource
             return AbstractSystemMessageSource.this;
         }
 
+        public MessageReference getMessageReference()
+        {
+            return _messageReference;
+        }
     }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message