Setting NON_PERSISTENT DeliveryMode on Spring AMQP messages

[monkeyhands]

Hi

I don't want messages to be persisted, so I'd like them to be discarded if the consumer is not running.

Using the MessageDeliveryMode in the message header I specify messages in my ticketing system to be non-persistent.

    public void openTicket(Ticket ticket) {
    	Message<Ticket> msg = 
    			MessageBuilder
    				.withPayload(ticket)
    				.setHeader(AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.NON_PERSISTENT)
    				.build();
        ticketChannel.send(msg);
        System.out.println("Ticket Sent - " + ticket.toString());
    }

However the consumer is still getting the tickets even though it was shutdown when I run the producer.
In addition on the RabbitConsole I can 'get the messages' from the queue after the client has shutdown.


I'm using:
Spring AMQP 2.1.0.BUILD-SNAPSHOT
Spring Integration 2.0.5.RELEASE
Rabbit 2.6.1
Erlang 5.8.4

[Mark Fisher]

I assume "ticketChannel" is connected to an AMQP outbound-channel-adapter? Can you provide the config excerpt for that as well?

Also, how recent is the snapshot you are using (i.e. is it POST-RC1?).

Thanks,
Mark

[monkeyhands]

Hi

Config for the output bound adapter is below (I'm using rabbit example from Pro Spring Integration book).
I downloaded the snapshot on Friday, the 9th Dec, sorry but I don't know whether this is more recent than POST-RC1 or not.

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:int-amqp="http://www.springframework.org/schema/integration/amqp"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
		http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
		http://www.springframework.org/schema/context
		http://www.springframework.org/schema/context/spring-context-3.0.xsd
    http://www.springframework.org/schema/integration/amqp
    http://www.springframework.org/schema/integration/amqp/spring-integration-amqp-2.1.xsd">

  <context:component-scan
      base-package="com.apress.prospringintegration.messaging.rabbitmq.jms.adapter"/>

  <int-amqp:outbound-channel-adapter routing-key="ticket.queue"
                                     amqp-template="amqpTemplate"
                                     channel="ticketChannel"/>

</beans>

Thanks
David/

[monkeyhands]

Hi

In case anyone is following this thread I managed to make messages NON_PERSISTENT by passing the message through a Spring Integration header enricher to add the header 'ampq_deliveryMode' withe value MessageDeliveryMode.NON_PERSISTENT.

This is an enum value and did cause me problems setting it from the context xml, in the end I got round this by using a bean to return the enum value.

I used the context xml to create the enricher and specified the header field and bean method to use for the value.

	<int:header-enricher input-channel="nonPersistentChannel"
		output-channel="toRabbitChannel">
		<int:header name="amqp_deliveryMode" ref="ampqHeaderEnricher" method="getNonPersistentDeliveryMode" />
	</int:header-enricher>

        <!-- Bean def -->
	<bean id="ampqHeaderEnricher" class="org.example.ampq.AMPQHeaderEnricher"/>

The Java code looks like:

package org.example.ampq;

import org.springframework.amqp.core.MessageDeliveryMode;

public class AMPQHeaderEnricher {
	public MessageDeliveryMode getNonPersistentDeliveryMode() {
		return MessageDeliveryMode.NON_PERSISTENT;
	}
}

[Mark Fisher]

You should be able to use a SpEL expression there instead (possibly including the "type" attribute).

[monkeyhands]

Hi Mark

Thanks for the advice, I'd missed the type attribute - doh!
It's now much cleaner no special bean/class is needed.

<int:header-enricher input-channel="nonPersistentChannel"
		output-channel="toRabbitChannel">
	<int:header name="amqp_deliveryMode" value="NON_PERSISTENT" type="org.springframework.amqp.core.MessageDeliveryMode"/>
</int:header-enricher>

Thanks
David/