Rabbit MQ Java Service Consumer - Not picking up messages

, ,
1

Hi There,

We are trying to connect to a Rabbit MQ platform in AWS, via a Java Service Consumer client code in 10.15 IS.
We are able to successfully connect and push the message, while reading the message it seems to be NOT working.
The requirement is to create a JAVA consumer that can run all the time and read the message as soon as its published into that queue. Below is the code we have written for that.

public static final void consumeMessageV1(IData pipeline) throws ServiceException {
   	  
   	String queueName= "Queue";
   	
   	try
   	{
   		ConnectionFactory factory = new ConnectionFactory();
   		factory.setHost("abc.test.com");
   		factory.setPort(5671);
   		factory.setUsername("username");
   		factory.setPassword("pwd");
   		factory.setVirtualHost("virtualhost");
   		factory.setAutomaticRecoveryEnabled(true);
   		factory.setTopologyRecoveryEnabled(true);
   		factory.useSslProtocol();
   	
   		Connection connection = factory.newConnection();
   	    debugLog ("Connection Created");
   	    Channel channel;
   	
   		if (connection.isOpen()) {
   			channel = connection.createChannel();
   		} else {
   			throw new IOException("Unable to open connection");
   		}
   		
   		channel.queueDeclare(queueName, true, false, false, null);
   	    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
   	
   	    channel.basicQos(1);
   		
   	    DeliverCallback deliverCallback = (consumerTag, delivery) -> {
   	        String message = new String(delivery.getBody(), "UTF-8");
   	        try {
   	        	debugLog ("Trust Message is : " +message);
                               // DO something to message here
   	        	} 
   	        finally {
   	            System.out.println(" [x] Done");
   	            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
   	        	}
   	    };
   	    channel.basicConsume(queueName, false, deliverCallback, consumerTag -> { });		
   	}
   	catch (Exception e)
   	{
   		debugLog ("Unable to consume message.Error: " +e.toString());
   	}
   		
   }
   
   // --- <<IS-BEGIN-SHARED-SOURCE-AREA>> ---
   
   public static void debugLog(String message)
   {
   
   	// input
   	IData input = IDataFactory.create();
   	IDataCursor inputCursor = input.getCursor();
   	IDataUtil.put( inputCursor, "message", message );
   	inputCursor.destroy();
   
   	// output
   	IData 	output = IDataFactory.create();
   	try{
   		output = Service.doInvoke( "pub.flow", "debugLog", input ); 
   	}catch( Exception e){} 
   
   }
   
   // --- <<IS-END-SHARED-SOURCE-AREA>> ---

Below are issues:

  1. The deliverCallback method seems to be never invoked at all and we are not getting any errors. As a result the message is also not picked up from queue and not logged.
  2. How do we ensure this piece of consumer keeps on running and reads message as soon as it comes (like a JMS trigger).
  3. The connection is not closed here, so if I run this JAVA service will it keep on holding on to that connection to read messages ? Or will the thread/flow as soon exit as soon as the service completes execution once?

Can someone shed light on this please ?

Thanks
Ragav J