JBoss Messaging Core Specification Revision 1
-
Date: 06/07/2005
Author: Ovidiu Feodorov
Summary
Introduction
Channel State
Message Store
Persistence Manager
Message State
Acknowledgments
Channels
Filters
Message Order
Message Priority
-
1. Introduction
This document completes and further qualifies the original JBoss Messaging Core Specification. If you are not familiar with the Core Specification, you should start by reading it (http://wiki.jboss.org/wiki/Wiki.jsp?page=JBossMessagingCore) and then return to Revision 1.
2. Channel State
http://jira.jboss.org/jira/browse/JBMESSAGING-74
http://jira.jboss.org/jira/browse/JBMESSAGING-70
Channels are message delivery mechanisms that forward messages to registered receivers. A synchronous channel always attempts synchronous delivery and immediately returns the acknowledgment generated by the target receiver(s) to the sender. A synchronous channel does not store the message in case of unsuccessful delivery. It passes the decision of how to handle the rejected message to the original sender. In the same situation, an asynchronous channel may decide to positively acknowledge the message to the original sender, store the message and retry the delivery later. Maintaining a channel state only makes sense for asynchronous channels.
The channel state consists in the union of message states corresponding to all messages stored or being processed by the channel. More about the message state can be found in Section 5. "Message State". Because the channel state completely describes the state of all messages being processed by the channel it can be used together with a message store's state to recover the channel in case of virtual machine or hardware failure. The channel state is maintained by a ChannelState implementation, which acts as a state manager.
The channel state does not physically contain the messages whose states it represents. A message state includes a message reference, which maintains attributes such as messageID, expiration time, etc. The message itself is stored in a message store, which returns the corresponding message reference (see Section 3. "Message Store").
The ChannelState was formerly known as AcknowledgmentStore in the previous version of the specifications.
2.1 Channel State and Reliable Messages
The implementation of the channel state must clearly indicate if it is able to handle reliable messages or not. A channel must accept reliable messages if and only if its channel state implementation has access to a reliable persistence manager. Otherwise, the channel can only accept non-reliable messages, since it cannot offer sufficient reliability guarantees required to handle reliable messages. The separation between the channel state and the persistence manager is essential in case of distributed channels. For more details, see Section 2.3 "Channel State and Distributed Channels".
2.2 Channel State and Transactional Message Handling
It should be possible to transactionally modify the state of a message. A channel state implementation must offer support for transactional message handling or clearly indicate that is not able to provide this functionality.
Transactional capabilities includes at least two cases:
2.2.1 Possibility to transactionally submit new messages to the channel. The channel state should maintain these messages in a NOT_COMMITTED state until the transaction commits. After commit, the messages become eligible for delivery to the channel's receivers and their delivery must be attempted by the channel.
2.2.2 Possibility to transactionally change the state of a message. It should be possible to transactionally submit acknowledgments to the channel. In this situation, the message's state change will become effective on transaction's commit.
2.3 Channel State and Distributed Channels
The channel state implementation should be able to handle the situation when the channel is distributed (spans across multiple address spaces). Any distributed channel should be able to handle non-reliable messages. Due to the distributed nature of the channel, the actual degree of reliability achieved handling non-reliable messages is higher than handling the same message in a single address space channel. Increased reliability is made possible by in-memory replication across multiple address spaces.
If all peers of a distributed channels have access to the same persistent storage, the distributed channel should be also able to handle reliable messages.
2.4 The ChannelState interface
The state of a channel is maintained by an implementation of the ChannelState interface:
public interface ChannelState
{
public boolean isReliable();
public void setPersistenceManager(PersistenceManager pm);
public boolean isTransactional();
public void setTransactionManager(TransactionManager tm);
/**
* Add the message transactionally in presence of an active JTA transaction.
*/
public add(MessageReference m);
/**
* Change the message state transactionally in presence of an active JTA transaction.
*/
public update(Serializable messageID, boolean positiveAcknowledgment);
public remove(Serializable messageID);
// other accessors
}
3. Message Store
A message store is an in-memory repository for messages. Once a message is submitted for delivery to a system based on the Messaging Core, the message is converted into a message reference as soon as possible. The components of the system (the channels) try to forward the message reference for as long as possible, before the reference is converted back into the original message. For more details on messages and message references, see the original JBoss Messaging Core specification document
The responsibilities of a message store instance are:
Convert a message to its corresponding message reference
Restore the message from its corresponding message reference
Transfer in-memory messages to persistent storage, when necessary
A channel must have access to a message store in order to delegate the conversion and the physical storage of the messages in transit. A message store can be shared by multiple channels co-existing in the same address space, allowing the channels to pass to each other message references instead of messages. This configuration avoids the necessity of handling messages with large bodies, which leads to performance improvements.
3.1 Message Store and Reliable Messages
If a message store has access to a reliable persistence manager, the message store is considered reliable, thus can be trusted to store reliable messages. If no persistence manager is available, the message store can only handle non-reliable messages. The implementation of a message store must clearly indicate if it is able to handle reliable messages or not.
Reliable messages are always transferred to persistent storage. Non-reliable message could also be transferred to persistent storage under certain memory usage constraints.
3.2 Message Store and Transactional Message Handling
An message store implementation must offer support for transactional message handling or clearly indicate that is not able to provide this functionality.
3.3 Message Store and Distributed Channels
A distributed Message Core configuration requires a distributed message store implementation. Such an implementation will help convert message references back to the original messages at destination, even if the source and destination channels are not in the same address space.
TODO: to further qualify.
3.4 The MessageStore Interface
public interface MessageStore
{
public Serializable getStoreID();
public boolean isReliable();
public void setPersistenceManager(PersistenceManager pm);
public boolean isTransactional();
public void setTransactionManager(TransactionManager tm);
/**
* Add the message transactionally in presence of an active JTA transaction.
*/
public MessageReference store(Message m);
public Message retrieve(Serializable messageID);
public void remove(Serializable messageID);
}
4. Persistence Manager
http://jira.jboss.org/jira/browse/JBMESSAGING-18
5. Message State
The state of a message being processed by an asynchronous channel is part of the channel's state. This paragraph describes the internal representation of a message state, as maintained by the channel state. From this perspective, a message can be not in channel, NOT_COMMITTED, NEW or DELIVERED.
5.1. NOT_COMMITTED Message State
A message is in a NOT_COMMITTED state if the message is accepted by the channel in a transaction that has not been committed yet. The channel does not attempt delivery on a NOT_COMMITTED messages. If the message is declared reliable, the channel must insure that the message's state is managed by a reliable state manager and the message's state and content are recoverable in case of a channel failure.
5.1.1 NOT_COMMITTED State Attributes
The transaction ID is maintained as an attributed of the NOT_COMMITTED state.
TODO: Can a message be part of multiple transactions at the same time, relative to the same channel?
5.1.2 Transitions to NOT_COMMITTED State
The channel state maintains a message as NOT_COMMITTED when the message is passed as argument of Channel.handle() in a transacted context.
5.1.3 Transitions from NOT_COMMITTED State
A message ceases to be NOT_COMMITTED in the following situations:
The message becomes NEW on transaction commit.
The message is removed from the channel upon expiration. The message's expiration should trigger a transaction rollback.
The message is removed from the channel on transaction rollback.
5.2 NEW Message State
A message is in a NEW state if the message was accepted by the channel, but the channel has not delivered the message to any receiver yet. Examples of situations when delivery is not possible or desirable are:
The channel has no receivers.
The channel has receivers and attempts delivery, but no receiver accepts the message.
The channel has receivers, but all receivers are "broken". A receiver is "broken" whether invoking Receiver.handle() throws an unchecked exception.
If the message is declared reliable, the channel must insure that the message's state is managed by a reliable state manager and the message's state and content are recoverable in case of a channel failure.
The ChannelNACK state's semantic, as defined by the current implementation, is similar to NEW. ChannelNACK state will be deprecated.
5.2.1 NEW attributes
There are no attributes to further qualify this state.
5.2.2 Transitions to NEW State
A message transitions to a NEW state in the following situations:
The message is passed as argument to the Channel.handle() method in a non-transacted context and the synchronous delivery fails.
The message is NOT_COMMITTED and the transaction it belongs to commits.
A delivery attempt of the message (currently in the NEW state) fails because the delivery is not possible for one of the reasons listed above.
5.2.3 Transitions from NEW State
A message ceases to be NEW in the following situations:
The message transitions to a DELIVERED state if there is at least one negative acknowledgment at delivery (see below).
The message is removed from the channel upon successful delivery. Successful delivery means that positive acknowledgments are received from all receivers the message is delivered to.
The message is removed from the channel upon expiration.
5.2.4 Channel behavior in the presence of undeliverable NEW messages
If a channel has no receivers, or all existing receivers do not accept the message, a specific channel implementation may choose between two possible outcomes:
Accept the message, positively acknowledge it to its sender and store it as NEW, pending further delivery.
Accept the message, positively acknowledge and discard it.
The first situation is common for a Point to Point messaging domain. A queue holds a message even if there are no receivers. The second situation is common for a Publish/Subscribe messaging domain. A topic discards a message if there are no receivers, but also positively acknowledges the message. Since this is an arbitrary choice, the channel must let its senders know what to expect in this situation. (See Section 7.3 Channel.isStoringUndeliverableMessages()).
5.3 DELIVERED Message State
A message transitions to a DELIVERED state if the channel attempts delivery, the message is accepted by receiver(s) but there is at least one receiver that negatively acknowledges the message. In this situation, the asynchronous channel continues to keep the message. If the message is declared reliable, the channel must insure that the message's state is managed by a reliable state manager and the message's state and content are recoverable in case of a channel failure.
5.3.1 DELIVERED Attributes
The DELIVERED state maintains a Set of receiver IDs for all receivers that negatively acknowledged the message. Since the channel state manager should allow transactional message state updates, a mechanism to logically record an acknowledgment (positive or negative) and then commit it should be available.
5.3.2 Transitions to DELIVERED State
A message becomes DELIVERED only by transition from a NEW state, on delivery, when the channel receives at least one negative acknowledgment.
5.3.3 Transitions from DELIVERED State
A message ceases to be in a DELIVERED state in the following situations
The message is removed from the channel upon expiration.
The message is removed from the channel when all outstanding negative acknowledgments are canceled out by positive acknowlegments.
An outstanding negative acknowledgment can be canceled in the following situations:
The receiver that initially negatively acknowledged the message asynchronously positively acknowledges the message later, using the Channel.acknowledge() method.
The receiver requests redelivery of the message and during redelivery it synchronously positively acknowledges the message.
The way a channel deals with acknowledgments if further specified in the section 6 (Acknowledgments).
A Channel does never take the initiative of redelivering a DELIVERED message. It is the receiver that negatively acknowledged the message that must explicitly requests the redelivery via Channel.deliver(Receiver r).
5.3.4 Positive acknowledgment race condition
A positive acknowledgment can be sent asynchronously by a receiver that initially negatively acknowledged the message. This behavior could lead to the following situation: the channel is in process of delivering the message to the receiver and during delivery the receiver sends the positive acknowledgment before the channel regains control over the delivery thread and associates the negative acknowledgment with the message state. Thus, the positive acknowledgment is lost in a race condition and the channel keeps the DELIVERED message forever (or until expires), because the target receiver will never request re-delivery.
There are at least two solutions to this problem:
Positive acknowledgments are explicitly stored by the channel state. Once stored, a positive acknowledgment can be canceled out when the corresponding negative acknowledgment is finally received by the channel. Upon storing an asynchronous positive acknowledgment, the message state will change from NEW to DELIVERED. The DELIVERED state should be able to explicitly store positive acknowledgments.
It is not possible to receive an early asynchronous positive acknowledgment if there is no corresponding NEW message in the channel. This should be handled as an error. If an as early acknowledgment is received for a NEW message, this should change the message state to DELIVERED.
The channel does not allow asynchronous positive acknowledgments while it is busy performing the delivery. However, this situation can lead to deadlock in case of a poorly written receiver.
6. Acknowledgments
A channel can receive the acknowledgment for a message either synchronously at delivery and asynchronously.
6.1 Synchronous Acknowledgment at Delivery
A channel synchronously receives a positive or negative acknowledgment when it attempts to deliver a message to a receiver using Receiver.handle().
If the message's state is NEW, it changes to DELIVERED. If the message's state is DELIVERED, the state doesn't change. A positive acknowledgment cancels out a pre-existing negative acknowledgment, or it is simply discarded if there is no preexisting negative acknowledgment. A negative acknowledgment is either stored or it cancels a pre-existing positive acknowledgment (this is only possible if we store positive acknowledgments - See Section 5.3.4 Positive acknowledgment race condition).
6.2 Asynchronous Positive Acknowledgments
A channel also accepts asynchronous positive acknowledgments. They can be submitted using Channel.acknowledge() method. If a pre-existing negative acknowledgment exists, it is canceled out. For the description of a possible complication related to asynchronous positive acknowledgments, see Section 5.3.4 Positive acknowledgment race condition.
6.2.1 Transacted positive acknowledgments
Positive acknowledgments can be submitted transactionally:
transactionManager.begin() channel.acknowledge(); ... transactionManager.commit();
It is the channel's responsibility to logically store acknowledgments before the transaction commit and only apply the acknowledgments on commit or discard them on rollback.
7. Channels
A channel is a message delivery mechanisms that forwards messages to registered receivers, and an essential component of the Messaging Core. The Channel interface presented in the original version of the Messaging Core specifications change as follows:
public interface Channel extends Receiver
{
public boolean isSynchronous();
public boolean setSynchronous(boolean b);
public boolean isReliable();
public void setPersistenceManager(PersistenceManager pm);
public boolean isTransactional();
public void setTransactionManager(TransactionManager tm);
public void setMessageStore(MessageStore store);
public boolean isStoringUndeliverableMessages();
public boolean deliver();
public boolean redeliver(Receiver t);
public Set browse(Filter filter);
public void acknowledge(Serializable messageID, Serializable receiverID);
public void close();
}
7.1 public boolean isReliable()
The method returns true if the channel is able to handle reliable messages. A channel is normally able to handle reliable messages if it has access to a valid persistence manager and the message store is reliable.
7.2 public boolean isTransactional()
The method returns true if the channel is able to transactionally handle messages and acknowledgments.
7.3 public boolean isStoringUndeliverableMessage()
See Section 5.2.4 "Channel behavior in the rpesence of undeliverable NEW messages" for further details of the channel behavior in the presence of undeliverable messages.
7.4 public boolean deliver()
The semantic of this method changes as follows: the channel attempts delivery for all NEW messages available at the moment of the invocation. After completion, the method returns false whether the channel still holds NEW messages that could not be delivered or true otherwise. The method does not interact with NOT_COMMITTED or DELIVERED messages. Invoking deliver() on a channel that has no NEW messages should return true.
7.5 public boolean redeliver(Receiver receiver)
The channel re-delivers all DELIVERED messages that have not been acknowledged yet by the receiver. The "redelivered" flag is set on true on all redelivered messages. The receiver has again a choice between synchronous positive acknowledgment or synchronous negative acknowledgment followed by later asynchronous positive acknowledgment.
The method requires a Receiver reference and not a receiver ID because it should be possible for a receiver to demand the redelivery of the negatively acknowledged messages even after it has been disconnected from the channel. For a disconnected receiver, providing just the receiverID is not enough to get a reference to the receiver.
TODO: What happens if in between invocations new NEW messages appear? Will they be delivered together with the previously negatively acknowledged messages or not?
7.6 public List browse(Filter filter)
Browse the channel. Returns a List of Messages. To browse the channel without a filter, use a null filter reference.
7.7 Additional protected methods
ChannelSupport is an abstract class that implements Channel and offers protected functionality that should be available to all Channel implementations. It exposes protected convenience methods that do not belong to Channel public interface.
protected boolean hasMessages(int state) protected List getMessages(int state)
The implementation of the hasMessages(int) method can be optimized if the channel maintains cached flags per message state.
8. Filters
http://jira.jboss.org/jira/browse/JBMESSAGING-64
http://jira.jboss.org/jira/browse/JBMESSAGING-65
A receiver must be able to selectively reject a message, even if it is connected to the sender and it is available for handling.
A sender should be able to find out in advance whether a receiver would accept a certain message or not. This is possible by invoking the Receiver's method accept(Routable r). If the method invocation returns true, the receiver declares its availability to handle the message and return a positive or negative acknowledgment. It is expected that a receiver will delegate the decision whether to accept or not a message to a Filter implementation. However, a Receiver implementation may choose to accept a message or not using a different mechanism.
The new Filter interface follows:
public interface Filter
{
public boolean accept(Routable routable);
}
The Receiver interface changes as follows
public interface Receiver extends Filter
{
....
8.1 Race condition when invoking accept()/handle()
A possible problem with using the filtering extension of the Receiver interface is that the receiver may change its state between the accept() invocation and the handle() invocation. Even if accept() returns true, handle() could be invoked on a receiver that does not accept that specific message anymore. There are several solutions to this problem, two of them being presented below.
8.1.1 Lock the Receiver
Provide additional locking and unlocking methods for Receiver. A possible sequence of calls in this situation would be the following:
Receiver receiver;
boolean acknowledgment;
receiver.lock();
try
{
if (!r.accept(routable))
{
return;
}
acknowledgment = receiver.handle(routable);
}
finally
{
receiver.unlock();
}
// deal with the acknowledgment
This solution is potentially dangerous because it allows external code to lock a Receiver and this may lead to a situation when the lock is never released. It also leads to cumbersome code.
8.1.2 Modify handle() to throw a checked exception
Another solution (preferred) would be to pass the locking responsibility to the receiver, and have handle() method throw a RoutableNotAcceptedException checked exception.
boolean acknowledgment;
try
{
acknowledgment = receiver.handle(routable);
}
catch(RoutableNotAcceptedException e)
{
// the receiver doesn't want the message
return;
}
// deal with the acknowledgment
Even if this solution is chosen, accept() method remains useful. It allows the sender to check in advance receiver's willingness to handle the specific message without being forced to actually attempt the delivery of the message.
9. Message Order
http://jira.jboss.org/jira/browse/JBMESSAGING-19
9.1 Order Guarantees
A channel provides the following partial order guarantees:
9.1.1 Messages submitted to the channel on the same thread of control are delivered to eligible receivers in the order they were submitted. If two messages M1 and M2 are delivered to the channel on the same thread, in this order, then all channel's receivers, regardless to the fact that are in the same address space of not, are guaranteed to receive messages in the order (M1, M2).
9.1.2 If messages are submitted to the channel on two different threads of control, in the same address space and in sequential order (the first thread had exited channel.handle() at the time the second thread enters it), the channel guarantees that the message received on the first thread will be delivered to eligible receivers before the message received on the second thread.
9.1.3 The delivery order of two messages submitted to the channel on concurrent threads, even if in the same address space, is unspecified.
9.1.4 The delivery order of two messages submitted to the channel in different address spaces is unspecified.
9.2 Message Order and Transactions
A channel preserves the order of the messages sent within the boundaries of the same transaction. If two messages M1 and M2 are sent to the channel in a transaction, in this order, they will be delivered to all eligible receivers in the order (M1, M2) when transaction commits.
9.3 Message Order and Reliability Class
The order guarantees specified so far are only valid for messages in the same reliability class. If a reliable message M1 is submitted to the channel before an unreliable message M2, the delivery order of messages M1 and M2 is undefined. There is no guarantee that the channel's receivers will receive M1 before M2.
9.4 Message Order and Priority
The order guarantees specified so far are only valid for message within the same priority group.
9.5 Message Order and Redelivery
The order guarantees specified so far are only valid for first time delivery. If two messages M1 and M2 are received by the Channel in this order but they are negatively acknowledged by a receiver, the order (M1, M2) is not guaranteed at redelivery.
9.6 Stronger guarantees
The partial order guarantees specified above should be sufficient to use channels as base for a JMS compliant implementation, without need for additional message order enforcement. If a messaging system based on JBoss Messaging Core needs stronger order guarantees (such as total order), it should be relatively easy to add those guarantees to a Channel extension.
10. Message Priority
http://jira.jboss.org/jira/browse/JBMESSAGING-59
Routable interface will be changed to implement java.lang.Comparable. The total ordering relation between Routable instances will reflect their priority.
-
Referenced by:
Comments