You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@qpid.apache.org by Shahbaz Chaudhary <sc...@marcopolonetwork.com> on 2008/06/10 22:09:17 UTC
RE: Excel (RTD) add-in?
Sorry for the late reply. Unfortunately I don't think .NET stuff helps
much with Excel's RTD.
I am already using Java and C# apis with the latest stable release (M2 I
believe).
Ironically, I get the impression that a windows C++ programmer could
link Qpid's C++ api and Excel's RTD protocol in a few hours.
The ability to allow Java/JMS, C#, C++ program, on any operating system,
to talk to Excel/RTD will make a hero out of any one :)
(http://support.microsoft.com/kb/327215)
-----Original Message-----
From: Robert Greig [mailto:robert.j.greig@gmail.com]
Sent: Friday, April 25, 2008 3:25 AM
To: qpid-users@incubator.apache.org
Subject: Re: Excel (RTD) add-in?
On 25/04/2008, Shahbaz Chaudhary <sc...@marcopolonetwork.com>
wrote:
> I'm surprised to see that there isn't an Excel add-in yet for Qpid or
> any other AMQP implementation. Are there any plans to develop one?
There is a Qpid .NET client, which can be easily used from within
Excel. There is also a WCF client produced by Rabbit
(http://www.rabbitmq.com/dotnet.html).
Would either of those meet your requirements?
Robert
Re: Maximum count on queue threshold breached
Posted by Carl Trieloff <cc...@redhat.com>.
>
> We would strongly recommend you use M2.1.
>
> RedHat MRG is based on trunk, and AFAIK does not have a compatible
> .NET client (can someone confirm that?).
>
>
yes, trunk & MRG is 0-10 -- .NET is still 0-8/0-9 so to use .NET you
need to use
Java broker with M2.1. Options increase once the 0-10 update for Java broker
and .NET are complete.
Carl.
Re: Maximum count on queue threshold breached
Posted by Robert Greig <ro...@gmail.com>.
2008/7/18 Shahbaz Chaudhary <sc...@marcopolonetwork.com>:
> Thanks. You are right, the server doesn't actually die. However, after
> I receive the "queue breached" message, my clients are not able to
> successfully connect.
That is worrying. What error do you see on the clients and/or broker?
> If the broker has this information, perhaps the error message could
> print what topic/exchange the queue was bound to.
Yes, that is a good idea.
> By the way, if I want to use my C# client as well, what is the latest
> usable release I should be using? Anything more recent than the M2? Is
> the RedHat MRG release based on M2 or later?
We would strongly recommend you use M2.1.
RedHat MRG is based on trunk, and AFAIK does not have a compatible
.NET client (can someone confirm that?).
RG
RE: Maximum count on queue threshold breached
Posted by Shahbaz Chaudhary <sc...@marcopolonetwork.com>.
Thanks. You are right, the server doesn't actually die. However, after
I receive the "queue breached" message, my clients are not able to
successfully connect.
If the broker has this information, perhaps the error message could
print what topic/exchange the queue was bound to.
By the way, if I want to use my C# client as well, what is the latest
usable release I should be using? Anything more recent than the M2? Is
the RedHat MRG release based on M2 or later?
Thanks again.
-----Original Message-----
From: Robert Greig [mailto:robert.j.greig@gmail.com]
Sent: Thursday, July 17, 2008 4:50 PM
To: qpid-users@incubator.apache.org
Subject: Re: Maximum count on queue threshold breached
2008/7/17 Shahbaz Chaudhary <sc...@marcopolonetwork.com>:
> I'm trying to make sense of these errors.
> --How can I find out which queue breached? tmp_0234282whatever doesn't
> mean anything to me, it is created by one of the clients. I can't use
> JMX to find out where this queue is bound since the server is already
> dead by the time I see this.
This is something that has been discussed previously. I am not sure if
this made it into the M2.x codebase or whether it is slated for trunk.
Maybe someone who was involved in that discussion can give us an
update?
However I don't understand why "the server is dead"? The message is
informational and the server should be working normally.
> --Why does this happen? Clearly the queue is getting overfilled. Is
> this because data is not being read by the client fast enough?
Yes.
> I do notice that when I disconnect my client (I think only JMS
client),
> I get a "session closed implictly: java.io.IOException: Connection
reset
> by peer" error. This is despite the fact that I explicitly close the
> connection with calls to .close() on Connection. Is it possible that
> qpid doesn't realize that the client is no longer connected and keeps
> populating the queue...eventually causing these errors?
I don't think we have seen that error before. The server has clearly
detected the client has disconnected so it should clean up the temp
queue. Even if you were not calling close() it should still work fine.
It would be interesting to check using the management console (or JMX
directly) that the number of connections and queues/bindings to the
topic exchange are as you expect.
RG
Re: Maximum count on queue threshold breached
Posted by Robert Greig <ro...@gmail.com>.
2008/7/17 Shahbaz Chaudhary <sc...@marcopolonetwork.com>:
> I'm trying to make sense of these errors.
> --How can I find out which queue breached? tmp_0234282whatever doesn't
> mean anything to me, it is created by one of the clients. I can't use
> JMX to find out where this queue is bound since the server is already
> dead by the time I see this.
This is something that has been discussed previously. I am not sure if
this made it into the M2.x codebase or whether it is slated for trunk.
Maybe someone who was involved in that discussion can give us an
update?
However I don't understand why "the server is dead"? The message is
informational and the server should be working normally.
> --Why does this happen? Clearly the queue is getting overfilled. Is
> this because data is not being read by the client fast enough?
Yes.
> I do notice that when I disconnect my client (I think only JMS client),
> I get a "session closed implictly: java.io.IOException: Connection reset
> by peer" error. This is despite the fact that I explicitly close the
> connection with calls to .close() on Connection. Is it possible that
> qpid doesn't realize that the client is no longer connected and keeps
> populating the queue...eventually causing these errors?
I don't think we have seen that error before. The server has clearly
detected the client has disconnected so it should clean up the temp
queue. Even if you were not calling close() it should still work fine.
It would be interesting to check using the management console (or JMX
directly) that the number of connections and queues/bindings to the
topic exchange are as you expect.
RG
Maximum count on queue threshold breached
Posted by Shahbaz Chaudhary <sc...@marcopolonetwork.com>.
I'm using QPID M2 with a C# client which publishes lots of data to it,
one JMS client which publishes much smaller amount of data and Java
and/or C# clients which read off some data.
I publish approximately the same amount of data every day; however, I
receive the following errors just once in a while (I restart all my
servers every morning).
I'm trying to make sense of these errors.
--How can I find out which queue breached? tmp_0234282whatever doesn't
mean anything to me, it is created by one of the clients. I can't use
JMX to find out where this queue is bound since the server is already
dead by the time I see this.
--Why does this happen? Clearly the queue is getting overfilled. Is
this because data is not being read by the client fast enough? This
sounds unlikely since my largest volume data is prices and quotes for
about 100, mid to low liquidity stocks. My server hardware is medium
sized server level hardware.
I do notice that when I disconnect my client (I think only JMS client),
I get a "session closed implictly: java.io.IOException: Connection reset
by peer" error. This is despite the fact that I explicitly close the
connection with calls to .close() on Connection. Is it possible that
qpid doesn't realize that the client is no longer connected and keeps
populating the queue...eventually causing these errors?
--Errors--
2008-07-17 16:25:29,793 INFO [pool-2-thread-6] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 5000: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:25:59,835 INFO [pool-2-thread-2] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 5386: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:26:29,883 INFO [pool-2-thread-2] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 5786: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:26:59,924 INFO [pool-2-thread-3] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 6186: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:27:29,966 INFO [pool-2-thread-1] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 6586: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:28:00,005 INFO [pool-2-thread-8] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 6986: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:28:30,046 INFO [pool-2-thread-4] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 7386: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:29:00,086 INFO [pool-2-thread-2] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 7786: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:29:30,129 INFO [pool-2-thread-6] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 8186: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:30:00,167 INFO [pool-2-thread-2] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 8586: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:30:30,207 INFO [pool-2-thread-4] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 8986: Maximum count on queue
threshold (5000) breached.
2008-07-17 16:31:00,252 INFO [pool-2-thread-5] queue.AMQQueueMBean
(AMQQueueMBean.java:274) - MESSAGE_COUNT_ALERT On Queue
tmp_033d4cd6-f0d7-42c4-8b20-ce4aa4d61378 - 9386: Maximum count on queue
threshold (5000) breached.
Thanks
Re: JMS: No Route (error 312)
Posted by Robert Greig <ro...@gmail.com>.
2008/7/7 Shahbaz Chaudhary <sc...@marcopolonetwork.com>:
> In the link, someone advises that Qpid specific session should be used
> instead of JMS specific session. When topic producer is created
> (createProducer(...)), pass in false as the mandatory flag.
Yes, the mandatory flag is part of the underlying AMQP protocol.
> Two problems:
>
> Where do I use the qpid specific session? Is there a qpid specific
> version of TopicSession?
We have an "extended JMS" API which has interfaces that are subclasses
of javax.jms.Connection, Session etc but provide some extra methods
and fields.
To access these, cast your javax.jms.Session to an
org.apache.qpid.jms.Session and then you should see some extra methods
that allow you to specify the mandatory flag on the producer.
> Secondly, I am creating topic producer by using
> "session.createPublisher(mytopic)". I haven't found any document which
> describes the difference between .createPublisher and .createProducer.
> Which is the right one to use?
A TopicPublisher is a JMS 1.0 concept, it was unified in JMS 1.1 to a
producer. Use a producer since that is uniform across Queues and
Topics.
RG
JMS: No Route (error 312)
Posted by Shahbaz Chaudhary <sc...@marcopolonetwork.com>.
I am getting a 'no route' error when I publish to a topic which does not
yet have a subscriber. This problem is exactly the same as this:
http://markmail.org/message/qmtayydcdlacgl7f?q=qpid+%22no+route%22+error
+312&page=1&refer=zj4xsyfskjqgv5wb
In the link, someone advises that Qpid specific session should be used
instead of JMS specific session. When topic producer is created
(createProducer(...)), pass in false as the mandatory flag.
Two problems:
Where do I use the qpid specific session? Is there a qpid specific
version of TopicSession?
Secondly, I am creating topic producer by using
"session.createPublisher(mytopic)". I haven't found any document which
describes the difference between .createPublisher and .createProducer.
Which is the right one to use?
Thanks
RE: Excel RTD Server: Proof of Concept
Posted by Shahbaz Chaudhary <sc...@marcopolonetwork.com>.
Will do.
-----Original Message-----
From: Gordon Sim [mailto:gsim@redhat.com]
Sent: Friday, June 20, 2008 3:58 AM
To: qpid-users@incubator.apache.org
Subject: Re: Excel RTD Server: Proof of Concept
Shahbaz Chaudhary wrote:
> In excel, you just have to type the following formula:
>
=rtd("rtd.test",,"amqp://guest:guest@1/test?brokerlist='tcp://<host>:567
> 2'","<topic>","<field>")
> This will subscribe to a topic, each time an update is received, it
will
> retrieve a field and display it in Excel.
> Example:
>
>
=rtd("rtd.test",,"amqp://guest:guest@1/test?brokerlist='tcp://<host>:567
> 2'","md.bidsoffers","price")
>
> This will just display the price for all incoming bidsoffers (so
> MSFT/ORCL/EBAY will be mixed in)
Great job; this looks nice and to my mind the potential is obvious and
exciting! If you want to contribute code the preferred way is to create
a JIRA and attach the relevant files to that, checking the box that
officially grants the code to the ASF for use.
Re: Excel RTD Server: Proof of Concept
Posted by Gordon Sim <gs...@redhat.com>.
Shahbaz Chaudhary wrote:
> In excel, you just have to type the following formula:
> =rtd("rtd.test",,"amqp://guest:guest@1/test?brokerlist='tcp://<host>:567
> 2'","<topic>","<field>")
> This will subscribe to a topic, each time an update is received, it will
> retrieve a field and display it in Excel.
> Example:
>
> =rtd("rtd.test",,"amqp://guest:guest@1/test?brokerlist='tcp://<host>:567
> 2'","md.bidsoffers","price")
>
> This will just display the price for all incoming bidsoffers (so
> MSFT/ORCL/EBAY will be mixed in)
Great job; this looks nice and to my mind the potential is obvious and
exciting! If you want to contribute code the preferred way is to create
a JIRA and attach the relevant files to that, checking the box that
officially grants the code to the ASF for use.
Excel RTD Server: Proof of Concept
Posted by Shahbaz Chaudhary <sc...@marcopolonetwork.com>.
Hi All,
The following email contains an Excel RTD server which is able to
subscribe to information from an M2 Qpid server.
This is just a proof of concept, there are almost no optimizations,
check for leaks, etc. I'm not really a .NET programmer and I have never
done any COM programming (and no C++ since college).
In any case, try it out. It works for me. I haven't figured out how to
run it outside Visual Studio 2008 yet (no idea how to create install
packages, register assemblies, etc.).
QPID C# folks are welcome to use it as they wish. I probably won't
maintain this, hopefully someone else will.
In excel, you just have to type the following formula:
=rtd("rtd.test",,"amqp://guest:guest@1/test?brokerlist='tcp://<host>:567
2'","<topic>","<field>")
This will subscribe to a topic, each time an update is received, it will
retrieve a field and display it in Excel.
Example:
=rtd("rtd.test",,"amqp://guest:guest@1/test?brokerlist='tcp://<host>:567
2'","md.bidsoffers","price")
This will just display the price for all incoming bidsoffers (so
MSFT/ORCL/EBAY will be mixed in)
You can also add 'filters' to the RTD function. Just append two
parameters to the end of the previous RTD function, the first parameter
refers to the field you wish to use in comparison, the second parameter
refers to the value the first parameter must have.
Example:
=rtd("rtd.test",,"amqp://guest:guest@1/test?brokerlist='tcp://<host>:567
2'","md.bidsoffers","price",
"symbol","MSFT")
This will subscribe to the same information as the last RTD function,
but will only display prices for MSFT.
You can add as many filters as you like, just make sure you append the
RTD function with a filter field and a filter value.
--8<------------------------------RTDTest.cs----------------------------
-----
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;
using System.Runtime.InteropServices;
using Microsoft.Office.Interop.Excel;
namespace RTDTest
{
[ComVisible(true), ProgId("RTD.Test")]
public class RTDTest : IRtdServer
{
//QPID CACHE
Dictionary<string, IChannel> channelCache;//url, channel
Dictionary<string, int> channelCacheCount;
Dictionary<string, IMessageConsumer> topicCache;//url+topic,
consumer
Dictionary<string, int> topicCacheCount;
Dictionary<int, string> topicIDCache;//url+topic+field, topicid
Dictionary<string, IList<Tuple2<int, string>>>
topicTopicIDFieldCache;
Dictionary<int, Tuple2<string[], string[]>> filtersCache;
//END QPID CACHE
//IRTDServer Globals
IRTDUpdateEvent updateEvent;
Queue<Tuple2<int, object>> refreshQ;
//END IRTDServer Globals
public RTDTest()
{
channelCache = new Dictionary<string, IChannel>();
topicCache = new Dictionary<string, IMessageConsumer>();
channelCacheCount = new Dictionary<string, int>();
topicCacheCount = new Dictionary<string, int>();
topicIDCache = new Dictionary<int, string>();
topicTopicIDFieldCache = new Dictionary<string,
IList<Tuple2<int, string>>>();
filtersCache = new Dictionary<int, Tuple2<string[],
string[]>>();
refreshQ = new Queue<Tuple2<int, object>>();
}
//QPID METHODS
private IChannel getChannel(string url)
{
IChannel chan;
if (channelCache.ContainsKey(url))
{
chan = channelCache[url];
}
else
{
IConnectionInfo connectionInfo =
QpidConnectionInfo.FromUrl(url);
Apache.Qpid.Messaging.IConnection connection = new
AMQConnection(connectionInfo);
IChannel channel = connection.CreateChannel(false,
AcknowledgeMode.AutoAcknowledge, 1);
connection.Start();
chan = channel;
channelCache[url] = chan;
}
return chan;
}
private IMessageConsumer getTopicConsumer(string url, string
topic)
{
IMessageConsumer cons;
string key = url + topic;
if (topicCache.ContainsKey(key))
{
cons = topicCache[key];
}
else
{
IChannel channel = getChannel(url);
string tempQ = channel.GenerateUniqueName();
channel.DeclareQueue(tempQ, false, true, true);
cons = channel.CreateConsumerBuilder(tempQ).Create();
channel.Bind(tempQ, ExchangeNameDefaults.TOPIC, topic);
topicCache[key] = cons;
}
return cons;
}
private IList<Tuple2<int, string>> getFields(string topic)
{
if (!topicTopicIDFieldCache.ContainsKey(topic))
{
topicTopicIDFieldCache[topic] = new List<Tuple2<int,
string>>();
}
return topicTopicIDFieldCache[topic];
}
private void onMessage(IMessage msg, string url, string topic,
string field, int topicid)
{
foreach (Tuple2<int, string> f in getFields(topic))//?
{
int id = f.a;
object value = msg.Headers[f.b];
//Dictionary<int, object> d = new Dictionary<int,
object>();
//d.Add(id, value);
string[] filterFields = filtersCache[id].a;
string[] filterVals = filtersCache[id].b;
if(allFiltersTrue(filterFields,filterVals,msg)){
refreshQ.Enqueue(new
Tuple2<int,object>(id,value));
}
}
try
{
updateEvent.UpdateNotify();
}
catch (COMException e)
{
}
}
void registerTopicID(string url, string topic, string field, int
topicid)
{
string val = url + "|" + topic + "|" + field;
topicIDCache.Add(topicid, val);
Tuple2<int, string> dict = new Tuple2<int,
string>(topicid,field);
getFields(topic).Add(dict);
if (!channelCacheCount.ContainsKey(url))
channelCacheCount[url] = 0;
channelCacheCount[url]++;
if (!topicCacheCount.ContainsKey(url + "|" + topic))
topicCacheCount[url + "|" + topic] = 0;
topicCacheCount[url + "|" + topic]++;
getTopicConsumer(url, topic).OnMessage += msg => {
onMessage(msg,url, topic, field, topicid); };
}
private bool allFiltersTrue(string[] filterKeys, string[]
filterVals, IMessage msg)
{
for (int i = 0; i < filterKeys.Length; i++)
{
if
(!msg.Headers[filterKeys[i]].ToString().Equals(filterVals[i]))
{
return false;
}
}
return true;
}
public void removeRegisteredTopic(int topicid)
{
string vals = topicIDCache[topicid];
string[] keys = vals.Split(new char[] { '|' });
string url = keys[0];
string topic = keys[1];
string field = keys[2];
channelCacheCount[url]--;
topicCacheCount[url + "|" + topic]--;
if (channelCacheCount[url] <= 0)
{
channelCacheCount.Remove(url);
channelCache[url].Dispose();
channelCache.Remove(url);
}
if (topicCacheCount[url + "|" + topic] <= 0)
{
topicCacheCount.Remove(url + "|" + topic);
topicCache[url + "|" + topic].Dispose();
topicCache.Remove(url + "|" + topic);
topicTopicIDFieldCache.Remove(topic);
}
filtersCache.Remove(topicid);
}
//END QPID METHODS
//----------------------------------------------------------------------
-----------------------------------
//IRTDServer METHODS
#region IRtdServer Members
public int ServerStart(IRTDUpdateEvent CallbackObject)
{
updateEvent = CallbackObject;
return 1;
}
public object ConnectData(int TopicID, ref Array Strings, ref
bool GetNewValues)
{
int size = Strings.Length;
int conditions = (int)Math.Floor((double)(size - 3) / 2);
string url;
string topic;
string field;
string[] filterKeys = new string[conditions];
string[] filterVals = new string[conditions];
url = (string)Strings.GetValue(0);
topic = (string)Strings.GetValue(1);
field = (string)Strings.GetValue(2);
for (int i = 0; i < conditions; i = i + 2)
{
filterKeys[i] = (string)Strings.GetValue(i + 3);
filterVals[i] = (string)Strings.GetValue(i + 1 + 3);
}
Tuple2<string[], string[]> filters = new Tuple2<string[],
string[]>(filterKeys,filterVals);
filtersCache.Add(TopicID, filters);
registerTopicID(url, topic, field, TopicID);
return "Getting data...";
}
public void DisconnectData(int TopicID)
{
removeRegisteredTopic(TopicID);
}
public int Heartbeat()
{
return 1;
}
public Array RefreshData(ref int TopicCount)
{
Tuple2<int, object> data;
object[,] result = new object[2, refreshQ.Count];
TopicCount = 0;
for (int i = 0; i < refreshQ.Count; i++)
{
data = refreshQ.Dequeue();
TopicCount++;
result[0, i] = data.a;
result[1, i] = data.b;
}
return result;
}
public void ServerTerminate()
{
foreach (IChannel c in channelCache.Values)
{
c.Dispose();
}
}
#endregion
//END IRTDServer METHODS
}
class Tuple2<T, U>
{
public Tuple2(T t, U u)
{
a = t;
b = u;
}
public T a { get; set; }
public U b { get; set; }
}
class Tuple3<T, U, V>
{
public Tuple3(T t, U u, V v)
{
a = t;
b = u;
c = v;
}
public T a { get; set; }
public U b { get; set; }
public V c { get; set; }
}
}