Ensure AMQP protocol marshals messages before passing to broker (#1859)

This change brings the behaior inline with other protocols and will
prevent a future race condition during copy/dispatch as the data will
already be marshaled.

This closes #1858

Author: cshannon

activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/protocol/AmqpReceiver.java

@@ -205,6 +205,11 @@ protected void processDelivery(final Delivery delivery, Buffer deliveryBytes) th
             }
 
             message.onSend();
+            // GH-1851 - Ensure we marshal the message before passing to the broker
+            // This prevents a race condition later if the message is copyied/marshaled
+            // at this same time. This behavior is in line with how messages are received
+            // using OpenWire (when not using VM) and Stomp.
+            message.beforeMarshall(null);
 
             sendsInFlight++;
 

activemq-amqp/src/test/java/org/apache/activemq/transport/amqp/JMSClientTest.java

@@ -26,6 +26,7 @@
 import java.util.ArrayList;
 import java.util.Enumeration;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -52,6 +53,10 @@
 import jakarta.jms.TopicSession;
 import jakarta.jms.TopicSubscriber;
 
+import org.apache.activemq.broker.Broker;
+import org.apache.activemq.broker.BrokerFilter;
+import org.apache.activemq.broker.BrokerPlugin;
+import org.apache.activemq.broker.ProducerBrokerExchange;
 import org.apache.activemq.broker.jmx.BrokerView;
 import org.apache.activemq.broker.jmx.BrokerViewMBean;
 import org.apache.activemq.broker.jmx.ConnectorViewMBean;
@@ -1360,4 +1365,26 @@ public void onMessage(Message message) {
 
         assertTrue("Did not receive all messages: " + MSG_COUNT, done.await(15, TimeUnit.SECONDS));
     }
+
+    @Override
+    protected void addAdditionalPlugins(List<BrokerPlugin> plugins) throws Exception {
+        super.addAdditionalPlugins(plugins);
+        plugins.add(new BrokerPlugin() {
+
+            @Override
+            public Broker installPlugin(Broker broker) {
+                return new BrokerFilter(broker) {
+
+                    @Override
+                    public void send(ProducerBrokerExchange producerExchange,
+                            org.apache.activemq.command.Message messageSend) throws Exception {
+                        super.send(producerExchange, messageSend);
+
+                        // The message should always be passed to the broker in a marshaled state
+                        assertTrue(messageSend.isMarshalled());
+                    }
+                };
+            }
+        });
+    }
 }