qpid-users mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Benn <mr.b...@gmail.com>
Subject c# multiple producers on seperate threads
Date Fri, 03 Jul 2009 16:39:34 GMT
Hi There,
I'm hoping someone has come accross this issue before or can maybe shed some
light on why this is happening.

I have ported a test application i wrote for RabbitMQ (which worked fine)
but it throws a SessionClosedException when multiple producers are started.
Perhaps i am going about this all the wrong way for Qpid - but it seemed
right to me... any assistance would be appreciated :)

Here is my (very) simple app:

--------------------------------------------------------------------------------------------

using System;
using System.Collections;
using System.Text;
using System.Threading;
using org.apache.qpid.client;
using org.apache.qpid.transport;

namespace QTest
{
class Program
{
 static Queue testQueue = Queue.Synchronized(new Queue());
static bool run = true;
 public static void Main(string[] args)
{
int messages = int.Parse(args[0]);
int producers = int.Parse(args[1]);
 PopulateQueue(messages);
 DateTime start = DateTime.Now;
 for (int x = 0; x < producers ; x++ ) {
Thread thd = new Thread(new ThreadStart(Runner));
thd.IsBackground = true;
thd.Start();
}
 while (testQueue.Count > 0) {
Thread.Sleep(10);
}
 TimeSpan end = DateTime.Now.Subtract(start);
 Console.WriteLine("{0} producers took {1} seconds to enqueue {2} messages",
producers, end.TotalSeconds, messages);
 run = false;
 Console.ReadLine();
}
 public static void PopulateQueue(int count)
{
for (int i = 0; i < count ; i++) {
testQueue.Enqueue(string.Format("Message: {0}", i));
}
}
 public static void Runner()
{
Client connection = new Client();
connection.connect("redrabbits.co.uk", 5672, string.Empty, "guest",
"guest");
ClientSession session = connection.createSession(50000);
 IMessage message = new Message();
message.DeliveryProperties.setRoutingKey("testing");
while (run) {
if(testQueue.Count > 0) {
object thisMsg = testQueue.Dequeue();
if(thisMsg != null) {
 // put it in message queue
message.clearData();
message.appendData(Encoding.UTF8.GetBytes(thisMsg.ToString()));
session.messageTransfer("amq.direct", message);
}
}
Thread.Sleep(10);
}
 session.sync();
connection.close();
}
}
}


-------------------------------------------------------------------------------------

Many thanks in advance :)

-- 
Cheers,
Benn

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message