blob: fd83825a61a81415bf61b76cb671fd4b9f329e96 [file] [log] [blame]
/*
*
* 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.Text;
using log4net;
using Apache.Qpid.Messaging;
namespace Apache.Qpid.Integration.Tests.interop.TestCases
{
///
/// Implements test case 4, from the interop test specification. This test sets up the TC2_P2PMessageSize test for 50
/// messages, and a variety of message sizes. It checks that the sender and receivers reports both indicate that all
/// the test messages were transmitted successfully.
///
/// <p><table id="crc"><caption>CRC Card</caption>
/// <tr><th> Responsibilities <th> Collaborations
/// <tr><td> Setup p2p test parameters and compare with test output. <td> {@link FrameworkBaseCase}
/// </table>
///
public class TestCase4P2PMessageSize : InteropClientTestCase
{
/// Used for debugging.
private static ILog log = LogManager.GetLogger(typeof(TestCase4P2PMessageSize));
/// <summary> The role to be played by the test. </summary>
private Roles role;
/// <summary> Holds the count of test messages received. </summary>
private int messageCount;
///<summary>The size of the message to be sent </summary>
private int messageSize;
/// <summary> The number of test messages to send. </summary>
private int numMessages;
/// <summary> The number of receiver connection to use. </summary>
private int numReceivers;
/// <summary> The routing key to send them to on the default direct exchange. </summary>
private string sendDestination;
/// <summary> The connections to send/receive the test messages on. </summary>
private IConnection[] connection;
/// <summary> The sessions to send/receive the test messages on. </summary>
private IChannel[] channel;
/// <summary> The producer to send the test messages with. </summary>
IMessagePublisher publisher;
/// <summary>
/// Creates a new coordinating test case with the specified name.
///</summary>
/// <returns>The test case name.</returns>
///
public String GetName()
{
log.Info("public String GetName(): called");
return "TC4_P2PMessageSize";
}
/// <summary>
/// Determines whether the test invite that matched this test case is acceptable.
/// </summary>
///
/// <param name="inviteMessage"> The invitation to accept or reject. </param>
///
/// <returns> <tt>true</tt> to accept the invitation, <tt>false</tt> to reject it. </returns>
public bool AcceptInvite(IMessage inviteMessage)
{
log.Info("public boolean AcceptInvite(IMessage inviteMessage = " + inviteMessage + "): called");
// All invites are acceptable.
return true;
}
public void Start()
{
log.Info("public void start(): called");
// Assuming numMessages = 1
Start(1);
}
public void Start(int numMessages)
{
log.Info("public void start("+numMessages+"): called");
// Check that the sender role is being performed.
if (role == Roles.SENDER)
{
IMessage testMessage = createMessageOfSize(messageSize);
for (int i = 0; i < numMessages; i++)
{
publisher.Send(testMessage);
// Increment the message count.
messageCount++;
}
}
}
private IMessage createMessageOfSize(int size)
{
IBytesMessage message = channel[0].CreateBytesMessage();
string messageStr = "Test Message -- Test Message -- Test Message -- Test Message -- Test Message -- Test Message -- Test Message -- ";
System.Text.ASCIIEncoding encoding = new System.Text.ASCIIEncoding();
byte[] messageBytes = encoding.GetBytes(messageStr);
if (size > 0)
{
int div = size / messageBytes.Length;
int mod = size % messageBytes.Length;
for (int i = 0; i < div; i++)
{
message.WriteBytes(messageBytes);
}
if (mod != 0)
{
message.WriteBytes(messageBytes, 0, mod);
}
}
return message;
}
public void AssignRole(Roles role, IMessage assignRoleMessage)
{
log.Info("public void assignRole(Roles role = " + role + ", IMessage assignRoleMessage = " + assignRoleMessage
+ "): called");
// Reset the message count for a new test.
messageCount = 0;
// Take note of the role to be played.
this.role = role;
// Extract and retain the test parameters.
numMessages = assignRoleMessage.Headers.GetInt("P2P_NUM_MESSAGES");
messageSize = assignRoleMessage.Headers.GetInt("messageSize");
string sendKey = assignRoleMessage.Headers.GetString("P2P_QUEUE_AND_KEY_NAME");
sendDestination = sendKey;
log.Info("numMessages = " + numMessages);
log.Info("messageSize = " + messageSize);
log.Info("sendKey = " + sendKey);
log.Info("role = " + role);
switch (role)
{
// Check if the sender role is being assigned, and set up a single message producer if so.
case Roles.SENDER:
// Create a new connection to pass the test messages on.
connection = new IConnection[1];
channel = new IChannel[1];
connection[0] =
TestClient.CreateConnection(TestClient.brokerUrl, TestClient.virtualHost);
channel[0] = connection[0].CreateChannel(false, AcknowledgeMode.AutoAcknowledge);
// Extract and retain the test parameters.
publisher = channel[0].CreatePublisherBuilder()
.WithExchangeName(ExchangeNameDefaults.TOPIC)
.WithRoutingKey(sendDestination)
.WithMandatory(false)
.WithImmediate(false)
.Create();
break;
// Otherwise the receiver role is being assigned, so set this up to listen for messages on the required number
// of receiver connections.
case Roles.RECEIVER:
// Create the required number of receiver connections.
connection = new IConnection[1];
channel = new IChannel[1];
connection[0] = TestClient.CreateConnection(TestClient.brokerUrl, TestClient.virtualHost);
channel[0] = connection[0].CreateChannel(false, AcknowledgeMode.AutoAcknowledge);
IMessageConsumer consumer = channel[0].CreateConsumerBuilder(sendDestination).Create();
consumer.OnMessage += new MessageReceivedDelegate(OnMessage);
break;
}
// Start all the connection dispatcher threads running.
foreach (IConnection con in connection)
{
con.Start();
}
}
public IMessage GetReport(IChannel channel)
{
log.Info("public Message GetReport(IChannel channel): called");
// Close the test connection.
//connection.Stop();
// Generate a report message containing the count of the number of messages passed.
IMessage report = channel.CreateMessage();
//report.Headers.SetString("CONTROL_TYPE", "REPORT");
report.Headers.SetInt("MESSAGE_COUNT", messageCount);
return report;
}
/// <summary>
/// Counts incoming test messages.
/// </summary>
///
/// <param name="message"> The incoming test message. </param>
public void OnMessage(IMessage message)
{
log.Info("public void onMessage(IMessage message = " + message + "): called");
// Increment the message count.
messageCount++;
}
}
}