org.springframework.amqp.rabbit.listener

Class BlockingQueueConsumer

java.lang.Object

org.springframework.amqp.rabbit.listener.BlockingQueueConsumer

public class BlockingQueueConsumer
extends java.lang.Object

Specialized consumer encapsulating knowledge of the broker connections and having its own lifecycle (start and stop).

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
class	BlockingQueueConsumer.DeclarationException
private static class	BlockingQueueConsumer.Delivery Encapsulates an arbitrary message - simple "bean" holder structure.
private class	BlockingQueueConsumer.InternalConsumer

Field Summary

Fields

Modifier and Type	Field and Description
private AcknowledgeMode	acknowledgeMode
private ActiveObjectCounter<BlockingQueueConsumer>	activeObjectCounter
private java.util.concurrent.atomic.AtomicBoolean	cancelled
private java.util.concurrent.atomic.AtomicBoolean	cancelReceived
private com.rabbitmq.client.Channel	channel
private ConnectionFactory	connectionFactory
private BlockingQueueConsumer.InternalConsumer	consumer
private java.util.Map<java.lang.String,java.lang.Object>	consumerArgs
private java.util.Collection<java.lang.String>	consumerTags
private int	declarationRetries
private boolean	defaultRequeuRejected
private java.util.Set<java.lang.Long>	deliveryTags
private boolean	exclusive
private long	failedDeclarationRetryInterval
private long	lastRetryDeclaration
private static org.apache.commons.logging.Log	logger
private MessagePropertiesConverter	messagePropertiesConverter
private java.util.Set<java.lang.String>	missingQueues
private int	prefetchCount
private java.util.concurrent.BlockingQueue<BlockingQueueConsumer.Delivery>	queue
private java.lang.String[]	queues
private RabbitResourceHolder	resourceHolder
private long	retryDeclarationInterval
private com.rabbitmq.client.ShutdownSignalException	shutdown
private long	shutdownTimeout
private java.util.concurrent.CountDownLatch	suspendClientThread
private boolean	transactional

Constructor Summary

Constructors

Constructor and Description
BlockingQueueConsumer(ConnectionFactory connectionFactory, MessagePropertiesConverter messagePropertiesConverter, ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, boolean defaultRequeueRejected, java.util.Map<java.lang.String,java.lang.Object> consumerArgs, boolean exclusive, java.lang.String... queues) Create a consumer.
BlockingQueueConsumer(ConnectionFactory connectionFactory, MessagePropertiesConverter messagePropertiesConverter, ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, boolean defaultRequeueRejected, java.util.Map<java.lang.String,java.lang.Object> consumerArgs, java.lang.String... queues) Create a consumer.
BlockingQueueConsumer(ConnectionFactory connectionFactory, MessagePropertiesConverter messagePropertiesConverter, ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, boolean defaultRequeueRejected, java.lang.String... queues) Create a consumer.
BlockingQueueConsumer(ConnectionFactory connectionFactory, MessagePropertiesConverter messagePropertiesConverter, ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter, AcknowledgeMode acknowledgeMode, boolean transactional, int prefetchCount, java.lang.String... queues) Create a consumer.

Method Summary

Modifier and Type	Method and Description
private void	attemptPassiveDeclarations()
private void	checkMissingQueues()
private void	checkShutdown() Check if we are in shutdown mode and if so throw an exception.
boolean	commitIfNecessary(boolean locallyTransacted) Perform a commit or message acknowledgement, as appropriate.
private void	consumeFromQueue(java.lang.String queue)
com.rabbitmq.client.Channel	getChannel()
java.lang.String	getConsumerTag()
private Message	handle(BlockingQueueConsumer.Delivery delivery) If this is a non-POISON non-null delivery simply return it.
Message	nextMessage() Main application-side API: wait for the next message delivery and return it.
Message	nextMessage(long timeout) Main application-side API: wait for the next message delivery and return it.
void	rollbackOnExceptionIfNecessary(java.lang.Throwable ex) Perform a rollback, handling rollback exceptions properly.
void	setDeclarationRetries(int declarationRetries) Set the number of retries after passive queue declaration fails.
void	setFailedDeclarationRetryInterval(long failedDeclarationRetryInterval) Set the interval between passive queue declaration attempts in milliseconds.
void	setQuiesce(long shutdownTimeout) Stop receiving new messages; drain the queue of any prefetched messages.
void	setRetryDeclarationInterval(long retryDeclarationInterval) When consuming multiple queues, set the interval between declaration attempts when only a subset of the queues were available (milliseconds).
void	start()
void	stop()
java.lang.String	toString()

Methods inherited from class java.lang.Object

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

Field Detail

logger

private static org.apache.commons.logging.Log logger

queue

private final java.util.concurrent.BlockingQueue<BlockingQueueConsumer.Delivery> queue

shutdown

private volatile com.rabbitmq.client.ShutdownSignalException shutdown

queues

private final java.lang.String[] queues

prefetchCount

private final int prefetchCount

transactional

private final boolean transactional

channel

private com.rabbitmq.client.Channel channel

resourceHolder

private RabbitResourceHolder resourceHolder

consumer

private BlockingQueueConsumer.InternalConsumer consumer

cancelled

private final java.util.concurrent.atomic.AtomicBoolean cancelled

shutdownTimeout

private volatile long shutdownTimeout

cancelReceived

private final java.util.concurrent.atomic.AtomicBoolean cancelReceived

acknowledgeMode

private final AcknowledgeMode acknowledgeMode

connectionFactory

private final ConnectionFactory connectionFactory

messagePropertiesConverter

private final MessagePropertiesConverter messagePropertiesConverter

activeObjectCounter

private final ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter

consumerArgs

private final java.util.Map<java.lang.String,java.lang.Object> consumerArgs

exclusive

private final boolean exclusive

deliveryTags

private final java.util.Set<java.lang.Long> deliveryTags

defaultRequeuRejected

private final boolean defaultRequeuRejected

suspendClientThread

private final java.util.concurrent.CountDownLatch suspendClientThread

consumerTags

private final java.util.Collection<java.lang.String> consumerTags

missingQueues

private final java.util.Set<java.lang.String> missingQueues

retryDeclarationInterval

private long retryDeclarationInterval

failedDeclarationRetryInterval

private long failedDeclarationRetryInterval

declarationRetries

private int declarationRetries

lastRetryDeclaration

private long lastRetryDeclaration

Constructor Detail

BlockingQueueConsumer

public BlockingQueueConsumer(ConnectionFactory connectionFactory,
                             MessagePropertiesConverter messagePropertiesConverter,
                             ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter,
                             AcknowledgeMode acknowledgeMode,
                             boolean transactional,
                             int prefetchCount,
                             java.lang.String... queues)

Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker until it is started. RequeueRejected defaults to true.

Parameters:

connectionFactory - The connection factory.

messagePropertiesConverter - The properties converter.

activeObjectCounter - The active object counter; used during shutdown.

acknowledgeMode - The acknowledgemode.

transactional - Whether the channel is transactional.

prefetchCount - The prefetch count.

queues - The queues.

BlockingQueueConsumer

public BlockingQueueConsumer(ConnectionFactory connectionFactory,
                             MessagePropertiesConverter messagePropertiesConverter,
                             ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter,
                             AcknowledgeMode acknowledgeMode,
                             boolean transactional,
                             int prefetchCount,
                             boolean defaultRequeueRejected,
                             java.lang.String... queues)

Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker until it is started.

Parameters:

connectionFactory - The connection factory.

messagePropertiesConverter - The properties converter.

activeObjectCounter - The active object counter; used during shutdown.

acknowledgeMode - The acknowledge mode.

transactional - Whether the channel is transactional.

prefetchCount - The prefetch count.

defaultRequeueRejected - true to reject requeued messages.

queues - The queues.

BlockingQueueConsumer

public BlockingQueueConsumer(ConnectionFactory connectionFactory,
                             MessagePropertiesConverter messagePropertiesConverter,
                             ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter,
                             AcknowledgeMode acknowledgeMode,
                             boolean transactional,
                             int prefetchCount,
                             boolean defaultRequeueRejected,
                             java.util.Map<java.lang.String,java.lang.Object> consumerArgs,
                             java.lang.String... queues)

Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker until it is started.

Parameters:

connectionFactory - The connection factory.

messagePropertiesConverter - The properties converter.

activeObjectCounter - The active object counter; used during shutdown.

acknowledgeMode - The acknowledge mode.

transactional - Whether the channel is transactional.

prefetchCount - The prefetch count.

defaultRequeueRejected - true to reject requeued messages.

consumerArgs - The consumer arguments (e.g. x-priority).

queues - The queues.

BlockingQueueConsumer

public BlockingQueueConsumer(ConnectionFactory connectionFactory,
                             MessagePropertiesConverter messagePropertiesConverter,
                             ActiveObjectCounter<BlockingQueueConsumer> activeObjectCounter,
                             AcknowledgeMode acknowledgeMode,
                             boolean transactional,
                             int prefetchCount,
                             boolean defaultRequeueRejected,
                             java.util.Map<java.lang.String,java.lang.Object> consumerArgs,
                             boolean exclusive,
                             java.lang.String... queues)

Create a consumer. The consumer must not attempt to use the connection factory or communicate with the broker until it is started.

Parameters:

connectionFactory - The connection factory.

messagePropertiesConverter - The properties converter.

activeObjectCounter - The active object counter; used during shutdown.

acknowledgeMode - The acknowledge mode.

transactional - Whether the channel is transactional.

prefetchCount - The prefetch count.

defaultRequeueRejected - true to reject requeued messages.

consumerArgs - The consumer arguments (e.g. x-priority).

exclusive - true if the consumer is to be exclusive.

queues - The queues.

Method Detail

getChannel

public com.rabbitmq.client.Channel getChannel()

getConsumerTag

public java.lang.String getConsumerTag()

setQuiesce

public final void setQuiesce(long shutdownTimeout)

Stop receiving new messages; drain the queue of any prefetched messages.

Parameters:

shutdownTimeout - how long (ms) to suspend the client thread.

setDeclarationRetries

public void setDeclarationRetries(int declarationRetries)

Set the number of retries after passive queue declaration fails.

Parameters:

declarationRetries - The number of retries, default 3.

Since:

1.3.9

See Also:

setFailedDeclarationRetryInterval(long)

setFailedDeclarationRetryInterval

public void setFailedDeclarationRetryInterval(long failedDeclarationRetryInterval)

Set the interval between passive queue declaration attempts in milliseconds.

Parameters:

failedDeclarationRetryInterval - the interval, default 5000.

Since:

1.3.9

See Also:

setDeclarationRetries(int)

setRetryDeclarationInterval

public void setRetryDeclarationInterval(long retryDeclarationInterval)

When consuming multiple queues, set the interval between declaration attempts when only a subset of the queues were available (milliseconds).

Parameters:

retryDeclarationInterval - the interval, default 60000.

Since:

1.3.9

checkShutdown

private void checkShutdown()

Check if we are in shutdown mode and if so throw an exception.

handle

private Message handle(BlockingQueueConsumer.Delivery delivery)
                throws java.lang.InterruptedException

If this is a non-POISON non-null delivery simply return it. If this is POISON we are in shutdown mode, throw shutdown. If delivery is null, we may be in shutdown mode. Check and see.

Throws:

java.lang.InterruptedException

nextMessage

public Message nextMessage()
                    throws java.lang.InterruptedException,
                           com.rabbitmq.client.ShutdownSignalException

Main application-side API: wait for the next message delivery and return it.

Returns:

the next message

Throws:

java.lang.InterruptedException - if an interrupt is received while waiting

com.rabbitmq.client.ShutdownSignalException - if the connection is shut down while waiting

nextMessage

public Message nextMessage(long timeout)
                    throws java.lang.InterruptedException,
                           com.rabbitmq.client.ShutdownSignalException

Main application-side API: wait for the next message delivery and return it.

Parameters:

timeout - timeout in millisecond

Returns:

the next message or null if timed out

Throws:

java.lang.InterruptedException - if an interrupt is received while waiting

com.rabbitmq.client.ShutdownSignalException - if the connection is shut down while waiting

checkMissingQueues

private void checkMissingQueues()

start

public void start()
           throws AmqpException

Throws:

AmqpException

consumeFromQueue

private void consumeFromQueue(java.lang.String queue)
                       throws java.io.IOException

Throws:

java.io.IOException

attemptPassiveDeclarations

private void attemptPassiveDeclarations()

stop

public void stop()

toString

public java.lang.String toString()

Overrides:

toString in class java.lang.Object

rollbackOnExceptionIfNecessary

public void rollbackOnExceptionIfNecessary(java.lang.Throwable ex)
                                    throws java.lang.Exception

Perform a rollback, handling rollback exceptions properly.

Parameters:

ex - the thrown application exception or error

Throws:

java.lang.Exception - in case of a rollback error

commitIfNecessary

public boolean commitIfNecessary(boolean locallyTransacted)
                          throws java.io.IOException

Perform a commit or message acknowledgement, as appropriate.

Parameters:

locallyTransacted - Whether the channel is locally transacted.

Returns:

true if at least one delivery tag exists.

Throws:

java.io.IOException - Any IOException.