You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ri...@apache.org on 2007/08/30 14:19:42 UTC

svn commit: r571129 [8/15] - in /incubator/qpid/trunk/qpid/java: ./ broker/ broker/bin/ broker/distribution/src/main/assembly/ broker/etc/ broker/src/main/java/org/apache/log4j/ broker/src/main/java/org/apache/qpid/configuration/ broker/src/main/java/o...

Added: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedClientTestCase.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedClientTestCase.java?rev=571129&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedClientTestCase.java (added)
+++ incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedClientTestCase.java Thu Aug 30 05:19:31 2007
@@ -0,0 +1,906 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.sustained;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.client.AMQNoConsumersException;
+import org.apache.qpid.client.AMQNoRouteException;
+import org.apache.qpid.test.framework.distributedtesting.TestClient;
+import org.apache.qpid.interop.clienttestcases.TestCase3BasicPubSub;
+import org.apache.qpid.test.framework.TestUtils;
+
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * Implements test case 3, basic pub/sub. Sends/received a specified number of messages to a specified route on the
+ * default topic exchange, using the specified number of receivers connections. Produces reports on the actual number of
+ * messages sent/received.
+ *
+ * <p><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * <tr><td> Supply the name of the test case that this implements.
+ * <tr><td> Accept/Reject invites based on test parameters.
+ * <tr><td> Adapt to assigned roles.
+ * <tr><td> Send required number of test messages using pub/sub. <tr><td> Generate test reports.
+ * </table>
+ */
+public class SustainedClientTestCase extends TestCase3BasicPubSub implements ExceptionListener, MessageListener
+{
+    /** Used for debugging. */
+    private static final Logger log = Logger.getLogger(SustainedClientTestCase.class);
+
+    /** Used to log to the console. */
+    private static final Logger console = Logger.getLogger("SustainedTest");
+
+    /** The role to be played by the test. */
+    private Roles role;
+
+    /** The number of receivers connection to use. */
+    private int numReceivers;
+
+    /** The routing key to send them to on the default direct exchange. */
+    private Destination sendDestination;
+
+    /** The routing key to send updates to on the default direct exchange. */
+    private Destination sendUpdateDestination;
+
+    /** The connections to send/receive the test messages on. */
+    private Connection[] connection;
+
+    /** The sessions to send/receive the test messages on. */
+    private Session[] session;
+
+    /** The producer to send the test messages with. */
+    MessageProducer producer;
+
+    /** Adapter that adjusts the send rate based on the updates from clients. */
+    SustainedRateAdapter _rateAdapter;
+
+    /**  */
+    int _batchSize;
+
+    private static final long TEN_MILLI_SEC = 10000000;
+    private static final int DEBUG_LOG_UPATE_INTERVAL = 10;
+    private static final int LOG_UPATE_INTERVAL = 10;
+    private static final boolean SLEEP_PER_MESSAGE = Boolean.getBoolean("sleepPerMessage");
+
+    /**
+     * Should provide the name of the test case that this class implements. The exact names are defined in the interop
+     * testing spec.
+     *
+     * @return The name of the test case that this implements.
+     */
+    public String getName()
+    {
+        log.debug("public String getName(): called");
+
+        return "Perf_SustainedPubSub";
+    }
+
+    /**
+     * Assigns the role to be played by this test case. The test parameters are fully specified in the assignment
+     * message. When this method return the test case will be ready to execute.
+     *
+     * @param role              The role to be played; sender or receivers.
+     * @param assignRoleMessage The role assingment message, contains the full test parameters.
+     *
+     * @throws JMSException Any JMSException resulting from reading the message are allowed to fall through.
+     */
+    public void assignRole(Roles role, Message assignRoleMessage) throws JMSException
+    {
+        log.debug("public void assignRole(Roles role = " + role + ", Message assignRoleMessage = " + assignRoleMessage
+            + "): called");
+
+        // Take note of the role to be played.
+        this.role = role;
+
+        // Extract and retain the test parameters.
+        numReceivers = assignRoleMessage.getIntProperty("SUSTAINED_NUM_RECEIVERS");
+        _batchSize = assignRoleMessage.getIntProperty("SUSTAINED_UPDATE_INTERVAL");
+        String sendKey = assignRoleMessage.getStringProperty("SUSTAINED_KEY");
+        String sendUpdateKey = assignRoleMessage.getStringProperty("SUSTAINED_UPDATE_KEY");
+        int ackMode = assignRoleMessage.getIntProperty("ACKNOWLEDGE_MODE");
+        String clientName = assignRoleMessage.getStringProperty("CLIENT_NAME");
+
+        if (log.isDebugEnabled())
+        {
+            log.debug("numReceivers = " + numReceivers);
+            log.debug("_batchSize = " + _batchSize);
+            log.debug("ackMode = " + ackMode);
+            log.debug("sendKey = " + sendKey);
+            log.debug("sendUpdateKey = " + sendUpdateKey);
+            log.debug("role = " + role);
+        }
+
+        switch (role)
+        {
+        // Check if the sender role is being assigned, and set up a single message producer if so.
+        case SENDER:
+            console.info("Creating Sender");
+            // Create a new connection to pass the test messages on.
+            connection = new Connection[1];
+            session = new Session[1];
+
+            connection[0] = TestUtils.createConnection(TestClient.testContextProperties);
+            session[0] = connection[0].createSession(false, ackMode);
+
+            // Extract and retain the test parameters.
+            sendDestination = session[0].createTopic(sendKey);
+
+            connection[0].setExceptionListener(this);
+
+            producer = session[0].createProducer(sendDestination);
+
+            sendUpdateDestination = session[0].createTopic(sendUpdateKey);
+            MessageConsumer updateConsumer = session[0].createConsumer(sendUpdateDestination);
+
+            _rateAdapter = new SustainedRateAdapter(this);
+            updateConsumer.setMessageListener(_rateAdapter);
+
+            break;
+
+        // Otherwise the receivers role is being assigned, so set this up to listen for messages on the required number
+        // of receivers connections.
+        case RECEIVER:
+            console.info("Creating Receiver");
+            // Create the required number of receivers connections.
+            connection = new Connection[numReceivers];
+            session = new Session[numReceivers];
+
+            for (int i = 0; i < numReceivers; i++)
+            {
+                connection[i] = TestUtils.createConnection(TestClient.testContextProperties);
+                session[i] = connection[i].createSession(false, ackMode);
+
+                sendDestination = session[i].createTopic(sendKey);
+
+                sendUpdateDestination = session[i].createTopic(sendUpdateKey);
+
+                MessageConsumer consumer = session[i].createConsumer(sendDestination);
+
+                consumer.setMessageListener(new SustainedListener(clientName + "-" + i, _batchSize, session[i],
+                        sendUpdateDestination));
+            }
+
+            break;
+        }
+
+        // Start all the connection dispatcher threads running.
+        for (int i = 0; i < connection.length; i++)
+        {
+            connection[i].start();
+        }
+    }
+
+    /** Performs the test case actions.
+     * @param numMessages*/
+    public void start(int numMessages) throws JMSException
+    {
+        log.debug("public void start(): called");
+
+        // Check that the sender role is being performed.
+        switch (role)
+        {
+        // Check if the sender role is being assigned, and set up a single message producer if so.
+        case SENDER:
+            _rateAdapter.run();
+            break;
+        case RECEIVER:
+
+        }
+
+        // return from here when you have finished the test.. this will signal the controller and
+    }
+
+    public void terminate() throws JMSException, InterruptedException
+    {
+        if (_rateAdapter != null)
+        {
+            _rateAdapter.stop();
+        }
+    }
+
+    /**
+     * Gets a report on the actions performed by the test case in its assigned role.
+     *
+     * @param session The controlSession to create the report message in.
+     *
+     * @return The report message.
+     *
+     * @throws JMSException Any JMSExceptions resulting from creating the report are allowed to fall through.
+     */
+    public Message getReport(Session session) throws JMSException
+    {
+        log.debug("public Message getReport(Session controlSession): called");
+
+        // Close the test connections.
+        for (int i = 0; i < connection.length; i++)
+        {
+            connection[i].close();
+        }
+
+        Message report = session.createMessage();
+        report.setStringProperty("CONTROL_TYPE", "REPORT");
+
+        return report;
+    }
+
+    public void onException(JMSException jmsException)
+    {
+        Exception linked = jmsException.getLinkedException();
+
+        if (linked != null)
+        {
+            if (log.isDebugEnabled())
+            {
+                log.debug("Linked Exception:" + linked);
+            }
+
+            if ((linked instanceof AMQNoRouteException) || (linked instanceof AMQNoConsumersException))
+            {
+                if (log.isDebugEnabled())
+                {
+                    if (linked instanceof AMQNoConsumersException)
+                    {
+                        log.warn("No clients currently available for message:"
+                            + ((AMQNoConsumersException) linked).getUndeliveredMessage());
+                    }
+                    else
+                    {
+                        log.warn("No route for message");
+                    }
+                }
+
+                // Tell the rate adapter that there are no clients ready yet
+                _rateAdapter.NO_CLIENTS = true;
+            }
+        }
+        else
+        {
+            log.warn("Exception:" + linked);
+        }
+    }
+
+    /**
+     * Inner class that listens for messages and sends a report for the time taken between receiving the 'start' and
+     * 'end' messages.
+     */
+    class SustainedListener implements MessageListener
+    {
+        /** Number of messages received */
+        private long _received = 0;
+        /** The number of messages in the batch */
+        private int _batchSize = 0;
+        /** Record of the when the 'start' messagse was sen */
+        private Long _startTime;
+        /** Message producer to use to send reports */
+        MessageProducer _updater;
+        /** Session to create the report message on */
+        Session _session;
+        /** Record of the client ID used for this SustainedListnener */
+        String _client;
+
+        /**
+         * Main Constructor
+         *
+         * @param clientname      The _client id used to identify this connection.
+         * @param batchSize       The number of messages that are to be sent per batch. Note: This is not used to
+         *                        control the interval between sending reports.
+         * @param session         The controlSession used for communication.
+         * @param sendDestination The destination that update reports should be sent to.
+         *
+         * @throws JMSException My occur if creatingthe Producer fails
+         */
+        public SustainedListener(String clientname, int batchSize, Session session, Destination sendDestination)
+            throws JMSException
+        {
+            _batchSize = batchSize;
+            _client = clientname;
+            _session = session;
+            _updater = session.createProducer(sendDestination);
+        }
+
+        public void onMessage(Message message)
+        {
+            if (log.isTraceEnabled())
+            {
+                log.trace("Message " + _received + "received in listener");
+            }
+
+            if (message instanceof TextMessage)
+            {
+                try
+                {
+                    _received++;
+                    if (((TextMessage) message).getText().equals("start"))
+                    {
+                        log.debug("Starting Batch");
+                        _startTime = System.nanoTime();
+                    }
+                    else if (((TextMessage) message).getText().equals("end"))
+                    {
+                        if (_startTime != null)
+                        {
+                            long currentTime = System.nanoTime();
+                            sendStatus(currentTime - _startTime, _received, message.getIntProperty("BATCH"));
+                            log.debug("End Batch");
+                        }
+                    }
+                }
+                catch (JMSException e)
+                {
+                    // ignore error
+                }
+            }
+
+        }
+
+        /**
+         * sendStatus creates and sends the report back to the publisher
+         *
+         * @param time     taken for the the last batch
+         * @param received Total number of messages received.
+         * @param batchNumber the batch number
+         * @throws JMSException if an error occurs during the send
+         */
+        private void sendStatus(long time, long received, int batchNumber) throws JMSException
+        {
+            Message updateMessage = _session.createTextMessage("update");
+            updateMessage.setStringProperty("CLIENT_ID", ":" + _client);
+            updateMessage.setStringProperty("CONTROL_TYPE", "UPDATE");
+            updateMessage.setLongProperty("RECEIVED", received);
+            updateMessage.setIntProperty("BATCH", batchNumber);
+            updateMessage.setLongProperty("DURATION", time);
+
+            if (log.isInfoEnabled())
+            {
+                log.info("**** SENDING [" + batchNumber + "]**** " + "CLIENT_ID:" + _client + " RECEIVED:" + received
+                    + " BATCH:" + batchNumber + " DURATION:" + time);
+            }
+
+            // Output on the main console.info the details of this batch
+            if ((batchNumber % 10) == 0)
+            {
+                console.info("Sending Report [" + batchNumber + "] " + "CLIENT_ID:" + _client + " RECEIVED:" + received
+                    + " BATCH:" + batchNumber + " DURATION:" + time);
+            }
+
+            _updater.send(updateMessage);
+        }
+    }
+
+    /**
+     * This class is used here to adjust the _delay value which in turn is used to control the number of messages/second
+     * that are sent through the test system.
+     *
+     * By keeping a record of the messages recevied and the average time taken to process the batch size can be
+     * calculated and so the delay can be adjusted to maintain that rate.
+     *
+     * Given that delays of < 10ms can be rounded up the delay is only used between messages if the _delay > 10ms * no
+     * messages in the batch. Otherwise the delay is used at the end of the batch.
+     */
+    class SustainedRateAdapter implements MessageListener, Runnable
+    {
+        private SustainedClientTestCase _client;
+        private long _batchVariance = Integer.getInteger("batchVariance", 3); // no. batches to allow drifting
+        private long _timeVariance = TEN_MILLI_SEC * 5; // no. nanos between send and report delay (10ms)
+        private volatile long _delay; // in nanos
+        private long _sent;
+        private Map<String, Long> _slowClients = new HashMap<String, Long>();
+        private static final long PAUSE_SLEEP = TEN_MILLI_SEC / 1000; // 10 ms
+        private static final long NO_CLIENT_SLEEP = 1000; // 1s
+        private volatile boolean NO_CLIENTS = true;
+        private int _delayShifting;
+        private final int REPORTS_WITHOUT_CHANGE = Integer.getInteger("stableReportCount", 5);
+        private boolean _warmedup = false;
+        private static final long EXPECTED_TIME_PER_BATCH = 100000L;
+        private int _warmUpBatches = Integer.getInteger("warmUpBatches", 10);
+
+        SustainedRateAdapter(SustainedClientTestCase client)
+        {
+            _client = client;
+        }
+
+        public void onMessage(Message message)
+        {
+            if (log.isDebugEnabled())
+            {
+                log.debug("SustainedRateAdapter onMessage(Message message = " + message + "): called");
+            }
+
+            try
+            {
+                String controlType = message.getStringProperty("CONTROL_TYPE");
+
+                // Check if the message is a test invite.
+                if ("UPDATE".equals(controlType))
+                {
+                    NO_CLIENTS = false;
+                    long duration = message.getLongProperty("DURATION");
+                    long totalReceived = message.getLongProperty("RECEIVED");
+                    String client = message.getStringProperty("CLIENT_ID");
+                    int batchNumber = message.getIntProperty("BATCH");
+
+                    if (log.isInfoEnabled() && ((batchNumber % DEBUG_LOG_UPATE_INTERVAL) == 0))
+                    {
+                        log.info("Update Report: CLIENT_ID:" + client + " RECEIVED:" + totalReceived + " Recevied BATCH:"
+                            + batchNumber + " DURATION:" + duration);
+                    }
+
+                    recordSlow(client, totalReceived, batchNumber);
+
+                    adjustDelay(client, batchNumber, duration);
+
+                    // Warm up completes when:
+                    // we haven't warmed up
+                    // and the number of batches sent to each client is at least half of the required warmup batches
+                    if (!_warmedup && (batchNumber >= _warmUpBatches))
+                    {
+                        _warmedup = true;
+                        _warmup.countDown();
+
+                    }
+                }
+            }
+            catch (JMSException e)
+            {
+                //
+            }
+        }
+
+        CountDownLatch _warmup = new CountDownLatch(1);
+
+        int _numBatches = 10000;
+
+        // long[] _timings = new long[_numBatches];
+        private boolean _running = true;
+
+        public void run()
+        {
+            console.info("Warming up");
+
+            doBatch(_warmUpBatches);
+
+            try
+            {
+                // wait for warmup to complete.
+                _warmup.await();
+
+                // set delay to the average length of the batches
+                _delay = _totalDuration / _warmUpBatches / delays.size();
+
+                console.info("Warmup complete delay set : " + _delay + " based on _totalDuration: " + _totalDuration
+                    + " over no. batches: " + _warmUpBatches + " with client count: " + delays.size());
+
+                _totalDuration = 0L;
+                _totalReceived = 0L;
+                _sent = 0L;
+            }
+            catch (InterruptedException e)
+            {
+                //
+            }
+
+            doBatch(_numBatches);
+
+        }
+
+        private void doBatch(int batchSize) // long[] timings,
+        {
+            TextMessage testMessage = null;
+            try
+            {
+                testMessage = _client.session[0].createTextMessage("start");
+
+                for (int batch = 0; batch <= batchSize; batch++)
+                // while (_running)
+                {
+                    long start = System.nanoTime();
+
+                    testMessage.setText("start");
+                    testMessage.setIntProperty("BATCH", batch);
+
+                    _client.producer.send(testMessage);
+                    _rateAdapter.sentMessage();
+
+                    testMessage.setText("test");
+                    // start at 2 so start and end count as part of batch
+                    for (int m = 2; m < _batchSize; m++)
+                    {
+                        _client.producer.send(testMessage);
+                        _rateAdapter.sentMessage();
+                    }
+
+                    testMessage.setText("end");
+                    _client.producer.send(testMessage);
+                    _rateAdapter.sentMessage();
+
+                    long end = System.nanoTime();
+
+                    long sendtime = end - start;
+
+                    if (log.isDebugEnabled())
+                    {
+                        log.info("Sent batch[" + batch + "](" + _batchSize + ") in " + sendtime); // timings[batch]);
+                    }
+
+                    if ((batch % LOG_UPATE_INTERVAL) == 0)
+                    {
+                        console.info("Sent Batch[" + batch + "](" + _batchSize + ")" + status());
+                    }
+
+                    _rateAdapter.sleepBatch();
+
+                }
+            }
+            catch (JMSException e)
+            {
+                console.error("Runner ended");
+            }
+        }
+
+        private String status()
+        {
+            return " TotalDuration: " + _totalDuration + " for " + delays.size() + " consumers" + " Delay is " + _delay
+                + " resulting in "
+                + ((_delay > (TEN_MILLI_SEC * _batchSize)) ? ((_delay / _batchSize) + "/msg") : (_delay + "/batch"));
+        }
+
+        private void sleepBatch()
+        {
+            if (checkForSlowClients())
+            { // if there werwe slow clients we have already slept so don't sleep anymore again.
+                return;
+            }
+
+            if (!SLEEP_PER_MESSAGE)
+            {
+                // per batch sleep.. if sleep is to small to spread over the batch.
+                if (_delay <= (TEN_MILLI_SEC * _batchSize))
+                {
+                    sleepLong(_delay);
+                }
+                else
+                {
+                    log.info("Not sleeping _delay > ten*batch is:" + _delay);
+                }
+            }
+        }
+
+        public void stop()
+        {
+            _running = false;
+        }
+
+        Map<String, Long> delays = new HashMap<String, Long>();
+        Long _totalReceived = 0L;
+        Long _totalDuration = 0L;
+        int _skipUpdate = 0;
+
+        /**
+         * Adjust the delay for sending messages based on this update from the client
+         *
+         * @param client        The client that send this update
+         * @param duration      The time taken for the last batch of messagse
+         * @param batchNumber   The reported batchnumber from the client
+         */
+        private void adjustDelay(String client, int batchNumber, long duration)
+        {
+            // Retrieve the current total time taken for this client.
+            Long currentTime = delays.get(client);
+
+            // Add the new duration time to this client
+            if (currentTime == null)
+            {
+                currentTime = duration;
+            }
+            else
+            {
+                currentTime += duration;
+            }
+
+            delays.put(client, currentTime);
+
+            long batchesSent = _sent / _batchSize;
+
+            // ensure we don't divide by zero
+            if (batchesSent == 0)
+            {
+                batchesSent = 1L;
+            }
+
+            _totalReceived += _batchSize;
+            _totalDuration += duration;
+
+            // calculate average duration accross clients per batch
+            long averageDuration = _totalDuration / delays.size() / batchesSent;
+
+            // calculate the difference between current send delay and average report delay
+            long diff = (duration) - averageDuration;
+
+            if (log.isInfoEnabled() && ((batchNumber % DEBUG_LOG_UPATE_INTERVAL) == 0))
+            {
+                log.info("TotalDuration:" + _totalDuration + " for " + delays.size() + " consumers." + " on batch: "
+                    + batchesSent + " received batch: " + batchNumber + " Batch Duration: " + duration + " Average: "
+                    + averageDuration + " so diff: " + diff + " for : " + client + " Delay is " + _delay + " resulting in "
+                    + ((_delay > (TEN_MILLI_SEC * _batchSize)) ? ((_delay / _batchSize) + "/msg") : (_delay + "/batch")));
+            }
+
+            // if the averageDuration differs from the current by more than the specified variane then adjust delay.
+            if (Math.abs(diff) > _timeVariance)
+            {
+
+                // if the the _delay is larger than the required duration to send report
+                // speed up
+                if (diff > TEN_MILLI_SEC)
+                {
+                    _delay -= TEN_MILLI_SEC;
+
+                    if (_delay < 0)
+                    {
+                        _delay = 0;
+                        log.info("Reset _delay to 0");
+                        delayStable();
+                    }
+                    else
+                    {
+                        delayChanged();
+                    }
+
+                }
+                else if (diff < 0) // diff < 0 diff cannot be 0 as it is > _timeVariance
+                {
+                    // the report took longer
+                    _delay += TEN_MILLI_SEC;
+                    delayChanged();
+                }
+            }
+            else
+            {
+                delayStable();
+            }
+
+            // If we have a consumer that is behind with the batches.
+            if ((batchesSent - batchNumber) > _batchVariance)
+            {
+                log.debug("Increasing _delay as sending more than receiving");
+
+                _delay += 2 * TEN_MILLI_SEC;
+                delayChanged();
+            }
+
+        }
+
+        /** Reset the number of iterations before we say the delay has stabilised. */
+        private void delayChanged()
+        {
+            _delayShifting = REPORTS_WITHOUT_CHANGE;
+        }
+
+        /**
+         * Record the fact that delay has stabilised If delay has stablised for REPORTS_WITHOUT_CHANGE then it will
+         * output Delay stabilised
+         */
+        private void delayStable()
+        {
+            _delayShifting--;
+
+            if (_delayShifting < 0)
+            {
+                _delayShifting = 0;
+                console.debug("Delay stabilised:" + _delay);
+            }
+        }
+
+        /**
+         * Checks that the client has received enough messages. If the client has fallen behind then they are put in the
+         * _slowClients lists which will increase the delay.
+         *
+         * @param client   The client identifier to check
+         * @param received the number of messages received by that client
+         * @param batchNumber
+         */
+        private void recordSlow(String client, long received, int batchNumber)
+        {
+            if (Math.abs(batchNumber - (_sent / _batchSize)) > _batchVariance)
+            {
+                _slowClients.put(client, received);
+            }
+            else
+            {
+                _slowClients.remove(client);
+            }
+        }
+
+        /** Incrment the number of sent messages and then sleep, if required. */
+        public void sentMessage()
+        {
+
+            _sent++;
+
+            if (_delay > (TEN_MILLI_SEC * _batchSize))
+            {
+                long batchDelay = _delay / _batchSize;
+                // less than 10ms sleep doesn't always work.
+                // _delay is in nano seconds
+                // if (batchDelay < (TEN_MILLI_SEC))
+                // {
+                // sleep(0, (int) batchDelay);
+                // }
+                // else
+                {
+                    // if (batchDelay < 30000000000L)
+                    {
+                        sleepLong(batchDelay);
+                    }
+                }
+            }
+            else
+            {
+                if (SLEEP_PER_MESSAGE && (_delay > 0))
+                {
+                    sleepLong(_delay / _batchSize);
+                }
+            }
+        }
+
+        /**
+         * Check at the end of each batch and pause sending messages to allow slow clients to catch up.
+         *
+         * @return true if there were slow clients that caught up.
+         */
+        private boolean checkForSlowClients()
+        {
+            // This will allways be true as we are running this at the end of each batchSize
+            // if (_sent % _batchSize == 0)
+            {
+                // Cause test to pause when we have slow
+                if (!_slowClients.isEmpty() || NO_CLIENTS)
+                {
+
+                    while (!_slowClients.isEmpty())
+                    {
+                        if (log.isInfoEnabled() && ((_sent / _batchSize % DEBUG_LOG_UPATE_INTERVAL) == 0))
+                        {
+                            String clients = "";
+                            Iterator it = _slowClients.keySet().iterator();
+                            while (it.hasNext())
+                            {
+                                clients += it.next();
+                                if (it.hasNext())
+                                {
+                                    clients += ", ";
+                                }
+                            }
+
+                            log.info("Pausing for slow clients:" + clients);
+                        }
+
+                        if (console.isDebugEnabled() && ((_sent / _batchSize % LOG_UPATE_INTERVAL) == 0))
+                        {
+                            console.debug(_slowClients.size() + " slow clients.");
+                        }
+
+                        sleep(PAUSE_SLEEP);
+                    }
+
+                    if (NO_CLIENTS)
+                    {
+                        sleep(NO_CLIENT_SLEEP);
+                    }
+
+                    log.debug("Continuing");
+
+                    return true;
+                }
+                else
+                {
+                    if ((_sent / _batchSize % LOG_UPATE_INTERVAL) == 0)
+                    {
+                        console.info("Total Delay :" + _delay + " "
+                            + ((_delayShifting == 0) ? "Stablised" : ("Not Stablised(" + _delayShifting + ")")));
+                    }
+                }
+
+            }
+
+            return false;
+        }
+
+        /**
+         * Sleep normally takes micro-seconds this allows the use of a nano-second value.
+         *
+         * @param delay nanoseconds to sleep for.
+         */
+        private void sleepLong(long delay)
+        {
+            sleep(delay / 1000000, (int) (delay % 1000000));
+        }
+
+        /**
+         * Sleep for the specified micro-seconds.
+         * @param sleep microseconds to sleep for.
+         */
+        private void sleep(long sleep)
+        {
+            sleep(sleep, 0);
+        }
+
+        /**
+         * Perform the sleep , swallowing any InteruptException.
+         *
+         * NOTE: If a sleep request is > 10s then reset only sleep for 5s
+         *
+         * @param milli to sleep for
+         * @param nano sub miliseconds to sleep for
+         */
+        private void sleep(long milli, int nano)
+        {
+            try
+            {
+                log.debug("Sleep:" + milli + ":" + nano);
+                if (milli > 10000)
+                {
+
+                    if (_delay == milli)
+                    {
+                        _totalDuration = _totalReceived / _batchSize * EXPECTED_TIME_PER_BATCH;
+                        log.error("Sleeping for more than 10 seconds adjusted to 5s!:" + (milli / 1000)
+                            + "s. Reset _totalDuration:" + _totalDuration);
+                    }
+                    else
+                    {
+                        log.error("Sleeping for more than 10 seconds adjusted to 5s!:" + (milli / 1000) + "s");
+                    }
+
+                    milli = 5000;
+                }
+
+                Thread.sleep(milli, nano);
+            }
+            catch (InterruptedException e)
+            {
+                //
+            }
+        }
+
+        public void setClient(SustainedClientTestCase client)
+        {
+            _client = client;
+        }
+    }
+
+}

Propchange: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedClientTestCase.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedClientTestCase.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCase.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCase.java?rev=571129&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCase.java (added)
+++ incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCase.java Thu Aug 30 05:19:31 2007
@@ -0,0 +1,126 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.sustained;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.test.framework.DropInTest;
+import org.apache.qpid.test.framework.FrameworkBaseCase;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+
+import java.util.Properties;
+
+/**
+ * SustainedTestCase is a {@link FrameworkBaseCase} that runs the "Perf_SustainedPubSub" test case. This consists of one
+ * test client sending, and several receiving, and attempts to find the highest rate at which messages can be broadcast
+ * to the receivers. It is also a {@link DropInTest} to which more test clients may be added during a test run.
+ *
+ * <p/><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * <tr><td>
+ * </table>
+ */
+public class SustainedTestCase extends FrameworkBaseCase implements DropInTest
+{
+    /** Used for debugging. */
+    Logger log = Logger.getLogger(SustainedTestCase.class);
+
+    /** Holds the root name of the topic on which to send the test messages. */
+    private static final String SUSTAINED_KEY = "Perf_SustainedPubSub";
+
+    /**
+     * Creates a new coordinating test case with the specified name.
+     *
+     * @param name The test case name.
+     */
+    public SustainedTestCase(String name)
+    {
+        super(name);
+    }
+
+    /**
+     * Performs a single test run of the sustained test.
+     *
+     * @throws Exception Any exceptions are allowed to fall through and fail the test.
+     */
+    public void testBasicPubSub() throws Exception
+    {
+        log.debug("public void testSinglePubSubCycle(): called");
+
+        Properties testConfig = new Properties();
+        testConfig.put("TEST_NAME", "Perf_SustainedPubSub");
+        testConfig.put("SUSTAINED_KEY", SUSTAINED_KEY);
+        testConfig.put("SUSTAINED_NUM_RECEIVERS", Integer.getInteger("numReceives", 2));
+        testConfig.put("SUSTAINED_UPDATE_INTERVAL", Integer.getInteger("batchSize", 1000));
+        testConfig.put("SUSTAINED_UPDATE_KEY", SUSTAINED_KEY + ".UPDATE");
+        testConfig.put("ACKNOWLEDGE_MODE", Integer.getInteger("ackMode", AMQSession.AUTO_ACKNOWLEDGE));
+
+        log.info("Created Config: " + testConfig.entrySet().toArray());
+
+        getCircuitFactory().sequenceTest(null, null, testConfig);
+    }
+
+    /**
+     * Accepts a late joining client into this test case. The client will be enlisted with a control message
+     * with the 'CONTROL_TYPE' field set to the value 'LATEJOIN'. It should also provide values for the fields:
+     *
+     * <p/><table>
+     * <tr><td> CLIENT_NAME <td> A unique name for the new client.
+     * <tr><td> CLIENT_PRIVATE_CONTROL_KEY <td> The key for the route on which the client receives its control messages.
+     * </table>
+     *
+     * @param message The late joiners join message.
+     *
+     * @throws JMSException Any JMS Exception are allowed to fall through, indicating that the join failed.
+     */
+    public void lateJoin(Message message) throws JMSException
+    {
+        throw new RuntimeException("Not implemented.");
+        /*
+        // Extract the joining clients details from its join request message.
+        TestClientDetails clientDetails = new TestClientDetails();
+        clientDetails.clientName = message.getStringProperty("CLIENT_NAME");
+        clientDetails.privateControlKey = message.getStringProperty("CLIENT_PRIVATE_CONTROL_KEY");
+
+        // Register the joining client, but do block for confirmation as cannot do a synchronous receivers during this
+        // method call, as it may have been called from an 'onMessage' method.
+        assignReceiverRole(clientDetails, new Properties(), false);
+         */
+    }
+
+    /**
+     * Should provide a translation from the junit method name of a test to its test case name as known to the test
+     * clients that will run the test. The purpose of this is to convert the JUnit method name into the correct test
+     * case name to place into the test invite. For example the method "testP2P" might map onto the interop test case
+     * name "TC2_BasicP2P".
+     *
+     * @param methodName The name of the JUnit test method.
+     *
+     * @return The name of the corresponding interop test case.
+     */
+    public String getTestCaseNameForTestMethod(String methodName)
+    {
+        return "Perf_SustainedPubSub";
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCase.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/sustained/SustainedTestCase.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Added: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/test/framework/distributedtesting/TestClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/test/framework/distributedtesting/TestClient.java?rev=571129&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/test/framework/distributedtesting/TestClient.java (added)
+++ incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/test/framework/distributedtesting/TestClient.java Thu Aug 30 05:19:31 2007
@@ -0,0 +1,464 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.test.framework.distributedtesting;
+
+import org.apache.log4j.Logger;
+import org.apache.log4j.NDC;
+
+import org.apache.qpid.interop.clienttestcases.TestCase1DummyRun;
+import org.apache.qpid.interop.clienttestcases.TestCase2BasicP2P;
+import org.apache.qpid.interop.clienttestcases.TestCase3BasicPubSub;
+import org.apache.qpid.sustained.SustainedClientTestCase;
+import org.apache.qpid.test.framework.MessagingTestConfigProperties;
+import org.apache.qpid.test.framework.TestUtils;
+import org.apache.qpid.test.framework.clocksynch.ClockSynchThread;
+import org.apache.qpid.test.framework.clocksynch.ClockSynchronizer;
+import org.apache.qpid.test.framework.clocksynch.UDPClockSynchronizer;
+import org.apache.qpid.test.framework.distributedcircuit.TestClientCircuitEnd;
+
+import uk.co.thebadgerset.junit.extensions.SleepThrottle;
+import uk.co.thebadgerset.junit.extensions.Throttle;
+import uk.co.thebadgerset.junit.extensions.util.ParsedProperties;
+import uk.co.thebadgerset.junit.extensions.util.TestContextProperties;
+
+import javax.jms.*;
+
+import java.util.*;
+
+/**
+ * Implements a test client as described in the interop testing spec
+ * (http://cwiki.apache.org/confluence/display/qpid/Interop+Testing+Specification). A test client is an agent that
+ * reacts to control message sequences send by the test {@link Coordinator}.
+ *
+ * <p/><table><caption>Messages Handled by SustainedTestClient</caption>
+ * <tr><th> Message               <th> Action
+ * <tr><td> Invite(compulsory)    <td> Reply with Enlist.
+ * <tr><td> Invite(test case)     <td> Reply with Enlist if test case available.
+ * <tr><td> AssignRole(test case) <td> Reply with Accept Role if matches an enlisted test. Keep test parameters.
+ * <tr><td> Start                 <td> Send test messages defined by test parameters. Send report on messages sent.
+ * <tr><td> Status Request        <td> Send report on messages received.
+ * <tr><td> Terminate             <td> Terminate the test client.
+ * <tr><td> ClockSynch            <td> Synch clock against the supplied UDP address.
+ * </table>
+ *
+ * <p><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * <tr><td> Handle all incoming control messages. <td> {@link TestClientControlledTest}
+ * <tr><td> Configure and look up test cases by name. <td> {@link TestClientControlledTest}
+ * </table>
+ */
+public class TestClient implements MessageListener
+{
+    /** Used for debugging. */
+    private static final Logger log = Logger.getLogger(TestClient.class);
+
+    /** Used for reporting to the console. */
+    private static final Logger console = Logger.getLogger("CONSOLE");
+
+    /** Holds the default identifying name of the test client. */
+    public static final String CLIENT_NAME = "java";
+
+    /** Holds the URL of the broker to run the tests on. */
+    public static String brokerUrl;
+
+    /** Holds the virtual host to run the tests on. If <tt>null</tt>, then the default virtual host is used. */
+    public static String virtualHost;
+
+    /**
+     * Holds the test context properties that provides the default test parameters, plus command line overrides.
+     * This is initialized with the default test parameters, to which command line overrides may be applied.
+     */
+    public static ParsedProperties testContextProperties =
+        TestContextProperties.getInstance(MessagingTestConfigProperties.defaults);
+
+    /** Holds all the test cases loaded from the classpath. */
+    Map<String, TestClientControlledTest> testCases = new HashMap<String, TestClientControlledTest>();
+
+    /** Holds the test case currently being run by this client. */
+    protected TestClientControlledTest currentTestCase;
+
+    /** Holds the connection to the broker that the test is being coordinated on. */
+    protected Connection connection;
+
+    /** Holds the message producer to hold the test coordination over. */
+    protected MessageProducer producer;
+
+    /** Holds the JMS controlSession for the test coordination. */
+    protected Session session;
+
+    /** Holds the name of this client, with a default value. */
+    protected String clientName = CLIENT_NAME;
+
+    /** This flag indicates that the test client should attempt to join the currently running test case on start up. */
+    protected boolean join;
+
+    /** Holds the clock synchronizer for the test node. */
+    ClockSynchThread clockSynchThread;
+
+    /**
+     * Creates a new interop test client, listenting to the specified broker and virtual host, with the specified client
+     * identifying name.
+     *
+     * @param brokerUrl   The url of the broker to connect to.
+     * @param virtualHost The virtual host to conect to.
+     * @param clientName  The client name to use.
+     * @param join        Flag to indicate that this client should attempt to join running tests.
+     */
+    public TestClient(String brokerUrl, String virtualHost, String clientName, boolean join)
+    {
+        log.debug("public TestClient(String brokerUrl = " + brokerUrl + ", String virtualHost = " + virtualHost
+            + ", String clientName = " + clientName + ", boolean join = " + join + "): called");
+
+        // Retain the connection parameters.
+        this.brokerUrl = brokerUrl;
+        this.virtualHost = virtualHost;
+        this.clientName = clientName;
+        this.join = join;
+    }
+
+    /**
+     * The entry point for the interop test coordinator. This client accepts the following command line arguments:
+     *
+     * <p/><table>
+     * <tr><td> -b         <td> The broker URL.       <td> Optional.
+     * <tr><td> -h         <td> The virtual host.     <td> Optional.
+     * <tr><td> -n         <td> The test client name. <td> Optional.
+     * <tr><td> name=value <td> Trailing argument define name/value pairs. Added to system properties. <td> Optional.
+     * </table>
+     *
+     * @param args The command line arguments.
+     */
+    public static void main(String[] args)
+    {
+        log.debug("public static void main(String[] args = " + Arrays.toString(args) + "): called");
+        console.info("Qpid Distributed Test Client.");
+
+        // Override the default broker url to be localhost:5672.
+        testContextProperties.setProperty(MessagingTestConfigProperties.BROKER_PROPNAME, "tcp://localhost:5672");
+
+        // Use the command line parser to evaluate the command line with standard handling behaviour (print errors
+        // and usage then exist if there are errors).
+        // Any options and trailing name=value pairs are also injected into the test context properties object,
+        // to override any defaults that may have been set up.
+        ParsedProperties options =
+            new ParsedProperties(uk.co.thebadgerset.junit.extensions.util.CommandLineParser.processCommandLine(args,
+                    new uk.co.thebadgerset.junit.extensions.util.CommandLineParser(
+                        new String[][]
+                        {
+                            { "b", "The broker URL.", "broker", "false" },
+                            { "h", "The virtual host to use.", "virtual host", "false" },
+                            { "o", "The name of the directory to output test timings to.", "dir", "false" },
+                            { "n", "The name of the test client.", "name", "false" },
+                            { "j", "Join this test client to running test.", "false" }
+                        }), testContextProperties));
+
+        // Extract the command line options.
+        String brokerUrl = options.getProperty("b");
+        String virtualHost = options.getProperty("h");
+        String clientName = options.getProperty("n");
+        clientName = (clientName == null) ? CLIENT_NAME : clientName;
+        boolean join = options.getPropertyAsBoolean("j");
+
+        // To distinguish logging output set up an NDC on the client name.
+        NDC.push(clientName);
+
+        // Create a test client and start it running.
+        TestClient client = new TestClient(brokerUrl, virtualHost, clientName, join);
+
+        // Use a class path scanner to find all the interop test case implementations.
+        // Hard code the test classes till the classpath scanner is fixed.
+        Collection<Class<? extends TestClientControlledTest>> testCaseClasses =
+            new ArrayList<Class<? extends TestClientControlledTest>>();
+        // ClasspathScanner.getMatches(TestClientControlledTest.class, "^TestCase.*", true);
+        Collections.addAll(testCaseClasses, TestCase1DummyRun.class, TestCase2BasicP2P.class, TestCase3BasicPubSub.class,
+            SustainedClientTestCase.class, TestClientCircuitEnd.class);
+
+        try
+        {
+            client.start(testCaseClasses);
+        }
+        catch (Exception e)
+        {
+            log.error("The test client was unable to start.", e);
+            console.info(e.getMessage());
+            System.exit(1);
+        }
+    }
+
+    /**
+     * Starts the interop test client running. This causes it to start listening for incoming test invites.
+     *
+     * @param testCaseClasses The classes of the available test cases. The test case names from these are used to
+     *                        matchin incoming test invites against.
+     *
+     * @throws JMSException Any underlying JMSExceptions are allowed to fall through.
+     */
+    protected void start(Collection<Class<? extends TestClientControlledTest>> testCaseClasses) throws JMSException
+    {
+        log.debug("protected void start(Collection<Class<? extends TestClientControlledTest>> testCaseClasses = "
+            + testCaseClasses + "): called");
+
+        // Create all the test case implementations and index them by the test names.
+        for (Class<? extends TestClientControlledTest> nextClass : testCaseClasses)
+        {
+            try
+            {
+                TestClientControlledTest testCase = nextClass.newInstance();
+                testCases.put(testCase.getName(), testCase);
+            }
+            catch (InstantiationException e)
+            {
+                log.warn("Could not instantiate test case class: " + nextClass.getName(), e);
+                // Ignored.
+            }
+            catch (IllegalAccessException e)
+            {
+                log.warn("Could not instantiate test case class due to illegal access: " + nextClass.getName(), e);
+                // Ignored.
+            }
+        }
+
+        // Open a connection to communicate with the coordinator on.
+        connection = TestUtils.createConnection(testContextProperties);
+        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+        // Set this up to listen for control messages.
+        Topic privateControlTopic = session.createTopic("iop.control." + clientName);
+        MessageConsumer consumer = session.createConsumer(privateControlTopic);
+        consumer.setMessageListener(this);
+
+        Topic controlTopic = session.createTopic("iop.control");
+        MessageConsumer consumer2 = session.createConsumer(controlTopic);
+        consumer2.setMessageListener(this);
+
+        // Create a producer to send replies with.
+        producer = session.createProducer(null);
+
+        // If the join flag was set, then broadcast a join message to notify the coordinator that a new test client
+        // is available to join the current test case, if it supports it. This message may be ignored, or it may result
+        // in this test client receiving a test invite.
+        if (join)
+        {
+            Message joinMessage = session.createMessage();
+
+            joinMessage.setStringProperty("CONTROL_TYPE", "JOIN");
+            joinMessage.setStringProperty("CLIENT_NAME", clientName);
+            joinMessage.setStringProperty("CLIENT_PRIVATE_CONTROL_KEY", "iop.control." + clientName);
+            producer.send(controlTopic, joinMessage);
+        }
+
+        // Start listening for incoming control messages.
+        connection.start();
+    }
+
+    /**
+     * Handles all incoming control messages.
+     *
+     * @param message The incoming message.
+     */
+    public void onMessage(Message message)
+    {
+        NDC.push(clientName);
+        log.debug("public void onMessage(Message message = " + message + "): called");
+
+        try
+        {
+            String controlType = message.getStringProperty("CONTROL_TYPE");
+            String testName = message.getStringProperty("TEST_NAME");
+
+            log.debug("Received control of type '" + controlType + "' for the test '" + testName + "'");
+
+            // Check if the message is a test invite.
+            if ("INVITE".equals(controlType))
+            {
+                // Flag used to indicate that an enlist should be sent. Only enlist to compulsory invites or invites
+                // for which test cases exist.
+                boolean enlist = false;
+
+                if (testName != null)
+                {
+                    log.debug("Got an invite to test: " + testName);
+
+                    // Check if the requested test case is available.
+                    TestClientControlledTest testCase = testCases.get(testName);
+
+                    if (testCase != null)
+                    {
+                        log.debug("Found implementing class for test '" + testName + "', enlisting for it.");
+
+                        // Check if the test case will accept the invitation.
+                        enlist = testCase.acceptInvite(message);
+
+                        log.debug("The test case "
+                            + (enlist ? " accepted the invite, enlisting for it."
+                                      : " did not accept the invite, not enlisting."));
+
+                        // Make the requested test case the current test case.
+                        currentTestCase = testCase;
+                    }
+                    else
+                    {
+                        log.debug("Received an invite to the test '" + testName + "' but this test is not known.");
+                    }
+                }
+                else
+                {
+                    log.debug("Got a compulsory invite, enlisting for it.");
+
+                    enlist = true;
+                }
+
+                if (enlist)
+                {
+                    // Reply with the client name in an Enlist message.
+                    Message enlistMessage = session.createMessage();
+                    enlistMessage.setStringProperty("CONTROL_TYPE", "ENLIST");
+                    enlistMessage.setStringProperty("CLIENT_NAME", clientName);
+                    enlistMessage.setStringProperty("CLIENT_PRIVATE_CONTROL_KEY", "iop.control." + clientName);
+                    enlistMessage.setJMSCorrelationID(message.getJMSCorrelationID());
+
+                    log.debug("Sending enlist message '" + enlistMessage + "' to " + message.getJMSReplyTo());
+
+                    producer.send(message.getJMSReplyTo(), enlistMessage);
+                }
+                else
+                {
+                    // Reply with the client name in an Decline message.
+                    Message enlistMessage = session.createMessage();
+                    enlistMessage.setStringProperty("CONTROL_TYPE", "DECLINE");
+                    enlistMessage.setStringProperty("CLIENT_NAME", clientName);
+                    enlistMessage.setStringProperty("CLIENT_PRIVATE_CONTROL_KEY", "iop.control." + clientName);
+                    enlistMessage.setJMSCorrelationID(message.getJMSCorrelationID());
+
+                    log.debug("Sending decline message '" + enlistMessage + "' to " + message.getJMSReplyTo());
+
+                    producer.send(message.getJMSReplyTo(), enlistMessage);
+                }
+            }
+            else if ("ASSIGN_ROLE".equals(controlType))
+            {
+                // Assign the role to the current test case.
+                String roleName = message.getStringProperty("ROLE");
+
+                log.debug("Got a role assignment to role: " + roleName);
+
+                TestClientControlledTest.Roles role = Enum.valueOf(TestClientControlledTest.Roles.class, roleName);
+
+                currentTestCase.assignRole(role, message);
+
+                // Reply by accepting the role in an Accept Role message.
+                Message acceptRoleMessage = session.createMessage();
+                acceptRoleMessage.setStringProperty("CLIENT_NAME", clientName);
+                acceptRoleMessage.setStringProperty("CONTROL_TYPE", "ACCEPT_ROLE");
+                acceptRoleMessage.setJMSCorrelationID(message.getJMSCorrelationID());
+
+                log.debug("Sending accept role message '" + acceptRoleMessage + "' to " + message.getJMSReplyTo());
+
+                producer.send(message.getJMSReplyTo(), acceptRoleMessage);
+            }
+            else if ("START".equals(controlType) || "STATUS_REQUEST".equals(controlType))
+            {
+                if ("START".equals(controlType))
+                {
+                    log.debug("Got a start notification.");
+
+                    // Extract the number of test messages to send from the start notification.
+                    int numMessages;
+
+                    try
+                    {
+                        numMessages = message.getIntProperty("MESSAGE_COUNT");
+                    }
+                    catch (NumberFormatException e)
+                    {
+                        // If the number of messages is not specified, use the default of one.
+                        numMessages = 1;
+                    }
+
+                    // Start the current test case.
+                    currentTestCase.start(numMessages);
+                }
+                else
+                {
+                    log.debug("Got a status request.");
+                }
+
+                // Generate the report from the test case and reply with it as a Report message.
+                Message reportMessage = currentTestCase.getReport(session);
+                reportMessage.setStringProperty("CLIENT_NAME", clientName);
+                reportMessage.setStringProperty("CONTROL_TYPE", "REPORT");
+                reportMessage.setJMSCorrelationID(message.getJMSCorrelationID());
+
+                log.debug("Sending report message '" + reportMessage + "' to " + message.getJMSReplyTo());
+
+                producer.send(message.getJMSReplyTo(), reportMessage);
+            }
+            else if ("TERMINATE".equals(controlType))
+            {
+                console.info("Received termination instruction from coordinator.");
+
+                // Is a cleaner shutdown needed?
+                connection.close();
+                System.exit(0);
+            }
+            else if ("CLOCK_SYNCH".equals(controlType))
+            {
+                log.debug("Received clock synch command.");
+                String address = message.getStringProperty("ADDRESS");
+
+                log.debug("address = " + address);
+
+                // Re-create (if necessary) and start the clock synch thread to synch the clock every ten seconds.
+                if (clockSynchThread != null)
+                {
+                    clockSynchThread.terminate();
+                }
+
+                SleepThrottle throttle = new SleepThrottle();
+                throttle.setRate(0.1f);
+
+                clockSynchThread = new ClockSynchThread(new UDPClockSynchronizer(address), throttle);
+                clockSynchThread.start();
+            }
+            else
+            {
+                // Log a warning about this but otherwise ignore it.
+                log.warn("Got an unknown control message, controlType = " + controlType + ", message = " + message);
+            }
+        }
+        catch (JMSException e)
+        {
+            // Log a warning about this, but otherwise ignore it.
+            log.warn("Got JMSException whilst handling message: " + message, e);
+        }
+        // Log any runtimes that fall through this message handler. These are fatal errors for the test client.
+        catch (RuntimeException e)
+        {
+            log.error("The test client message handler got an unhandled exception: ", e);
+            console.info("The message handler got an unhandled exception, terminating the test client.");
+            System.exit(1);
+        }
+        finally
+        {
+            NDC.pop();
+        }
+    }
+}

Propchange: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/test/framework/distributedtesting/TestClient.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/qpid/trunk/qpid/java/integrationtests/src/main/java/org/apache/qpid/test/framework/distributedtesting/TestClient.java
------------------------------------------------------------------------------
    svn:keywords = Rev Date

Modified: incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/test/java/org/apache/qpid/management/ui/ManagementConsoleTest.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/test/java/org/apache/qpid/management/ui/ManagementConsoleTest.java?rev=571129&r1=571128&r2=571129&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/test/java/org/apache/qpid/management/ui/ManagementConsoleTest.java (original)
+++ incubator/qpid/trunk/qpid/java/management/eclipse-plugin/src/test/java/org/apache/qpid/management/ui/ManagementConsoleTest.java Thu Aug 30 05:19:31 2007
@@ -1,17 +1,21 @@
 /*
- * Copyright (c) 2006 The Apache Software Foundation
  *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
  *
- *    http://www.apache.org/licenses/LICENSE-2.0
+ *   http://www.apache.org/licenses/LICENSE-2.0
  *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
  *
  */
 package org.apache.qpid.management.ui;

Propchange: incubator/qpid/trunk/qpid/java/plugins/
------------------------------------------------------------------------------
--- svn:ignore (original)
+++ svn:ignore Thu Aug 30 05:19:31 2007
@@ -1 +1,2 @@
 target
+*.iml

Modified: incubator/qpid/trunk/qpid/java/plugins/pom.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/plugins/pom.xml?rev=571129&r1=571128&r2=571129&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/plugins/pom.xml (original)
+++ incubator/qpid/trunk/qpid/java/plugins/pom.xml Thu Aug 30 05:19:31 2007
@@ -13,6 +13,11 @@
         <relativePath>../pom.xml</relativePath>
     </parent>
 
+    <properties>
+        <!-- This is an assembly/distribution pom so no test code exists -->
+        <maven.test.skip>true</maven.test.skip>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>org.apache.maven</groupId>

Modified: incubator/qpid/trunk/qpid/java/pom.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/pom.xml?rev=571129&r1=571128&r2=571129&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/pom.xml (original)
+++ incubator/qpid/trunk/qpid/java/pom.xml Thu Aug 30 05:19:31 2007
@@ -154,7 +154,7 @@
         <module>cluster</module>
         <module>systests</module>
         <module>perftests</module>
-       <!-- <module>integrationtests</module> -->
+        <module>integrationtests</module>
         <module>management/eclipse-plugin</module>
         <module>client/example</module>
         <module>client-java14</module>
@@ -163,6 +163,18 @@
 
     <build>
         <resources>
+
+            <resource>
+                <targetPath>META-INF/</targetPath>
+                <filtering>false</filtering>
+                <directory>../resources</directory>
+                <includes>
+                    <include>DISCLAIMER</include>
+                    <include>LICENSE</include>
+                    <include>NOTICE</include>
+                </includes>
+            </resource>
+ 
             <resource>
                 <directory>src/main/java</directory>
                 <excludes>
@@ -172,26 +184,28 @@
             </resource>
             <resource>
                 <directory>src/main/resources</directory>
-                <includes>
-                    <include>**</include>
-                </includes>
             </resource>
             <resource>
                 <directory>src/main/resources-filtered</directory>
-                <includes>
-                    <include>**</include>
-                </includes>
                 <filtering>true</filtering>
             </resource>
             <resource>
                 <directory>target/generated/src/main/resources</directory>
-                <includes>
-                    <include>**</include>
-                </includes>
             </resource>
         </resources>
 
         <testResources>
+           <testResource>
+                <targetPath>META-INF/</targetPath>
+                <filtering>false</filtering>
+                <directory>../resources</directory>
+                <includes>
+                    <include>DISCLAIMER</include>
+                    <include>LICENSE</include>
+                    <include>NOTICE</include>
+                </includes>
+            </testResource>
+
             <testResource>
                 <directory>src/test/java</directory>
                 <excludes>
@@ -200,9 +214,6 @@
             </testResource>
             <testResource>
                 <directory>src/test/resources</directory>
-                <includes>
-                    <include>**</include>
-                </includes>
             </testResource>
             <testResource>
                 <directory>src/test/java</directory>
@@ -391,29 +402,31 @@
                     <version>0.5</version>
                 </plugin>
 
-                <plugin>
-                    <artifactId>maven-remote-resources-plugin</artifactId>
-                    <version>1.0-alpha-5</version>
-                    <executions>
-                        <execution>
-                            <goals>
-                                <goal>process</goal>
-                            </goals>
-                            <configuration>
-                                <resourceBundles>
-                                    <resourceBundle>org.apache:apache-incubator-disclaimer-resource-bundle:1.1</resourceBundle>
-                                    <resourceBundle>org.apache:apache-jar-resource-bundle:1.2</resourceBundle>
-                                </resourceBundles>
-                                <properties>
-                                    <addLicense>true</addLicense>
-                                    <projectName>Apache Qpid</projectName>
-                                </properties>
-                            </configuration>
-                        </execution>
-                    </executions>
-                </plugin>
+            <!-- Disabled as plugin crashes on the systest module. 
+             Also, the resulting NOTICE file doesn't include all license info due to missing data in dependant poms.
 
-            </plugins>
+            <plugin>
+               <artifactId>maven-remote-resources-plugin</artifactId>
+               <version>1.0-alpha-5</version>
+               <executions>
+                   <execution>
+                       <goals>
+                           <goal>process</goal>
+                       </goals>
+                       <configuration>
+                           <resourceBundles>
+                               <resourceBundle>org.apache:apache-incubator-disclaimer-resource-bundle:1.1</resourceBundle>
+                               <resourceBundle>org.apache:apache-jar-resource-bundle:1.2</resourceBundle>
+                           </resourceBundles>
+                           <properties>
+                               <addLicense>true</addLicense>
+                               <projectName>Apache Qpid</projectName>
+                           </properties>
+                       </configuration>
+                   </execution>
+               </executions>
+           </plugin-->
+        </plugins>
         </pluginManagement>
 
         <defaultGoal>install</defaultGoal>
@@ -526,7 +539,7 @@
             <dependency>
                 <groupId>uk.co.thebadgerset</groupId>
                 <artifactId>junit-toolkit</artifactId>
-                <version>0.6-20070718.144514-11</version>
+                <version>0.6-SNAPSHOT</version>
                 <scope>compile</scope>
             </dependency>
 

Modified: incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES.txt
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES.txt?rev=571129&r1=571128&r2=571129&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES.txt (original)
+++ incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES.txt Thu Aug 30 05:19:31 2007
@@ -1,8 +1,9 @@
-Apache Incubator Qpid Java M1 Release Notes
+Apache Incubator Qpid Java M2 Release Notes
 -------------------------------------------
 
-The Qpid M1 release is intended to serve as a useful baseline release
-for the project, allowing interested users to use the current featureset.
+The Qpid M2 release contains support the for AMQP 0-8 specification.
+You can access the 0-8 specification using the following link.
+http://www.amqp.org/tikiwiki/tiki-index.php?page=Download
 
 For full details of Qpid capabilities, as they currently stand, see our
 detailed project documentation at:
@@ -13,62 +14,92 @@
 and detailed developer documentation.
 
 
-Known Issues/OUtstanding Work
+Known Issues/Outstanding Work
 -----------------------------
 
 You can view the outstanding task list for Qpid by visiting our JIRA:
-
 http://issues.apache.org/jira/browse/QPID
 
+These issues are moved to M3 release.
 
-M1 Tasks Completed
--------------------
-
-The set of JIRA tasks completed as part of the M1 effort is available at:
-
-http://cwiki.apache.org/confluence/display/qpid/Qpid+Java+M1+Release+Notes
-
-This list is copied below for convenience:
-
-Bugs
-----
-http://issues.apache.org/jira/browse/QPID-4 - Remove '/' and ':' from generated queue names
-http://issues.apache.org/jira/browse/QPID-7 - Occasionally messages are ack'd more than once
-http://issues.apache.org/jira/browse/QPID-10- Broker throughput falls off with transactions
-http://issues.apache.org/jira/browse/QPID-56 - AMQQueueMBean - MessageCount on the management interface is not correct.
-http://issues.apache.org/jira/browse/QPID-58 - Creating a QueueReceiver results in ClassCastException
-http://issues.apache.org/jira/browse/QPID-66 - AMQSession implementation of TopicSession and QueueSession interfaces not JMS compliant
-http://issues.apache.org/jira/browse/QPID-68 - Ant build system fails if the project path contains a space
-http://issues.apache.org/jira/browse/QPID-69 - Race condition in Delivery Manager  
-
-Improvements
-------------
-http://issues.apache.org/jira/browse/QPID-36 - Add high and low watermark to flow control
-http://issues.apache.org/jira/browse/QPID-44 - Add high and low watermark to flow control
-http://issues.apache.org/jira/browse/QPID-57 - AMQQueueMBean - Message header attributes should be sent along with message content.
-
-    
-New Features
-------------
-http://issues.apache.org/jira/browse/QPID-13 - Add option to include prefix and suffix in log file name for broker
-http://issues.apache.org/jira/browse/QPID-23 - Extend JNDI support provided to include initial context factory
-http://issues.apache.org/jira/browse/QPID-29 - Provide support for using Qpid JMX with Tivoli for application monitoring
-http://issues.apache.org/jira/browse/QPID-30 - Allow configuration of working/log directories written to by broker
-http://issues.apache.org/jira/browse/QPID-40 - Implement tx.select, tx.commit & tx.rollback from AMQP
-
-            
-Tasks
------
-http://issues.apache.org/jira/browse/QPID-18 - Update Java client and broker to MINA 1.0 release
-http://issues.apache.org/jira/browse/QPID-73 - Create Build Artifacts for release process using ant/maven
-http://issues.apache.org/jira/browse/QPID-74 - Create source distribtuion using build system
-http://issues.apache.org/jira/browse/QPID-75 - Create Standard Binary distribution using build system
-
-
+Here is a filtered list for your convinience
+---------------------------------------
+New Feature	QPID-274 	add connection configuratble timeout on waituntilStateHasChanged
+New Feature	QPID-156 	Implement persistence to disk for Qpid
+New Feature	QPID-155 	Add ability to configure (on/off) queue creation on demand
+New Feature	QPID-43 	Multiple-AMQP version support in the broker
+New Feature	QPID-28 	Allow user to select policy for undeliverable message handling
+New Feature	QPID-27 	Introduce user configurable redlivery delay
+New Feature	QPID-22 	Provide run scripts for clustered broker
+
+Improvement	QPID-430 	Message Age Alerting should not depend upon queue activity
+Improvement	QPID-19 	Add protocol logging capability to client and broker
+Improvement	QPID-11 	Move protocol literals from code to AMQConstant
+
+Bug		QPID-539 	HeadersExchange doesnot correctly implement isBound
+Bug		QPID-517 	Broker doesn't return NO_CONSUMERS code for an immediate message, when the consumer is disconnected.
+Bug		QPID-469 	Redelivered information is currently recorded per message it should be per message per queue.
+Bug		QPID-463 	Java client doesn't close connection gracefully when faced with broker with unsuported protocol version
+Bug		QPID-462 	Exclusive queues and with subscription that 'filtersMessages' will build up messages it doesn't hasInterest() in.
+
+Bug		QPID-397 	Client closeure can be processed before final message ack.
+Bug		QPID-396 	Broker OutOfMemory Error handling
+Bug		QPID-377 	NumberFormatException thrown by broker when running one performance test
+Bug		QPID-293 	setting MessageListener after connection has started can cause messages to be "lost" on a internal delivery queue.
 
+Bug		QPID-185 	Amend Java Broker handling of ifUnused & ifEmpty flags
+Bug		QPID-168 	qpid-server.bat needs updated to support same arguments/features as qpid-server (via qpid-run) bash script
 
 
+M2 Tasks Completed
+-------------------
 
+The set of JIRA tasks completed as part of the M2 effort is available at:
+https://issues.apache.org/jira/secure/ReleaseNote.jspa?projectId=12310520&styleName=Html&version=12312116
 
 
+Here is a filtered (by Java components) version of the JIRA items
 
+Task		QPID-190 	refactoring the java broker mbean classes
+Task		QPID-125 	Moving eclipse plugin for broker management to Maven
+Task		QPID-93 	Delete the old management modules (trunk\qpid\java\management)
+
+New Feature	QPID-428 	Add login functionality for a qpid server from the management console
+New Feature	QPID-418 	Add ability to save user preferences to Java Management Console
+New Feature	QPID-170 	Enhance Management features to support moving a message from one queue to another
+
+Improvement	QPID-482 	[Java] Small performance tweaks
+Improvement	QPID-466 	Create STRICT_AMQP System property to disable JMS extensions in Java client
+Improvement	QPID-453 	AMQShortString should implement Comparable
+Improvement	QPID-422 	Consolidate notification view to display all user configured notifications on one view
+Improvement	QPID-421 	Provide enumerated description for static constants including delivery mode
+Improvement	QPID-420 	Add client id to information displayed about connections on management console
+Improvement	QPID-419 	Introduce read-only and modify authorisation for all objects in a virtual host
+Improvement	QPID-129 	improving Broker MBeans
+
+Bug		QPID-540 	Transient Broker throws NullPointerException and locks up.
+Bug		QPID-538 	[Memory Leak] Connecting lots of consumers causes the broker memory to leak
+Bug		QPID-537 	Make AMQMessage.incrementReference public
+Bug		QPID-531 	[Memory Leak] Broker retains messages that are consumed with NO_ACK
+Bug		QPID-527 	encoding issue
+Bug		QPID-508 	[Memory Leak] Broker does not return mandatory messages sent outside of a transaction.
+Bug		QPID-476 	AMQProtocolSession channelId2SessionMap does not have sessions removed
+Bug		QPID-472 	Creation of TemporaryQueues will not guarantee unqiue queue names if created rapidly.
+Bug		QPID-471 	UserManagement panel lists all users but only after a View Users has been executed and is not updated on Create/Delete User
+
+Bug		QPID-467 	Complete Interop Testing
+Bug		QPID-465 	Incorrect Exception thrown from send() method.
+Bug		QPID-459 	Broker doesn't correctly handle noLocal consumers when messages are pre-exisiting on queues.
+Bug		QPID-458 	Queue Browsing Broken
+Bug		QPID-454 	Message 'taken' notion is per message. But should be per message per queue
+Bug		QPID-443 	Abruptly disconnecting client on transaction publish causes error
+Bug		QPID-440 	Can create dangling transactions on unroutable messages.
+Bug		QPID-436 	topic exchange doesn't obey the mandatory flag
+Bug		QPID-414 	Authentication requires plain text passwords in password file
+Bug		QPID-408 	Queue Depth data incorrect
+Bug		QPID-290 	Java broker does not honor maximum number of channels threshold
+Bug		QPID-276 	Potential race condition in AMQChannel
+Bug		QPID-200 	set/get Destination not implemented in JMSMessage impl
+Bug		QPID-166 	Check for pre conditions to satisfy JMS spec requirments
+Bug		QPID-162 	over 600 warnings when building under Eclipse
+Bug		QPID-159 	The following Interface implementations do not throw Exceptions as required by the spec

Added: incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES_M1.txt
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES_M1.txt?rev=571129&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES_M1.txt (added)
+++ incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES_M1.txt Thu Aug 30 05:19:31 2007
@@ -0,0 +1,74 @@
+Apache Incubator Qpid Java M1 Release Notes
+-------------------------------------------
+
+The Qpid M1 release is intended to serve as a useful baseline release
+for the project, allowing interested users to use the current featureset.
+
+For full details of Qpid capabilities, as they currently stand, see our
+detailed project documentation at:
+
+http://cwiki.apache.org/confluence/display/qpid/Qpid+Java+Documentation
+
+From the link above you can access our Getting Started Guide, FAQ, Build How To
+and detailed developer documentation.
+
+
+Known Issues/OUtstanding Work
+-----------------------------
+
+You can view the outstanding task list for Qpid by visiting our JIRA:
+
+http://issues.apache.org/jira/browse/QPID
+
+
+M1 Tasks Completed
+-------------------
+
+The set of JIRA tasks completed as part of the M1 effort is available at:
+
+http://cwiki.apache.org/confluence/display/qpid/Qpid+Java+M1+Release+Notes
+
+This list is copied below for convenience:
+
+Bugs
+----
+http://issues.apache.org/jira/browse/QPID-4 - Remove '/' and ':' from generated queue names
+http://issues.apache.org/jira/browse/QPID-7 - Occasionally messages are ack'd more than once
+http://issues.apache.org/jira/browse/QPID-10- Broker throughput falls off with transactions
+http://issues.apache.org/jira/browse/QPID-56 - AMQQueueMBean - MessageCount on the management interface is not correct.
+http://issues.apache.org/jira/browse/QPID-58 - Creating a QueueReceiver results in ClassCastException
+http://issues.apache.org/jira/browse/QPID-66 - AMQSession implementation of TopicSession and QueueSession interfaces not JMS compliant
+http://issues.apache.org/jira/browse/QPID-68 - Ant build system fails if the project path contains a space
+http://issues.apache.org/jira/browse/QPID-69 - Race condition in Delivery Manager  
+
+Improvements
+------------
+http://issues.apache.org/jira/browse/QPID-36 - Add high and low watermark to flow control
+http://issues.apache.org/jira/browse/QPID-44 - Add high and low watermark to flow control
+http://issues.apache.org/jira/browse/QPID-57 - AMQQueueMBean - Message header attributes should be sent along with message content.
+
+    
+New Features
+------------
+http://issues.apache.org/jira/browse/QPID-13 - Add option to include prefix and suffix in log file name for broker
+http://issues.apache.org/jira/browse/QPID-23 - Extend JNDI support provided to include initial context factory
+http://issues.apache.org/jira/browse/QPID-29 - Provide support for using Qpid JMX with Tivoli for application monitoring
+http://issues.apache.org/jira/browse/QPID-30 - Allow configuration of working/log directories written to by broker
+http://issues.apache.org/jira/browse/QPID-40 - Implement tx.select, tx.commit & tx.rollback from AMQP
+
+            
+Tasks
+-----
+http://issues.apache.org/jira/browse/QPID-18 - Update Java client and broker to MINA 1.0 release
+http://issues.apache.org/jira/browse/QPID-73 - Create Build Artifacts for release process using ant/maven
+http://issues.apache.org/jira/browse/QPID-74 - Create source distribtuion using build system
+http://issues.apache.org/jira/browse/QPID-75 - Create Standard Binary distribution using build system
+
+
+
+
+
+
+
+
+

Propchange: incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES_M1.txt
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/qpid/trunk/qpid/java/release-docs/RELEASE_NOTES_M1.txt
------------------------------------------------------------------------------
    svn:mime-type = text/plain