I am using qpid's last stable release: M2
I am trying to subscribe to a topic using the C# API, publishing using the Java
API (JMS) and the broker is also Java (rather than the C++ version).
It looks like my C# program is correctly listening to incoming messages (my
'OnMessage' function is called repeatedly), however, there doesn't seem to be
anything of value in the IMessage payload.
I get the following exception:
System.ArgumentNullException: Value cannot be null.
Parameter name: uriString
at System.Uri..ctor(String uriString)
at Apache.Qpid.Client.Message.BindingURL.Parse()
at Apache.Qpid.Client.Message.BindingURL..ctor(String url)
at Apache.Qpid.Client.Message.AbstractQmsMessage.ReadReplyToHeader()
at Apache.Qpid.Client.Message.AbstractQmsMessage.get_ReplyToExchangeName()
at Apache.Qpid.Client.Message.AbstractQmsMessage.ToString()
But I can't figure out where this is coming from. This is apparently not even
a qpid specific exception, people on the web have see something similar with
ASP.NET, web services, etc.
Following is the toString result of the message I send through my Java API:
Body:
{price=10.617051124572754, symbol=MSFT, size=8185460096350332850}
JMS Correlation ID: null
JMS timestamp: 0
JMS expiration: 0
JMS priority: 0
JMS delivery mode: 0
JMS reply to: null
JMS Redelivered: false
JMS Destination: null
JMS Type: null
JMS MessageID: ID:a3bb43c2-c51d-4c56-b55f-6bcab0f4f314
AMQ message number: -1
Properties:<NONE>
My code is a basically a simplified version of TopicListener.cs, available in
qpid's svn.
Please let me know if I miseed something here.
Following is my code:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Apache.Qpid.Messaging;
using Apache.Qpid.Client.Qms;
using Apache.Qpid.Client;
namespace QPidSubscriber
{
class QPidSubscriber
{
Apache.Qpid.Messaging.IConnection connection;
IConnectionInfo connectionInfo;
IChannel channel;
IMessageConsumer consumer;
string qpidServer = "amqp://guest:[EMAIL
PROTECTED]/test?brokerlist='tcp://localhost:5672'";
string topicName = "md.bids";
string fieldName = "size";
object fieldValue;
public QPidSubscriber()
{
initializeQpid(qpidServer);
initializeNewTopicField(topicName, fieldName);
}
void initializeQpid(string url)
{
connectionInfo = QpidConnectionInfo.FromUrl(url);
connection = new AMQConnection(connectionInfo);
channel = connection.CreateChannel(false, AcknowledgeMode.AutoAcknowledge, 1);
//connection.Start();
}
void initializeNewTopicField(string topic, string field)
{
string tempQ = channel.GenerateUniqueName();
channel.DeclareQueue(tempQ, false, true, true);
channel.Bind(tempQ, ExchangeNameDefaults.TOPIC, topic);
consumer = channel.CreateConsumerBuilder(tempQ).Create();
consumer.OnMessage += new MessageReceivedDelegate(OnMessage);
connection.Start();
}
public void OnMessage(IMessage message)
{
//fieldValue = message.Headers.GetLong(fieldName).ToString();
fieldValue = message.Headers[fieldName];
Console.WriteLine(fieldValue);//<== fieldValue is always null!!
Console.WriteLine(message);
}
static void Main(string[] args)
{
QPidSubscriber qps = new QPidSubscriber();
Console.ReadLine();
}
}
}
