[jboss-user] [JBoss Messaging] - Re: All messages are not persisted in Database.

[sajankn]

I tried to merge both subscriber and publisher code into a single file, but I'm not able to reproduce the same situation for this single file. It can be reproduced only if both run in separate jvm and in a clustered mode.


  | import javax.jms.JMSException;
  | import javax.jms.TextMessage;
  | import javax.jms.Topic;
  | import javax.jms.TopicConnection;
  | import javax.jms.TopicConnectionFactory;
  | import javax.jms.TopicPublisher;
  | import javax.jms.TopicSession;
  | import javax.jms.TopicSubscriber;
  | import javax.naming.InitialContext;
  | import javax.naming.NamingException;
  | 
  | public class MyTest 
  | {
  | 	private String destinationName;
  | 	private TopicConnectionFactory cf;
  | 	private Topic topic;
  | 	private Topic [] topicArr;
  | 	private TopicConnection connection;
  | 	private TopicSession session;
  | 	private TopicSession sessionPub;
  | 	private TopicSubscriber subscriber;
  | 	private TopicPublisher [] publisher;
  | 	private int publisherCount = 5;
  | 
  | 	public static void main(String[] args) 
  | 	{
  | 		MyTest obj = new MyTest();
  | 		obj.init();
  | 		obj.subscribe();
  | 		obj.publish();
  | 		//obj.cleanup();
  | 	}
  | 	
  | 	public void init()
  | 	{
  | 		try
  | 		{
  | 			destinationName = "/topic/testTopic";
  | 			InitialContext ic = new InitialContext();
  | 			cf = (TopicConnectionFactory)ic.lookup("/ClusteredConnectionFactory");
  | 			topic = (Topic)ic.lookup(destinationName);
  | 			connection = cf.createTopicConnection("admin", "admin");
  | 			connection.setClientID("MyClientID");
  | 			session = connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
  | 			sessionPub = connection.createTopicSession(false, TopicSession.AUTO_ACKNOWLEDGE);
  | 			subscriber = session.createDurableSubscriber(topic, "MyName");
  | 			publisher = new TopicPublisher[publisherCount];
  | 			topicArr = new Topic[publisherCount];
  | 			for (int j=0; j<publisherCount; j++)
  | 			{
  | 				topicArr[j] = (Topic)ic.lookup(destinationName);
  | 				publisher[j] = sessionPub.createPublisher(topicArr[j]);
  | 			}
  | 			ic = null;
  | 			connection.start();
  | 		}
  | 		catch(JMSException jmse)
  | 		{
  | 			jmse.printStackTrace();
  | 		}
  | 		catch(NamingException ne)
  | 		{
  | 			ne.printStackTrace();
  | 		}
  | 		catch(Exception e)
  | 		{
  | 			e.printStackTrace();
  | 		}
  | 	}
  | 	
  | 	private class SubscriberMessages extends Thread
  | 	{
  | 		public void run()
  | 		{
  | 			try
  | 			{
  | 				while (true)
  | 				{
  | 					TextMessage message = (TextMessage)subscriber.receive(5000);
  | 					if (message != null)
  | 					{
  | 						String msgStr = message.getText();
  | 						System.out.println(msgStr);
  | 					}				
  | 					message = null;
  | 				}
  | 			}
  | 			catch(JMSException jmse)
  | 			{
  | 				jmse.printStackTrace();
  | 			}
  | 			catch(Exception e)
  | 			{
  | 				e.printStackTrace();
  | 			}
  | 		}
  | 		}
  | 	
  | 	public void subscribe()
  | 	{
  | 		SubscriberMessages sObj = new SubscriberMessages();
  | 		sObj.start();
  | 	}
  | 	
  | 	private class PublishMessages extends Thread
  | 	{
  | 		private int pubID = 0;
  | 		public PublishMessages(int id)
  | 		{
  | 			pubID = id;
  | 		}
  | 		public void run()
  | 		{
  | 			try
  | 			{
  | 				String msgStr = "Hello World";
  | 				for (int i = 0; i< 1000; i++)
  | 				{
  | 					TextMessage tm = session.createTextMessage(msgStr + i);
  | 					publisher[pubID].publish(tm);
  | 				}			
  | 			}
  | 			catch(JMSException jmse)
  | 			{
  | 				jmse.printStackTrace();
  | 			}
  | 			catch(Exception e)
  | 			{
  | 				e.printStackTrace();
  | 			}
  | 		}
  | 	}
  | 	
  | 	public void publish() 
  | 	{
  | 		for (int i=0; i<publisherCount; i++)
  | 		{
  | 			PublishMessages pObj = new PublishMessages(i);
  | 			pObj.start();
  | 		}
  | 		
  | 	}
  | 	
  | 	public void cleanup()
  | 	{
  | 		try
  | 		{
  | 			try
  | 			{
  | 				if (subscriber != null)
  | 					subscriber.close();
  | 			}
  | 			catch(Exception e)
  | 			{
  | 			}
  | 			try
  | 			{
  | 				for (int j=0; j<publisherCount; j++)
  | 				{
  | 					if (publisher[j] != null)
  | 						publisher[j].close();
  | 				}
  | 			}
  | 			catch(Exception e)
  | 			{
  | 			}
  | 			try
  | 			{
  | 				if (connection != null)
  | 					connection.stop();
  | 			}
  | 			catch(Exception e)
  | 			{
  | 			}				
  | 			try
  | 			{
  | 				if (session != null)
  | 					session.close();
  | 			}
  | 			catch(Exception e)
  | 			{
  | 			}
  | 			try
  | 			{
  | 				if (connection != null)
  | 					connection.close();
  | 			}
  | 			catch(Exception e)
  | 			{
  | 			}
  | 		}
  | 
  | 		catch(Exception e)
  | 		{
  | 			e.printStackTrace();
  | 		}
  | 	}
  | }
  | 

View the original post : http://www.jboss.com/index.html?module=bb&op=viewtopic&p=4168598#4168598

Reply to the post : http://www.jboss.com/index.html?module=bb&op=posting&mode=reply&p=4168598