Rabbitmq consumer does not process/return processed message to client

[Irfan Sayed]

hi,
i have created the publisher - consumer "reply-to pattern" using rabbitmq tutorial 6 guide.
the issue , publisher and consumer works fine when the publisher sends the message first time. for second time, consumer does not do anything.
i am not sure wheather issue with replyTo queue. I am not using any exchange.
basically, by using this url , i have developed my code : https://www.rabbitmq.com/tutorials/tutorial-six-java.html
please help ...
please let me know where i can paste the code

Regards,

[Michael Klishin]

The best option is to put your code in a public GitHub repository

with some basic instructions about how to run it.

You can also compare and contrast your code to our runnable tutorials:
https://github.com/rabbitmq/rabbitmq-tutorials/

--
You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

--

MK

Staff Software Engineer, Pivotal/RabbitMQ

[Irfan Sayed]

Hello,

Thanks for the reply. 

i have pasted the code on gist along with all details and issues 

https://gist.github.com/irfanjs/ee75538a75f7b4aed60b4a3dbc2fa409

please suggest.

On Friday, February 23, 2018 at 10:33:35 PM UTC+1, Michael Klishin wrote:

The best option is to put your code in a public GitHub repository

with some basic instructions about how to run it.

You can also compare and contrast your code to our runnable tutorials:
https://github.com/rabbitmq/rabbitmq-tutorials/

On Sat, Feb 24, 2018 at 12:06 AM, Irfan Sayed <irfu....@gmail.com> wrote:

hi,
i have created the publisher - consumer "reply-to pattern" using rabbitmq tutorial 6 guide.
the issue , publisher and consumer works fine when the publisher sends the message first time. for second time, consumer does not do anything.
i am not sure wheather issue with replyTo queue. I am not using any exchange.
basically, by using this url , i have developed my code : https://www.rabbitmq.com/tutorials/tutorial-six-java.html
please help ...
please let me know where i can paste the code

Regards,

--
You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-user...@googlegroups.com.
To post to this group, send email to rabbitm...@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

[Michael Klishin]

Please provide something runnable, e.g. a small Gradle or Maven project

that can be started and queried via curl. Asking others to stitch a few gists together into

a project is too much: we have no idea what dependencies are involved and there's a non-trivial

chance that we won't end up with the same exact app as you are.

You also can inspect message flow using

 * Tracing (http://www.rabbitmq.com/firehose.html)

 * Wireshark (http://www.rabbitmq.com/amqp-wireshark.html)

There are things in the code that are confusing at first glance. Why do you use "requestQueueName"

for queue name and then assign it to a "replyQueue"?

Both apps seem to consume from the same queue, one in automatic acknowledgement mode,

which will consume all messages there are, without any limit [1]. Tutorial 6 on rabbitmq.com uses

two separate queues, one of which is "anonymous" (with a generated name) and used for

response collection. My best guess is that the first run works because consumer that uses QoS prefetch = 1

is started first and has a chance to consume its delivery but on the second run it is not the case

(with competing consumers, delivery is round robin only if their prefetch settings are equal, which is not the case here).

1. http://www.rabbitmq.com/confirms.html

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

[Irfan Sayed]

Thanks. 

I have checked in entire code.

First project : 

https://github.com/irfanjs/unit

second project :

https://github.com/irfanjs/unitdb

both are spring-boot based micro-services. have checked in the pom.xml as well for maven 

in readme file , the command to execute these services ..

before running, please change the values in application properties file and create the fresh package 

please suggest 

Regards,

[Irfan Sayed]

Further,

I have updated the readme file as well to reflect the exact endpoint to be hitted in the web browser ... 

once you hit the endpoint first time, it works properly. second time, it just hangs or waiting for some response which i am not getting 

please suggest 

[Irfan Sayed]

please suggest.. any issue in the code ? 

Regards,

[Michael Klishin]

No need to triple post. It's a Saturday or Saturday night in most of the world.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

[Michael Klishin]

Your code has a number of issues.

First, you have hardcoded IPs (I'm not going to post them here).

Next, you start a consumer in an HTTP API request handler and never stop it. It will stick around

consuming messages even after the request has completed, interfering with future [RPC] client requests.

Lastly, you declare response queues which are meant to be transient as durable.

Perhaps you want to hide all of that stuff from Boot's HTTP request handlers and avoid starting a consumer

and stopping it right in the request handler body.

When I hit the HTTP endpoint with curl the 2nd time, I get

java.lang.IllegalStateException: Invalid configuration: 'routingKey' must be non-null.
    at com.rabbitmq.client.impl.AMQImpl$Basic$Publish.<init>(AMQImpl.java:2280) ~[amqp-client-4.0.3.jar!/:4.0.3]
    at com.rabbitmq.client.AMQP$Basic$Publish$Builder.build(AMQP.java:1219) ~[amqp-client-4.0.3.jar!/:4.0.3]
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:691) ~[amqp-client-4.0.3.jar!/:4.0.3]
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:668) ~[amqp-client-4.0.3.jar!/:4.0.3]
    at com.rabbitmq.client.impl.ChannelN.basicPublish(ChannelN.java:658) ~[amqp-client-4.0.3.jar!/:4.0.3]
    at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.basicPublish(AutorecoveringChannel.java:192) ~[amqp-client-4.0.3.jar!/:4.0.3]
    at com.infy.ci.unitdbamqpservice.RPCServer$1.handleDelivery(RPCServer.java:52) ~[classes!/:0.0.1-SNAPSHOT]

which suggests that `properties.getReplyTo()` is null in that case.

Please take it from here.

[Arnaud Cogoluègnes]

I suggest you clean the code to use only Spring AMQP on top of Spring Boot AMQP auto-configuration (you don't need to use the RabbitMQ Java client API directly).

The server part (unitdb module) would look like the following:

@SpringBootApplication
public class UnitdbamqpserviceApplication {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        SpringApplication.run(UnitdbamqpserviceApplication.class, args);
    }

    private static final String RPC_QUEUE_NAME = "rpc_queue1";

    @Bean
    public Queue queue() {
        return new Queue(RPC_QUEUE_NAME);
    }

    @Component
    public static class RpcListener {

        @RabbitListener(queues = RPC_QUEUE_NAME)
        public String reply() {
            String response = "{" + "\n" + "" + "\"glossary\"" + ":" +  "{" + "\n" + "\"title\":" + "\"example glossary\"" + "\n" + "}" + "\n" + "}";
            return response;
        }

    }

}

And the controller part (unit module) like the following:

    @Autowired
    RabbitTemplate rabbitTemplate;

    @RequestMapping(value="/{projectid}/ut/aggregate",  
            method = RequestMethod.GET,
            produces = MediaType.TEXT_HTML_VALUE)   
    public @ResponseBody String getAggregatedDataForSectionOfNightlyBuild (@PathVariable("projectid") int projectid,
            @RequestParam("buildtype") String buildtype,
            @RequestParam("build") String build) throws Exception {
       
        buildtype = "nightly";
       
       
        if(build.toLowerCase().equals("latest") && buildtype.equals("nightly")){
           
             String message = String.format("aggregate");

             return rabbitTemplate.convertSendAndReceive("", requestQueueName, message).toString();

        else {

          return null;
       }

This is the bare minimum to make the projects work together. Have also a look at the Spring Boot and Spring AMQP respective documentation to understand what's going on.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

[Irfan Sayed]

Thanks Michel and Arnaud for great help.

To Michel :

I have removed the hardcoding and now reading the IP address from properties files 

i have removed the properties.getReplyTo() as it was containing null. and added the routing_key in the basicPublish method

channel.basicPublish( "", routing_key, replyProps, response.getBytes("UTF-8"));

but still the issue is same. i am using httpie command line client to test the endpoint . it is giving me 406 error code. something unexpected response 

i kept the value of routing_key and queue_name string same. but still the behavior is same  

not getting exception and second time when i hit the endpoint , just hangs .... still reading some doc... 

can you please give ,me your full curl command , i can try here ..

have updated the code and pushed to github. please have a look. i am sure , must be a stupid mistake but not hitting where .....

To arnaud,

i have created the code using your suggestion : 

unit module : 

https://github.com/irfanjs/unit1

unitdb module :

https://github.com/irfanjs/unitdb1

after hitting the endpoint, it is simply not reachable.404 status code. i checked the code but seems OK ... can you please have a look 

Regards,

Irfan

[Michael Klishin]

This is not a Spring Boot forum. Please figure out why the HTTP API of your app responds with a 404 or 406

(both of which are client errors) on your own.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

[Irfan Sayed]

Thats true. and i am trying to resolve that error. 

i m just struggling with some sample example where i can get the response based on some event. as of now, i am generating this event by hitting the HTTP endpoint, 

if you please help me with sample app which generates events and get the reply back to same queue as response, that would be great 

Regards,

[Irfan Sayed]

while reading the doc :  in the RPC section

https://spring.io/blog/2010/06/14/understanding-amqp-the-protocol-used-by-rabbitmq/

1: Client sends message to the queue, specifying: (a) a routing key that matches the service; and (b) the name of a queue to pick the response up from.

what is the meaning of : routing key that matches the service ?  is this like name matching ? any example plz 

2: Exchange passes the message to the service's queue ("ops_q" in this case).

so in my case , service which has HTTP endpoint ? and the queue which is declared in the code is a service's queue ? 

3: in my entire example, haven't declare any exchange. so declaring exchange and binding it with queue is important ?  if yes, i have to do that in my http endpoint code rite  ?

please help with these queries 

regards,

[Michael Klishin]

Irfan,

1. "Matches the service" means "that will make sure the request will be routed to a queue used by the service". In other words, client

requests reach the server. RabbitMQ supports different exchange types and topologies, so how *exactly* that is done is up to you.

Tutorial 6 uses a well known server name and the default exchange.

2. The app that handles HTTP requests in your case is the client. It has to communicate with a different service, which acts as a server (responds to client requests).

3. I suggest that you read http://www.rabbitmq.com/tutorials/amqp-concepts.html before doing any further work on this project.

Keep in mind that the post you are reading is from 2010.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

For more options, visit https://groups.google.com/d/optout.

[Michael Klishin]

So the request/reply pattern works more or less like this (I'm trying to provide an alternative explanation

in order to see if anything obviously missing in our tutorials pops up):

 * You have a server of some kind that declares a well known queue (or set of queues, e.g. one per server function or just as a unit of concurrency) and consumes client requests from it

 * Clients — which is a very general term and often that may mean services that expose an HTTP API, so in a way they are servers as well — publish requests in a way that reaches the server queue(s). There is no right or wrong way of doing that, e.g. our tutorials use the default exchange.

 * The server gets a delivery for the request and does some work, e.g. computes something

 * The server then has to communicate back to the client. For that clients usually provide a hint: include a queue name they use to collect responses in the message, the reply-to field was added to the protocol for this reason

 * The above means the client has to consume from at least one queue in order to collect responses

That's about it. The idea doesn't change significantly for a lot of workloads.

[Irfan Sayed]

Thanks.

I have gone through the JAVA API guide and now following is my understanding 

1: the binding between exchange and the queue happens through routing key. sometimes it called as binding key as well 

2: the meaning of the routing key is depends on the type of exchange . in case of fanout, rabbitmq simply ignores the binding key. however in my example , as i am using direct exchange type , routing key has special value. so having routing key as null can not be the good option in my opinion.

3: the value of routing key should be get from the envelope class which has the binding with the exchange so that messages can be delivered properly to the consumer

4: AMQP basic properties contains the correlationID which is key factor is delivering the messages.

5: reply-to is unique per connection. but if there are multiple calls per client , there has to be a request ID to track it and in my case , i guess , it is correlationID

according to above understanding , i made the changes to the code and published to github. 

however, unfortunately , the issue is still same. but now , i am not getting exception of having routing key as null. not getting what to read more to get it working properly :(

latest code : https://github.com/irfanjs/unitdb

please have a look on the code and suggest ... 

Regards,

[Gary Russell]

As Arnaud said last week; since you are using Spring Boot, why not let Spring do all the heavy lifting for you; it took me five minutes to write this sample [1]

[1]: https://gist.github.com/garyrussell/e003474483eb604ee2362d28125d0165

@SpringBootApplication
public class RabbitmqUsers1Application {

	private final Logger logger = LoggerFactory.getLogger(RabbitmqUsers1Application.class);

	public static void main(String[] args) {
		SpringApplication.run(RabbitmqUsers1Application.class, args);
	}

	@Bean
	public ApplicationRunner runner(RabbitTemplate template) {
		return args -> {
			ParameterizedTypeReference<String> returnTypeString = new ParameterizedTypeReference<String>() { };
			logger.info(template.convertSendAndReceiveAsType("myExchange", "rk", "foo", returnTypeString));
			logger.info(template.convertSendAndReceiveAsType("myExchange", "rk", "bar", returnTypeString));
			logger.info(template.convertSendAndReceiveAsType("myExchange", "rk", "baz", returnTypeString));
		};
	}

	@RabbitListener(queues = "myQueue")
	public String upCase(String in) {
		return in.toUpperCase();
	}

	@Bean
	public MessageConverter jsonConverter() {
		return new Jackson2JsonMessageConverter();
	}

	@Bean
	public DirectExchange exchange() {
		return new DirectExchange("myExchange");
	}

	@Bean
	public Queue queue() {
		return new Queue("myQueue");
	}

	@Bean
	public Binding binding() {
		return BindingBuilder.bind(queue()).to(exchange()).with("rk");
	}

}

2018-03-03 17:39:54.599  INFO 2182 --- [           main] com.example.RabbitmqUsers1Application    : FOO
2018-03-03 17:39:54.603  INFO 2182 --- [           main] com.example.RabbitmqUsers1Application    : BAR
2018-03-03 17:39:54.607  INFO 2182 --- [           main] com.example.RabbitmqUsers1Application    : BAZ

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

[Irfan Sayed]

Awesome !!! .

Thanks much to Gary, Arnaud and Michael. Indeed it worked. 

the issue was related to channel and connection. 

using convertSendAndReceive method and all using spring AMQP implementation (without using java API client) , it worked. I have to still understand more on JAVA client side.

now moving again till i reach the next block to bug you guys again  :)

regards,

irfan

[Irfan Sayed]

Hello,

Sorry to say, today faced issue of NPE. I haven't changed anything in the rabbit connection and rabbit template part. it is same as of month before. 

here are two projects : 

controller which accepts the response from rabbit :

https://github.com/irfanjs/unitamqpservice

server part of the code :

https://github.com/irfanjs/unitdbamqpservice

as per the exception , it is on the 83 line of UnitAmqpRestController.java

i invested hours but still not getting where is the issue. the rabbit and MySQL server is getting connected properly 

queues and exchanges are getting created.

no any issue related to connection but still not getting why NPE

Can you please have  a look on code ? really stuck now plz 

Regards,

[Gary Russell]

Please move this to Stack Overflow [spring-amqp] tag - this has nothing to do with RabbitMQ.

If you are getting an NPE on that line, it means there is something wrong with your Spring wiring.

You can debug spring bean wiring with DEBUG logging.

However, you are unconditionally calling toString() on the result; the request has likely timed out and `null` is received - you need to check for a null result and perhaps increase the reply timeout from its default 5 seconds.

If/when you ask your question on Stack Overflow, include the complete stack trace.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

[Irfan Sayed]

Thanks. Posted the query on stack overflow with complete stack trace. 

please have a look and suggest : 

https://stackoverflow.com/questions/50428087/getting-npe-while-receiving-the-reply-message-using-rabbittemplate

Regards,