import java.util.Random;
import javax.jms.*;
import javax.naming.*;

public class TopicSelectors 
{
    static String syncQueueName = "queue1";
    static String factoryName = "ConnectionFactory";    

    String topicName = null;
    static final String MESSAGE_TYPES[] = 
    {"Nation/World", "Metro/Region", "Business",
     "Sports", "Living/Arts", "Opinion",
     // always last type
     "Finished"
    };
    static final int NUM_OF_MSG_TYPES = MESSAGE_TYPES.length - 1;
    static final int NUM_SUBSCRIBERS = 3;

    static final String END_OF_MESSAGE_STREAM_TYPE = 
    	MESSAGE_TYPES[NUM_OF_MSG_TYPES];//"Finished"
  
    static public class Barrier{
        boolean done = false;
	
        public void waitTillDone(){
            synchronized (this){
            	while (! done){
				    try{
					    this.wait();
					} 
				    catch (InterruptedException ie) {}
            	}
            }
        }
        
        public void allDone(){
            synchronized (this){
            	done = true;
            	this.notify();
            }
        }
    }

    /**
     * receives a given number of messages (indicated by count) from the controlQueue 
     * (consequently blocks the caller until the expected number of messages are received on the queue)
     * 		> called by the Publisher
     * @param prefix: indicator of the entity (e.g. Publisher) waiting to receive the number of expected messages
     * @param count: the number of expected messages to receive
     * @throws Exception
     */
    public static void receiveSynchronizeMessages (String prefix, int count) throws Exception{
        
    	Context                 context = null;
    	QueueConnectionFactory  queueConnectionFactory = null;
        QueueConnection         queueConnection = null;
        QueueSession            queueSession = null;
        Queue                   controlQueue = null;
        QueueReceiver           queueReceiver = null;
	
        //create a queueConnection and obtain the controlQueue (e.g. named "queue1")
        try{	 
        	context = new InitialContext();
        	queueConnectionFactory = (javax.jms.QueueConnectionFactory) context.lookup(factoryName);
        	queueConnection = queueConnectionFactory.createQueueConnection( "admin", "joram" );
        	queueSession = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
        	controlQueue = (javax.jms.Queue) context.lookup (syncQueueName);
        	queueConnection.start();
	    } 
        catch (Exception e){
        	System.out.println("Connection problem: " + e.toString());
        	if (queueConnection != null){
				try{
					queueConnection.close();
				} 
				catch (JMSException ee) {}
        	}
        	throw e;
	    } 

        //create a queueReceiver for the controlQueue and 
        //	have the queueReceiver receive the expected number of messages (the receiver will be blocked while waiting)
        try {
        	System.out.println(prefix + "Receiving synchronize messages from "
				   + syncQueueName + "; count = " + count);
        	queueReceiver = queueSession.createReceiver(controlQueue);
        	while (count > 0){
        		queueReceiver.receive();
        		count--;
        		System.out.println(prefix 
					   + "Received synchronize message; expect " 
					   + count + " more");
		    }
	    } 
        catch (JMSException e){
        	System.out.println("Exception occurred: " + e.toString());
        	throw e;
	    } 
        finally{
        	if (queueConnection != null){
				try{
					queueConnection.close();
				} 
				catch (JMSException e) {}
        	}
	    }
    }
    
    /**
     * sends the message "synchronize" on the controlQueue (named "queue1")
     * 		> called by each Subscriber
     * @param prefix: indicator of who is sending this message - used for logging purposes only
     * @throws Exception
     */
    public static void sendSynchronizeMessage(String prefix) throws Exception{
    	
    	Context                 context = null;
    	QueueConnectionFactory  queueConnectionFactory = null;
        QueueConnection         queueConnection = null;
        QueueSession            queueSession = null;
        Queue                   controlQueue = null;
        QueueSender             queueSender = null;
        TextMessage             message = null;
	
        try{
			context = new InitialContext();
			queueConnectionFactory = 
			    (javax.jms.QueueConnectionFactory) context.lookup(factoryName);
			queueConnection = queueConnectionFactory.createQueueConnection( "admin", "joram" );
			queueSession = queueConnection.createQueueSession
			    (false, Session.AUTO_ACKNOWLEDGE);
			controlQueue = (javax.jms.Queue) context.lookup (syncQueueName);
	    } 
        catch (Exception e){
        	System.out.println("Connection problem: " + e.toString());
			if (queueConnection != null){
				try{
					queueConnection.close();
				} 
				catch (JMSException ee) {}
			}
			throw e;
	    } 
	
        try{
			queueSender = queueSession.createSender(controlQueue);
			message = queueSession.createTextMessage();
			message.setText("synchronize");
			System.out.println(prefix + "Sending synchronize message to " + syncQueueName);
			queueSender.send(message);
	    } 
        catch (JMSException e){
        	System.out.println("Exception occurred: " + e.toString());
        	throw e;
	    } 
        finally{
			if (queueConnection != null) 
			    {
				try{
					queueConnection.close();
				} 
				catch (JMSException e) {}
			}
		}
    }
    
    public class Publisher extends Thread{
        final int NUM_SUBSCRIBERS;
        
        public Publisher(int numSubscribers){
            NUM_SUBSCRIBERS = numSubscribers;
        }

        /* (non-Javadoc)
         * @see java.lang.Thread#run()
         * 
         * - waits until all Subscribers have initialised 
         * 		(by calling the method: receiveSynchronizeMessages(count) 
         * 			- blocking until the expected number of messages have been received from the controlQueue))
         * - creates a topicPublisher for the given topic 
         * - publishes a predefined number of messages with random message types on the given topic
         * - publishes the final message on the given topic
         */
        public void run(){ 
        	Context                 context = null;
            TopicConnectionFactory  topicConnectionFactory = null;
            TopicConnection         topicConnection = null;
            TopicSession            topicSession = null;
            Topic                   topic = null;
            TopicPublisher          topicPublisher = null;
            TextMessage             message = null;
            int                     numMsgs = NUM_OF_MSG_TYPES * 5;
            String                  messageType = null;
            Random                  rgen = new Random();
            
            //create a topicSession and obtain the topic with the given name (topicName)
            try{
            	context = new InitialContext();
            	topicConnectionFactory = (javax.jms.TopicConnectionFactory) context.lookup(factoryName);
            	topicConnection = topicConnectionFactory.createTopicConnection( "admin", "joram" );
            	topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
            	topic = (javax.jms.Topic) context.lookup(topicName);
            } 
            catch (Exception e){
            	System.out.println("Connection problem: " + e.toString());
            	if (topicConnection != null){
            		try{
            			topicConnection.close();
            		} 
            		catch (JMSException ee) {}
            	}
            	System.exit(1);
            }
            
            try{
            	//wait until all subscribers have been initiated by calling receiveSynchronizeMessages( NUM_SUBSCRIBERS )
            	try{
            		TopicSelectors.receiveSynchronizeMessages("PUBLISHER THREAD: ", NUM_SUBSCRIBERS);
            	} 
            	catch (Exception e){
            		System.out.println("Queue probably missing: " + e.toString());
            		if (topicConnection != null){
            			try{
            				topicConnection.close();
            			} 
            			catch (JMSException ee) {}
            		}
            		System.exit(1);
            	}

	            //create a Publisher for the given topic (with topicName)
			    topicPublisher = topicSession.createPublisher(topic);
			    message = topicSession.createTextMessage();
			    
			    //publishes a predefined number of messages of random message types on the given topic
			    for (int i = 0; i < numMsgs; i++){
				    //selects a random message type
			    	messageType = MESSAGE_TYPES[rgen.nextInt(NUM_OF_MSG_TYPES)];
			    	//creates a message of the selected type
				    message.setStringProperty("NewsType", messageType);
				    message.setText("Item " + i + ": " + messageType);
				    System.out.println("PUBLISHER THREAD: Setting message text to: " 
						       + message.getText());
				    topicPublisher.publish(message);
				}
		    
			    //publishes the final message on the given topic
			    message.setStringProperty("NewsType", END_OF_MESSAGE_STREAM_TYPE);
			    message.setText("That's all the news for today.");
			    System.out.println("PUBLISHER THREAD: Setting message text to: " + message.getText());
			    topicPublisher.publish(message);
            } 
            catch (JMSException e){
            	System.out.println("Exception occurred: " + e.toString());
            } 
            finally{
            	if (topicConnection != null){
            		try{
            			topicConnection.close();
            		} 
            		catch (JMSException e) {}
            	}
            }
        }//end run
    }//end Publisher
    
    
    public class Subscriber extends Thread{
        String whatKind; 
        int    subscriberNumber;
	
        /**
         * @author ada
         * The Subscriber's MessageListeners - receives all messages on the given topic
         * 	- prints all received messages
         *  - lifts the barrier upon receiving the final message
         */
        private class NewsListener implements MessageListener{
            final Barrier myBarrier = new Barrier();

            public void onMessage(Message inMessage){
                TextMessage msg = (TextMessage) inMessage;
                String newsType;
		
                try{
                	//print out the message
                	System.out.println("SUBSCRIBER " + subscriberNumber 
					   + " THREAD: Message received: " + msg.getText());
                	//get the message type
                	newsType = msg.getStringProperty("NewsType");
                	//if this is the final message then lift this listener's barrier
                	if (newsType.equals(TopicSelectors.END_OF_MESSAGE_STREAM_TYPE)){
                			System.out.println("SUBSCRIBER " + subscriberNumber 
                					+ " THREAD: Received finished-publishing message");
                			myBarrier.allDone();
                	}
                } 
                catch(JMSException e){
                	System.out.println("Exception in onMessage(): " + e.toString());
                }
            }
        }
        
        public Subscriber(String str, int num){
            this.whatKind = str;
            this.subscriberNumber = num + 1;
        }
 
        /* (non-Javadoc)
         * @see java.lang.Thread#run()
         * 
         * - creates a TopicSubscriber for a given topic and sets an instance of the newsListener as the topic's MessageListener
         * - sends a synchronization message to the controlQueue
         * - sets in the barrier (=> blocked/alive until final message arrives)
         */
        public void run(){
        	
        	Context                 context = null;
            TopicConnectionFactory  topicConnectionFactory = null;
            TopicConnection         topicConnection = null;
            TopicSession            topicSession = null;
            Topic                   topic = null;
            String                  selector = null;
            TopicSubscriber         topicSubscriber = null;
            NewsListener            newsListener = new NewsListener();
	    
            //create a topicSession and obtain the topic with the given topicName
            try{
			    context = new InitialContext();
			    topicConnectionFactory = (javax.jms.TopicConnectionFactory) context.lookup(factoryName);
			    topicConnection = topicConnectionFactory.createTopicConnection( "admin", "joram" );
			    topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
			    topic = (javax.jms.Topic) context.lookup(topicName);
			} 
		    catch (Exception e){
			    System.out.println("Connection problem: " + e.toString());
			    if (topicConnection != null){
				    try{
					    topicConnection.close();
					} 
				    catch (JMSException ee) {}
				}
			    System.exit(1);
			} 
            
		    //create a topicSubscriber for the obtained topic (with the given topicName) and set the newsListener as MessageListener
            try{
			    selector = new String("NewsType = '" + whatKind + "'" + 
						  " OR NewsType = '" + END_OF_MESSAGE_STREAM_TYPE + "'");
			    System.out.println("SUBSCRIBER " + subscriberNumber 
					       + " THREAD: selector is \"" + selector + "\"");
			    topicSubscriber = topicSession.createSubscriber(topic, selector, false);
			    topicSubscriber.setMessageListener(newsListener);
			    topicConnection.start();
		    
			    try{
				    TopicSelectors.sendSynchronizeMessage("SUBSCRIBER " + subscriberNumber + " THREAD: ");
				} 
			    catch (Exception e){
			    	System.out.println("Queue probably missing: " + e.toString());
			    	if (topicConnection != null){
					    try {
						    topicConnection.close();
						} 
					    catch (JMSException ee) {}
			    	}
			    	System.exit(1);
			    }
		    
			    
			    //waits until the final message arrives
			    newsListener.myBarrier.waitTillDone();
            } 
		    catch (JMSException e){
			    System.out.println("Exception occurred: " + e.toString());
			} 
		    finally{
		    	if (topicConnection != null){
				    try{
					    topicConnection.close();
					} 
				    catch (JMSException e) {}
		    	}
		    }  
		    
        }//end run
    }//end Subscriber class
    
    public void run_threads(){
    	
    	//create subscriber array
        Subscriber subscriberArray[] = new Subscriber[NUM_SUBSCRIBERS];
        //create one publisher
        Publisher  publisher = new Publisher(NUM_SUBSCRIBERS);
        
        //instantiate and start subscribers (listening to different messages)
		for (int i = 0; i < subscriberArray.length; i++){
			subscriberArray[i] = new Subscriber(MESSAGE_TYPES[2+i], i);
			subscriberArray[i].start();        
		}
		
		//start publisher
        publisher.start();
        
        for (int i = 0; i < subscriberArray.length; i++){
        	try{
        		//wait for all the subscribers' threads to terminate
        		subscriberArray[i].join();
		    } 
        	catch (InterruptedException e) {}
	    }
        
	        try{
	        	//wait for the publisher's thread to terminate 
	        	publisher.join();
		    } 
	        catch (InterruptedException e) {}
    }
    
    public static void main(String[] args){
        TopicSelectors ts = new TopicSelectors();
	
        if (args.length != 1){
        	System.out.println("Usage: TopicSelectors <topic_name>");
        	System.exit(1);
	    }
    	
        ts.topicName = new String(args[0]);
        System.out.println("Topic name is " + ts.topicName);
	
    	ts.run_threads();

    	System.out.println ("Game Over");
    	System.exit (1);
    }
}
