Package org.apache.activemq.artemis.core.server.impl

Class ServerConsumerImpl

java.lang.Object

org.apache.activemq.artemis.core.server.impl.ServerConsumerImpl

All Implemented Interfaces:

PriorityAware, Consumer, ConsumerInfo, ServerConsumer, ReadyListener

public class ServerConsumerImpl
extends Object
implements ServerConsumer, ReadyListener

Concrete implementation of a ClientConsumer.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
protected class	ServerConsumerImpl.BrowserDeliverer

Field Summary

Fields

Modifier and Type	Field	Description
protected ServerConsumerImpl.BrowserDeliverer	browserDeliverer
protected final Object	lock
protected final Queue	messageQueue

Constructor Summary

Constructors

Constructor	Description
ServerConsumerImpl(long id, ServerSession session, QueueBinding binding, Filter filter, boolean started, boolean browseOnly, StorageManager storageManager, SessionCallback callback, boolean preAcknowledge, boolean strictUpdateDeliveryCount, ManagementService managementService, boolean supportLargeMessage, Integer credits, ActiveMQServer server)
ServerConsumerImpl(long id, ServerSession session, QueueBinding binding, Filter filter, boolean started, boolean browseOnly, StorageManager storageManager, SessionCallback callback, boolean preAcknowledge, boolean strictUpdateDeliveryCount, ManagementService managementService, ActiveMQServer server)
ServerConsumerImpl(long id, ServerSession session, QueueBinding binding, Filter filter, int priority, boolean started, boolean browseOnly, StorageManager storageManager, SessionCallback callback, boolean preAcknowledge, boolean strictUpdateDeliveryCount, ManagementService managementService, boolean supportLargeMessage, Integer credits, ActiveMQServer server)

Method Summary

Modifier and Type	Method	Description
List<Long>	acknowledge(Transaction tx, long messageID)
void	addAttachment(String key, Object attachment)	Adds the given attachment to the ServerConsumer which will overwrite any previously assigned value with the same key.
boolean	allowReferenceCallback()	The current queue settings will allow use of the Reference Execution and callback.
void	backToDelivering(MessageReference reference)	Some protocols may choose to send the message back to delivering instead of redeliver.
List<MessageReference>	cancelRefs(boolean failed, boolean lastConsumedAsDelivered, Transaction tx)
void	close(boolean failed)
String	debug()
void	disconnect()	disconnect the consumer
void	errorProcessing(Throwable e, MessageReference deliveryObject)
void	failed(Throwable t)
void	fireSlowConsumer()
void	forceDelivery(long sequence)	Prompt delivery and send a "forced delivery" message to the consumer.
void	forceDelivery(long sequence, Runnable r)
Object	getAttachment(String key)	Gets any attachment that has been assigned to this ServerConsumer using the provided key.
Map<String,Object>	getAttachments()	Provides access to the full Map of consumer attachments in an unmodifiable Map instance.
AtomicInteger	getAvailableCredits()	To be used on tests only
Binding	getBinding()
SessionCallback	getCallback()
String	getConnectionClientID()
Object	getConnectionID()
String	getConnectionLocalAddress()	Returns a string representation of the local IP address this connection is connected to; useful when the server is configured at 0.0.0.0 (or multiple IPs).
String	getConnectionProtocolName()	Returns the name of the protocol for this Remoting Connection.
String	getConnectionRemoteAddress()	Returns a string representation of the remote address this connection is connected to.
long	getCreationTime()
List<MessageReference>	getDeliveringMessages()
Filter	getFilter()
SimpleString	getFilterString()
long	getID()
long	getLastAcknowledgedTime()	Returns the time in milliseconds that the last message was acknowledged by a consumer.
long	getLastDeliveredTime()	Returns the time in milliseconds that the last message was delivered to a consumer.
long	getMessagesAcknowledged()	Returns the number of messages acknowledged by this consumer since it was created.
int	getMessagesAcknowledgedAwaitingCommit()	Returns the number of acknowledged messages that are awaiting commit an a transaction.
long	getMessagesDelivered()	Returns the total number of messages sent to a consumer including redeliveries that have been acknowledged.
long	getMessagesDeliveredSize()	Returns the total size of all the messages delivered to the consumer including redelivered messages.
int	getMessagesInTransit()	Returns how many messages are out for delivery but not yet acknowledged.
long	getMessagesInTransitSize()	Returns the combined size of all the messages out for delivery but not yet acknowledged.
int	getPriority()
Object	getProtocolContext()	An object set by the Protocol implementation. it could be anything pre-determined by the implementation
Object	getProtocolData()	this is to be used with anything specific on a protocol head.
Queue	getQueue()
SimpleString	getQueueAddress()	Returns address of the queue that is being consumed.
SimpleString	getQueueName()	Returns name of the queue that is being consumed.
RoutingType	getQueueType()	Returns routing type of the queue that is being consumed.
float	getRate()
long	getSequentialID()	an unique sequential ID for this consumer
String	getSessionID()
String	getSessionName()
SlowConsumerDetectionListener	getSlowConsumerDetecion()
HandleStatus	handle(MessageReference ref)	There was a change on semantic during 2.3 here.
void	individualAcknowledge(Transaction tx, long messageID)
void	individualCancel(long messageID, boolean failed)
boolean	isBrowseOnly()
boolean	isClosed()
void	metricsAcknowledge(MessageReference ref, Transaction transaction)	This is needed when some protocols (OW) handle the acks themselves and need to update the metrics
void	proceedDeliver(MessageReference reference)	This will proceed with the actual delivery.
void	promptDelivery()	wakes up internal threads to deliver more messages
void	readyForWriting()
void	receiveCredits(int credits)
void	reject(long messageID)
void	removeAttachment(String key)	Remove the any attachment entry from the ServerConsumer clearing any previously assigned value
void	removeItself()	This method is just to remove itself from Queues.
MessageReference	removeReferenceByID(long messageID)
List<MessageReference>	scanDeliveringReferences(boolean remove, Function<MessageReference,Boolean> startFunction, Function<MessageReference,Boolean> endFunction)	Remove references based on the protocolData.
long	sequentialID()	an unique sequential ID for this consumer
void	setlowConsumerDetection(SlowConsumerDetectionListener listener)
void	setPreAcknowledge(boolean preAcknowledge)
void	setProtocolContext(Object protocolContext)
void	setProtocolData(Object protocolData)	this is to be used with anything specific on a protocol head.
void	setStarted(boolean started)
void	setTransferring(boolean transferring)
boolean	supportsDirectDelivery()	Whether this Consumer supports direct delivery.
String	toManagementString()	This method will create a string representation meant for management operations.
String	toString()
protected void	updateDeliveryCountForCanceledRef(MessageReference ref, boolean failed)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait

Field Details

messageQueue

protected final Queue messageQueue

lock

protected final Object lock

browserDeliverer

protected ServerConsumerImpl.BrowserDeliverer browserDeliverer

Constructor Details

ServerConsumerImpl

public ServerConsumerImpl(long id,
 ServerSession session,
 QueueBinding binding,
 Filter filter,
 boolean started,
 boolean browseOnly,
 StorageManager storageManager,
 SessionCallback callback,
 boolean preAcknowledge,
 boolean strictUpdateDeliveryCount,
 ManagementService managementService,
 ActiveMQServer server)
                   throws Exception

Throws:

Exception

ServerConsumerImpl

public ServerConsumerImpl(long id,
 ServerSession session,
 QueueBinding binding,
 Filter filter,
 boolean started,
 boolean browseOnly,
 StorageManager storageManager,
 SessionCallback callback,
 boolean preAcknowledge,
 boolean strictUpdateDeliveryCount,
 ManagementService managementService,
 boolean supportLargeMessage,
 Integer credits,
 ActiveMQServer server)
                   throws Exception

Throws:

Exception

ServerConsumerImpl

public ServerConsumerImpl(long id,
 ServerSession session,
 QueueBinding binding,
 Filter filter,
 int priority,
 boolean started,
 boolean browseOnly,
 StorageManager storageManager,
 SessionCallback callback,
 boolean preAcknowledge,
 boolean strictUpdateDeliveryCount,
 ManagementService managementService,
 boolean supportLargeMessage,
 Integer credits,
 ActiveMQServer server)
                   throws Exception

Throws:

Exception

Method Details

debug

public String debug()

Specified by:

debug in interface Consumer

isClosed

public boolean isClosed()

Specified by:

isClosed in interface Consumer

readyForWriting

public void readyForWriting()

Specified by:

readyForWriting in interface ReadyListener

allowReferenceCallback

public boolean allowReferenceCallback()

Description copied from interface: ServerConsumer

The current queue settings will allow use of the Reference Execution and callback.

Specified by:

allowReferenceCallback in interface ServerConsumer

sequentialID

public long sequentialID()

Description copied from interface: Consumer

an unique sequential ID for this consumer

Specified by:

sequentialID in interface Consumer

getProtocolData

public Object getProtocolData()

Description copied from interface: ServerConsumer

this is to be used with anything specific on a protocol head.

Specified by:

getProtocolData in interface ServerConsumer

setProtocolData

public void setProtocolData(Object protocolData)

Description copied from interface: ServerConsumer

this is to be used with anything specific on a protocol head.

Specified by:

setProtocolData in interface ServerConsumer

setlowConsumerDetection

public void setlowConsumerDetection(SlowConsumerDetectionListener listener)

Specified by:

setlowConsumerDetection in interface ServerConsumer

getSlowConsumerDetecion

public SlowConsumerDetectionListener getSlowConsumerDetecion()

Specified by:

getSlowConsumerDetecion in interface ServerConsumer

fireSlowConsumer

public void fireSlowConsumer()

Specified by:

fireSlowConsumer in interface ServerConsumer

getProtocolContext

public Object getProtocolContext()

Description copied from interface: ServerConsumer

An object set by the Protocol implementation. it could be anything pre-determined by the implementation

Specified by:

getProtocolContext in interface ServerConsumer

setProtocolContext

public void setProtocolContext(Object protocolContext)

Specified by:

setProtocolContext in interface ServerConsumer

getID

public long getID()

Specified by:

getID in interface ServerConsumer

isBrowseOnly

public boolean isBrowseOnly()

Specified by:

isBrowseOnly in interface ServerConsumer

getCreationTime

public long getCreationTime()

Specified by:

getCreationTime in interface ServerConsumer

getConnectionID

public Object getConnectionID()

Specified by:

getConnectionID in interface ServerConsumer

getSessionID

public String getSessionID()

Specified by:

getSessionID in interface ServerConsumer

metricsAcknowledge

public void metricsAcknowledge(MessageReference ref,
 Transaction transaction)

Description copied from interface: ServerConsumer

This is needed when some protocols (OW) handle the acks themselves and need to update the metrics

Specified by:

metricsAcknowledge in interface ServerConsumer

Parameters:

ref - the message reference

transaction - the tx

getDeliveringMessages

public List<MessageReference> getDeliveringMessages()

Specified by:

getDeliveringMessages in interface Consumer

supportsDirectDelivery

public boolean supportsDirectDelivery()

Description copied from interface: Consumer

Whether this Consumer supports direct delivery.

Specified by:

supportsDirectDelivery in interface Consumer

See Also:

• SessionCallback.supportsDirectDelivery()

errorProcessing

public void errorProcessing(Throwable e,
 MessageReference deliveryObject)

Specified by:

errorProcessing in interface Consumer

handle

public HandleStatus handle(MessageReference ref)
                    throws Exception

Description copied from interface: Consumer

There was a change on semantic during 2.3 here.

We now first accept the message, and the actual deliver is done as part of Consumer.proceedDeliver(MessageReference). This is to avoid holding a lock on the queues while the delivery is being accomplished To avoid a lock on the queue in case of misbehaving consumers.

This should return busy if handle is called before proceed deliver is called

Specified by:

handle in interface Consumer

Throws:

Exception

proceedDeliver

public void proceedDeliver(MessageReference reference)
                    throws Exception

Description copied from interface: Consumer

This will proceed with the actual delivery. Notice that handle should hold a readLock and proceedDelivery should release the readLock any lock operation on Consumer should also get a writeLock on the readWriteLock to guarantee there are no pending deliveries

Specified by:

proceedDeliver in interface Consumer

Throws:

Exception

getBinding

public Binding getBinding()

Specified by:

getBinding in interface Consumer

getFilter

public Filter getFilter()

Specified by:

getFilter in interface Consumer

getPriority

public int getPriority()

Specified by:

getPriority in interface Consumer

Specified by:

getPriority in interface PriorityAware

getFilterString

public SimpleString getFilterString()

Specified by:

getFilterString in interface ConsumerInfo

close

public void close(boolean failed)
           throws Exception

Specified by:

close in interface ServerConsumer

Throws:

Exception

removeItself

public void removeItself()
                  throws Exception

Description copied from interface: ServerConsumer

This method is just to remove itself from Queues. If for any reason during a close an exception occurred, the exception treatment will call removeItself what should take the consumer out of any queues.

Specified by:

removeItself in interface ServerConsumer

Throws:

Exception

forceDelivery

public void forceDelivery(long sequence)

Prompt delivery and send a "forced delivery" message to the consumer.

When the consumer receives such a "forced delivery" message, it discards it and knows that there are no other messages to be delivered.

Specified by:

forceDelivery in interface ServerConsumer

forceDelivery

public void forceDelivery(long sequence,
 Runnable r)

cancelRefs

public List<MessageReference> cancelRefs(boolean failed,
 boolean lastConsumedAsDelivered,
 Transaction tx)
                                  throws Exception

Specified by:

cancelRefs in interface ServerConsumer

Throws:

Exception

updateDeliveryCountForCanceledRef

protected void updateDeliveryCountForCanceledRef(MessageReference ref,
 boolean failed)

setStarted

public void setStarted(boolean started)

Specified by:

setStarted in interface ServerConsumer

setTransferring

public void setTransferring(boolean transferring)

Specified by:

setTransferring in interface ServerConsumer

receiveCredits

public void receiveCredits(int credits)

Specified by:

receiveCredits in interface ServerConsumer

getQueue

public Queue getQueue()

Specified by:

getQueue in interface ServerConsumer

scanDeliveringReferences

public List<MessageReference> scanDeliveringReferences(boolean remove,
 Function<MessageReference,Boolean> startFunction,
 Function<MessageReference,Boolean> endFunction)

Remove references based on the protocolData. There will be an interval defined between protocolDataStart and protocolDataEnd. This method will fetch the delivering references, remove them from the delivering list and return a list.

This will be useful for other protocols that will need this such as OpenWire or MQTT.

Specified by:

scanDeliveringReferences in interface ServerConsumer

acknowledge

public List<Long> acknowledge(Transaction tx,
 long messageID)
                       throws Exception

Specified by:

acknowledge in interface ServerConsumer

Throws:

Exception

individualAcknowledge

public void individualAcknowledge(Transaction tx,
 long messageID)
                           throws Exception

Specified by:

individualAcknowledge in interface ServerConsumer

Throws:

Exception

individualCancel

public void individualCancel(long messageID,
 boolean failed)
                      throws Exception

Specified by:

individualCancel in interface ServerConsumer

Throws:

Exception

reject

public void reject(long messageID)
            throws Exception

Specified by:

reject in interface ServerConsumer

Throws:

Exception

backToDelivering

public void backToDelivering(MessageReference reference)

Description copied from interface: ServerConsumer

Some protocols may choose to send the message back to delivering instead of redeliver. For example openwire will redeliver through the client, so messages will go back to delivering list after rollback.

Specified by:

backToDelivering in interface ServerConsumer

removeReferenceByID

public MessageReference removeReferenceByID(long messageID)
                                     throws Exception

Specified by:

removeReferenceByID in interface ServerConsumer

Throws:

Exception

getAvailableCredits

public AtomicInteger getAvailableCredits()

To be used on tests only

toString

public String toString()

Overrides:

toString in class Object

toManagementString

public String toManagementString()

Description copied from interface: Consumer

This method will create a string representation meant for management operations. This is different from the toString method that's meant for debugging and will contain information that regular users won't understand well

Specified by:

toManagementString in interface Consumer

disconnect

public void disconnect()

Description copied from interface: Consumer

disconnect the consumer

Specified by:

disconnect in interface Consumer

failed

public void failed(Throwable t)

Specified by:

failed in interface Consumer

getRate

public float getRate()

promptDelivery

public void promptDelivery()

Description copied from interface: Consumer

wakes up internal threads to deliver more messages

Specified by:

promptDelivery in interface Consumer

setPreAcknowledge

public void setPreAcknowledge(boolean preAcknowledge)

getSequentialID

public long getSequentialID()

Description copied from interface: ConsumerInfo

an unique sequential ID for this consumer

Specified by:

getSequentialID in interface ConsumerInfo

getQueueName

public SimpleString getQueueName()

Description copied from interface: ConsumerInfo

Returns name of the queue that is being consumed.

Specified by:

getQueueName in interface ConsumerInfo

Returns:

name of the queue that is being consumed

getQueueType

public RoutingType getQueueType()

Description copied from interface: ConsumerInfo

Returns routing type of the queue that is being consumed.

Specified by:

getQueueType in interface ConsumerInfo

Returns:

routing type of the queue that is being consumed

getQueueAddress

public SimpleString getQueueAddress()

Description copied from interface: ConsumerInfo

Returns address of the queue that is being consumed.

Specified by:

getQueueAddress in interface ConsumerInfo

Returns:

address of the queue that is being consumed

getSessionName

public String getSessionName()

Specified by:

getSessionName in interface ConsumerInfo

getConnectionClientID

public String getConnectionClientID()

Specified by:

getConnectionClientID in interface ConsumerInfo

getConnectionProtocolName

public String getConnectionProtocolName()

Description copied from interface: ConsumerInfo

Returns the name of the protocol for this Remoting Connection.

Specified by:

getConnectionProtocolName in interface ConsumerInfo

Returns:

the name of the protocol for this Remoting Connection

getConnectionLocalAddress

public String getConnectionLocalAddress()

Description copied from interface: ConsumerInfo

Returns a string representation of the local IP address this connection is connected to; useful when the server is configured at 0.0.0.0 (or multiple IPs).

Specified by:

getConnectionLocalAddress in interface ConsumerInfo

Returns:

a string representation of the local IP address this connection is connected to; useful when the server is configured at 0.0.0.0 (or multiple IPs)

getConnectionRemoteAddress

public String getConnectionRemoteAddress()

Description copied from interface: ConsumerInfo

Returns a string representation of the remote address this connection is connected to.

Specified by:

getConnectionRemoteAddress in interface ConsumerInfo

Returns:

a string representation of the remote address this connection is connected to

getMessagesInTransitSize

public long getMessagesInTransitSize()

Description copied from interface: ConsumerInfo

Returns the combined size of all the messages out for delivery but not yet acknowledged.

Specified by:

getMessagesInTransitSize in interface ConsumerInfo

Returns:

the combined size of all the messages out for delivery but not yet acknowledged

getMessagesInTransit

public int getMessagesInTransit()

Description copied from interface: ConsumerInfo

Returns how many messages are out for delivery but not yet acknowledged.

Specified by:

getMessagesInTransit in interface ConsumerInfo

Returns:

how many messages are out for delivery but not yet acknowledged

getLastDeliveredTime

public long getLastDeliveredTime()

Description copied from interface: ConsumerInfo

Returns the time in milliseconds that the last message was delivered to a consumer.

Specified by:

getLastDeliveredTime in interface ConsumerInfo

Returns:

the time in milliseconds that the last message was delivered to a consumer

getLastAcknowledgedTime

public long getLastAcknowledgedTime()

Description copied from interface: ConsumerInfo

Returns the time in milliseconds that the last message was acknowledged by a consumer.

Specified by:

getLastAcknowledgedTime in interface ConsumerInfo

Returns:

the time in milliseconds that the last message was acknowledged by a consumer

getMessagesAcknowledged

public long getMessagesAcknowledged()

Description copied from interface: ConsumerInfo

Returns the number of messages acknowledged by this consumer since it was created.

Specified by:

getMessagesAcknowledged in interface ConsumerInfo

Returns:

the number of messages acknowledged by this consumer since it was created

getMessagesDeliveredSize

public long getMessagesDeliveredSize()

Description copied from interface: ConsumerInfo

Returns the total size of all the messages delivered to the consumer including redelivered messages.

Specified by:

getMessagesDeliveredSize in interface ConsumerInfo

Returns:

the total size of all the messages delivered to the consumer including redelivered messages

getMessagesDelivered

public long getMessagesDelivered()

Description copied from interface: ConsumerInfo

Returns the total number of messages sent to a consumer including redeliveries that have been acknowledged.

Specified by:

getMessagesDelivered in interface ConsumerInfo

Returns:

the total number of messages sent to a consumer including redeliveries that have been acknowledged

getMessagesAcknowledgedAwaitingCommit

public int getMessagesAcknowledgedAwaitingCommit()

Description copied from interface: ConsumerInfo

Returns the number of acknowledged messages that are awaiting commit an a transaction.

Specified by:

getMessagesAcknowledgedAwaitingCommit in interface ConsumerInfo

Returns:

the number of acknowledged messages that are awaiting commit an a transaction

getCallback

public SessionCallback getCallback()

addAttachment

public void addAttachment(String key,
 Object attachment)

Description copied from interface: ServerConsumer

Adds the given attachment to the ServerConsumer which will overwrite any previously assigned value with the same key.

Specified by:

addAttachment in interface ServerConsumer

Parameters:

key - The key used to identify the attachment.

attachment - The actual value to store for the assigned key.

removeAttachment

public void removeAttachment(String key)

Description copied from interface: ServerConsumer

Remove the any attachment entry from the ServerConsumer clearing any previously assigned value

Specified by:

removeAttachment in interface ServerConsumer

Parameters:

key - The key used to identify the attachment.

getAttachment

public Object getAttachment(String key)

Description copied from interface: ServerConsumer

Gets any attachment that has been assigned to this ServerConsumer using the provided key. If no value was assigned a null is returned.

Specified by:

getAttachment in interface ServerConsumer

Parameters:

key - The key identifying the target attachment.

Returns:

the assigned value associated with the given key or null if nothing assigned.

getAttachments

public Map<String,Object> getAttachments()

Description copied from interface: ServerConsumer

Provides access to the full Map of consumer attachments in an unmodifiable Map instance. If no attachments are assigned to the consumer an empty Map instance is returned, never null.

Specified by:

getAttachments in interface ServerConsumer

Returns:

an unmodifiable Map that carries all consumer attachments.