Creating a branch for experimenting with some ideas for JMS client.rajith_jms_client

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/rajith_jms_client@1128369 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 381 insertions, 0 deletions

diff --git a/qpid/dotnet/Qpid.Integration.Tests/interop/TestClient.cs b/qpid/dotnet/Qpid.Integration.Tests/interop/TestClient.cs
new file mode 100644
index 0000000000..0def89c6da
--- /dev/null
+++ b/qpid/dotnet/Qpid.Integration.Tests/interop/TestClient.cs
@@ -0,0 +1,381 @@
+/*
+ *
+ * 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.
+ *
+ */
+using System;
+using System.Collections;
+using System.Text;
+using System.Threading;
+using Apache.Qpid.Messaging;
+using Apache.Qpid.Client.Qms;
+using Apache.Qpid.Client;
+using log4net;
+using Apache.Qpid.Integration.Tests.interop.TestCases;
+
+namespace Apache.Qpid.Integration.Tests.interop
+{
+    /// <summary>
+    /// 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 coordinator.
+    ///
+    /// <p/><table><caption>Messages Handled by TestClient</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.
+    /// </table>
+    ///
+    /// <p><table id="crc"><caption>CRC Card</caption>
+    /// <tr><th> Responsibilities <th> Collaborations
+    /// <tr><td> Handle all incoming control messages. <td> {@link InteropClientTestCase}
+    /// <tr><td> Configure and look up test cases by name. <td> {@link InteropClientTestCase}
+    /// </table>
+    /// </summary>
+    public class TestClient
+    {
+        private static ILog log = LogManager.GetLogger(typeof(TestClient));
+      
+        /// <summary> Defines the default broker for the tests, localhost, default port. </summary>
+        public static string DEFAULT_BROKER_URL = "amqp://guest:guest@clientid/?brokerlist='tcp://localhost:5672'";
+      
+        /// <summary> Defines the default virtual host to use for the tests, none. </summary>
+        public static string DEFAULT_VIRTUAL_HOST = "";
+      
+        /// <summary> Defines the default identifying name of this test client. </summary>
+        public static string DEFAULT_CLIENT_NAME = "dotnet";
+
+        /// <summary> Holds the URL of the broker to run the tests on. </summary>
+        public static string brokerUrl;
+
+        /// <summary> Holds the virtual host to run the tests on. If <tt>null</tt>, then the default virtual host is used. </summary>
+        public static string virtualHost;
+
+        /// <summary> The clients identifying name to print in test results and to distinguish from other clients. </summary>
+        private string clientName;
+
+        /// <summary> Holds all the test cases. </summary>
+        private IDictionary testCases = new Hashtable();
+
+        InteropClientTestCase currentTestCase;
+
+        private MessagePublisherBuilder publisherBuilder;
+
+        private IChannel channel;
+
+        /// <summary> Monitor to wait for termination events on. </summary>
+        private static object terminationMonitor = new Object();
+
+        /// <summary>
+        /// Creates a new interop test client, listenting to the specified broker and virtual host, with the specified
+        /// client identifying name.
+        /// </summary>
+        ///
+        /// <param name="brokerUrl"> The url of the broker to connect to. </param>
+        /// <param name="virtualHost"> The virtual host to conect to. </param>
+        /// <param name="clientName">  The client name to use. </param>
+        public TestClient(string brokerUrl, string virtualHost, string clientName)
+        {
+            log.Info("public TestClient(string brokerUrl = " + brokerUrl + ", string virtualHost = " + virtualHost
+                     + ", string clientName = " + clientName + "): called");
+
+            // Retain the connection parameters.
+            TestClient.brokerUrl = brokerUrl;
+            TestClient.virtualHost = virtualHost;
+            this.clientName = clientName;
+        }
+
+    
+        /// <summary>
+        /// The entry point for the interop test coordinator. This client accepts the following command line arguments:
+        /// </summary>
+        /// 
+        /// <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 name="args"> The command line arguments. </param>
+        public static void Main(string[] args)
+        {
+            // Extract the command line options (Not exactly Posix but it will do for now...).
+            string brokerUrl = DEFAULT_BROKER_URL;
+            string virtualHost = DEFAULT_VIRTUAL_HOST;
+            string clientName = DEFAULT_CLIENT_NAME;
+
+            foreach (string nextArg in args)
+            {
+                if (nextArg.StartsWith("-b"))
+                {
+                    brokerUrl = nextArg.Substring(2);
+                }
+                else if (nextArg.StartsWith("-h"))
+                {
+                    virtualHost = nextArg.Substring(2);
+                }
+                else if (nextArg.StartsWith("-n"))
+                {
+                    clientName = nextArg.Substring(2);
+                }
+            }
+            
+            NDC.Push(clientName);
+
+            // Create a test client and start it running.
+            TestClient client = new TestClient(brokerUrl, virtualHost, clientName);
+
+            try
+            {
+                client.Start();
+            }
+            catch (Exception e)
+            {
+                log.Error("The test client was unable to start.", e);
+                System.Environment.Exit(1);
+            }
+
+            // Wait for a signal on the termination monitor before quitting.
+            lock (terminationMonitor)
+            {
+                Monitor.Wait(terminationMonitor);
+            }
+
+            NDC.Pop();
+        }
+
+        /// <summary>
+        /// Starts the interop test client running. This causes it to start listening for incoming test invites.
+        /// </summary>
+        private void Start()
+        {
+            log.Info("private void Start(): called");
+
+            // Use a class path scanner to find all the interop test case implementations.
+            ArrayList testCaseClasses = new ArrayList();
+
+            // ClasspathScanner.getMatches(InteropClientTestCase.class, "^TestCase.*", true);
+            // Hard code the test classes till the classpath scanner is fixed.
+            testCaseClasses.Add(typeof(TestCase1DummyRun));
+            testCaseClasses.Add(typeof(TestCase2BasicP2P));
+            testCaseClasses.Add(typeof(TestCase3BasicPubSub));
+            testCaseClasses.Add(typeof(TestCase4P2PMessageSize));
+            testCaseClasses.Add(typeof(TestCase5PubSubMessageSize));
+
+            // Create all the test case implementations and index them by the test names.
+            foreach (Type testClass in testCaseClasses)
+            {
+                InteropClientTestCase testCase = (InteropClientTestCase)Activator.CreateInstance(testClass);
+                testCases.Add(testCase.GetName(), testCase);
+
+                log.Info("Found test case: " + testClass);
+            }
+
+            // Open a connection to communicate with the coordinator on.
+            log.Info("brokerUrl = " + brokerUrl);
+            IConnection connection = CreateConnection(brokerUrl, virtualHost);
+
+            channel = connection.CreateChannel(false, AcknowledgeMode.AutoAcknowledge);
+
+            // Set this up to listen for control messages.
+            string responseQueueName = channel.GenerateUniqueName();
+            channel.DeclareQueue(responseQueueName, false, true, true);
+
+            channel.Bind(responseQueueName, ExchangeNameDefaults.TOPIC, "iop.control." + clientName);
+            channel.Bind(responseQueueName, ExchangeNameDefaults.TOPIC, "iop.control");
+
+            IMessageConsumer consumer = channel.CreateConsumerBuilder(responseQueueName)
+                .Create();
+            consumer.OnMessage += new MessageReceivedDelegate(OnMessage);
+
+            // Create a publisher to send replies with.
+            publisherBuilder = channel.CreatePublisherBuilder()
+                .WithExchangeName(ExchangeNameDefaults.DIRECT);
+                
+
+            // Start listening for incoming control messages.
+            connection.Start();
+            Console.WriteLine("Test client " + clientName + " ready to receive test control messages...");
+        }
+
+        /// <summary>
+        /// Establishes an AMQ connection. This is a simple convenience method for code that does not anticipate handling connection failures. 
+        /// All exceptions that indicate that the connection has failed, are allowed to fall through.
+        /// </summary>
+        ///
+        /// <param name="brokerUrl">   The broker url to connect to, <tt>null</tt> to use the default from the properties. </param>
+        /// <param name="virtualHost"> The virtual host to connectio to, <tt>null</tt> to use the default. </param>
+        ///
+        /// <returns> A JMS conneciton. </returns>
+        public static IConnection CreateConnection(string brokerUrl, string virtualHost)
+        {
+            log.Info("public static Connection createConnection(string brokerUrl = " + brokerUrl + ", string virtualHost = " 
+                     + virtualHost + "): called");
+
+            // Create a connection to the broker.
+            IConnectionInfo connectionInfo = QpidConnectionInfo.FromUrl(brokerUrl);
+            connectionInfo.VirtualHost = virtualHost;
+            IConnection connection = new AMQConnection(connectionInfo);
+
+            return connection;
+        }
+        
+        /// <summary>
+        /// Handles all incoming control messages.
+        /// </summary>
+        ///
+        /// <param name="message"> The incoming message. </param>
+        public void OnMessage(IMessage message)
+        {
+            log.Info("public void OnMessage(IMessage message = " + message + "): called");
+
+            try
+            {
+                string controlType = message.Headers.GetString("CONTROL_TYPE");
+                string testName = message.Headers.GetString("TEST_NAME");
+
+                // Check if the message is a test invite.
+                if ("INVITE" == controlType)
+                {
+                    string testCaseName = message.Headers.GetString("TEST_NAME");
+
+                    // Flag used to indicate that an enlist should be sent. Only enlist to compulsory invites or invites
+                    // for which test cases exist.
+                    bool enlist = false;
+
+                    if (testCaseName != null)
+                    {
+                        log.Info("Got an invite to test: " + testCaseName);
+
+                        // Check if the requested test case is available.
+                        InteropClientTestCase testCase = (InteropClientTestCase)testCases[testCaseName];
+    
+                        if (testCase != null)
+                        {
+                            // Make the requested test case the current test case.
+                            currentTestCase = testCase;
+                            enlist = true;
+                        }
+                    }
+                    else
+                    {
+                        log.Info("Got a compulsory invite.");
+
+                        enlist = true;
+                    }
+
+                    log.Info("enlist = " + enlist);
+
+                    if (enlist)
+                    {
+                        // Reply with the client name in an Enlist message.
+                        IMessage enlistMessage = channel.CreateMessage();
+                        enlistMessage.Headers.SetString("CONTROL_TYPE", "ENLIST");
+                        enlistMessage.Headers.SetString("CLIENT_NAME", clientName);
+                        enlistMessage.Headers.SetString("CLIENT_PRIVATE_CONTROL_KEY", "iop.control." + clientName);
+                        enlistMessage.CorrelationId = message.CorrelationId;
+
+                        Send(enlistMessage, message.ReplyToRoutingKey);
+                    }                    
+                }
+                else if ("ASSIGN_ROLE" == controlType)
+                {
+                    // Assign the role to the current test case.
+                    string roleName = message.Headers.GetString("ROLE");
+    
+                    log.Info("Got a role assignment to role: " + roleName);
+    
+                    Roles role;
+
+                    if (roleName == "SENDER")
+                    {
+                        role = Roles.SENDER;
+                    }
+                    else
+                    {
+                        role = Roles.RECEIVER;
+                    }
+    
+                    currentTestCase.AssignRole(role, message);
+
+                    // Reply by accepting the role in an Accept Role message.
+                    IMessage acceptRoleMessage = channel.CreateMessage();
+                    acceptRoleMessage.Headers.SetString("CONTROL_TYPE", "ACCEPT_ROLE");
+                    acceptRoleMessage.CorrelationId = message.CorrelationId;
+
+                    Send(acceptRoleMessage, message.ReplyToRoutingKey);
+                }
+                else if ("START" == controlType || "STATUS_REQUEST" == controlType)
+                {
+                    if ("START" == controlType)
+                    {
+                        log.Info("Got a start notification.");
+
+                        // Start the current test case.
+                        currentTestCase.Start();
+                    }
+                    else
+                    {
+                        log.Info("Got a status request.");
+                    }
+
+                    // Generate the report from the test case and reply with it as a Report message.
+                    IMessage reportMessage = currentTestCase.GetReport(channel);
+                    reportMessage.Headers.SetString("CONTROL_TYPE", "REPORT");
+                    reportMessage.CorrelationId = message.CorrelationId;
+
+                    Send(reportMessage, message.ReplyToRoutingKey);
+                }
+                else if ("TERMINATE" == controlType)
+                {
+                    Console.WriteLine("Received termination instruction from coordinator.");
+
+                    // Is a cleaner shutdown needed?
+                    System.Environment.Exit(1);
+                }
+                else
+                {
+                    // Log a warning about this but otherwise ignore it.
+                    log.Warn("Got an unknown control message, controlType = " + controlType + ", message = " + message);
+                }
+            }
+            catch (QpidException e)
+            {
+                // Log a warning about this, but otherwise ignore it.
+                log.Warn("A QpidException occurred whilst handling a message.");
+                log.Info("Got QpidException whilst handling message: " + message, e);
+            }
+        }
+
+        /// <summary>
+        /// Send the specified message using the specified routing key on the direct exchange.
+        /// </summary>
+        /// 
+        /// <param name="message">    The message to send.</param>
+        /// <param name="routingKey"> The routing key to send the message with.</param>
+        public void Send(IMessage message, string routingKey)
+        {
+            IMessagePublisher publisher = publisherBuilder.WithRoutingKey(routingKey).Create();
+            publisher.Send(message);
+        }
+    }
+}