Merge pull request #45 from rabbitmq/rabbitmq-jms-client-42-message-customizer

Add AmqpPropertiesCustomiser interface

Author: michaelklishin

pom.xml

@@ -44,8 +44,8 @@
     <rabbitmq.version>4.6.0</rabbitmq.version>
     <slf4j-api.version>1.7.25</slf4j-api.version>
     <junit.version>4.12</junit.version>
-    <mockito-core.version>2.12.0</mockito-core.version>
-    <awaitility.version>3.0.0</awaitility.version>
+    <mockito-core.version>2.16.0</mockito-core.version>
+    <awaitility.version>3.1.0</awaitility.version>
 
     <maven.dependency.plugin.version>2.8</maven.dependency.plugin.version>
     <maven.compiler.plugin.version>3.5.1</maven.compiler.plugin.version>

src/main/java/com/rabbitmq/jms/admin/RMQConnectionFactory.java

@@ -1,7 +1,8 @@
-/* Copyright (c) 2013-2017 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2013-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.admin;
 
 import com.rabbitmq.client.Address;
+import com.rabbitmq.jms.client.AmqpPropertiesCustomiser;
 import com.rabbitmq.jms.client.ConnectionParams;
 import com.rabbitmq.jms.client.RMQConnection;
 import com.rabbitmq.jms.util.RMQJMSException;
@@ -72,6 +73,12 @@ public class RMQConnectionFactory implements ConnectionFactory, Referenceable, S
      */
     private boolean cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose = false;
 
+    /**
+     * Callback to customise properties of outbound AMQP messages.
+     * @since 1.9.0
+     */
+    private AmqpPropertiesCustomiser amqpPropertiesCustomiser;
+
     /** Default not to use ssl */
     private boolean ssl = false;
     private String tlsProtocol;
@@ -134,6 +141,7 @@ public Connection createConnection(String username, String password) throws JMSE
             .setPreferProducerMessageProperty(preferProducerMessageProperty)
             .setRequeueOnMessageListenerException(requeueOnMessageListenerException)
             .setCleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose(this.cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose)
+            .setAmqpPropertiesCustomiser(this.amqpPropertiesCustomiser)
         );
         conn.setTrustedPackages(this.trustedPackages);
         logger.debug("Connection {} created.", conn);
@@ -693,5 +701,10 @@ public void setCleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose(boolean
     public boolean isCleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose() {
         return this.cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose;
     }
+
+    public void setAmqpPropertiesCustomiser(AmqpPropertiesCustomiser amqpPropertiesCustomiser) {
+        this.amqpPropertiesCustomiser = amqpPropertiesCustomiser;
+    }
+
 }
 

src/main/java/com/rabbitmq/jms/client/AmqpPropertiesCustomiser.java

@@ -0,0 +1,23 @@
+/* Copyright (c) 2018 Pivotal Software, Inc. All rights reserved. */
+
+package com.rabbitmq.jms.client;
+
+import com.rabbitmq.client.AMQP;
+
+import javax.jms.Message;
+
+/**
+ * Callback to customise properties of outbound AMQP messages.
+ * @since 1.9.0
+ */
+public interface AmqpPropertiesCustomiser {
+
+    /**
+     * Customise AMQP message properties.
+     * @param builder the AMQP properties builder
+     * @param jmsMessage the outbound JMS message
+     * @return the customised or a new AMQP properties builder
+     */
+    AMQP.BasicProperties.Builder customise(AMQP.BasicProperties.Builder builder, Message jmsMessage);
+
+}

src/main/java/com/rabbitmq/jms/client/ConnectionParams.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2016-2017 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2016-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.client;
 
 import com.rabbitmq.client.Connection;
@@ -54,6 +54,12 @@ public class ConnectionParams {
      */
     private boolean cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose = false;
 
+    /**
+     * Callback to customise properties of outbound AMQP messages.
+     * @since 1.9.0
+     */
+    private AmqpPropertiesCustomiser amqpPropertiesCustomiser;
+
     public Connection getRabbitConnection() {
         return rabbitConnection;
     }
@@ -125,4 +131,13 @@ public ConnectionParams setCleanUpServerNamedQueuesForNonDurableTopicsOnSessionC
         this.cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose = cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose;
         return this;
     }
+
+    public AmqpPropertiesCustomiser getAmqpPropertiesCustomiser() {
+        return amqpPropertiesCustomiser;
+    }
+
+    public ConnectionParams setAmqpPropertiesCustomiser(AmqpPropertiesCustomiser amqpPropertiesCustomiser) {
+        this.amqpPropertiesCustomiser = amqpPropertiesCustomiser;
+        return this;
+    }
 }

src/main/java/com/rabbitmq/jms/client/RMQConnection.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2013 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2013-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.client;
 
 import java.io.IOException;
@@ -100,6 +100,12 @@ public class RMQConnection implements Connection, QueueConnection, TopicConnecti
      */
     private final boolean cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose;
 
+    /**
+     * Callback to customise properties of outbound AMQP messages.
+     * @since 1.9.0
+     */
+    private final AmqpPropertiesCustomiser amqpPropertiesCustomiser;
+
     /**
      * Classes in these packages can be transferred via ObjectMessage.
      *
@@ -123,6 +129,7 @@ public RMQConnection(ConnectionParams connectionParams) {
         this.preferProducerMessageProperty = connectionParams.willPreferProducerMessageProperty();
         this.requeueOnMessageListenerException = connectionParams.willRequeueOnMessageListenerException();
         this.cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose = connectionParams.isCleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose();
+        this.amqpPropertiesCustomiser = connectionParams.getAmqpPropertiesCustomiser();
     }
 
     /**
@@ -171,6 +178,7 @@ public Session createSession(boolean transacted, int acknowledgeMode) throws JMS
             .setPreferProducerMessageProperty(this.preferProducerMessageProperty)
             .setRequeueOnMessageListenerException(this.requeueOnMessageListenerException)
             .setCleanUpServerNamedQueuesForNonDurableTopics(this.cleanUpServerNamedQueuesForNonDurableTopicsOnSessionClose)
+            .setAmqpPropertiesCustomiser(this.amqpPropertiesCustomiser)
         );
         session.setTrustedPackages(this.trustedPackages);
         this.sessions.add(session);

src/main/java/com/rabbitmq/jms/client/RMQMessageProducer.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2013-2017 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2013-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.client;
 
 import java.io.IOException;
@@ -69,20 +69,27 @@ public class RMQMessageProducer implements MessageProducer, QueueSender, TopicPu
 
     private final SendingStrategy sendingStrategy;
 
-    /**
-     * Create a producer of messages.
-     * @param session which this producer uses
-     * @param destination to which this producer sends messages.
-     * @param preferProducerMessageProperty properties take precedence over respective message properties
-     */
-    public RMQMessageProducer(RMQSession session, RMQDestination destination, boolean preferProducerMessageProperty) {
+    private final AmqpPropertiesCustomiser amqpPropertiesCustomiser;
+
+    public RMQMessageProducer(RMQSession session, RMQDestination destination, boolean preferProducerMessageProperty, AmqpPropertiesCustomiser amqpPropertiesCustomiser) {
         this.session = session;
         this.destination = destination;
         if (preferProducerMessageProperty) {
             sendingStrategy = new PreferMessageProducerPropertySendingStategy();
         } else {
             sendingStrategy = new PreferMessagePropertySendingStrategy();
         }
+        this.amqpPropertiesCustomiser = amqpPropertiesCustomiser == null ? new NoOpAmqpPropertiesCustomiser() : amqpPropertiesCustomiser;
+    }
+
+    /**
+     * Create a producer of messages.
+     * @param session which this producer uses
+     * @param destination to which this producer sends messages.
+     * @param preferProducerMessageProperty properties take precedence over respective message properties
+     */
+    public RMQMessageProducer(RMQSession session, RMQDestination destination, boolean preferProducerMessageProperty) {
+        this(session, destination, preferProducerMessageProperty, new NoOpAmqpPropertiesCustomiser());
     }
 
     /**
@@ -292,6 +299,8 @@ private void sendAMQPMessage(RMQDestination destination, RMQMessage msg, int del
                 bob.expiration(rmqExpiration(timeToLive));
                 bob.headers(msg.toAmqpHeaders());
 
+                bob = amqpPropertiesCustomiser.customise(bob, msg);
+
                 byte[] data = msg.toAmqpByteArray();
 
                 this.session.getChannel().basicPublish(destination.getAmqpExchangeName(), destination.getAmqpRoutingKey(), bob.build(), data);
@@ -461,4 +470,13 @@ public void send(Destination destination, Message message, int deliveryMode, int
     private enum MessageExpirationType {
         TTL, EXPIRATION
     }
+
+    private static final class NoOpAmqpPropertiesCustomiser implements AmqpPropertiesCustomiser {
+
+        @Override
+        public AMQP.BasicProperties.Builder customise(AMQP.BasicProperties.Builder builder, Message jmsMessage) {
+            return builder;
+        }
+
+    }
 }

src/main/java/com/rabbitmq/jms/client/RMQSession.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2013-2017 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2013-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.client;
 
 import java.io.IOException;
@@ -103,6 +103,12 @@ public class RMQSession implements Session, QueueSession, TopicSession {
      */
     private final boolean cleanUpServerNamedQueuesForNonDurableTopics;
 
+    /**
+     * Callback to customise properties of outbound AMQP messages.
+     * @since 1.9.0
+     */
+    private final AmqpPropertiesCustomiser amqpPropertiesCustomiser;
+
     /** The main RabbitMQ channel we use under the hood */
     private final Channel channel;
     /** Set to true if close() has been called and completed */
@@ -195,6 +201,7 @@ public RMQSession(SessionParams sessionParams) throws JMSException {
         this.preferProducerMessageProperty = sessionParams.willPreferProducerMessageProperty();
         this.requeueOnMessageListenerException = sessionParams.willRequeueOnMessageListenerException();
         this.cleanUpServerNamedQueuesForNonDurableTopics = sessionParams.isCleanUpServerNamedQueuesForNonDurableTopics();
+        this.amqpPropertiesCustomiser = sessionParams.getAmqpPropertiesCustomiser();
 
         if (transacted) {
             this.acknowledgeMode = Session.SESSION_TRANSACTED;
@@ -604,7 +611,7 @@ public MessageProducer createProducer(Destination destination) throws JMSExcepti
         illegalStateExceptionIfClosed();
         RMQDestination dest = (RMQDestination) destination;
         declareDestinationIfNecessary(dest);
-        RMQMessageProducer producer = new RMQMessageProducer(this, dest, this.preferProducerMessageProperty);
+        RMQMessageProducer producer = new RMQMessageProducer(this, dest, this.preferProducerMessageProperty, this.amqpPropertiesCustomiser);
         this.producers.add(producer);
         return producer;
     }

src/main/java/com/rabbitmq/jms/client/SessionParams.java

@@ -1,4 +1,4 @@
-/* Copyright (c) 2016-2017 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2016-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.jms.client;
 
 import javax.jms.Message;
@@ -49,6 +49,12 @@ public class SessionParams {
      */
     private boolean cleanUpServerNamedQueuesForNonDurableTopics = false;
 
+    /**
+     * Callback to customise properties of outbound AMQP messages.
+     * @since 1.9.0
+     */
+    private AmqpPropertiesCustomiser amqpPropertiesCustomiser;
+
     public RMQConnection getConnection() {
         return connection;
     }
@@ -120,4 +126,13 @@ public SessionParams setCleanUpServerNamedQueuesForNonDurableTopics(boolean clea
         this.cleanUpServerNamedQueuesForNonDurableTopics = cleanUpServerNamedQueuesForNonDurableTopics;
         return this;
     }
+
+    public AmqpPropertiesCustomiser getAmqpPropertiesCustomiser() {
+        return amqpPropertiesCustomiser;
+    }
+
+    public SessionParams setAmqpPropertiesCustomiser(AmqpPropertiesCustomiser amqpPropertiesCustomiser) {
+        this.amqpPropertiesCustomiser = amqpPropertiesCustomiser;
+        return this;
+    }
 }

src/test/java/com/rabbitmq/integration/tests/AbstractAmqpITQueue.java

@@ -1,9 +1,10 @@
-/* Copyright (c) 2013 Pivotal Software, Inc. All rights reserved. */
+/* Copyright (c) 2013-2018 Pivotal Software, Inc. All rights reserved. */
 package com.rabbitmq.integration.tests;
 
 import javax.jms.QueueConnection;
 import javax.jms.QueueConnectionFactory;
 
+import com.rabbitmq.jms.admin.RMQConnectionFactory;
 import org.junit.After;
 import org.junit.Before;
 
@@ -23,12 +24,17 @@ public abstract class AbstractAmqpITQueue {
     public void setUp() throws Exception {
         this.connFactory = (QueueConnectionFactory) AbstractTestConnectionFactory.getTestConnectionFactory()
                 .getConnectionFactory();
+        customise((RMQConnectionFactory) connFactory);
         this.queueConn = connFactory.createQueueConnection();
 
         this.rabbitConn = rabbitConnFactory.newConnection();
         this.channel = rabbitConn.createChannel();
     }
 
+    protected void customise(RMQConnectionFactory connectionFactory) {
+
+    }
+
     @After
     public void tearDown() throws Exception {
         if (this.queueConn != null) this.queueConn.close();

src/test/java/com/rabbitmq/integration/tests/AmqpPropertiesCustomiserIT.java

@@ -0,0 +1,84 @@
+/* Copyright (c) 2018 Pivotal Software, Inc. All rights reserved. */
+package com.rabbitmq.integration.tests;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.GetResponse;
+import com.rabbitmq.jms.admin.RMQConnectionFactory;
+import com.rabbitmq.jms.admin.RMQDestination;
+import com.rabbitmq.jms.client.AmqpPropertiesCustomiser;
+import org.junit.Test;
+
+import javax.jms.BytesMessage;
+import javax.jms.DeliveryMode;
+import javax.jms.Message;
+import javax.jms.Queue;
+import javax.jms.QueueReceiver;
+import javax.jms.QueueSender;
+import javax.jms.QueueSession;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import java.util.Arrays;
+import java.util.Enumeration;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+/**
+ * Integration test to test the AMQP properties customiser is applied correctly.
+ */
+public class AmqpPropertiesCustomiserIT extends AbstractAmqpITQueue {
+
+    private static final String QUEUE_NAME = "test.queue."+AmqpPropertiesCustomiserIT.class.getCanonicalName();
+    public static final String MESSAGE = "hello";
+    public static final String TEXT_PLAIN = "text/plain";
+
+    @Override
+    protected void customise(RMQConnectionFactory connectionFactory) {
+        connectionFactory.setAmqpPropertiesCustomiser(new AmqpPropertiesCustomiser() {
+
+            @Override
+            public AMQP.BasicProperties.Builder customise(AMQP.BasicProperties.Builder builder, Message jmsMessage) {
+                builder.contentType(TEXT_PLAIN);
+                return builder;
+            }
+        });
+    }
+
+    @Test
+    public void customiserIsApplied() throws Exception {
+
+        channel.queueDeclare(QUEUE_NAME,
+                             false, // durable
+                             true,  // exclusive
+                             true,  // autoDelete
+                             null   // options
+                             );
+
+        queueConn.start();
+        QueueSession queueSession = queueConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+        Queue queue = new RMQDestination(QUEUE_NAME, "", QUEUE_NAME, null);  // write-only AMQP-mapped queue
+
+        QueueSender queueSender = queueSession.createSender(queue);
+
+        TextMessage message = queueSession.createTextMessage(MESSAGE);
+
+        queueSender.send(message);
+        queueConn.close();
+
+        GetResponse response = channel.basicGet(QUEUE_NAME, false);
+        assertNotNull("basicGet failed to retrieve a response", response);
+
+        byte[] body = response.getBody();
+        assertNotNull("body of response is null", body);
+
+        assertEquals("body of response is not correct", MESSAGE, new String(body));
+
+        assertEquals("body of response is not correct", TEXT_PLAIN, response.getProps().getContentType());
+    }
+
+}