Benn wrote:
Hi Gordon,
Just looking through the log and seen the following...
2009-jul-06 09:23:45 debug SessionState::SessionState
[email protected]: 0x9086108
2009-jul-06 09:23:45 debug
[email protected]: attached on broker.
2009-jul-06 09:23:45 debug Attached channel 1 to
[email protected]
2009-jul-06 09:23:45 debug
[email protected]: ready to send,
activating output.
2009-jul-06 09:23:45 trace 209.160.72.159:2144-12(local) activateOutput -
sending doOutput
2009-jul-06 09:23:45 trace MCAST 209.160.72.159:2144-12:
{ClusterConnectionDeliverDoOutputBody: bytes=41; }
2009-jul-06 09:23:45 trace 209.160.72.159:2144-12(local)Send doOutput
request for 41
2009-jul-06 09:23:45 trace SENT [217.41.62.170:7445]: Frame[BEbe; channel=1;
{SessionAttachedBody: name=1cf3d1fbf-1ac35-14db7-19fa0-113c720be1772; }]
2009-jul-06 09:23:45 trace SENT [217.41.62.170:7445]: Frame[BEbe; channel=1;
{SessionCommandPointBody: command-id=0; command-offset=0; }]
2009-jul-06 09:23:45 trace 209.160.72.159:2144(READY) DLVR:
Event[209.160.72.159:2144-12 control 22 bytes]
2009-jul-06 09:23:45 trace 209.160.72.159:2144(READY) DLVR: Frame[BEbe;
channel=0; {ClusterConnectionDeliverDoOutputBody: bytes=41; }] control
209.160.72.159:2144-12
2009-jul-06 09:23:45 trace 209.160.72.159:2144-12(local) delivereDoOutput:
requested=41 sent=0 more=0
2009-jul-06 09:23:45 trace MCAST Event[209.160.72.159:2144-15 data 59 bytes]
2009-jul-06 09:23:45 trace 209.160.72.159:2144(READY) DLVR:
Event[209.160.72.159:2144-15 data 59 bytes]
2009-jul-06 09:23:45 trace 209.160.72.159:2144(READY) DLVR: Frame[BEbe;
channel=1; {SessionAttachBody:
name=1cf3d1fbf-1ac35-14db7-19fa0-113c720be1772; }] data 209.160.72.159:2144-15
read-credit=1
2009-jul-06 09:23:45 debug Exception constructed: Session already attached:
[email protected](qpid/broker/SessionManager.cpp:55)
2009-jul-06 09:23:45 error Channel exception: session-busy: Session already
attached:
[email protected](qpid/broker/SessionManager.cpp:55)
2009-jul-06 09:23:45 trace SENT [217.41.62.170:13894]: Frame[BEbe;
channel=1; {SessionDetachedBody:
name=1cf3d1fbf-1ac35-14db7-19fa0-113c720be1772; code=1; }]
2009-jul-06 09:23:45 trace MCAST Event[209.160.72.159:2144-13 data 59 bytes]
Any help much appreciated.
From this it looks like either the same uuid is being used for two
different sessions, or somehow the attach control is being sent (or
received twice). Could you include some of the lines just before this
snippet to try and work out which of these it is?
Also, do you see the same issue with a standalone broker (i.e.
non-clustered)?
2009/7/6 Gordon Sim <[email protected]>
Benn wrote:
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 :)
Do you see any errors logged on the broker?
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 :)
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:[email protected]
