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 and Description
protected class	ServerConsumerImpl.BrowserDeliverer

Field Summary

Fields

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

Constructor Summary

Constructors

Constructor and Description
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, 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, 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 and Description
List<Long>	acknowledge(Transaction tx, long messageID)
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)
void	close(boolean failed, boolean sorted)
String	debug()
void	disconnect() disconnect the consumer
void	errorProcessing(Throwable e, MessageReference deliveryObject)
void	fireSlowConsumer()
void	forceDelivery(long sequence) Prompt delivery and send a "forced delivery" message to the consumer.
void	forceDelivery(long sequence, Runnable r)
AtomicInteger	getAvailableCredits() To be used on tests only
String	getConnectionClientID()
String	getConnectionID()
String	getConnectionLocalAddress() Returns a string representation of the local address this connection is connected to.
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()
int	getPriority()
Object	getProtocolContext() An object set by the Protocol implementation.
Object	getProtocolData() this is to be used with anything specific on a protocol head.
Queue	getQueue()
SimpleString	getQueueAddress()
SimpleString	getQueueName()
RoutingType	getQueueType()
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. We now first accept the message, and the actual deliver is done as part of Consumer.proceedDeliver(MessageReference).
void	individualAcknowledge(Transaction tx, long messageID)
void	individualCancel(long messageID, boolean failed, boolean sorted)
boolean	isBrowseOnly()
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	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() i
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 Detail

messageQueue

protected final Queue messageQueue

lock

protected final Object lock

browserDeliverer

protected ServerConsumerImpl.BrowserDeliverer browserDeliverer

Constructor Detail

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 Detail

debug

public String debug()

Specified by:

debug 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. This is because

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

See Also:

ServerConsumer.getProtocolContext()

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 String getConnectionID()

Specified by:

getConnectionID in interface ServerConsumer

getSessionID

public String getSessionID()

Specified by:

getSessionID in interface ServerConsumer

getDeliveringMessages

public List<MessageReference> getDeliveringMessages()

Specified by:

getDeliveringMessages in interface Consumer

Returns:

the list of messages being delivered

supportsDirectDelivery

public boolean supportsDirectDelivery()

i

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

Returns:

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

getFilter

public Filter getFilter()

Specified by:

getFilter in interface Consumer

getPriority

public int getPriority()

Specified by:

getPriority in interface PriorityAware

Specified by:

getPriority in interface Consumer

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

close

public void close(boolean failed,
                  boolean sorted)
           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,
                             boolean sorted)
                      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

Returns:

disconnect

public void disconnect()

Description copied from interface: Consumer

disconnect the consumer

Specified by:

disconnect 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()

Specified by:

getQueueName in interface ConsumerInfo

Returns:

name of the queue that is being consumed

getQueueType

public RoutingType getQueueType()

Specified by:

getQueueType in interface ConsumerInfo

Returns:

routing type of the queue that is being consumed

getQueueAddress

public SimpleString getQueueAddress()

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 protocol

getConnectionLocalAddress

public String getConnectionLocalAddress()

Description copied from interface: ConsumerInfo

Returns a string representation of the local address this connection is connected to. This is useful when the server is configured at 0.0.0.0 (or multiple IPs). This will give you the actual IP that's being used.

Specified by:

getConnectionLocalAddress in interface ConsumerInfo

Returns:

the local address

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:

the remote address