1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
|
/*
*
* 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.Threading;
using log4net;
using Apache.Qpid.Messaging;
using Apache.Qpid.Client.Qms;
namespace Apache.Qpid.Client.Tests.interop
{
public class TopicListener
{
private static ILog log = LogManager.GetLogger(typeof(TopicListener));
/// <summary> The default AMQ connection URL to use for tests. </summary>
const string DEFAULT_URI = "amqp://guest:guest@default/test?brokerlist='tcp://localhost:5672'";
/// <summary> Holds the routing key for the topic to receive test messages on. </summary>
public static string CONTROL_ROUTING_KEY = "topic_control";
/// <summary> Holds the routing key for the queue to send reports to. </summary>
public static string RESPONSE_ROUTING_KEY = "response";
/// <summary> Holds the connection to listen on. </summary>
private IConnection connection;
/// <summary> Holds the channel for all test messages.</summary>
private IChannel channel;
/// <summary> Holds the producer to send report messages on. </summary>
private IMessagePublisher publisher;
/// <summary> A monitor used to wait for shutdown. </summary>
private AutoResetEvent shutdownReceivedEvt = new AutoResetEvent(false);
/// <summary> Holds the default test timeout for communications . </summary>
const int TIMEOUT = 60000;
/// <summary> Holds a flag to indicate that a timer has begun on the first message. Reset when report is sent. </summary> */
private bool init;
/// <summary> Holds the count of messages received by this listener. </summary> */
private int count;
/// <summary> Creates a topic listener using the specified broker URL. </summary>
///
/// <param name="connectionUri">The broker URL to listen on.</param>
TopicListener(string connectionUri)
{
log.Debug("TopicListener(string connectionUri = " + connectionUri + "): called");
// Create a connection to the broker.
IConnectionInfo connectionInfo = QpidConnectionInfo.FromUrl(connectionUri);
connection = new AMQConnection(connectionInfo);
// Establish a session on the broker.
channel = connection.CreateChannel(false, AcknowledgeMode.AutoAcknowledge, 1);
// Set up a queue to listen for test messages on.
string topicQueueName = channel.GenerateUniqueName();
channel.DeclareQueue(topicQueueName, false, true, true);
// Set this listener up to listen for incoming messages on the test topic queue.
channel.Bind(topicQueueName, ExchangeNameDefaults.TOPIC, CONTROL_ROUTING_KEY);
IMessageConsumer consumer = channel.CreateConsumerBuilder(topicQueueName)
.Create();
consumer.OnMessage += new MessageReceivedDelegate(OnMessage);
// Set up this listener with a producer to send the reports on.
publisher = channel.CreatePublisherBuilder()
.WithExchangeName(ExchangeNameDefaults.DIRECT)
.WithRoutingKey(RESPONSE_ROUTING_KEY)
.Create();
connection.Start();
Console.WriteLine("Waiting for messages...");
if (shutdownReceivedEvt.WaitOne(TIMEOUT, true))
{
Console.WriteLine("Shutting down - shut down message was received");
}
else
{
Console.WriteLine("Shutting down - timeout elapsed");
}
}
public static void Main(String[] argv)
{
// Create an instance of this listener with the command line parameters.
new TopicListener(DEFAULT_URI);
}
/// <summary>
/// Handles all message received by this listener. Test messages are counted, report messages result in a report being sent and
/// shutdown messages result in this listener being terminated.
/// </summary>
///
/// <param name="message">The received message.</param>
public void OnMessage(IMessage message)
{
log.Debug("public void onMessage(Message message = " + message + "): called");
// Take the start time of the first message if this is the first message.
if (!init)
{
count = 0;
init = true;
}
// Check if the message is a control message telling this listener to shut down.
if (IsShutdown(message))
{
log.Debug("Got a shutdown message.");
Shutdown();
}
// Check if the message is a report request message asking this listener to respond with the message count.
else if (IsReport(message))
{
log.Debug("Got a report request message.");
// Send the message count report.
SendReport();
// Reset the initialization flag so that the next message is considered to be the first.
init = false;
}
// Otherwise it is an ordinary test message, so increment the message count.
else
{
count++;
}
}
/// <summary> Checks a message to see if it is a shutdown control message. </summary>
///
/// <param name="m">The message to check.</param>
///
/// <returns><tt>true</tt> if it is a shutdown control message, <tt>false</tt> otherwise.</returns>
private bool IsShutdown(IMessage m)
{
bool result = CheckTextField(m, "TYPE", "TERMINATION_REQUEST");
//log.Debug("isShutdown = " + result);
return result;
}
/// <summary> Checks a message to see if it is a report request control message. </summary>
///
/// <param name="m">The message to check.</param>
///
/// <returns><tt>true</tt> if it is a report request control message, <tt>false</tt> otherwise.</returns>
private bool IsReport(IMessage m)
{
bool result = CheckTextField(m, "TYPE", "REPORT_REQUEST");
//log.Debug("isReport = " + result);
return result;
}
/// <summary> Checks whether or not a text field on a message has the specified value. </summary>
///
/// <param name="e">The message to check.</param>
/// <param name="e">The name of the field to check.</param>
/// <param name="e">The expected value of the field to compare with.</param>
///
/// <returns> <tt>true</tt>If the specified field has the specified value, <tt>fals</tt> otherwise. </returns>
private static bool CheckTextField(IMessage m, string fieldName, string value)
{
/*log.Debug("private static boolean checkTextField(Message m = " + m + ", String fieldName = " + fieldName
+ ", String value = " + value + "): called");*/
string comp = m.Headers.GetString(fieldName);
return (comp != null) && comp == value;
}
/// <summary> Stops the message consumer and closes the connection. </summary>
private void Shutdown()
{
connection.Stop();
channel.Dispose();
connection.Dispose();
shutdownReceivedEvt.Set();
}
/// <summary> Sends the report message to the response location. </summary>
private void SendReport()
{
string report = "Received " + count + ".";
IMessage reportMessage = channel.CreateTextMessage(report);
reportMessage.Headers.SetBoolean("BOOLEAN", false);
//reportMessage.Headers.SetByte("BYTE", 5);
reportMessage.Headers.SetDouble("DOUBLE", 3.141);
reportMessage.Headers.SetFloat("FLOAT", 1.0f);
reportMessage.Headers.SetInt("INT", 1);
reportMessage.Headers.SetLong("LONG", 1);
reportMessage.Headers.SetString("STRING", "hello");
reportMessage.Headers.SetShort("SHORT", 2);
publisher.Send(reportMessage);
Console.WriteLine("Sent report: " + report);
}
}
}
|
