transactedexample.java

来自「j2ee API 开发重要工具 免费下载 欢迎使用」· Java 代码 · 共 878 行 · 第 1/3 页

                return;
            }

            /* 
             * Message is an order confirmation message from a 
             * supplier.
             */
            int orderNumber = -1;

            try {
                MapMessage component = (MapMessage) message;

                /* 
                 * Process the order confirmation message and 
                 * commit the transaction.
                 */
                orderNumber = 
                    component.getInt("VendorOrderNumber");
                Order order = 
                    Order.getOrder(orderNumber).processSubOrder(component);
                session.commit();
                
                /*
                 * If this message is the last supplier message, 
                 * send message to Retailer and commit
                 * transaction.
                 */
                if (! order.isPending()) {
                    System.out.println("Vendor: Completed " +
                        "processing for order " +
                        order.orderNumber);
                    Queue replyQueue = 
                        (Queue) order.order.getJMSReplyTo();
                    MessageProducer producer = 
                        session.createProducer(replyQueue);
                    MapMessage retailerConfirmMessage = 
                        session.createMapMessage();
                    if (order.isFulfilled()) {
                        retailerConfirmMessage.setBoolean("OrderAccepted",
                            true);
                        System.out.println("Vendor: sent " + 
                            order.quantity + " computer(s)");
                    } else if (order.isCancelled()) {
                        retailerConfirmMessage.setBoolean("OrderAccepted",
                            false);
                        System.out.println("Vendor: unable to " +
                            "send " + order.quantity + 
                            " computer(s)");
                    }
                    producer.send(retailerConfirmMessage);
                    session.commit();
                    System.out.println("  Vendor: committed " +
                        "transaction 2");
                }
            } catch (JMSException je) {
                je.printStackTrace();
                try {
                    session.rollback();
                } catch (JMSException je2) {}
            } catch (Exception e) {
                e.printStackTrace();
                try {
                    session.rollback();
                } catch (JMSException je2) {}
            }
        }
    }

    /**
     * The GenericSupplier class receives an item order from the
     * vendor and sends a message accepting or refusing it.
     */
    public static class GenericSupplier extends Thread {
        final String  PRODUCT_NAME;
        final String  IN_ORDER_TOPIC;
        int           quantity = 0;

        /**
         * Constructor.  Instantiates the supplier as the
         * supplier for the kind of item to be ordered.
         *
         * @param itemName    the name of the item being ordered
         * @param inTopic     the topic from which the order is
         *                    obtained
         */
        public GenericSupplier(String itemName, String inTopic) {
            PRODUCT_NAME = itemName;
            IN_ORDER_TOPIC = inTopic;
        }

        /**
         * Checks to see if there are enough items in inventory.
         * Rather than go to a database, it generates a random
         * number related to the order quantity, so that some of
         * the time there won't be enough in stock.
         *
         * @return    the number of items in inventory
         */
        public int checkInventory() {
            Random  rgen = new Random();

            return (rgen.nextInt(quantity * 5));
        }

        /**
         * Runs the thread.
         */
        public void run() {
            ConnectionFactory  connectionFactory = null;
            Connection         connection = null;
            Session            session = null;
            Topic              orderTopic = null;
            MessageConsumer    receiver = null;
            Message            inMessage = null;
            MapMessage         orderMessage = null;
            MapMessage         outMessage = null;

            try {
                connectionFactory =
                    SampleUtilities.getConnectionFactory();
                connection = 
                    connectionFactory.createConnection();
                session = connection.createSession(true, 0);
                orderTopic =
                    SampleUtilities.getTopic(IN_ORDER_TOPIC, 
                        session);
            } catch (Exception e) {
                System.err.println("Connection problem: " + 
                    e.toString());
                System.err.println("Program assumes three " +
                    "queues named AQueue BQueue CQueue and " +
                    "one topic named OTopic");
                if (connection != null) {
                    try {
                        connection.close();
                    } catch (JMSException ee) {}
                }
                System.exit(1);
            }
            
            /*
             * Create receiver for order topic and start message 
             * delivery.
             */
            try {
                receiver = session.createConsumer(orderTopic);
                connection.start();
            } catch (JMSException je) {}

            /*
             * Keep checking supplier order topic for order
             * request until end-of-message-stream message is
             * received. Receive order and send an order
             * confirmation as one transaction.
             */
            while (true) {
                try {
                    inMessage = receiver.receive();
                    if (inMessage instanceof MapMessage) {
                        orderMessage = (MapMessage) inMessage;
                    } else {
                        /*
                         * Message is an end-of-message-stream 
                         * message. Send a similar message to
                         * reply queue, commit transaction, then
                         * stop processing orders by breaking out
                         * of loop.
                         */
                        MessageProducer producer =
                            session.createProducer(inMessage.getJMSReplyTo());
                        producer.send(session.createMessage());
                        session.commit();
                        break;
                    }

                    /*
                     * Extract quantity ordered from order
                     * message.
                     */
                    quantity = orderMessage.getInt("Quantity");
                    System.out.println(PRODUCT_NAME + 
                        " Supplier: Vendor ordered " + quantity +
                        " " + PRODUCT_NAME + "(s)");

                    /*
                     * Create sender and message for reply queue.
                     * Set order number and item; check inventory
                     * and set quantity available.
                     * Send message to vendor and commit
                     * transaction.
                     */
                    MessageProducer producer =
                        session.createProducer((Queue) 
                            orderMessage.getJMSReplyTo());
                    outMessage = session.createMapMessage();
                    outMessage.setInt("VendorOrderNumber",
                       orderMessage.getInt("VendorOrderNumber"));
                    outMessage.setString("Item", PRODUCT_NAME);
                    int numAvailable = checkInventory();
                    if (numAvailable >= quantity) {
                        outMessage.setInt("Quantity", quantity);
                    } else {
                        outMessage.setInt("Quantity", 
                            numAvailable);
                    }
                    producer.send(outMessage);
                    System.out.println(PRODUCT_NAME + 
                        " Supplier: sent " +
                        outMessage.getInt("Quantity") + " " +
                        outMessage.getString("Item") + "(s)");
                    session.commit();
                    System.out.println("  " + PRODUCT_NAME +
                        " Supplier: committed transaction");
                } catch (Exception e) {
                    System.err.println(PRODUCT_NAME +  
                        " Supplier: Exception occurred: " + 
                        e.toString());
                    e.printStackTrace();
                } 
            }

            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {}
            }
        }
    }

    /**
     * Creates the Retailer and Vendor classes and the two
     * supplier classes, then starts the threads.
     *
     * @param quantity    the quantity specified on the command
     *                    line
     */
    public static void run_threads(int quantity) {
        Retailer         r = new Retailer(quantity);
        Vendor           v = new Vendor();
        GenericSupplier  ms = new GenericSupplier("Monitor", 
                                 supplierOrderTopicName);
        GenericSupplier  ss = new GenericSupplier("Hard Drive",
                                 supplierOrderTopicName);

        ms.start();
        ss.start();
        r.start();
        v.start();
        try {
            ms.join();
            ss.join();
            r.join();
            v.join();
        } catch (InterruptedException e) {}
    }

    /**
     * Reads the order quantity from the command line, then
     * calls the run_threads method to execute the program
     * threads.
     *
     * @param args    the quantity of computers being ordered
     */
    public static void main(String[] args) {
        TransactedExample  te = new TransactedExample();
        int                quantity = 0;

        if (args.length != 1) {
            System.out.println("Program takes numerical argument.");
            System.err.println("Program assumes three " +
                "queues named AQueue BQueue CQueue and " +
                "one topic named OTopic");
            System.exit(1);
        }
        te.vendorOrderQueueName = new String("AQueue");
        te.retailerConfirmQueueName = new String("BQueue");
        te.supplierOrderTopicName = new String("OTopic");
        te.vendorConfirmQueueName = new String("CQueue");
        quantity = (new Integer(args[0])).intValue();
        System.out.println("Quantity to be ordered is " + 
            quantity);
        if (quantity > 0) {
            te.run_threads(quantity);
            System.exit(0);
        } else {
            System.out.println("Quantity must be positive and " +
                "nonzero");
            System.exit(1);
        }
    }
}