So if I understand it right, while a message is in its redelivery delay, other messages in that queue can be redelivered.

After that delay, the message is put back into the queue. Right?

What I need is indeed a pausing so that the order of messages in the queue is maintained. How could I achieve this?

Thanks,

Sebastian

You're right, that was a typo indeed.

But how to block delivery of other messages as long as there is a redelivery pending?

Hmm that's strange.

I only have one consumer for that queue (see the JMX info):

And I also found out about the message-grouping feature which I use also as said in the docs:

Messages in a message group are always consumed by the same consumer, even if there are many consumers on a queue. They pin all messages with the same group id to the same consumer. If that consumer closes another consumer is chosen and will receive all messages with the same group id.

I will show you some code maybe it clearer what I am doing with it:

       public void run()
          {
                    ClientMessage clientMessage = null;
                    try
                    {
                              // Make sure not to wait since this blocks the complete thread pool
                              while ((clientMessage = getConsumerSession().getConsumer().receive(-1)) != null)
                              {
                                        LogMF.fatal(Logger.getRootLogger(), "ID={0}, Session={1}, Consumer={2}", new Object[] { clientMessage.getMessageID(), getConsumerSession().getSession(),
                                                            getConsumerSession().getConsumer() });
                                        // sent message
                                        UID uid = UID.parse(clientMessage.getBytesProperty(MessageQueueClient.MESSAGE_PROPERTY_DISPATCHER_UID));
                                        IoSession ioSession = SessionLookupFilter.getSession(uid);


                                        if ((ioSession != null) && ioSession.isConnected())
                                        {
                                                  BaseLTV ltv;
                                                  try
                                                  {
                                                            ltv = MessageQueueClient.clientMessageToLTV(clientMessage);
                                                  }
                                                  catch (Exception e)
                                                  {
                                                            LogMF.error(log, e, "Unable to convert {0} from queue {1} into LTV. Abandoning message!", new Object[] { clientMessage, queueName });
                                                            // acknowledge message to get rid of queue blocker
                                                            clientMessage.acknowledge();


                                                            continue;
                                                  }


                                                  if (!checkAcknowleged(uid, clientMessage, ltv, ioSession, 0))
                                                  {
                                                            ioSession.write(ltv);
                                                            LogMF.trace(log, "Dispatched message {0} for {1} to client {2}.", new Object[] { ltv, uid, ioSession.getRemoteAddress() });


                                                            // give some short possibility to immediately receive a client ack
                                                            if (!checkAcknowleged(uid, clientMessage, ltv, ioSession, 200))
                                                            {
                                                                      // ATTENTION:
                                                                      // Do not acknowledge now since we need to wait for client's low level/high level ack
                                                                      // clientMessage.acknowledge();
                                                                      getConsumerSession().getSession().rollback();
                                                            }
                                                  }
                                        }
                                        else
                                        {
                                                  LogMF.debug(log, "Client with {0} currently disconnected.", new Object[] { uid });
                                                  // don't acknowledge the message so we will get it again
                                                  // FIXME: invite the client to establish a connection via SIP server
                                                  // Set redelivery timeout of message to a serious value so we don't bother the client
                                                  // with subsequent invitations.
                                                  getConsumerSession().getSession().rollback();


                                                  break;
                                        }
                              }
                    }
                    catch (HornetQException e)
                    {
                              log.error("Error while trying to read message from queue.", e);
                    }
                    catch (Exception e)
                    {
                              LogMF.error(log, e, "Error during dispatching of mq-message {0} from queue {1} to client.", new Object[] { clientMessage, queueName });
                    }
                    finally
                    {
                              messageQueueOutbox.schedule(this, messageQueueOutbox.getScheduleDelay(), TimeUnit.MILLISECONDS);
                              LogMF.trace(log, "Rescheduled {0} with {1} seconds delay.", new Object[] { this, messageQueueOutbox.getScheduleDelay() });
                    }
          }

The above code belongs to a runnable that reschedules itself in a ScheduledThreadPoolExecutor. This runnable has a single instance of a ClientSession and one single ClientProducer for that session. As you can see I am rolling back the session in some cases. But the behavior is as posted initially.

Do I have to do something else after a rollback?

Dear Clebert,

sorry but I was off yesterday (three little children and a wife complaining :-9). Its a mess with that timezine offsets.

In bed I thought of the problem again and again. Today I am going to write a simple test to ensure I have no problem with multi-threading.

Thanks so far,

Sebastian

Thanks to Clebert,

the hint with the concurrency brought me onto the right road.

I had a mistake in my code in the end had me rollback the false session. After fixing that issue it works like a charm

Greets,

Sebastian