Package org.objectweb.joram.mom.proxies
Class ClientSubscription
- java.lang.Object
-
- org.objectweb.joram.mom.proxies.ClientSubscription
-
- All Implemented Interfaces:
Encodable,Serializable,ClientSubscriptionMBean
public final class ClientSubscription extends Object implements ClientSubscriptionMBean, Serializable, Encodable
TheClientSubscriptionclass holds the data of a client subscription, and the methods managing the delivery and acknowledgement of the messages.- See Also:
- Serialized Form
-
-
Nested Class Summary
Nested Classes Modifier and Type Class Description static classClientSubscription.ClientSubscriptionFactory
-
Field Summary
Fields Modifier and Type Field Description private intactiveGive the maximum number of messages per request if the subscription is active, 0 if the subscription is passive.private StringclientIDprivate intcontextIdIdentifier of the subscription context.private static booleanDEBUGprivate Map<String,String>deliveredIdsTable of delivered messages identifiers.private Map<String,Integer>deniedMsgsTable keeping the denied messages identifiers.private AgentIddmqIdIdentifier of the subscriber's dead message queue,nullfor DMQ not set.private booleandurabletrueif the subscription is durable.private static Loggerloggerstatic intMAX_NUMBER_OF_MSG_PER_REQUESTstatic StringMAX_NUMBER_OF_MSG_PER_REQUEST_PROPERTYstatic StringMESSAGE_ID_LIST_PREFIXstatic StringMESSAGE_TIME_ID_LIST_PREFIXprivate MessageIdListmessageIdsList of identifiers of the messages to deliver.private MessageTablemessagesTableProxy messages table.private MessageIdListmessageTimeIdsList of identifiers of the messages delivery time.private booleanmodifiedTransient should be useless, should only be used via the engine thread..private StringnameThe subscription name.private intnbMaxMsgMax number of Message stored in the queue (-1 no limit).private longnbMsgsDeliveredSinceCreationthe number of delivered messagesprivate longnbMsgsSentToDMQSinceCreationthe number of erroneous messages forwarded to the DMQprivate booleannoFilteringtrueif the subscription does not filter messages in any way.private booleannoLocaltrueif the subscriber does not wish to consume messages published in the same context.private ProxyAgentItfproxyAn handle to proxy Agentprivate AgentIdproxyIdThe proxy's agent identifier.private longrequestExpTimeExpiration time of the "receive" request, if any.private intrequestIdIdentifier of the request requesting messages, either the listener's request, or a "receive" request.private StringselectorThe selector for filtering messages.private static longserialVersionUIDdefine serialVersionUID for interoperabilityprivate intsubRequestIdIdentifier of the subscription request.private intthresholdThreshold above which messages are considered as undeliverable because constantly denied. 0 stands for no threshold, -1 for value not set (use servers' default value).private booleantoListenertrueif the messages are destinated to a listener.private AgentIdtopicIdThe topic identifier.private StringtxNameTransient should be useless, should only be used via the engine thread..-
Fields inherited from interface fr.dyade.aaa.common.encoding.Encodable
BOOLEAN_ENCODED_SIZE, BYTE_ENCODED_SIZE, DOUBLE_ENCODED_SIZE, FLOAT_ENCODED_SIZE, INT_ENCODED_SIZE, LONG_ENCODED_SIZE, SHORT_ENCODED_SIZE
-
-
Constructor Summary
Constructors Constructor Description ClientSubscription()ClientSubscription(AgentId proxyId, int contextId, int reqId, boolean durable, AgentId topicId, String name, String selector, boolean noLocal, AgentId dmqId, int threshold, int nbMaxMsg, MessageTable messagesTable, String clientID)Constructs aClientSubscriptioninstance.
-
Method Summary
All Methods Static Methods Instance Methods Concrete Methods Modifier and Type Method Description (package private) voidacknowledge(String id)(package private) voidacknowledge(Iterator<String> acks)Acknowledges messages.(package private) voidbrowseNewMessages(List<Message> newMessages)Browses messages and keeps those which will have to be delivered to the subscriber.voidclear()Deletes all messages(package private) voiddeactivate(boolean denies)De-activates the subscription.voiddecode(Decoder decoder)Decodes the content of this objectprivate voiddecrAckCounters(String msgId, Message message)voiddelete()voiddeleteMessage(String msgId)Deletes a particular pending message in the subscription.(package private) voiddeleteMessages()Decreases the subscription's messages acknowledgement expectations, deletes those not to be consumed anymore.(package private) ConsumerMessagesdeliver()Launches a delivery sequence, either for a listener, or for a receiver.(package private) voiddeny(Iterator<String> denies, boolean redelivered)Denies messages.private voiddeny(Iterator<String> denies, boolean remove, boolean redelivered)Denies the messages.voidencode(Encoder encoder)Encodes the content of this objectintgetActive()Returns the maximum number of messages per request if the subscription is active.(package private) StringgetClientID()Only used in tests.intgetContextId()Returns the subscription's context identifier.(package private) Map<String,String>getDeliveredIds()Only used in tests.intgetDeliveredMessageCount()Returns the number of messages delivered and waiting for acknowledge.(package private) Map<String,Integer>getDeniedMsgs()Only used in tests.booleangetDurable()Returnstrueif the subscription is durable.intgetEncodableClassId()Returns a unique class identifier.intgetEncodedSize()Returns the size of the byte array that results from the encoding of this object.CompositeDatagetMessage(String msgId)Returns the description of a particular pending message.String[]getMessageIds()Returns the list of message's identifiers for the subscription.TabularDatagetMessages()Returns the description of all pending messages.List<MessageView>getMessagesView()Only used from shell-mom MOMCommandsImpl.StringgetName()Returns the name of the subscription.intgetNbMaxMsg()Returns the maximum number of message for the subscription.longgetNbMsgsDeliveredSinceCreation()Returns the number of messages delivered to the client since creation time of this subscription.longgetNbMsgsSentToDMQSinceCreation()Returns the number of erroneous messages forwarded to the DMQ since creation time of this subscription.intgetPendingDeliveryTimeMessageCount()Returns the number of pending delivery time messages for the subscription.intgetPendingMessageCount()Returns the number of pending messages for the subscription.StringgetSelector()Returns the selector.(package private) MessagegetSubMessage(String msgId)intgetSubRequestId()Returns the identifier of the subscribing request.(package private) MessagegetSubscriptionMessage(String msgId)intgetThreshold()Returns the threshold above which messages are considered undeliverable because constantly denied.AgentIdgetTopicId()Returns the identifier of the subscription topic.StringgetTopicIdAsString()Returns the identifier of the subscription topic.static StringgetTransactionPrefix(AgentId proxyId)private StringgetTxName()voidinitMessageIds()private booleanisUndeliverable(int deliveryAttempts)Returnstrueif a given message is considered as undeliverable, because its delivery count matches the subscription's threshold, if any, or the server's default threshold value (if any).static ClientSubscriptionload(String txName)voidloadMessageIds()(package private) voidreactivate(int contextId, int reqId, AgentId topicId, String selector, boolean noLocal)Reactivates the subscription.(package private) voidreinitialize(MessageTable messagesTable, List<Message> persistedMessages, boolean denyDeliveredMessages)Re-initializes the client subscription.(package private) MessageremoveMessage(String msgId)Removes a particular pending message in the subscription.(package private) voidremoveMessagesTimeIds(String msgId)voidsave()(package private) voidscheduleDeliveryTimeMessage(Message message)schedule the delivery time message.(package private) voidsetActive(int active)(package private) voidsetDMQId(AgentId dmqId)Sets the subscription's dead message queue identifier.(package private) voidsetListener(int requestId)Sets a listener.private voidsetModified()voidsetNbMaxMsg(int nbMaxMsg)Sets the maximum number of message for the subscription.(package private) voidsetProxyAgent(ProxyAgentItf proxy)(package private) voidsetReceiver(int requestId, long timeToLive)Sets a receiver request.voidsetThreshold(int threshold)Sets the subscription's threshold value.StringtoString()(package private) voidunsetListener()Unsets the listener.(package private) voidunsetReceiver()Unsets a receiver request.
-
-
-
Field Detail
-
serialVersionUID
private static final long serialVersionUID
define serialVersionUID for interoperability- See Also:
- Constant Field Values
-
logger
private static final Logger logger
-
DEBUG
private static final boolean DEBUG
-
MESSAGE_ID_LIST_PREFIX
public static final String MESSAGE_ID_LIST_PREFIX
- See Also:
- Constant Field Values
-
MESSAGE_TIME_ID_LIST_PREFIX
public static final String MESSAGE_TIME_ID_LIST_PREFIX
- See Also:
- Constant Field Values
-
proxyId
private AgentId proxyId
The proxy's agent identifier.
-
proxy
private transient ProxyAgentItf proxy
An handle to proxy Agent
-
durable
private boolean durable
trueif the subscription is durable.
-
topicId
private AgentId topicId
The topic identifier.
-
name
private String name
The subscription name.
-
selector
private String selector
The selector for filtering messages.
-
dmqId
private AgentId dmqId
Identifier of the subscriber's dead message queue,nullfor DMQ not set.
-
threshold
private int threshold
Threshold above which messages are considered as undeliverable because constantly denied. 0 stands for no threshold, -1 for value not set (use servers' default value).
-
nbMaxMsg
private int nbMaxMsg
Max number of Message stored in the queue (-1 no limit).
-
messageIds
private transient MessageIdList messageIds
List of identifiers of the messages to deliver.
-
messageTimeIds
private transient MessageIdList messageTimeIds
List of identifiers of the messages delivery time.
-
contextId
private transient int contextId
Identifier of the subscription context.
-
subRequestId
private transient int subRequestId
Identifier of the subscription request.
-
noLocal
private transient boolean noLocal
trueif the subscriber does not wish to consume messages published in the same context.
-
noFiltering
private transient boolean noFiltering
trueif the subscription does not filter messages in any way.
-
active
private transient int active
Give the maximum number of messages per request if the subscription is active, 0 if the subscription is passive.
-
requestId
private transient int requestId
Identifier of the request requesting messages, either the listener's request, or a "receive" request.
-
toListener
private transient boolean toListener
trueif the messages are destinated to a listener.
-
requestExpTime
private transient long requestExpTime
Expiration time of the "receive" request, if any.
-
messagesTable
private transient MessageTable messagesTable
Proxy messages table. Be careful: currently this table is shared between all subscription.
-
nbMsgsSentToDMQSinceCreation
private long nbMsgsSentToDMQSinceCreation
the number of erroneous messages forwarded to the DMQ
-
nbMsgsDeliveredSinceCreation
private long nbMsgsDeliveredSinceCreation
the number of delivered messages
-
MAX_NUMBER_OF_MSG_PER_REQUEST_PROPERTY
public static final String MAX_NUMBER_OF_MSG_PER_REQUEST_PROPERTY
- See Also:
- Constant Field Values
-
MAX_NUMBER_OF_MSG_PER_REQUEST
public static final int MAX_NUMBER_OF_MSG_PER_REQUEST
-
txName
private transient String txName
Transient should be useless, should only be used via the engine thread..
-
modified
private transient boolean modified
Transient should be useless, should only be used via the engine thread..
-
clientID
private String clientID
-
-
Constructor Detail
-
ClientSubscription
ClientSubscription()
-
ClientSubscription
ClientSubscription(AgentId proxyId, int contextId, int reqId, boolean durable, AgentId topicId, String name, String selector, boolean noLocal, AgentId dmqId, int threshold, int nbMaxMsg, MessageTable messagesTable, String clientID) throws RequestException
Constructs aClientSubscriptioninstance.- Parameters:
proxyId- Proxy's identifier.contextId- Context identifier.reqId- Request identifier.durable-truefor a durable subscription.topicId- Topic identifier.name- Subscription's name.selector- Selector for filtering messages.noLocal-truefor not consuming messages published within the same proxy's context.dmqId- Identifier of the proxy's dead message queue, if any.threshold- Proxy's threshold value, if any.messagesTable- Proxy's messages table.clientID- the clientID- Throws:
RequestException
-
-
Method Detail
-
setProxyAgent
void setProxyAgent(ProxyAgentItf proxy)
-
getDurable
public boolean getDurable()
Returnstrueif the subscription is durable.- Specified by:
getDurablein interfaceClientSubscriptionMBean
-
getTopicId
public AgentId getTopicId()
Returns the identifier of the subscription topic.
-
getName
public String getName()
Returns the name of the subscription.- Specified by:
getNamein interfaceClientSubscriptionMBean
-
getSelector
public String getSelector()
Returns the selector.- Specified by:
getSelectorin interfaceClientSubscriptionMBean
-
getThreshold
public int getThreshold()
Returns the threshold above which messages are considered undeliverable because constantly denied.- Specified by:
getThresholdin interfaceClientSubscriptionMBean- Returns:
- the threshold if set; -1 otherwise.
-
setThreshold
public void setThreshold(int threshold)
Sets the subscription's threshold value.
-
getNbMaxMsg
public int getNbMaxMsg()
Returns the maximum number of message for the subscription. If the limit is unset the method returns -1.- Specified by:
getNbMaxMsgin interfaceClientSubscriptionMBean- Returns:
- the maximum number of message for subscription if set; -1 otherwise.
-
setNbMaxMsg
public void setNbMaxMsg(int nbMaxMsg)
Sets the maximum number of message for the subscription.- Specified by:
setNbMaxMsgin interfaceClientSubscriptionMBean- Parameters:
nbMaxMsg- the maximum number of message for subscription (-1 set no limit).
-
getPendingMessageCount
public int getPendingMessageCount()
Returns the number of pending messages for the subscription.- Specified by:
getPendingMessageCountin interfaceClientSubscriptionMBean- Returns:
- The number of pending message for the subscription.
-
getPendingDeliveryTimeMessageCount
public int getPendingDeliveryTimeMessageCount()
Returns the number of pending delivery time messages for the subscription.- Returns:
- The number of pending delivery time message for the subscription.
-
getContextId
public int getContextId()
Returns the subscription's context identifier.- Specified by:
getContextIdin interfaceClientSubscriptionMBean
-
getSubRequestId
public int getSubRequestId()
Returns the identifier of the subscribing request.- Specified by:
getSubRequestIdin interfaceClientSubscriptionMBean
-
getActive
public int getActive()
Returns the maximum number of messages per request if the subscription is active.- Specified by:
getActivein interfaceClientSubscriptionMBean
-
setActive
void setActive(int active)
-
setModified
private void setModified()
-
getClientID
String getClientID()
Only used in tests.
-
getDeliveredMessageCount
public int getDeliveredMessageCount()
Returns the number of messages delivered and waiting for acknowledge.- Specified by:
getDeliveredMessageCountin interfaceClientSubscriptionMBean- Returns:
- The number of messages delivered and waiting for acknowledge.
-
getMessageIds
public String[] getMessageIds()
Returns the list of message's identifiers for the subscription.- Specified by:
getMessageIdsin interfaceClientSubscriptionMBean- Returns:
- the list of message's identifiers for the subscription.
-
reinitialize
void reinitialize(MessageTable messagesTable, List<Message> persistedMessages, boolean denyDeliveredMessages)
Re-initializes the client subscription.- Parameters:
messagesTable- Proxy's table where storing the messages.persistedMessages- Proxy's persisted messages.denyDeliveredMessages- Denies already delivered messages.
-
reactivate
void reactivate(int contextId, int reqId, AgentId topicId, String selector, boolean noLocal)Reactivates the subscription.- Parameters:
contextId- Re-activation context.reqId- Re-activation request identifier.topicId- Topic identifier.selector- Selector for filtering messages.noLocal-truefor not consuming messages published within the same proxy's context.
-
deactivate
void deactivate(boolean denies)
De-activates the subscription.- Parameters:
denies- denies the non acknowledged messages.
-
setListener
void setListener(int requestId)
Sets a listener.- Parameters:
requestId- Identifier of the listener request.
-
unsetListener
void unsetListener()
Unsets the listener.
-
setReceiver
void setReceiver(int requestId, long timeToLive)Sets a receiver request.- Parameters:
requestId- Identifier of the "receive" request.timeToLive- Request's time to live value.
-
unsetReceiver
void unsetReceiver()
Unsets a receiver request.
-
setDMQId
void setDMQId(AgentId dmqId)
Sets the subscription's dead message queue identifier.
-
scheduleDeliveryTimeMessage
void scheduleDeliveryTimeMessage(Message message)
schedule the delivery time message.- Parameters:
message- the message.
-
removeMessagesTimeIds
void removeMessagesTimeIds(String msgId)
-
browseNewMessages
void browseNewMessages(List<Message> newMessages)
Browses messages and keeps those which will have to be delivered to the subscriber.
-
deliver
ConsumerMessages deliver()
Launches a delivery sequence, either for a listener, or for a receiver.
-
acknowledge
void acknowledge(String id)
-
deny
void deny(Iterator<String> denies, boolean redelivered)
Denies messages.- Parameters:
denies- all ids of the messages to denyredelivered- true if redelivered.
-
deny
private void deny(Iterator<String> denies, boolean remove, boolean redelivered)
Denies the messages.- Parameters:
denies- all ids of the messages to denyremove- true to remove messages from deliveredIds map. Must be false when denies iterates over deliveredIds map keys, to avoid a ConcurrentModificationException.redelivered- true if redelivered.
-
deleteMessages
void deleteMessages()
Decreases the subscription's messages acknowledgement expectations, deletes those not to be consumed anymore.
-
isUndeliverable
private boolean isUndeliverable(int deliveryAttempts)
Returnstrueif a given message is considered as undeliverable, because its delivery count matches the subscription's threshold, if any, or the server's default threshold value (if any).
-
getNbMsgsSentToDMQSinceCreation
public long getNbMsgsSentToDMQSinceCreation()
Description copied from interface:ClientSubscriptionMBeanReturns the number of erroneous messages forwarded to the DMQ since creation time of this subscription.- Specified by:
getNbMsgsSentToDMQSinceCreationin interfaceClientSubscriptionMBean- Returns:
- the number of erroneous messages forwarded to the DMQ.
-
getNbMsgsDeliveredSinceCreation
public long getNbMsgsDeliveredSinceCreation()
Description copied from interface:ClientSubscriptionMBeanReturns the number of messages delivered to the client since creation time of this subscription.- Specified by:
getNbMsgsDeliveredSinceCreationin interfaceClientSubscriptionMBean- Returns:
- the number of delivered messages.
-
deleteMessage
public void deleteMessage(String msgId)
Description copied from interface:ClientSubscriptionMBeanDeletes a particular pending message in the subscription. The message is pointed out through its unique identifier.- Specified by:
deleteMessagein interfaceClientSubscriptionMBean- Parameters:
msgId- The unique message's identifier.
-
clear
public void clear()
Description copied from interface:ClientSubscriptionMBeanDeletes all messages- Specified by:
clearin interfaceClientSubscriptionMBean
-
removeMessage
Message removeMessage(String msgId)
Removes a particular pending message in the subscription. The message is pointed out through its unique identifier.- Parameters:
msgId- The unique message's identifier.
-
getMessage
public CompositeData getMessage(String msgId) throws Exception
Returns the description of a particular pending message. The message is pointed out through its unique identifier.- Specified by:
getMessagein interfaceClientSubscriptionMBean- Parameters:
msgId- The unique message's identifier.- Returns:
- the description of the message.
- Throws:
Exception- See Also:
MessageJMXWrapper
-
getMessages
public TabularData getMessages() throws Exception
Returns the description of all pending messages.- Specified by:
getMessagesin interfaceClientSubscriptionMBean- Returns:
- the description of the message.
- Throws:
Exception- See Also:
MessageJMXWrapper
-
getMessagesView
public List<MessageView> getMessagesView()
Only used from shell-mom MOMCommandsImpl.
-
getTopicIdAsString
public String getTopicIdAsString()
Returns the identifier of the subscription topic. Only used from shell-mom MOMCommandsImpl.- Specified by:
getTopicIdAsStringin interfaceClientSubscriptionMBean
-
getTxName
private final String getTxName()
-
load
public static final ClientSubscription load(String txName) throws Exception
- Throws:
Exception
-
save
public void save()
-
delete
public void delete()
-
getEncodableClassId
public int getEncodableClassId()
Description copied from interface:EncodableReturns a unique class identifier.- Specified by:
getEncodableClassIdin interfaceEncodable- Returns:
- a unique class identifier
-
getEncodedSize
public int getEncodedSize() throws ExceptionDescription copied from interface:EncodableReturns the size of the byte array that results from the encoding of this object.- Specified by:
getEncodedSizein interfaceEncodable- Returns:
- the size of the encoded byte array
- Throws:
Exception- if an error occurs
-
encode
public void encode(Encoder encoder) throws Exception
Description copied from interface:EncodableEncodes the content of this object
-
-