Mailing-List: contact activemq-users-help@geronimo.apache.org; run by ezmlm
Precedence: bulk
Reply-To: activemq-users@geronimo.apache.org
Received-SPF: pass (herse.apache.org: local policy)
Content-class: urn:content-classes:message
MIME-Version: 1.0
Content-Type: text/plain;
	charset="US-ASCII"
Content-Transfer-Encoding: quoted-printable
Subject: RE: above topic
Date: Thu, 16 Nov 2006 07:12:34 -0500
Message-ID: 
 <5A211522579EBD4A83155ED7FA153063013B6317@corpatsmail1.corp.sensis.com>
Thread-Topic: above topic
Thread-Index: AccJWzSRbkyBLQCCQKGygCXyQDk/zQAHBj2w
From: "Bish, Tim" <Tim.Bish@Sensis.com>
To: <activemq-users@geronimo.apache.org>

>=20
>=20
> follow program,why cann't receive data?
> How should I modify code?

Looks like you produce before the durable consumer has been connected
once.=20

See this link:
http://www.activemq.org/site/how-do-durable-queues-and-topics-work.html

A durable consumer must have connected and then disconnected before
messages are persisted in anticipation of the consumer reconnecting.
That's why its important to use the same subscription name when
reconnecting, so that the broker know that the consumer that was
subscribed as durable is now back and it can deliver the messages that
were stored in its absence. =20

>=20
> #include <activemq/concurrent/Thread.h>
> #include <activemq/concurrent/Runnable.h>
> #include <activemq/core/ActiveMQConnectionFactory.h>
> #include <activemq/util/Integer.h>
> #include <cms/Connection.h>
> #include <cms/Session.h>
> #include <cms/TextMessage.h>
> #include <cms/ExceptionListener.h>
> #include <cms/MessageListener.h>
> #include <stdlib.h>
>=20
> using namespace activemq::core;
> using namespace activemq::util;
> using namespace activemq::concurrent;
> using namespace cms;
> using namespace std;
>=20
> class HelloWorldProducer : public Runnable {
> private:
>=20
> 	Connection* connection;
> 	Session* session;
> 	Topic* destination;
> 	MessageProducer* producer;
> 	int numMessages;
>=20
> public:
>=20
> 	HelloWorldProducer( int numMessages ){
> 		connection =3D NULL;
>     	session =3D NULL;
>     	destination =3D NULL;
>     	producer =3D NULL;
>     	this->numMessages =3D numMessages;
> 	}
>=20
> 	virtual ~HelloWorldProducer(){
> 		cleanup();
> 	}
>=20
>     virtual void run() {
>         try {
> 			string user,passwd,sID;
> 			user=3D"default";
> 			passwd=3D"";
> 			sID=3D"lsgID";
>             ActiveMQConnectionFactory* connectionFactory =3D new
> ActiveMQConnectionFactory("tcp://localhost:61613",user,passwd,sID);
>=20
>             connection =3D
> connectionFactory->createConnection(user,passwd,sID);
>             connection->start();
>=20
> 			string sss=3Dconnection->getClientId();
> 			cout << sss << endl;
>=20
>             session =3D connection->createSession(
Session::AUTO_ACKNOWLEDGE
> );
> 			destination =3D session->createTopic( "mytopic" );
>=20
> 			producer =3D session->createProducer( destination
);
>             producer->setDeliveryMode( DeliveryMode::PERSISTANT );
>=20
> 			producer->setTimeToLive(100000000);
>             string threadIdStr =3D Integer::toString( Thread::getId() =
);
>=20
>             // Create a messages
>             string text =3D (string)"Hello world! from thread " +
> threadIdStr;
>=20
>             for( int ix=3D0; ix<numMessages; ++ix ){
> 	            TextMessage* message =3D session->createTextMessage(
text
> );
>=20
> 				string messageID=3D"messageID";
> 				message->setCMSExpiration(10000000000);
> 				message->setCMSMessageId(messageID);
>=20
>     	        // Tell the producer to send the message
>         	    printf( "Sent message from thread %s\n",
> threadIdStr.c_str() );
>         		producer->send( message );
>=20
>             	delete message;
>             }
>=20
>         }catch ( CMSException& e ) {
>             e.printStackTrace();
>         }
>     }
>=20
> private:
>=20
>     void cleanup(){
>=20
> 			// Destroy resources.
> 			try{
>             	if( destination !=3D NULL ) delete destination;
> 			}catch ( CMSException& e ) {}
> 			destination =3D NULL;
>=20
> 			try{
> 	            if( producer !=3D NULL ) delete producer;
> 			}catch ( CMSException& e ) {}
> 			producer =3D NULL;
>=20
>     		// Close open resources.
>     		try{
>     			if( session !=3D NULL ) session->close();
>     			if( connection !=3D NULL ) connection->close();
> 			}catch ( CMSException& e ) {}
>=20
> 			try{
>             	if( session !=3D NULL ) delete session;
> 			}catch ( CMSException& e ) {}
> 			session =3D NULL;
>=20
>             try{
>             	if( connection !=3D NULL ) delete connection;
> 			}catch ( CMSException& e ) {}
>     		connection =3D NULL;
>     }
> };
>=20
> class HelloWorldConsumer : public ExceptionListener,
>                            public MessageListener,
>                            public Runnable {
>=20
> private:
>=20
> 	Connection* connection;
> 	Session* session;
> 	Topic* destination;
> 	MessageConsumer* consumer;
> 	long waitMillis;
>=20
> public:
>=20
> 	HelloWorldConsumer( long waitMillis ){
> 		connection =3D NULL;
>     	session =3D NULL;
>     	destination =3D NULL;
>     	consumer =3D NULL;
>     	this->waitMillis =3D waitMillis;
> 	}
>     virtual ~HelloWorldConsumer(){
>     	cleanup();
>     }
>=20
>     virtual void run() {
>=20
>         try {
>=20
> 			string user,passwd,sID;
> 			user=3D"default";
> 			passwd=3D"";
> 			sID=3D"lsgID";
>             // Create a ConnectionFactory
>             ActiveMQConnectionFactory* connectionFactory =3D
>                 new ActiveMQConnectionFactory(
> "tcp://localhost:61613",user,passwd,sID);
>=20
>             // Create a Connection
>             connection =3D
> connectionFactory->createConnection();//user,passwd,sID);
>             delete connectionFactory;
>             connection->start();
>=20
>             connection->setExceptionListener(this);
>=20
>             // Create a Session
>             session =3D connection->createSession(
Session::AUTO_ACKNOWLEDGE
> );
> 			destination =3D session->createTopic( "mytopic" );
> 			consumer =3D session->createDurableConsumer(
destination ,
> user ,
> "",false);
>=20
>             consumer->setMessageListener( this );
>=20
>             Thread::sleep( waitMillis );
>=20
>         } catch (CMSException& e) {
>             e.printStackTrace();
>         }
>     }
>=20
>     virtual void onMessage( const Message* message ){
>=20
>         try
>         {
>     	    const TextMessage* textMessage =3D
>                 dynamic_cast< const TextMessage* >( message );
>             string text =3D textMessage->getText();
>             printf( "Received: %s\n", text.c_str() );
>         } catch (CMSException& e) {
>             e.printStackTrace();
>         }
>     }
>=20
>     virtual void onException( const CMSException& ex ) {
>         printf("JMS Exception occured.  Shutting down client.\n");
>     }
>=20
> private:
>=20
>     void cleanup(){
>=20
> 		// Destroy resources.
> 		try{
>         	if( destination !=3D NULL ) delete destination;
> 		}catch (CMSException& e) {}
> 		destination =3D NULL;
>=20
> 		try{
>             if( consumer !=3D NULL ) delete consumer;
> 		}catch (CMSException& e) {}
> 		consumer =3D NULL;
>=20
> 		// Close open resources.
> 		try{
> 			if( session !=3D NULL ) session->close();
> 			if( connection !=3D NULL ) connection->close();
> 		}catch (CMSException& e) {}
>=20
>         try{
>         	if( session !=3D NULL ) delete session;
> 		}catch (CMSException& e) {}
> 		session =3D NULL;
>=20
> 		try{
>         	if( connection !=3D NULL ) delete connection;
> 		}catch (CMSException& e) {}
> 		connection =3D NULL;
>     }
> };
>   void Produce()
>  {
>      HelloWorldProducer producer( 2 );
>      Thread producerThread( &producer );
>      producerThread.start();
>      producerThread.join();
>  }
>  void Consumer()
>  {
>       HelloWorldConsumer consumer( 10000 );
>       Thread consumerThread( &consumer );
>       consumerThread.start();
>       consumerThread.join();
>  }
>  int main(int argc, char* argv[])
>  {
>       Produce();
>       Consumer();
>  }
> --
> View this message in context: http://www.nabble.com/above-topic-
> tf2641566.html#a7373622
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.