379 lines
16 KiB
C#
379 lines
16 KiB
C#
/*
|
|
*
|
|
* 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));
|
|
|
|
// 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);
|
|
}
|
|
}
|
|
}
|