Set destination ID for DELAYED_DELIVERY producer

Author: vavrtom

broker-core/src/main/java/org/apache/qpid/server/model/Producer.java

@@ -23,13 +23,37 @@
 import java.util.UUID;
 
 import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.qpid.server.message.MessageDestination;
 
 @ManagedObject(creatable = false, amqpName = "org.apache.qpid.Producer")
 public interface Producer<X extends Producer<X>> extends ConfiguredObject<X>
 {
     enum DeliveryType { DELAYED_DELIVERY, STANDARD_DELIVERY }
 
-    enum DestinationType { EXCHANGE, QUEUE }
+    enum DestinationType
+    {
+        EXCHANGE,
+        QUEUE;
+
+        public static DestinationType from(MessageDestination messageDestination)
+        {
+            if (messageDestination instanceof Exchange)
+            {
+                return EXCHANGE;
+            }
+            else if (messageDestination instanceof Queue)
+            {
+                return QUEUE;
+            }
+            return null;
+        }
+
+        public static UUID getId(MessageDestination messageDestination)
+        {
+            final DestinationType destinationType = from(messageDestination);
+            return destinationType == null ? null : ((ConfiguredObject<?>) messageDestination).getId();
+        }
+    }
 
     void registerMessageDelivered(long messageSize);
 

broker-core/src/main/java/org/apache/qpid/server/model/ProducerImpl.java

@@ -67,25 +67,16 @@ public ProducerImpl(final AbstractAMQPSession<?, ?> session,
         _sessionName = session.getName();
         _principal = session.getAMQPConnection().getPrincipal();
         _remoteAddress = session.getAMQPConnection().getRemoteAddress();
-        _destination = messageDestination == null ? null : messageDestination.getName();
         if (messageDestination == null)
         {
             _deliveryType = DeliveryType.DELAYED_DELIVERY;
-            _destinationType = null;
         }
         else
         {
             _deliveryType = DeliveryType.STANDARD_DELIVERY;
-            if (messageDestination instanceof Exchange)
-            {
-                _destinationId = ((Exchange<?>) messageDestination).getId();
-                _destinationType = DestinationType.EXCHANGE;
-            }
-            else if (messageDestination instanceof Queue)
-            {
-                _destinationId = ((Queue<?>) messageDestination).getId();
-                _destinationType = DestinationType.QUEUE;
-            }
+            _destination = messageDestination.getName();
+            _destinationType = DestinationType.from(messageDestination);
+            _destinationId = DestinationType.getId(messageDestination);
         }
 
         registerWithParents();

broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/StandardReceivingLinkEndpoint.java

@@ -296,14 +296,8 @@ public void onRollback()
                                     {
                                         final MessageDestination messageDestination = getAddressSpace()
                                                 .getAttainedMessageDestination(serverMessage.getTo(), false);
-                                        if (messageDestination != null)
-                                        {
-                                            final Producer.DestinationType destinationType =
-                                                    messageDestination instanceof Exchange
-                                                            ? Producer.DestinationType.EXCHANGE
-                                                            : Producer.DestinationType.QUEUE;
-                                            _producer.setDestinationType(destinationType);
-                                        }
+                                        _producer.setDestinationType(Producer.DestinationType.from(messageDestination));
+                                        _producer.setDestinationId(Producer.DestinationType.getId(messageDestination));
                                         _producer.setDestination(to);
                                     }
                                 }