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 int NUM_OF_MSG_TYPES = MESSAGE_TYPES.length - 1;
    static int NUM_SUBSCRIBERS = 3;

    static final String END_OF_MESSAGE_STREAM_TYPE = 
	MESSAGE_TYPES[NUM_OF_MSG_TYPES];
  
    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();
		}
        }
    }

    //TODO
    //    public static void receiveSynchronizeMessages (String prefix, int count) 
    //    public static void sendSynchronizeMessage(String prefix) throws Exception 
    
    public class Publisher extends Thread 
    {
        final int NUM_SUBSCRIBERS;
        
        public Publisher(int numSubscribers) 
	{
            NUM_SUBSCRIBERS = numSubscribers;
        }

        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();

            try 
		{
		    context = new InitialContext();
		    topicConnectionFactory = 
			(javax.jms.TopicConnectionFactory) context.lookup(factoryName);
		    topicConnection = 
			topicConnectionFactory.createTopicConnection();
		    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 
		{
		    topicPublisher = topicSession.createPublisher(topic);
		    message = topicSession.createTextMessage();
		    for (int i = 0; i < numMsgs; i++) 
			{
			    messageType = MESSAGE_TYPES[rgen.nextInt(NUM_OF_MSG_TYPES)];
			    message.setStringProperty("NewsType", messageType);
			    message.setText("Item " + i + ": " + messageType);
			    System.out.println("PUBLISHER THREAD: Setting message text to: " 
					       + message.getText());
			    topicPublisher.publish(message);
			}
		    
		    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) {}
			}
		}
        }
    }
    
    public class Subscriber extends Thread 
    {
        String whatKind;
        int    subscriberNumber;
	
        private class NewsListener implements MessageListener 
	{
            final Barrier myBarrier = new Barrier();
	    
            public void onMessage(Message inMessage) 
	    {
        //TODO
		// Implementer le listener
            }
        }
        
        public Subscriber(String str, int num) 
	{
            whatKind = str;
            subscriberNumber = num + 1;
        }
 
        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();
	    
            try 
		{
		    context = new InitialContext();
		    topicConnectionFactory = 
			(javax.jms.TopicConnectionFactory) context.lookup(factoryName);
		    topicConnection = 
			topicConnectionFactory.createTopicConnection();
		    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 
		{
		    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();
		    
		    //TODO
		    // Synchronisation avec l'emetteur
		    // Synchronisation avec le listener
		} 
	    catch (JMSException e) 
		{
		    System.out.println("Exception occurred: " + e.toString());
		} 
	    finally 
		{
		    if (topicConnection != null) 
			{
			    try 
				{
				    topicConnection.close();
				} 
			    catch (JMSException e) {}
			}
		}   	    
        }
    }
    
    public void run_threads() 
    {
        Subscriber subscriberArray[] = new Subscriber[NUM_SUBSCRIBERS];
        Publisher  publisher = new Publisher(NUM_SUBSCRIBERS);

	for (int i = 0; i < subscriberArray.length; i++) 
	    {
		subscriberArray[i] = new Subscriber(MESSAGE_TYPES[2+i], i);
		subscriberArray[i].start();        
	    }

        publisher.start();
        
        for (int i = 0; i < subscriberArray.length; i++) 
	    {
		try 
		    {
			subscriberArray[i].join();
		    } 
		catch (InterruptedException e) {}
	    }
        
        try 
	    {
		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);
    }
}
