Spring Community Forums - AMQP

SimpleContainer include multiple listener but only one consumer is vaild

zone — Mon, 13 May 2013 06:31:32 GMT

hi,all. I configure multiple listener in a SimpleContainer ,but I found one is valild
here is configure file

Code:


                host="127.0.0.1" channel-cache-size="25" />

        

        

        
                durable="true">

                

                        

                        

                

        

        
                exchange="ex_direct" />

        

        

        

        
                 connection-factory="connectionFactory">

                 
                        method="handleMessage" />

            
                        method="handleMessage" />

now only msghandler1 is vaild ,I only receive message of "DEMO.TEST1" queue

Start/stop consuming messages from queues using a single listener-container

goyalk — Thu, 09 May 2013 20:34:12 GMT

My web-application is consuming messages sent by thousands of other publishing applications. Until now, I have had a single queue design; all publishing applications send messages to a single queue from where my application consumes them. But, I am now realizing the need for a multi-queue design; if a misbehaving application sends data extremely fast, messages from other applications may get stuck behind messages from the misbehaving application.

I am trying to implement one-queue-per-publishing-application. I want to have a fixed number of inbound-adapters consuming messages from all these queues. Challenge is that these publishing applications get created/destroyed while my web-application is consuming messages. I want to be able to divide the queues available at start-up to the inbound-adapters and then add/remove queues from these adapters while they are consuming messages. What's the best way to accomplish this?

How to use separate connections for sending and receiving data in Spring-AMQP?

goyalk — Tue, 07 May 2013 22:22:23 GMT

While exploring flow-control in RabbitMQ, I found this:

�The intent here is to introduce a flow control mechanism that throttles producers but lets consumers continue unaffected. However, since AMQP permits producers and consumers to operate on the same channel, and on different channels of a single connection, this logic is necessarily imperfect. In practice that does not pose any problems for most applications since the throttling is observable merely as a delay. Nevertheless, other design considerations permitting, it is advisable to only use individual AMQP connections for either producing or consuming.�

I am using Spring Integration in my web-application to send/receive data to/from RabbitMQ broker. I declare connection-factory, amqp-template, rabbit-admin, channels and inbound/outbound adapters using XML. What is the recommended way to ensure separate connections are used for sending and receiving? My observation is that by default, both sending and receiving are done on the same connection but on different channels.

Missing message ID for stateful retry operations

marcio.gh — Mon, 06 May 2013 17:54:17 GMT

I'm trying to configure StatefulRetryOperationsInterceptorFactoryBean to simulate retry and dead letter queues as I have on a JMS implementation. But the retry is rasing a fatal exception due to the absence of the ID header.

How do I add message Id header on my messages, backed by an Spring Integration amqp outbound channel?

Code:

Spring Integration and Java Configuration

ZaphodB — Tue, 30 Apr 2013 04:18:01 GMT

I managed to get my spring integration working with xml config. I would now like to convert the xml config into its java equivalent. It seems like spring-amqp supports this as they give the following example :

HTML Code:

@Configuration

public class RabbitConfiguration {



  @Bean

  public ConnectionFactory connectionFactory() {

    CachingConnectionFactory connectionFactory = new CachingConnectionFactory("localhost");

    return connectionFactory;

  }



  @Bean

  public AmqpAdmin amqpAdmin() {

    return new RabbitAdmin(connectionFactory());

  }



  @Bean

  public RabbitTemplate rabbitTemplate() {

    return new RabbitTemplate(connectionFactory());

  }



   @Bean

   public Queue myQueue() {

     return new Queue("myqueue");

   }

 }

It would seam that Spring Integration is still dependent on xml configuration as the reference documentation for 3.0.0.M1 doesn't include Java Configuration examples.

Undeterred I had a go at configurating it via java

HTML Code:

**

 * AMQP Configuration

 */

@Configuration

public class AmqpConfig {



    public static final int DEFAULT_REPLY_TIMEOUT = 5000;

    private static final String EXCHANGE_NAME = "profile.manager.exchange";

    private static final String QUEUE_NAME = "profile.manager.queue";



    @Bean

    public MessageChannel profileManagerRequestChannel() {

        DirectChannel channel = new DirectChannel();

        return channel;

    }



    @Bean

    public MessageChannel profileManagerReplyChannel() {

        return new DirectChannel();

    }



    @Bean

    public MessageChannel profileManagerErrorChannel() {

        return new DirectChannel();

    }



    @Bean

    public ProfileManagerService profileManagerService() {

        return new ProfileManagerServiceImpl();

    }



    @Bean

    public JsonMessageConverter jsonMessageConverter() {

        return new JsonMessageConverter();

    }



    @Bean

    public GatewayProxyFactoryBean gatewayProxyFactoryBean() {

        GatewayProxyFactoryBean gatewayProxyFactoryBean = new GatewayProxyFactoryBean();

        gatewayProxyFactoryBean.setServiceInterface(ProfileManagerClient.class);

        gatewayProxyFactoryBean.setDefaultRequestChannel(profileManagerRequestChannel());

        gatewayProxyFactoryBean.setDefaultReplyTimeout(DEFAULT_REPLY_TIMEOUT);

        return gatewayProxyFactoryBean;

    }



    @Bean

    public AmqpOutboundEndpoint outboundGateway() {

        AmqpOutboundEndpoint outboundGateway = new AmqpOutboundEndpoint(rabbitTemplate());

        

        //how do I sent the input

        outboundGateway.setExpectReply(true);

        outboundGateway.setExchangeName(EXCHANGE_NAME);

        outboundGateway.setRoutingKey(QUEUE_NAME);



        return outboundGateway;

    }



    @Bean

    public AmqpInboundGateway inboundGateway() {

        //AbstractMessageListenerContainer listenerContainer ???

        AmqpInboundGateway inboundGateway = new AmqpInboundGateway();

        inboundGateway.setRequestChannel(profileManagerReplyChannel());

        inboundGateway.setMessageConverter(jsonMessageConverter());

        inboundGateway.setErrorChannel(profileManagerErrorChannel());

        inboundGateway.setReplyTimeout(DEFAULT_REPLY_TIMEOUT);



        return  inboundGateway;

    }



    @Bean

    public ConnectionFactory rabbitConnectionFactory() {

       return new CachingConnectionFactory();

    }



    @Bean

    public RabbitTemplate rabbitTemplate() {

        RabbitTemplate rabbitTemplate = new RabbitTemplate();

        rabbitTemplate.setConnectionFactory(rabbitConnectionFactory());

        rabbitTemplate.setMessageConverter(jsonMessageConverter());



        return rabbitTemplate;

    }



    @Bean

    public AmqpAdmin amqpAdmin() {

        return new RabbitAdmin(rabbitConnectionFactory());

    }



    @Bean

    public Queue profileManagerQueue() {

        return new Queue(QUEUE_NAME);

    }



}

To get this far I had to dig into the Namespace resolver code to work out the mappings between xml elements and classes.

Its strange that there is an AmqpInboundGateway class but NO AmqpOutboundGateway - instead you create an AmqpOutboundEndpoint and the replyExpected flag causes it to behave as either an endpointAdapter or a gateway.

The next thing that seems odd is that you cant specify the requestChannel for either the AmqpInboundGateway or AmqpOutboundEndpoint ???

When I try and create an AmqpInboundGateway the constructor wants a referenced to a AbstractMessageListenerContainer ????

I guess I am not supposed to see these innards as they are hidden from me by the Xml configuration ? If you can advise me how to fix my attempt at configuring spring integration via Java Configuration that would be great. If its all too hard at the moment do you have an idea when spring integration will support Java Configuration ??

I am trying to add spring integration into a production system and they are pretty keen to minimize the amount of xml configuration.

Cheers

Spring Integration Asynchronize Gateway Exception callback Problem

ericaZ — Thu, 25 Apr 2013 10:40:26 GMT

Hi All, I have a spring Rabbitmq integration project.

A system send request message to RabbitmqExchange then go to B system. This logic flow is using Asynchronize Gateway. Message from A's outboundGateway is able to reach B . B receive it will access to a server C then do the business job and then return the result msg back to A by using Asynchronize gateway config. [show in below diagram]

Now, A send request --> Rabbitmq --> B, B access another system C but throws a LoginException. I need B's exception return to A. How can I receive Exception from B which pass through rabbitmq(using async callback)? ps. Without exception, A is able to get callback from B.
description.jpg

Anyone can help me ? : )

Attached Images

Show.jpg (8.2 KB)
description.jpg (8.7 KB)

Error queue retry

tec — Thu, 25 Apr 2013 08:37:30 GMT

Hi,
I'm using spring-rabbit in my message consumer. I've implemented a dead-letter exchange and queue system to collect any messages that could not be processed. Please suggest me a good mechanism to re-queue the messages from dead-letter queue back to main queue after some time, so that it can try to process the message after some time.

GatewayProxyFactoryBean and AMQP

ZaphodB — Tue, 23 Apr 2013 07:37:19 GMT

Hi there,

First off let me admit that I am pretty new to Spring Integration and AMQP so I apologize in advance if I am completely off track.

I am trying to get an example working using a GatewayProxyFactoryBean and AMQP. Here is an snippet from my spring configuration:

-------

service-interface="com.innovisionsystems.service.StringCon versionService">

request-channel="toRabbit"
amqp-template="rabbitTemplate"
exchange-name="si.test.exchange"/>

-------

As I understand it the GatewayProxyFactory bean will automatically create a temporary reply queue. When I step through my example in the debugger I can see the message arrive at the AmqpOutboundEndPoint::sendAndReceive() method with a replyChannel set.

Here is an extract of the code:

private Message sendAndReceive(String exchangeName, String routingKey, Message requestMessage) {
// TODO: add a convertSendAndReceive method that accepts a MessagePostProcessor so we can map headers?
Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for send and receive");
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
MessageProperties amqpMessageProperties = new MessageProperties();
this.headerMapper.fromHeadersToRequest(requestMess age.getHeaders(), amqpMessageProperties);
org.springframework.amqp.core.Message amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
org.springframework.amqp.core.Message amqpReplyMessage = this.amqpTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage);
if (amqpReplyMessage == null) {
return null;
}

By the time we get to the amqpTemplate.sendAndReceive() method call the amqpMessage does not appear to have any reply / replyChannel set.

Is this expected behaviour ? Is there anyway for the temporary reply queue generated by the Gateway to be passed through AMQP ???

Cheers

stateful retry

raghukishoreb — Tue, 23 Apr 2013 07:26:44 GMT

my configuration looks like this.

Code:

but i am getting below exception..

Caused by: org.springframework.beans.factory.BeanCreationExce ption: Error creating bean with name 'org.springframework.integration.amqp.outboun
d.AmqpOutboundEndpoint#5d0ee8': Cannot resolve reference to bean 'retryInterceptor' while setting bean property 'adviceChain' with key [0]; nes
ted exception is org.springframework.beans.factory.BeanCreationExce ption: Error creating bean with name 'retryInterceptor' defined in ServletCo
ntext resource [/WEB-INF/beans-messaging.xml]: Error setting property values; nested exception is org.springframework.beans.NotWritableProperty
Exception: Invalid property 'messageKeyGenerator' of bean class [org.springframework.amqp.rabbit.config.StatefulRet ryOperationsInterceptorFacto
ryBean]: Bean property 'messageKeyGenerator' is not writable or has an invalid setter method. Did you mean 'messageKeyGeneretor'?
at org.springframework.beans.factory.support.BeanDefi nitionValueResolver.resolveReference(BeanDefinitio nValueResolver.java:329)
at org.springframework.beans.factory.support.BeanDefi nitionValueResolver.resolveValueIfNecessary(BeanDe finitionValueResolver.java:107)
at org.springframework.beans.factory.support.BeanDefi nitionValueResolver.resolveManagedList(BeanDefinit ionValueResolver.java:353)
at org.springframework.beans.factory.support.BeanDefi nitionValueResolver.resolveValueIfNecessary(BeanDe finitionValueResolver.java:154)

Multiple RabbitAdmin instances force dubious declarations.

patelm5 — Fri, 19 Apr 2013 13:38:06 GMT

Hi Guys,

I'm a little confused on the queue declaration policy for spring rabbit dist of amqp. Documentation suggests following

2.12.1 Automatic Declaration of Exchanges, Queues and Bindings

The RabbitAdmin component can declare exchanges, queues and bindings on startup. It does this lazily, through a ConnectionListener, so if the broker is not present on startup it doesn't matter. The first time a Connection is used (e.g. by sending a message) the listener will fire and the admin features will be applied. A further benefit of doing the auto declarations in a listener is that if the connection is dropped for any reason (e.g. broker death, network glitch, etc.) they will be applied again the next time they are needed.

However, currently there is no mechanism for assigning a queue to a particular connection factory, so if I wanted to declare a queue on a particular broker with RabbitAdmin, it is currently not possible.

The code at fault imo is the following :-

Code:

final Collection exchanges = applicationContext.getBeansOfType(Exchange.class).values();

                final Collection queues = applicationContext.getBeansOfType(Queue.class).values();

                final Collection bindings = applicationContext.getBeansOfType(Binding.class).values();



... 



rabbitTemplate.execute(new ChannelCallback() {

                        public Object doInRabbit(Channel channel) throws Exception {

                                declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()]));

                                declareQueues(channel, queues.toArray(new Queue[queues.size()]));

                                declareBindings(channel, bindings.toArray(new Binding[bindings.size()]));

                                return null;

                        }

                });

Which obviously is indiscriminate in its effort to create bindings.

Are there any plans to fix this or any workaround ?

Cheers,
Mike.