activemq-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Edward Ye <edwar...@21cn.com>
Subject Consumer Listener stop receving message until ActiveMQ restart
Date Thu, 06 Nov 2008 05:07:14 GMT

Hi,All.
I am facing a simliar problem like this
thread(http://www.nabble.com/Pending-Messages-are-shown-in-ActiveMQ-to20241332.html)
but in a different environment.

My PC:
ActiveMQ 5.1.0
Java SE 6 Update 10
Windows 2003 SP2
client api: activemq-cpp-2.2.1

Development tool:
Visual Studio 2005

I have a simple test:
A producer sent  thousands of  persistent messages (each 1k bytes )into a
queue in async way. and then stopped.
A consumer received message with listener start and tried to run up of the
pending messages.

Sometimes the consumer stopped to received even there were still many
messages pending in the queue. I have tried to start the client and
reconnect to the queue successfully. But the messages still remained in the
queue (from webconsole I can see)and the onMessage handler was never called
until the ActiveMQ was restarted.

Just like the above thread saying, sometimes the Number Of Pending Message
in webconsole is negative and not proper. If I restart ActiveMQ , the number
become correct.

here is my code of producer:
ActiveMQMessageProducer is a producer wrapper class.
int ActiveMQMessageProducer::connect(char* cbrokerURI,char* cqueueName)
{
	ConnectionFactory* connectionFactory = NULL;

	try
	{
		string brokerURI=cbrokerURI;
		string queueName=cqueueName;
		
		// Create a ConnectionFactory
		connectionFactory = ConnectionFactory::createCMSConnectionFactory(
brokerURI);//brokerURI must contain UseAsyncSend=true

		// Create a Connection
                         connection = connectionFactory->createConnection();
                         ((Connection*)connection)->start();

		// free the factory, we are done with it.
                         delete connectionFactory;
                         connectionFactory = NULL;

		// Create an Auto_Acknowledge Session
		session = ((Connection*)connection)->createSession(
Session::AUTO_ACKNOWLEDGE );

		destination =((Session*) session)->createQueue( queueName );

		// Create a MessageProducer from the Session to the Topic or Queue
                         producer = ((Session*)session)->createProducer(
(Destination*)destination );
                         ((MessageProducer*)producer)->setDeliveryMode(
DeliveryMode::PERSISTENT );
	}
	catch( CMSException& e ) 
	{
                         if (connectionFactory)
			delete connectionFactory;
                         connectionFactory = NULL;

		
		if (logCallbackFunc)//a function pointer
		{
			logCallbackFunc((char*)e.getMessage().c_str());
		}

		return  -1;
	}
	return 0;

}

int ActiveMQMessageProducer::sendMessage(char *message,size_t sizeOfMessage)
{
	BytesMessage* bytesMessage=NULL;
	try
	{
		bytesMessage=((Session*)session)->createBytesMessage((unsigned
char*)message,sizeOfMessage);
		((MessageProducer*)producer)->send(bytesMessage);

	}
	catch(CMSException& e)
	{
		
		if (logCallbackFunc)
		{
			logCallbackFunc((char*)(e.getMessage().c_str()));
		}
		if (bytesMessage)
			delete bytesMessage;

		bytesMessage=NULL;

		return  -1;
	}

	if (bytesMessage)
		delete bytesMessage;

	bytesMessage=NULL;

	return 0;
}

And below is code for consumer:
ActiveMQConsumer is a wrapper class for consumer

int ActiveMQMessageConsumer::connect(char *cbrokerURI,char *cqueueName)
{
	ConnectionFactory* connectionFactory = NULL;

	try
	{
		string brokerURI=cbrokerURI;
		string queueName=cqueueName;
		
		// Create a ConnectionFactory
		connectionFactory = ConnectionFactory::createCMSConnectionFactory(
brokerURI);

		// Create a Connection
                         connection = connectionFactory->createConnection();
                        ((Connection*)connection)->start();

	
((Connection*)connection)->setExceptionListener(((ConsumerListener*)listener));

		// free the factory, we are done with it.
                        delete connectionFactory;
                        connectionFactory = NULL;

		// Create an Auto_Acknowledge Session
		session = ((Connection*)connection)->createSession(
Session::AUTO_ACKNOWLEDGE );

		destination =((Session*) session)->createQueue( queueName );

		// Create a MessageProducer from the Session to the Topic or Queue        
                         consumer = ((Session*) session)->createConsumer(
((Destination*)destination ));
                        
((MessageConsumer*)consumer)->setMessageListener(((ConsumerListener*)listener));
	}
	catch( CMSException& e ) 
	{
                        if (connectionFactory)
			delete connectionFactory;
                        connectionFactory = NULL;

		
		if (logCallbackFunc)
		{
			logCallbackFunc((char*)e.getMessage().c_str());
		}

		return  -1;
	}
	return 0;

}

ConsumerListener is implementation for MessageListener and ExceptionListener

class ConsumerListener: public ExceptionListener,
		public MessageListener
{
private:
	ActiveMQMessageConsumer* consumer;

public:
	ConsumerListener(ActiveMQMessageConsumer* messageConsumer)
	{
		consumer=messageConsumer;
	}

	virtual ~ConsumerListener()
	{
		consumer=NULL;
	}

	// Called from the consumer since this class is a registered
MessageListener.
    virtual void onMessage( const Message* message )
	{
		
		try
		{
			const BytesMessage* bytesMessage =
                dynamic_cast< const BytesMessage* >( message );
			
			char* tempCharArray=new char(bytesMessage->getBodyLength());
		
memcpy(tempCharArray,(char*)(bytesMessage->getBodyBytes()),bytesMessage->getBodyLength());
                                       delete []tempCharArray;
                                       tempCharArray=NULL;
				
			
		}
		catch(CMSException& e)
		{
			// do some stuff
		}

		return ;
	}

	// If something bad happens you see it here as this class is also been
    // registered as an ExceptionListener with the connection.
    virtual void onException( const CMSException& ex )
	{
		try
		{			
			//do some error handler stuff
			
		}
		catch(CMSException& e)
		{
			//
		}
		return ;

    }
};

Anyone could help ?
Thanks a lot.


-- 
View this message in context: http://www.nabble.com/Consumer-Listener-stop-receving-message-until-ActiveMQ-restart-tp20355247p20355247.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.


Mime
View raw message