import javax.jms.*;
import javax.naming.*;

public class P2PTest {
    
    public static Connection connection;
    public static Destination destination;
    public static long nMsgs = 1234567890;

    public static void main(String[] args) {

        String smqpURL = "smqp://127.0.0.1:4001";
        String qcfName = "QueueConnectionFactory";
        String queueName = "testqueue";

        if (args.length > 0 && args.length < 4) {
            System.out.println();
            System.out.println("Usage: java P2PTest <smqpURL> <qcfName> <queueName> <nMsgs>\n");
            System.out.println("You may either specify all parameters or none. If you specify none, ");
            System.out.println("the following defaults are used: \n");
            System.out.println("   "+smqpURL+" "+qcfName+" "+queueName+" "+nMsgs+"\n");
            System.exit(-1);
        }

        if (args.length == 4) {
            smqpURL = args[0];
            qcfName = args[1];
            queueName = args[2];
            nMsgs = Long.parseLong(args[3]);
        }

        System.out.println();
        System.out.println("The following parameters are used:\n");
        System.out.println("SMQP-URL       : "+smqpURL);
        System.out.println("QCF Name       : "+qcfName);
        System.out.println("Queue Name     : "+queueName);
        System.out.println("Number Messages: "+nMsgs);
        System.out.println();

        final ConnectionFactory connectionFactory;
        try {
            final java.util.Properties env = new java.util.Properties();
            env.put(Context.INITIAL_CONTEXT_FACTORY,"com.swiftmq.jndi.InitialContextFactoryImpl");
            env.put(Context.PROVIDER_URL,smqpURL);
            final InitialContext ctx = new InitialContext(env);
            try {
                destination =(Destination)ctx.lookup(queueName);
                connectionFactory = (ConnectionFactory)ctx.lookup(qcfName);
            } finally {
                ctx.close();
            }
        } catch (NamingException e) {
            System.err.println("NamingException: "+e);
            return;
        }

        try {
            connection = connectionFactory.createConnection();
            connection.start();
            connection.setExceptionListener(new OnException());
        } catch (JMSException e) {
            System.err.println("JMSException while creating connection: "+e);
            return;
        }

        new Thread(new Consumer(), "Consumer").start();
        new Thread(new Producer(), "Producer").start();
        System.out.println("Main finished");
    }

    public static class Producer implements Runnable {
        public void run() {
            try {
                final Session session = connection.createSession(true, 0);
                final MessageProducer messageProducer = session.createProducer(destination);
                for (long i=1; i<=nMsgs; i++) {
                    messageProducer.send(session.createTextMessage("request #"+i));
                    Thread.sleep(1000);
                    System.out.println("Producer committing request #"+i);
                    session.commit();
                    System.out.println("Producer committed");
                }
            } catch (JMSException e) {
                System.err.println("JMSException from Producer: "+e);
            } catch (InterruptedException e) {
                System.err.println("Producer interrupted: "+e);
            }
            System.out.println("Producer finished");
        }
    }

    public static class Consumer implements Runnable {
        public void run() {
            try {
                final Session session = connection.createSession(true, 0);
                final MessageConsumer messageConsumer = session.createConsumer(destination);
                for (;;) {
                    TextMessage msg = (TextMessage)messageConsumer.receive(1500);
                    if (msg == null) {
                        System.out.println("Consumer received null!");
                    } else {
                        System.out.println("Consumer received "+msg.getText());
                    }
                    session.commit();
                    System.out.println("Consumer committed");
                }
            } catch (JMSException e) {
                System.err.println("JMSException from Consumer: "+e);
            }
            System.out.println("Consumer finished");
        }
    }

    public static class OnException implements ExceptionListener {
        public void onException(JMSException e) {
            System.out.println("onException: "+e);
        }
    }
}