Spring AMQP 1.0.0.M3 silently fails a project that worked with 1.0.0.M1

[skram]

I had a sample that works for Spring AMQP 1.0.0.M1

I ported it to 1.0.0.M3. There are classes that had been removed. That's fine. I managed to adapt it with the new changes.

Using the 1.0.0.M1 sample project I'm able to send and receive messages. But with 1.0.0.M3, I can only send but I cannot receive.

I know that my 1.0.0.M3 app is able to send because running a receiver app (based on 1.0.0.M1) is able to receive that message.

So I think the problem is the configuration of the receiver or the listener for 1.0.0.M3

Here are the configs:

1.0.0.M1 Sample Project
Client.java

package org.spring.amqp.client;

import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Queue;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class Client {
	public static void main(String[] args) {
		ApplicationContext applicationContext =
            new ClassPathXmlApplicationContext("applicationContext.xml", Client.class);
		
		MessageSender sender = (MessageSender) applicationContext.getBean("messageSender");
		sender.send("Hello Spring AMQP!");
	}

}

MessageSender.java

package org.spring.amqp.client;

import javax.annotation.Resource;

import org.apache.log4j.Logger;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

public class MessageSender {

	protected Logger logger = Logger.getLogger("client");
	
    @Resource(name="rabbitTemplate")
    private RabbitTemplate rabbitTemplate;

    public void send(String text) {
    	rabbitTemplate.convertAndSend(text);
    	logger.debug("Message sent: " + text);
    }

}

MessageHandler.java

package org.spring.amqp.client;

import org.apache.log4j.Logger;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;

public class MessageHandler implements MessageListener {

	protected Logger logger = Logger.getLogger("client");
	
	@Override
    public void onMessage(Message message) {
		logger.debug("Client: Message received!");
        System.out.println("Received message: " + message);
        System.out.println("Text: " + new String(message.getBody()));
    }

}

applicationContext.xml

<bean id="rabbitConnectionFactory" class="org.springframework.amqp.rabbit.connection.SingleConnectionFactory"
		p:username="guest" p:password="guest" p:port="5672">
		<constructor-arg value="localhost" />
	</bean>

	<bean id="rabbitTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate"
		p:connectionFactory-ref="rabbitConnectionFactory" p:routingKey="hello"/>

	<bean id="messageSender" class="org.spring.amqp.client.MessageSender" />

	<bean class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer"
		p:connectionFactory-ref="rabbitConnectionFactory" p:queueName="hello"
		p:messageListener-ref="messageListener" />

	<bean id="messageListener" class="org.spring.amqp.client.MessageHandler" />

This assumes you have a queue name "hello"

1.0.0.M3 Sample Project

Everything is the same. I just had to add the aopalliance.jar

Note:
There are no errors or exceptions. Setting the logger to DEBUG level doesn't show any odd output. It just shows that it's able to send. As mentioned earlier, a receiver based on 1.0.0.M1 is able to receive messages published by 1.0.0.M3

I'm using the latest RabbitMQ Server 2.3.1.

[Dave Syer]

I can't see where you declare any Queues, so in principle you could be just sending a message into a black hole (that's what happened to me when I tried your code). In practice I assume your broker has a durable queue set up from a previous run. You also didn't set the rabbit template in your sender, so I assume there is some config missing.

The real problem is that the default value for concurrentConsumers in the SimpleMessageListenerContainer changed in M3, so it defaults to 0, meaning there are none. If you make it explicit you will see your messages. Maybe the default should be 1, or maybe we should copy Spring JMS (6?)?

Note that from M3 we now recommend using CachingConnectionFactory instead of SingleConnectionFactory (it has better resilience if there is a protocol error).

[skram]

Dave, thanks for the reply. I tried quickly your suggestion.

I added concurrentConsumers="1" in the SimpleMessageListenerContainer and I'm able to get my messages.

<bean class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer"
		p:connectionFactory-ref="rabbitConnectionFactory" 
		p:queueName="hello"
		p:messageListener-ref="messageListenerAdapter"
		p:concurrentConsumers="1" />

For the queue it's actually declared in another class which I forgot to post. I know that sample I posted works because it works for 1.0.0.M1. But when switching to 1.0.0.M3 it fails.

Well, now I know why. It's because of that concurrentConsumers. I'm just wondering if there was a documentation indicating that change and the possible impact it might did? Because if you haven't told me, I would be left still guessing.

Right after typing that paragraph, I tried the CachingConnectionFactory, and it worked fine. Thanks a lot.

[skram]

By the way, the helloworld sample included in the 1.0.0.M3 also fails silently. It's the same behavior as what happened to my application.

To solve it, I have to set the concurrentConsumers to 1

@Bean
	public SimpleMessageListenerContainer listenerContainer() {
		SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
		container.setConnectionFactory(connectionFactory());
		container.setQueueName(this.helloWorldQueueName);
		container.setConcurrentConsumers(1);
		container.setMessageListener(new MessageListenerAdapter(new HelloWorldHandler()));
		return container;
	}

Maybe the sample code needs to be updated?

[deepgood2001]

Here: QueueName: alarmQueue is binded to fanout exchange. Queue is getting updated on each event as expected. Which clearly says, there is no problem in Sending the messages after upgrading to M3, but listener is just not consuming.
I tried with MeesageListener implementation as well. Attached is by using MessageListenerAdapter. Not only this queue/listener, all the listeners stopped working.

Please share valuable inputs on the same.

Receiver class:
==========

public class AlarmNotificationListener {

/** Reference of AlarmDataManager* */
private AlarmDataManager m_AlarmData;
SimpleMessageListenerContainer con = null;
/**
* constructor
*/
public AlarmNotificationListener(AlarmDataManager alarmData, String moduleName) throws Exception{

SingleConnectionFactory connectionFactory = new SingleConnectionFactory(AlarmMgmtApplet.HOST_IP); // IP is coming properly
connectionFactory.setUsername(guest);
connectionFactory.setPassword(guest);
connectionFactory.setPort(5672);
con = new SimpleMessageListenerContainer(connectionFactory);
con.setQueueName ("alarmQueue");
con.setConcurrentConsumers(1);
con.setMessageListener(new MessageListenerAdapter(new AlarmNotificationListenerhandler(m_AlarmData,modul eName)));
con.start();
}

}

Class AlarmNotificationListenerhandler.java
===============================

public class AlarmNotificationListenerhandler {
/** Reference of AlarmDataManager* */
private AlarmDataManager m_AlarmData;
/**
* constructor
*/
AlarmNotificationListenerhandler(AlarmDataManager alarmData, String moduleName) {

m_AlarmData = alarmData;
}


public void handleMessage(Notification n) {
System.out.println("Received:sff =>" + n);
//recieveMessage(n);
}
}


=======

[Dave Syer]

Please just upgrade to a snapshot (we are close to 1.0 now, so even RC1 has bugs that are fixed). M3 is too old to spend any time worrying about it.