Saturday, September 13, 2008

How to read messages from queue


The problem is how to read messages without removing them from the queue when NMS (C#) client is used.
Unfortunately ActiveMQ does not expose its administrative interfaces so it is not possible to obtain list of queues, or enumerate messages in the queue without removing them.


But knowledge of how queue server works gives us opportunity to cheat. If we open session with ClientAcknowledge mode then server will wait for clien issued confirmation before deleting message from queue. If we intentionally skip confirming than we can keep reading messages up to some limit.


using(IConnection conn = new ConnectionFactory("tcp://localhost:61616").CreateConnection())
{
conn.Start();
ISession session = conn.CreateSession(AcknowledgementMode.ClientAcknowledge);
IQueue queue = session.GetQueue("aaa");
for(int i=0; i<10; i++)
producer.Send(producer.CreateTextMessage("a test "+i));

List<IMessage> list = new List<IMessage>();
using(IMessageConsumer iterator = session.CreateConsumer(queue)) {
iterator.Listener += delegate(IMessage m) {list.Add(m);};
System.Threading.Thread.Sleep(250);
}
Console.WriteLine("First attempt: {0}", list.Count);

// walk through messages again to make sure we did not remove them first time
list.Clear();
using(IMessageConsumer iterator = session.CreateConsumer(queue)) {
iterator.Listener += delegate(IMessage m) {list.Add(m);};
System.Threading.Thread.Sleep(250);
}
Console.WriteLine("Second attempt: {0}", list.Count);



It is important to open session in ClientAcknowledge and not acknowledge messages to avoid their removal from the queue. Also I wrapped consumer into "using" clause in order to call Dispose which will inform queue server that messages sent to this consumer are free. We want to do it as soon as we are done and not relay on garbage collector which can kick in who knows when.


The sleep for 250ms is to give time to consumer to fulfill its buffer. See my notes about problem here.


The method described is a trick and there is a limit on how many messages you can receive. ActiveMQ will throttle consumer which receives messages but does not confirm their processing. But you hould be fine with showning first hundred or so messages. It would serve displaying purposes.

NMS: Consuming ActiveMQ messages Asynchronously



There are some articles around which demonstrate how to implement async message consumer in C# but they employ Spring Framework. In my opinion it is not justified on many accounts. Spring may be good for your project or may be not.

So here we go:

using System;
using Apache.NMS;
using Apache.NMS.ActiveMQ;

namespace NMSTestConsole
{
class MainClass
{
public static void Main(string[] args)
{
using(IConnection conn = new ConnectionFactory("tcp://localhost:61616").CreateConnection())
{
conn.Start();
ISession session = conn.CreateSession();
IQueue queue = session.GetQueue("aaa");
IMessageProducer producer = session.CreateProducer(queue);

IMessageConsumer consumer3 = session.CreateConsumer(session.GetQueue("aaa"));
consumer3.Listener += delegate(IMessage m) {Console.WriteLine("aaa: \n{0}",m);};

producer.Send(producer.CreateTextMessage("a test"));

Console.WriteLine("Press any key...");
Console.ReadKey();
}
}
}
}


Look ma, no Spring!
Please pay attention to conn.Start() call. Without it your consumers will remain in synchronous mode and will not serve messages to their listeners.
But if you do not assign a listener to your consumer then you still can do it synchronous way: "Console.WriteLine(consumer3.Receive());" and connection can remain in asynchronous mode.

Saturday, August 30, 2008

Language complexity irony



I'm in process of learning Java deeper and I discover Generics. No, I don't mean "Java has generics. Wow!" I mean corner cases like this:

class MyClass<T extends MyClass<T>>

Or this one:

What is the difference between a Collection<Pair<String,Object>>, a Collection<Pair<String,?>> and a Collection<? extends Pair<String,?>>?


It takes significant effort to imagine what such class is and what are its usages and limitations.
The irony I see is that Java initially was introduced as simplified C++. Not just C++ with garbage collector but the structure of language was simplified in order to become programming language for masses. And what we see 10 years later? Java became no less difficult then C++ in some aspects. Programming is too complex area to do it with simple tools.

Lesson learned: do things as simple as possible but not simpler :)

Thursday, August 28, 2008

Vadim's blog

Vadim
I am guessing that log4net wasn't included in order to simplify binary dependencies. But including it makes a lot of sense, log4net is quite popular in c# community.

Tracing in NMS



It easy to turn protocol tracing on in NMS client.
First of all you need to implement your trace class. I am lazy so I wrote an adapter for log4net.


/*
* ----------------------------------------------------------------------------
* "THE BEER-WARE LICENSE"
* Vadim Chekan wrote this file. As long as you retain this notice you
* can do whatever you want with this stuff. If we meet some day, and you think
* this stuff is worth it, you can buy me a beer in return
* ----------------------------------------------------------------------------
*/

using System;
using log4net;
using Apache.NMS;

namespace NMSTestConsole
{
public class TraceAdapter : Apache.NMS.ITrace
{
// keep logs on behalf of NMS Tracer
static readonly ILog _log = LogManager.GetLogger(typeof(Apache.NMS.Tracer));

public bool IsDebugEnabled {
get {return _log.IsDebugEnabled;}
}

public bool IsInfoEnabled {
get {return _log.IsInfoEnabled;}
}

public bool IsWarnEnabled {
get {return _log.IsWarnEnabled;}
}

public bool IsErrorEnabled {
get {return _log.IsErrorEnabled;}
}

public bool IsFatalEnabled {
get {return _log.IsFatalEnabled;}
}

public void Debug (string message)
{
_log.Debug(message);
}

public void Info (string message)
{
_log.Info(message);
}

public void Warn (string message)
{
_log.Warn(message);
}

public void Error (string message)
{
_log.Error(message);
}

public void Fatal (object message)
{
_log.Fatal(message);
}
}
}


Now you need hook it up somwhere at the very beginning of your program.

using Apache.NMS;
using Apache.NMS.ActiveMQ;
...
Tracer.Trace = new TraceAdapter();
log4net.Config.BasicConfigurator.Configure();


That's it. Run your program which communicates to ActiveMQ and see the result.

Getting more info


There is not too much information right now. Output looks like:

Parsing type: 1 with: Apache.NMS.ActiveMQ.OpenWire.V1.WireFormatInfoMarshaller
Parsing type: 2 with: Apache.NMS.ActiveMQ.OpenWire.V2.BrokerInfoMarshaller
Parsing type: 30 with: Apache.NMS.ActiveMQ.OpenWire.V2.ResponseMarshaller


Those numbers are ID of OpenWire packet types. Probably you would prefer something more human readable. You can achieve it by setting UseLogging property of TcpTransportFactory. The best wat of doing it is via broker url parameters:

IConnection conn = new ConnectionFactory("tcp://localhost:61616?transport.UseLogging=true").CreateConnection();



SENDING: WireFormatInfo[ Magic=System.Byte[] Version=2 MarshalledProperties={CacheEnabled=False, SizePrefixDisabled=False, StackTraceEnabled=False, TcpNoDelayEnabled=False, TightEncodingEnabled=False} ]
414 [-1225462896] DEBUG Apache.NMS.Tracer (null) - Parsing type: 1 with: Apache.NMS.ActiveMQ.OpenWire.V1.WireFormatInfoMarshaller
424 [-1225462896] INFO Apache.NMS.Tracer (null) - RECEIVED: WireFormatInfo[ Magic=System.Byte[] Version=3 MarshalledProperties={CacheEnabled=True, CacheSize=1024, SizePrefixDisabled=False, TightEncodingEnabled=True, MaxInactivityDuration=30000, MaxInactivityDurationInitalDelay=10000, StackTraceEnabled=True, TcpNoDelayEnabled=True} ]

SENDING: ConnectionInfo[ ConnectionId=ConnectionId[ Value=2957a0f0-a8b7-4abc-96ea-e5638b6d0baa ] ClientId=49b903c4-f55e-40b0-a78c-bde850b77ef8 Password= UserName= BrokerPath= BrokerMasterConnector=False Manageable=False ClientMaster=True ]
453 [-1225462896] DEBUG Apache.NMS.Tracer (null) - Parsing type: 2 with: Apache.NMS.ActiveMQ.OpenWire.V2.BrokerInfoMarshaller
463 [-1225462896] INFO Apache.NMS.Tracer (null) - RECEIVED: BrokerInfo[ BrokerId=BrokerId[ Value=ID:ubuntu-47413-1219891509239-0:0 ] BrokerURL=tcp://ubuntu:61616 PeerBrokerInfos=Apache.NMS.ActiveMQ.Commands.BrokerInfo[] BrokerName=localhost SlaveBroker=False MasterBroker=False FaultTolerantConfiguration=False DuplexConnection=False NetworkConnection=False ConnectionId=0 ]
463 [-1225462896] DEBUG Apache.NMS.Tracer (null) - Parsing type: 30 with: Apache.NMS.ActiveMQ.OpenWire.V2.ResponseMarshaller
463 [-1225462896] INFO Apache.NMS.Tracer (null) - RECEIVED: Response[ CorrelationId=1 ]
474 [-1211062496] INFO Apache.NMS.Tracer (null) - SENDING: SessionInfo[ SessionId=SessionId[ ConnectionId=2957a0f0-a8b7-4abc-96ea-e5638b6d0baa Value=1 ] ]
475 [-1225462896] DEBUG Apache.NMS.Tracer (null) - Parsing type: 30 with: Apache.NMS.ActiveMQ.OpenWire.V2.ResponseMarshaller
475 [-1225462896] INFO Apache.NMS.Tracer (null) - RECEIVED: Response[ CorrelationId=2 ]



Now you can have much better idea what your client is chatting about with the server.

Wednesday, August 27, 2008

NMS gotcha



I use NMS C# client to get messages from ActiveMQ and I faced a problem: despite there were messages in the queue my "consumer.ReceiveNoWait()" returned nothing.


After some code reading I realized the reason. My code looked like


using (IMessageConsumer consumer = session.CreateConsumer(queue, filter))
IMessage oldMessage;
while ((oldMessage = consumer.ReceiveNoWait() != null)
...



Apparently ReceiveNoWait does not send any request to the queue server. What it does is checking its own buffer of received messages.
When you create a Message Consumer then you notify the queue server that you want to receive messages from certain Target (Queue). But messages are not sent as registration response. So you have consumer registered but no messages yet.
Now, if you try to call consumer.ReceiveNoWait() right after registration you will get null. You need to wait a little bit before queue will send messages to the consumer.

/
Adding 100ms wait helps to address this problem:


using (IMessageConsumer consumer = session.CreateConsumer(queue, filter))
IMessage oldMessage;
while ((oldMessage = consumer.Receive(TimeSpan.FromMilliseconds(100))) != null)
...

Monday, August 25, 2008

Camel internals

I'm playing with Apache Camel project and boy, it is not easy to understand it internals!
Lets see how it is initialized.

SpringCamelContext(DefaultCamelContext).doStart() line: 543
SpringCamelContext.maybeDoStart() line: 165
SpringCamelContext.doStart() line: 160
SpringCamelContext(ServiceSupport).start() line: 47
SpringCamelContext.maybeStart() line: 95
SpringCamelContext.onApplicationEvent(ApplicationEvent) line: 114

So maybeStart calls start which calls doStart which calls maybeDoStart which calls doStart.
Now question from combinatorics: how many function names can we generate from 3 words: "maybe", "do", "start" :)))