You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@activemq.apache.org by Sebastjan Trepca <tr...@gmail.com> on 2008/01/28 21:02:10 UTC
Stomp protocol problems
Hi,
I'm using ActiveMQ(both versions are affected) with STOMP protocol and
noticed a problem with it. At least I hope it's a problem.
When you ACK a message through stomp, all messages until the right one
gets ACKed too.
If we check the code:
protected void onStompAck(StompFrame command) throws ProtocolException {
checkConnected();
// TODO: acking with just a message id is very bogus
// since the same message id could have been sent to 2 different
// subscriptions
// on the same stomp connection. For example, when 2 subs are created on
// the same topic.
Map<String, String> headers = command.getHeaders();
String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
if (messageId == null) {
throw new ProtocolException("ACK received without a
message-id to acknowledge!");
}
TransactionId activemqTx = null;
String stompTx = headers.get(Stomp.Headers.TRANSACTION);
if (stompTx != null) {
activemqTx = transactions.get(stompTx);
if (activemqTx == null) {
throw new ProtocolException("Invalid transaction id: "
+ stompTx);
}
}
boolean acked = false;
for (Iterator<StompSubscription> iter =
subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
StompSubscription sub = iter.next();
MessageAck ack = sub.onStompMessageAck(messageId);
if (ack != null) {
ack.setTransactionId(activemqTx);
sendToActiveMQ(ack, createResponseHandler(command));
acked = true;
break;
}
}
if (!acked) {
throw new ProtocolException("Unexpected ACK received for
message-id [" + messageId + "]");
}
}
synchronized MessageAck onStompMessageAck(String messageId) {
if (!dispatchedMessage.containsKey(messageId)) {
return null;
}
MessageAck ack = new MessageAck();
ack.setDestination(consumerInfo.getDestination());
ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
ack.setConsumerId(consumerInfo.getConsumerId());
int count = 0;
for (Iterator iter = dispatchedMessage.entrySet().iterator();
iter.hasNext();) {
Map.Entry entry = (Entry)iter.next();
String id = (String)entry.getKey();
MessageId msgid = (MessageId)entry.getValue();
if (ack.getFirstMessageId() == null) {
ack.setFirstMessageId(msgid);
}
iter.remove();
count++;
if (id.equals(messageId)) {
ack.setLastMessageId(msgid);
break;
}
}
ack.setMessageCount(count);
return ack;
}
It loops through all messages and ACKs them until it finds the correct
message id. So when you send few messages and the last one is the
processed first all previous ones get ACKed?
Is this a feature of STOMP protocol or is it a bug?
Thanks, Sebastjan
Re: Stomp protocol problems
Posted by Dejan Bosanac <de...@nighttale.net>.
Hi,
can you file a Jira issue for this?
Thanks
--
Dejan Bosanac
www.scriptinginjava.net
On Jan 28, 2008 9:02 PM, Sebastjan Trepca <tr...@gmail.com> wrote:
> Hi,
>
> I'm using ActiveMQ(both versions are affected) with STOMP protocol and
> noticed a problem with it. At least I hope it's a problem.
>
> When you ACK a message through stomp, all messages until the right one
> gets ACKed too.
>
> If we check the code:
>
> protected void onStompAck(StompFrame command) throws ProtocolException {
> checkConnected();
>
> // TODO: acking with just a message id is very bogus
> // since the same message id could have been sent to 2 different
> // subscriptions
> // on the same stomp connection. For example, when 2 subs are created on
> // the same topic.
>
> Map<String, String> headers = command.getHeaders();
> String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
> if (messageId == null) {
> throw new ProtocolException("ACK received without a
> message-id to acknowledge!");
> }
>
> TransactionId activemqTx = null;
> String stompTx = headers.get(Stomp.Headers.TRANSACTION);
> if (stompTx != null) {
> activemqTx = transactions.get(stompTx);
> if (activemqTx == null) {
> throw new ProtocolException("Invalid transaction id: "
> + stompTx);
> }
> }
>
> boolean acked = false;
> for (Iterator<StompSubscription> iter =
> subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
> StompSubscription sub = iter.next();
> MessageAck ack = sub.onStompMessageAck(messageId);
> if (ack != null) {
> ack.setTransactionId(activemqTx);
> sendToActiveMQ(ack, createResponseHandler(command));
> acked = true;
> break;
> }
> }
>
> if (!acked) {
> throw new ProtocolException("Unexpected ACK received for
> message-id [" + messageId + "]");
> }
>
> }
>
>
> synchronized MessageAck onStompMessageAck(String messageId) {
>
> if (!dispatchedMessage.containsKey(messageId)) {
> return null;
> }
>
> MessageAck ack = new MessageAck();
> ack.setDestination(consumerInfo.getDestination());
> ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
> ack.setConsumerId(consumerInfo.getConsumerId());
>
> int count = 0;
> for (Iterator iter = dispatchedMessage.entrySet().iterator();
> iter.hasNext();) {
>
> Map.Entry entry = (Entry)iter.next();
> String id = (String)entry.getKey();
> MessageId msgid = (MessageId)entry.getValue();
>
> if (ack.getFirstMessageId() == null) {
> ack.setFirstMessageId(msgid);
> }
>
> iter.remove();
> count++;
>
> if (id.equals(messageId)) {
> ack.setLastMessageId(msgid);
> break;
> }
>
> }
>
> ack.setMessageCount(count);
> return ack;
> }
>
> It loops through all messages and ACKs them until it finds the correct
> message id. So when you send few messages and the last one is the
> processed first all previous ones get ACKed?
>
> Is this a feature of STOMP protocol or is it a bug?
>
> Thanks, Sebastjan
>
Re: Stomp protocol problems
Posted by Dejan Bosanac <de...@nighttale.net>.
I'm not sure what are you trying to achieve, but if you need to group
messages in one queue, you can use selectors
(http://activemq.apache.org/selectors.html). You have an example of
how to use them in test cases. Another option would be to send
messages to different queues.
On Jan 30, 2008 1:19 PM, Sebastjan Trepca <tr...@gmail.com> wrote:
> Yes, sorry for not RTFM, it's not very logical though, to ACK all
> messages when you ack one.
>
> So what are my options then? Stomp protocol doesn't seem to support sessions:
>
> "The session-id header is a unique identifier for this session (though
> it isn't actually used yet).
> "
> Should I create separate transaction for each message and group them with that?
>
> I currently have only one listener that fills a local queue that gets
> processed by consumer threads. Should I create a listener/consumer for
> each thread?
>
> Thanks, Sebastjan
>
>
> On 1/30/08, gustav.mauer <gu...@iona.com> wrote:
> >
> > I am under the impression this is correct behaviour. See for example:
> > http://java.sun.com/products/jms/tutorial/ page 63
> >
> >
> > Sebastjan Trepca wrote:
> > >
> > > Hi,
> > >
> > > I'm using ActiveMQ(both versions are affected) with STOMP protocol and
> > > noticed a problem with it. At least I hope it's a problem.
> > >
> > > When you ACK a message through stomp, all messages until the right one
> > > gets ACKed too.
> > >
> > > If we check the code:
> > >
> > > protected void onStompAck(StompFrame command) throws ProtocolException
> > > {
> > > checkConnected();
> > >
> > > // TODO: acking with just a message id is very bogus
> > > // since the same message id could have been sent to 2 different
> > > // subscriptions
> > > // on the same stomp connection. For example, when 2 subs are
> > > created on
> > > // the same topic.
> > >
> > > Map<String, String> headers = command.getHeaders();
> > > String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
> > > if (messageId == null) {
> > > throw new ProtocolException("ACK received without a
> > > message-id to acknowledge!");
> > > }
> > >
> > > TransactionId activemqTx = null;
> > > String stompTx = headers.get(Stomp.Headers.TRANSACTION);
> > > if (stompTx != null) {
> > > activemqTx = transactions.get(stompTx);
> > > if (activemqTx == null) {
> > > throw new ProtocolException("Invalid transaction id: "
> > > + stompTx);
> > > }
> > > }
> > >
> > > boolean acked = false;
> > > for (Iterator<StompSubscription> iter =
> > > subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
> > > StompSubscription sub = iter.next();
> > > MessageAck ack = sub.onStompMessageAck(messageId);
> > > if (ack != null) {
> > > ack.setTransactionId(activemqTx);
> > > sendToActiveMQ(ack, createResponseHandler(command));
> > > acked = true;
> > > break;
> > > }
> > > }
> > >
> > > if (!acked) {
> > > throw new ProtocolException("Unexpected ACK received for
> > > message-id [" + messageId + "]");
> > > }
> > >
> > > }
> > >
> > >
> > > synchronized MessageAck onStompMessageAck(String messageId) {
> > >
> > > if (!dispatchedMessage.containsKey(messageId)) {
> > > return null;
> > > }
> > >
> > > MessageAck ack = new MessageAck();
> > > ack.setDestination(consumerInfo.getDestination());
> > > ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
> > > ack.setConsumerId(consumerInfo.getConsumerId());
> > >
> > > int count = 0;
> > > for (Iterator iter = dispatchedMessage.entrySet().iterator();
> > > iter.hasNext();) {
> > >
> > > Map.Entry entry = (Entry)iter.next();
> > > String id = (String)entry.getKey();
> > > MessageId msgid = (MessageId)entry.getValue();
> > >
> > > if (ack.getFirstMessageId() == null) {
> > > ack.setFirstMessageId(msgid);
> > > }
> > >
> > > iter.remove();
> > > count++;
> > >
> > > if (id.equals(messageId)) {
> > > ack.setLastMessageId(msgid);
> > > break;
> > > }
> > >
> > > }
> > >
> > > ack.setMessageCount(count);
> > > return ack;
> > > }
> > >
> > > It loops through all messages and ACKs them until it finds the correct
> > > message id. So when you send few messages and the last one is the
> > > processed first all previous ones get ACKed?
> > >
> > > Is this a feature of STOMP protocol or is it a bug?
> > >
> > > Thanks, Sebastjan
> > >
> > >
> >
> > --
> > View this message in context: http://www.nabble.com/Stomp-protocol-problems-tp15144123s2354p15177216.html
> > Sent from the ActiveMQ - User mailing list archive at Nabble.com.
> >
> >
>
--
Dejan Bosanac
www.scriptinginjava.net
Re: Stomp protocol problems
Posted by Sebastjan Trepca <tr...@gmail.com>.
On 1/30/08, gustav.mauer <gu...@iona.com> wrote:
>
> I assume you only want to ACK a message once the consumer/worker thread has
> processed it? I have also been wondering what the best pattern is for that.
> I was also wondering if the correct way to proceed is to create a listener
> for each worker thread? And if a listener thread is busy, with the request
> be delivered to one of the idle listener threads, so that multiple messages
> can be worked on in parallel in the server.
Yes, thats exactly what I try to do. I'll try with listeners.
>
>
> Sebastjan Trepca wrote:
> >
> > Yes, sorry for not RTFM, it's not very logical though, to ACK all
> > messages when you ack one.
> >
> > So what are my options then? Stomp protocol doesn't seem to support
> > sessions:
> >
> > "The session-id header is a unique identifier for this session (though
> > it isn't actually used yet).
> > "
> > Should I create separate transaction for each message and group them with
> > that?
> >
> > I currently have only one listener that fills a local queue that gets
> > processed by consumer threads. Should I create a listener/consumer for
> > each thread?
> >
> > Thanks, Sebastjan
> >
> > On 1/30/08, gustav.mauer <gu...@iona.com> wrote:
> >>
> >> I am under the impression this is correct behaviour. See for example:
> >> http://java.sun.com/products/jms/tutorial/ page 63
> >>
> >>
> >> Sebastjan Trepca wrote:
> >> >
> >> > Hi,
> >> >
> >> > I'm using ActiveMQ(both versions are affected) with STOMP protocol and
> >> > noticed a problem with it. At least I hope it's a problem.
> >> >
> >> > When you ACK a message through stomp, all messages until the right one
> >> > gets ACKed too.
> >> >
> >> > If we check the code:
> >> >
> >> > protected void onStompAck(StompFrame command) throws
> >> ProtocolException
> >> > {
> >> > checkConnected();
> >> >
> >> > // TODO: acking with just a message id is very bogus
> >> > // since the same message id could have been sent to 2
> >> different
> >> > // subscriptions
> >> > // on the same stomp connection. For example, when 2 subs are
> >> > created on
> >> > // the same topic.
> >> >
> >> > Map<String, String> headers = command.getHeaders();
> >> > String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
> >> > if (messageId == null) {
> >> > throw new ProtocolException("ACK received without a
> >> > message-id to acknowledge!");
> >> > }
> >> >
> >> > TransactionId activemqTx = null;
> >> > String stompTx = headers.get(Stomp.Headers.TRANSACTION);
> >> > if (stompTx != null) {
> >> > activemqTx = transactions.get(stompTx);
> >> > if (activemqTx == null) {
> >> > throw new ProtocolException("Invalid transaction id: "
> >> > + stompTx);
> >> > }
> >> > }
> >> >
> >> > boolean acked = false;
> >> > for (Iterator<StompSubscription> iter =
> >> > subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
> >> > StompSubscription sub = iter.next();
> >> > MessageAck ack = sub.onStompMessageAck(messageId);
> >> > if (ack != null) {
> >> > ack.setTransactionId(activemqTx);
> >> > sendToActiveMQ(ack, createResponseHandler(command));
> >> > acked = true;
> >> > break;
> >> > }
> >> > }
> >> >
> >> > if (!acked) {
> >> > throw new ProtocolException("Unexpected ACK received for
> >> > message-id [" + messageId + "]");
> >> > }
> >> >
> >> > }
> >> >
> >> >
> >> > synchronized MessageAck onStompMessageAck(String messageId) {
> >> >
> >> > if (!dispatchedMessage.containsKey(messageId)) {
> >> > return null;
> >> > }
> >> >
> >> > MessageAck ack = new MessageAck();
> >> > ack.setDestination(consumerInfo.getDestination());
> >> > ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
> >> > ack.setConsumerId(consumerInfo.getConsumerId());
> >> >
> >> > int count = 0;
> >> > for (Iterator iter = dispatchedMessage.entrySet().iterator();
> >> > iter.hasNext();) {
> >> >
> >> > Map.Entry entry = (Entry)iter.next();
> >> > String id = (String)entry.getKey();
> >> > MessageId msgid = (MessageId)entry.getValue();
> >> >
> >> > if (ack.getFirstMessageId() == null) {
> >> > ack.setFirstMessageId(msgid);
> >> > }
> >> >
> >> > iter.remove();
> >> > count++;
> >> >
> >> > if (id.equals(messageId)) {
> >> > ack.setLastMessageId(msgid);
> >> > break;
> >> > }
> >> >
> >> > }
> >> >
> >> > ack.setMessageCount(count);
> >> > return ack;
> >> > }
> >> >
> >> > It loops through all messages and ACKs them until it finds the correct
> >> > message id. So when you send few messages and the last one is the
> >> > processed first all previous ones get ACKed?
> >> >
> >> > Is this a feature of STOMP protocol or is it a bug?
> >> >
> >> > Thanks, Sebastjan
> >> >
> >> >
> >>
> >> --
> >> View this message in context:
> >> http://www.nabble.com/Stomp-protocol-problems-tp15144123s2354p15177216.html
> >> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
> >>
> >>
> >
> >
>
> --
> View this message in context: http://www.nabble.com/Stomp-protocol-problems-tp15144123s2354p15182191.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>
>
--
Sebastjan
Re: Stomp protocol problems
Posted by "gustav.mauer" <gu...@iona.com>.
I assume you only want to ACK a message once the consumer/worker thread has
processed it? I have also been wondering what the best pattern is for that.
I was also wondering if the correct way to proceed is to create a listener
for each worker thread? And if a listener thread is busy, with the request
be delivered to one of the idle listener threads, so that multiple messages
can be worked on in parallel in the server.
Sebastjan Trepca wrote:
>
> Yes, sorry for not RTFM, it's not very logical though, to ACK all
> messages when you ack one.
>
> So what are my options then? Stomp protocol doesn't seem to support
> sessions:
>
> "The session-id header is a unique identifier for this session (though
> it isn't actually used yet).
> "
> Should I create separate transaction for each message and group them with
> that?
>
> I currently have only one listener that fills a local queue that gets
> processed by consumer threads. Should I create a listener/consumer for
> each thread?
>
> Thanks, Sebastjan
>
> On 1/30/08, gustav.mauer <gu...@iona.com> wrote:
>>
>> I am under the impression this is correct behaviour. See for example:
>> http://java.sun.com/products/jms/tutorial/ page 63
>>
>>
>> Sebastjan Trepca wrote:
>> >
>> > Hi,
>> >
>> > I'm using ActiveMQ(both versions are affected) with STOMP protocol and
>> > noticed a problem with it. At least I hope it's a problem.
>> >
>> > When you ACK a message through stomp, all messages until the right one
>> > gets ACKed too.
>> >
>> > If we check the code:
>> >
>> > protected void onStompAck(StompFrame command) throws
>> ProtocolException
>> > {
>> > checkConnected();
>> >
>> > // TODO: acking with just a message id is very bogus
>> > // since the same message id could have been sent to 2
>> different
>> > // subscriptions
>> > // on the same stomp connection. For example, when 2 subs are
>> > created on
>> > // the same topic.
>> >
>> > Map<String, String> headers = command.getHeaders();
>> > String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
>> > if (messageId == null) {
>> > throw new ProtocolException("ACK received without a
>> > message-id to acknowledge!");
>> > }
>> >
>> > TransactionId activemqTx = null;
>> > String stompTx = headers.get(Stomp.Headers.TRANSACTION);
>> > if (stompTx != null) {
>> > activemqTx = transactions.get(stompTx);
>> > if (activemqTx == null) {
>> > throw new ProtocolException("Invalid transaction id: "
>> > + stompTx);
>> > }
>> > }
>> >
>> > boolean acked = false;
>> > for (Iterator<StompSubscription> iter =
>> > subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
>> > StompSubscription sub = iter.next();
>> > MessageAck ack = sub.onStompMessageAck(messageId);
>> > if (ack != null) {
>> > ack.setTransactionId(activemqTx);
>> > sendToActiveMQ(ack, createResponseHandler(command));
>> > acked = true;
>> > break;
>> > }
>> > }
>> >
>> > if (!acked) {
>> > throw new ProtocolException("Unexpected ACK received for
>> > message-id [" + messageId + "]");
>> > }
>> >
>> > }
>> >
>> >
>> > synchronized MessageAck onStompMessageAck(String messageId) {
>> >
>> > if (!dispatchedMessage.containsKey(messageId)) {
>> > return null;
>> > }
>> >
>> > MessageAck ack = new MessageAck();
>> > ack.setDestination(consumerInfo.getDestination());
>> > ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
>> > ack.setConsumerId(consumerInfo.getConsumerId());
>> >
>> > int count = 0;
>> > for (Iterator iter = dispatchedMessage.entrySet().iterator();
>> > iter.hasNext();) {
>> >
>> > Map.Entry entry = (Entry)iter.next();
>> > String id = (String)entry.getKey();
>> > MessageId msgid = (MessageId)entry.getValue();
>> >
>> > if (ack.getFirstMessageId() == null) {
>> > ack.setFirstMessageId(msgid);
>> > }
>> >
>> > iter.remove();
>> > count++;
>> >
>> > if (id.equals(messageId)) {
>> > ack.setLastMessageId(msgid);
>> > break;
>> > }
>> >
>> > }
>> >
>> > ack.setMessageCount(count);
>> > return ack;
>> > }
>> >
>> > It loops through all messages and ACKs them until it finds the correct
>> > message id. So when you send few messages and the last one is the
>> > processed first all previous ones get ACKed?
>> >
>> > Is this a feature of STOMP protocol or is it a bug?
>> >
>> > Thanks, Sebastjan
>> >
>> >
>>
>> --
>> View this message in context:
>> http://www.nabble.com/Stomp-protocol-problems-tp15144123s2354p15177216.html
>> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>>
>>
>
>
--
View this message in context: http://www.nabble.com/Stomp-protocol-problems-tp15144123s2354p15182191.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.
Re: Stomp protocol problems
Posted by Sebastjan Trepca <tr...@gmail.com>.
Yes, sorry for not RTFM, it's not very logical though, to ACK all
messages when you ack one.
So what are my options then? Stomp protocol doesn't seem to support sessions:
"The session-id header is a unique identifier for this session (though
it isn't actually used yet).
"
Should I create separate transaction for each message and group them with that?
I currently have only one listener that fills a local queue that gets
processed by consumer threads. Should I create a listener/consumer for
each thread?
Thanks, Sebastjan
On 1/30/08, gustav.mauer <gu...@iona.com> wrote:
>
> I am under the impression this is correct behaviour. See for example:
> http://java.sun.com/products/jms/tutorial/ page 63
>
>
> Sebastjan Trepca wrote:
> >
> > Hi,
> >
> > I'm using ActiveMQ(both versions are affected) with STOMP protocol and
> > noticed a problem with it. At least I hope it's a problem.
> >
> > When you ACK a message through stomp, all messages until the right one
> > gets ACKed too.
> >
> > If we check the code:
> >
> > protected void onStompAck(StompFrame command) throws ProtocolException
> > {
> > checkConnected();
> >
> > // TODO: acking with just a message id is very bogus
> > // since the same message id could have been sent to 2 different
> > // subscriptions
> > // on the same stomp connection. For example, when 2 subs are
> > created on
> > // the same topic.
> >
> > Map<String, String> headers = command.getHeaders();
> > String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
> > if (messageId == null) {
> > throw new ProtocolException("ACK received without a
> > message-id to acknowledge!");
> > }
> >
> > TransactionId activemqTx = null;
> > String stompTx = headers.get(Stomp.Headers.TRANSACTION);
> > if (stompTx != null) {
> > activemqTx = transactions.get(stompTx);
> > if (activemqTx == null) {
> > throw new ProtocolException("Invalid transaction id: "
> > + stompTx);
> > }
> > }
> >
> > boolean acked = false;
> > for (Iterator<StompSubscription> iter =
> > subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
> > StompSubscription sub = iter.next();
> > MessageAck ack = sub.onStompMessageAck(messageId);
> > if (ack != null) {
> > ack.setTransactionId(activemqTx);
> > sendToActiveMQ(ack, createResponseHandler(command));
> > acked = true;
> > break;
> > }
> > }
> >
> > if (!acked) {
> > throw new ProtocolException("Unexpected ACK received for
> > message-id [" + messageId + "]");
> > }
> >
> > }
> >
> >
> > synchronized MessageAck onStompMessageAck(String messageId) {
> >
> > if (!dispatchedMessage.containsKey(messageId)) {
> > return null;
> > }
> >
> > MessageAck ack = new MessageAck();
> > ack.setDestination(consumerInfo.getDestination());
> > ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
> > ack.setConsumerId(consumerInfo.getConsumerId());
> >
> > int count = 0;
> > for (Iterator iter = dispatchedMessage.entrySet().iterator();
> > iter.hasNext();) {
> >
> > Map.Entry entry = (Entry)iter.next();
> > String id = (String)entry.getKey();
> > MessageId msgid = (MessageId)entry.getValue();
> >
> > if (ack.getFirstMessageId() == null) {
> > ack.setFirstMessageId(msgid);
> > }
> >
> > iter.remove();
> > count++;
> >
> > if (id.equals(messageId)) {
> > ack.setLastMessageId(msgid);
> > break;
> > }
> >
> > }
> >
> > ack.setMessageCount(count);
> > return ack;
> > }
> >
> > It loops through all messages and ACKs them until it finds the correct
> > message id. So when you send few messages and the last one is the
> > processed first all previous ones get ACKed?
> >
> > Is this a feature of STOMP protocol or is it a bug?
> >
> > Thanks, Sebastjan
> >
> >
>
> --
> View this message in context: http://www.nabble.com/Stomp-protocol-problems-tp15144123s2354p15177216.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>
>
Re: Stomp protocol problems
Posted by Dejan Bosanac <de...@nighttale.net>.
I guess I responded instinctively without thinking it through. This is
definitely a correct behavior.
Cheers
On Jan 30, 2008 9:39 AM, gustav.mauer <gu...@iona.com> wrote:
>
> I am under the impression this is correct behaviour. See for example:
> http://java.sun.com/products/jms/tutorial/ page 63
>
>
>
> Sebastjan Trepca wrote:
> >
> > Hi,
> >
> > I'm using ActiveMQ(both versions are affected) with STOMP protocol and
> > noticed a problem with it. At least I hope it's a problem.
> >
> > When you ACK a message through stomp, all messages until the right one
> > gets ACKed too.
> >
> > If we check the code:
> >
> > protected void onStompAck(StompFrame command) throws ProtocolException
> > {
> > checkConnected();
> >
> > // TODO: acking with just a message id is very bogus
> > // since the same message id could have been sent to 2 different
> > // subscriptions
> > // on the same stomp connection. For example, when 2 subs are
> > created on
> > // the same topic.
> >
> > Map<String, String> headers = command.getHeaders();
> > String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
> > if (messageId == null) {
> > throw new ProtocolException("ACK received without a
> > message-id to acknowledge!");
> > }
> >
> > TransactionId activemqTx = null;
> > String stompTx = headers.get(Stomp.Headers.TRANSACTION);
> > if (stompTx != null) {
> > activemqTx = transactions.get(stompTx);
> > if (activemqTx == null) {
> > throw new ProtocolException("Invalid transaction id: "
> > + stompTx);
> > }
> > }
> >
> > boolean acked = false;
> > for (Iterator<StompSubscription> iter =
> > subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
> > StompSubscription sub = iter.next();
> > MessageAck ack = sub.onStompMessageAck(messageId);
> > if (ack != null) {
> > ack.setTransactionId(activemqTx);
> > sendToActiveMQ(ack, createResponseHandler(command));
> > acked = true;
> > break;
> > }
> > }
> >
> > if (!acked) {
> > throw new ProtocolException("Unexpected ACK received for
> > message-id [" + messageId + "]");
> > }
> >
> > }
> >
> >
> > synchronized MessageAck onStompMessageAck(String messageId) {
> >
> > if (!dispatchedMessage.containsKey(messageId)) {
> > return null;
> > }
> >
> > MessageAck ack = new MessageAck();
> > ack.setDestination(consumerInfo.getDestination());
> > ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
> > ack.setConsumerId(consumerInfo.getConsumerId());
> >
> > int count = 0;
> > for (Iterator iter = dispatchedMessage.entrySet().iterator();
> > iter.hasNext();) {
> >
> > Map.Entry entry = (Entry)iter.next();
> > String id = (String)entry.getKey();
> > MessageId msgid = (MessageId)entry.getValue();
> >
> > if (ack.getFirstMessageId() == null) {
> > ack.setFirstMessageId(msgid);
> > }
> >
> > iter.remove();
> > count++;
> >
> > if (id.equals(messageId)) {
> > ack.setLastMessageId(msgid);
> > break;
> > }
> >
> > }
> >
> > ack.setMessageCount(count);
> > return ack;
> > }
> >
> > It loops through all messages and ACKs them until it finds the correct
> > message id. So when you send few messages and the last one is the
> > processed first all previous ones get ACKed?
> >
> > Is this a feature of STOMP protocol or is it a bug?
> >
> > Thanks, Sebastjan
> >
> >
>
> --
> View this message in context: http://www.nabble.com/Stomp-protocol-problems-tp15144123s2354p15177216.html
> Sent from the ActiveMQ - User mailing list archive at Nabble.com.
>
>
--
Dejan Bosanac
www.scriptinginjava.net
Re: Stomp protocol problems
Posted by "gustav.mauer" <gu...@iona.com>.
I am under the impression this is correct behaviour. See for example:
http://java.sun.com/products/jms/tutorial/ page 63
Sebastjan Trepca wrote:
>
> Hi,
>
> I'm using ActiveMQ(both versions are affected) with STOMP protocol and
> noticed a problem with it. At least I hope it's a problem.
>
> When you ACK a message through stomp, all messages until the right one
> gets ACKed too.
>
> If we check the code:
>
> protected void onStompAck(StompFrame command) throws ProtocolException
> {
> checkConnected();
>
> // TODO: acking with just a message id is very bogus
> // since the same message id could have been sent to 2 different
> // subscriptions
> // on the same stomp connection. For example, when 2 subs are
> created on
> // the same topic.
>
> Map<String, String> headers = command.getHeaders();
> String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
> if (messageId == null) {
> throw new ProtocolException("ACK received without a
> message-id to acknowledge!");
> }
>
> TransactionId activemqTx = null;
> String stompTx = headers.get(Stomp.Headers.TRANSACTION);
> if (stompTx != null) {
> activemqTx = transactions.get(stompTx);
> if (activemqTx == null) {
> throw new ProtocolException("Invalid transaction id: "
> + stompTx);
> }
> }
>
> boolean acked = false;
> for (Iterator<StompSubscription> iter =
> subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
> StompSubscription sub = iter.next();
> MessageAck ack = sub.onStompMessageAck(messageId);
> if (ack != null) {
> ack.setTransactionId(activemqTx);
> sendToActiveMQ(ack, createResponseHandler(command));
> acked = true;
> break;
> }
> }
>
> if (!acked) {
> throw new ProtocolException("Unexpected ACK received for
> message-id [" + messageId + "]");
> }
>
> }
>
>
> synchronized MessageAck onStompMessageAck(String messageId) {
>
> if (!dispatchedMessage.containsKey(messageId)) {
> return null;
> }
>
> MessageAck ack = new MessageAck();
> ack.setDestination(consumerInfo.getDestination());
> ack.setAckType(MessageAck.STANDARD_ACK_TYPE);
> ack.setConsumerId(consumerInfo.getConsumerId());
>
> int count = 0;
> for (Iterator iter = dispatchedMessage.entrySet().iterator();
> iter.hasNext();) {
>
> Map.Entry entry = (Entry)iter.next();
> String id = (String)entry.getKey();
> MessageId msgid = (MessageId)entry.getValue();
>
> if (ack.getFirstMessageId() == null) {
> ack.setFirstMessageId(msgid);
> }
>
> iter.remove();
> count++;
>
> if (id.equals(messageId)) {
> ack.setLastMessageId(msgid);
> break;
> }
>
> }
>
> ack.setMessageCount(count);
> return ack;
> }
>
> It loops through all messages and ACKs them until it finds the correct
> message id. So when you send few messages and the last one is the
> processed first all previous ones get ACKed?
>
> Is this a feature of STOMP protocol or is it a bug?
>
> Thanks, Sebastjan
>
>
--
View this message in context: http://www.nabble.com/Stomp-protocol-problems-tp15144123s2354p15177216.html
Sent from the ActiveMQ - User mailing list archive at Nabble.com.