testbug705640.java

实现了Jms的服务器源码,支持多种适配器,DB,FTP,支持多种数据库

        }
    }

    /** 
     * Test the bug for topics
     */
    public void testTopicSimpleOrder() 
    throws Exception {
        final Object waitForReceiver = new Object();
        final String name = "testTopic705640";
        final TopicConnection connection = ClientHelper.getTopicConnection();
        try {
            connection.start();

            // set up and start the consumer before the publisher
            Thread consumer = new Thread(new Runnable() {
                    // implementation of Runnable.run
                    public void run() {          
                        try {
                            TopicSession session = connection.createTopicSession(
                                false, Session.AUTO_ACKNOWLEDGE);
                            Topic topic = session.createTopic(name);
                            TopicSubscriber subscriber = session.createSubscriber(topic);

                            // receive 10 messages and ensure they are in
                            // the correct order
                            int last_index = -1;
                            for (int index = 0; index < 10; index++) {
                                MapMessage msg = (MapMessage)subscriber.receive();
                                int value = msg.getInt("index");
                                assertTrue(value == (last_index + 1));
                                last_index = value;
                            }

                            Thread.sleep(3000);
                            synchronized(waitForReceiver) {
                                waitForReceiver.notifyAll();
                            }
                        } catch (Exception exception) {
                            exception.printStackTrace();
                            assertTrue("TestBug705640.testTopic : " + 
                                       exception.getMessage(), false);
                        }
                    }
                    });

            // start the receiver and then wait before starting the 
            // publisher
            consumer.start();  
            Thread.sleep(2000);

            // set up the publisher to publish the 10 messages
            TopicSession session = connection.createTopicSession(
                false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(name);
            TopicPublisher publisher = session.createPublisher(topic);
            publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            // send 10 messages and wait for 3 seconds
            for (int index = 0; index < 10; index++) {
                MapMessage msg = session.createMapMessage();
                msg.setInt("index", index);
                publisher.publish(msg);
            }

            try {
                synchronized(waitForReceiver) {
                    waitForReceiver.wait();
                }
            } catch (InterruptedException exception) {
                exception.printStackTrace();
                assertTrue("TestBug705640.testTopicSimpleOrder : Failed to receive the " +
                           "10 messages in the specified time frame",
                           false);
            }
        } catch (Exception exception) {
            exception.printStackTrace();
            assertTrue("TestBug705640.testTopicSimpleOrder : " + 
                       exception.getMessage(), false);
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException ignore) {
                }
            }
        }
    }


    /** 
     * Test the bug for topic using priority
     */
    public void testTopicPrtyOrder() 
    throws Exception {
        final Object waitForReceiver = new Object();
        final String name = "testTopic705640";
        final TopicConnection connection = ClientHelper.getTopicConnection();
        try {
            connection.start();

            // set up and start the consumer before the publisher
            Thread consumer = new Thread(new Runnable() {
                    // implementation of Runnable.run
                    public void run() {          
                        try {
                            TopicSession session = connection.createTopicSession(
                                false, Session.AUTO_ACKNOWLEDGE);
                            Topic topic = session.createTopic(name);
                            TopicSubscriber subscriber = session.createSubscriber(topic);

                            // receive 10 messages and ensure they are in
                            // the correct order
                            boolean first = true;
                            int last_priority = 0;
                            for (int index = 0; index < 10; index++) {
                                MapMessage msg = (MapMessage)subscriber.receive();
                                int priority = msg.getJMSPriority();
                                if (first) {
                                    last_priority = priority;
                                    first = false;
                                } else {
                                    assertTrue(priority <= last_priority);
                                }
                            }

                            Thread.sleep(3000);
                            synchronized(waitForReceiver) {
                                waitForReceiver.notifyAll();
                            }
                        } catch (Exception exception) {
                            exception.printStackTrace();
                            assertTrue("TestBug705640.testTopic : " + 
                                       exception.getMessage(), false);
                        }
                    }
                    });

            // start the receiver and then wait before starting the 
            // publisher
            consumer.start();  
            Thread.sleep(2000);

            // set up the publisher to publish the 10 messages
            TopicSession session = connection.createTopicSession(
                false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(name);
            TopicPublisher publisher = session.createPublisher(topic);
            publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            // send 10 messages and wait for 3 seconds
            for (int index = 0; index < 10; index++) {
                MapMessage msg = session.createMapMessage();
                msg.setJMSPriority(index);
                publisher.publish(msg);
            }

            try {
                synchronized(waitForReceiver) {
                    waitForReceiver.wait();
                }
            } catch (InterruptedException exception) {
                exception.printStackTrace();
                assertTrue("TestBug705640.testTopicPrtyOrder : Failed to receive the " +
                           "10 messages in the specified time frame",  false);
            }
        } catch (Exception exception) {
            exception.printStackTrace();
            assertTrue("TestBug705640.testTopicPrtyOrder : " + 
                       exception.getMessage(), false);
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException ignore) {
                }
            }
        }
    }

    /** 
     * Test the bug for topic using selector
     **/
    public void testTopicSelector() 
    throws Exception {
        final Object waitForReceiver = new Object();
        final String name = "testTopic705640";
        final TopicConnection connection = ClientHelper.getTopicConnection();
        try {
            connection.start();

            // set up and start the consumer before the publisher
            Thread consumer = new Thread(new Runnable() {
                    // implementation of Runnable.run
                    public void run() {          
                        try {
                            TopicSession session = connection.createTopicSession(
                                false, Session.AUTO_ACKNOWLEDGE);
                            Topic topic = session.createTopic(name);
                            TopicSubscriber subscriber = session.createSubscriber(
                                topic, "1=1", false);

                            // receive 10 messages and ensure they are in
                            // the correct order
                            boolean first = true;
                            int last_priority = 0;
                            for (int index = 0; index < 10; index++) {
                                MapMessage msg = (MapMessage)subscriber.receive();
                                int priority = msg.getJMSPriority();
                                if (first) {
                                    last_priority = priority;
                                    first = false;
                                } else {
                                    assertTrue(priority <= last_priority);
                                }
                            }

                            Thread.sleep(3000);
                            synchronized(waitForReceiver) {
                                waitForReceiver.notifyAll();
                            }
                        } catch (Exception exception) {
                            exception.printStackTrace();
                            assertTrue("TestBug705640.testTopic : " + 
                                       exception.getMessage(), false);
                        }
                    }
                    });

            // start the receiver and then wait before starting the 
            // publisher
            consumer.start();  
            Thread.sleep(2000);

            // set up the publisher to publish the 10 messages
            TopicSession session = connection.createTopicSession(
                false, Session.AUTO_ACKNOWLEDGE);
            Topic topic = session.createTopic(name);
            TopicPublisher publisher = session.createPublisher(topic);
            publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            // send 10 messages and wait for 3 seconds
            for (int index = 0; index < 10; index++) {
                MapMessage msg = session.createMapMessage();
                msg.setJMSPriority(index);
                publisher.publish(msg);
            }

            try {
                synchronized(waitForReceiver) {
                    waitForReceiver.wait();
                }
            } catch (InterruptedException exception) {
                exception.printStackTrace();
                assertTrue("TestBug705640.testTopicSelector : Failed to receive the " +
                           "10 messages in the specified time frame",  false);
            }
        } catch (Exception exception) {
            exception.printStackTrace();
            assertTrue("TestBug705640.testTopicSelector : " + 
                       exception.getMessage(), false);
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException ignore) {
                }
            }
        }
    }
}