org.hornetq.core.client.impl

Class ClientConsumerImpl

java.lang.Object

org.hornetq.core.client.impl.ClientConsumerImpl

All Implemented Interfaces:

ClientConsumer, ClientConsumerInternal

public class ClientConsumerImpl
extends Object
implements ClientConsumerInternal

Version:

$Revision: 3603 $ $Id: ClientConsumerImpl.java 3603 2008-01-21 18:49:20Z timfox $

Author:

Tim Fox, Ovidiu Feodorov, Jeff Mesnil, Clebert Suconic, Andy Taylor

Field Summary

Fields

Modifier and Type	Field and Description
static long	CLOSE_TIMEOUT_MILLISECONDS
static SimpleString	FORCED_DELIVERY_MESSAGE
static int	NUM_PRIORITIES

Constructor Summary

Constructors

Constructor and Description
ClientConsumerImpl(ClientSessionInternal session, long id, SimpleString queueName, SimpleString filterString, boolean browseOnly, int clientWindowSize, int ackBatchSize, TokenBucketLimiter rateLimiter, Executor executor, Executor flowControlExecutor, Channel channel, SessionQueueQueryResponseMessage queueInfo, ClassLoader contextClassLoader)

Method Summary

Methods

Modifier and Type	Method and Description
void	acknowledge(ClientMessage message)
void	cleanUp()
void	clear(boolean waitForOnMessage)
void	clearAtFailover()
void	close() Closes the consumer.
void	flowControl(int messageBytes, boolean discountSlowConsumer) LargeMessageBuffer will call flowcontrol here, while other handleMessage will also be calling flowControl.
void	flushAcks()
int	getBufferSize()
int	getClientWindowSize()
SimpleString	getFilterString()
long	getID()
Exception	getLastException() Returns the last exception thrown by a call to this consumer's MessageHandler.
MessageHandler	getMessageHandler() Returns the MessageHandler associated to this consumer.
SessionQueueQueryResponseMessage	getQueueInfo()
SimpleString	getQueueName()
ClientSessionInternal	getSession()
void	handleLargeMessage(SessionReceiveLargeMessage packet)
void	handleLargeMessageContinuation(SessionReceiveContinuationMessage chunk)
void	handleMessage(ClientMessageInternal message)
boolean	isBrowseOnly()
boolean	isClosed() Returns whether the consumer is closed or not.
ClientMessage	receive() Receives a message from a queue.
ClientMessage	receive(long timeout) Receives a message from a queue.
ClientMessage	receiveImmediate() Receives a message from a queue.
void	setMessageHandler(MessageHandler theHandler) Sets the MessageHandler for this consumer to consume messages asynchronously.
void	start()
void	stop()
void	stop(boolean waitForOnMessage)

Methods inherited from class java.lang.Object

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

Field Detail

CLOSE_TIMEOUT_MILLISECONDS

public static final long CLOSE_TIMEOUT_MILLISECONDS

See Also:

Constant Field Values

NUM_PRIORITIES

public static final int NUM_PRIORITIES

See Also:

Constant Field Values

FORCED_DELIVERY_MESSAGE

public static final SimpleString FORCED_DELIVERY_MESSAGE

Constructor Detail

ClientConsumerImpl

public ClientConsumerImpl(ClientSessionInternal session,
                  long id,
                  SimpleString queueName,
                  SimpleString filterString,
                  boolean browseOnly,
                  int clientWindowSize,
                  int ackBatchSize,
                  TokenBucketLimiter rateLimiter,
                  Executor executor,
                  Executor flowControlExecutor,
                  Channel channel,
                  SessionQueueQueryResponseMessage queueInfo,
                  ClassLoader contextClassLoader)

Method Detail

receive

public ClientMessage receive(long timeout)
                      throws HornetQException

Description copied from interface: ClientConsumer

Receives a message from a queue. This call will block until a message is received or the given timeout expires Calling this method on a closed consumer will throw a HornetQException.

Specified by:

receive in interface ClientConsumer

Parameters:

timeout - time (in milliseconds) to wait to receive a message

Returns:

a message or null if the time out expired

Throws:

HornetQException - if an exception occurs while waiting to receive a message

receive

public ClientMessage receive()
                      throws HornetQException

Description copied from interface: ClientConsumer

Receives a message from a queue. This call will block indefinitely until a message is received. Calling this method on a closed consumer will throw a HornetQException.

Specified by:

receive in interface ClientConsumer

Returns:

a ClientMessage

Throws:

HornetQException - if an exception occurs while waiting to receive a message

receiveImmediate

public ClientMessage receiveImmediate()
                               throws HornetQException

Description copied from interface: ClientConsumer

Receives a message from a queue. This call will force a network trip to HornetQ server to ensure that there are no messages in the queue which can be delivered to this consumer. This call will never wait indefinitely for a message, it will return null if no messages are available for this consumer. Note however that there is a performance cost as an additional network trip to the server may required to check the queue status. Calling this method on a closed consumer will throw a HornetQException.

Specified by:

receiveImmediate in interface ClientConsumer

Returns:

a message or null if there are no messages in the queue for this consumer

Throws:

HornetQException - if an exception occurs while waiting to receive a message

getMessageHandler

public MessageHandler getMessageHandler()
                                 throws HornetQException

Description copied from interface: ClientConsumer

Returns the MessageHandler associated to this consumer. Calling this method on a closed consumer will throw a HornetQException.

Specified by:

getMessageHandler in interface ClientConsumer

Returns:

the MessageHandler associated to this consumer or null

Throws:

HornetQException - if an exception occurs while getting the MessageHandler

setMessageHandler

public void setMessageHandler(MessageHandler theHandler)
                       throws HornetQException

Description copied from interface: ClientConsumer

Sets the MessageHandler for this consumer to consume messages asynchronously. Calling this method on a closed consumer will throw a HornetQException.

Specified by:

setMessageHandler in interface ClientConsumer

Parameters:

theHandler - a MessageHandler

Throws:

HornetQException - if an exception occurs while setting the MessageHandler

close

public void close()
           throws HornetQException

Description copied from interface: ClientConsumer

Closes the consumer. Once this consumer is closed, it can not receive messages, whether synchronously or asynchronously.

Specified by:

close in interface ClientConsumer

Throws:

HornetQException

cleanUp

public void cleanUp()

Specified by:

cleanUp in interface ClientConsumerInternal

isClosed

public boolean isClosed()

Description copied from interface: ClientConsumer

Returns whether the consumer is closed or not.

Specified by:

isClosed in interface ClientConsumer

Returns:

true if this consumer is closed, false else

stop

public void stop()
          throws HornetQException

Specified by:

stop in interface ClientConsumerInternal

Throws:

HornetQException

stop

public void stop(boolean waitForOnMessage)
          throws HornetQException

Specified by:

stop in interface ClientConsumerInternal

Throws:

HornetQException

clearAtFailover

public void clearAtFailover()

Specified by:

clearAtFailover in interface ClientConsumerInternal

start

public void start()

Specified by:

start in interface ClientConsumerInternal

getLastException

public Exception getLastException()

Description copied from interface: ClientConsumer

Returns the last exception thrown by a call to this consumer's MessageHandler.

Specified by:

getLastException in interface ClientConsumer

Returns:

the last exception thrown by a call to this consumer's MessageHandler or null

getSession

public ClientSessionInternal getSession()

Specified by:

getSession in interface ClientConsumerInternal

getQueueInfo

public SessionQueueQueryResponseMessage getQueueInfo()

Specified by:

getQueueInfo in interface ClientConsumerInternal

getID

public long getID()

Specified by:

getID in interface ClientConsumerInternal

getFilterString

public SimpleString getFilterString()

Specified by:

getFilterString in interface ClientConsumerInternal

getQueueName

public SimpleString getQueueName()

Specified by:

getQueueName in interface ClientConsumerInternal

isBrowseOnly

public boolean isBrowseOnly()

Specified by:

isBrowseOnly in interface ClientConsumerInternal

handleMessage

public void handleMessage(ClientMessageInternal message)
                   throws Exception

Specified by:

handleMessage in interface ClientConsumerInternal

Throws:

Exception

handleLargeMessage

public void handleLargeMessage(SessionReceiveLargeMessage packet)
                        throws Exception

Specified by:

handleLargeMessage in interface ClientConsumerInternal

Throws:

Exception

handleLargeMessageContinuation

public void handleLargeMessageContinuation(SessionReceiveContinuationMessage chunk)
                                    throws Exception

Specified by:

handleLargeMessageContinuation in interface ClientConsumerInternal

Throws:

Exception

clear

public void clear(boolean waitForOnMessage)
           throws HornetQException

Specified by:

clear in interface ClientConsumerInternal

Throws:

HornetQException

getClientWindowSize

public int getClientWindowSize()

Specified by:

getClientWindowSize in interface ClientConsumerInternal

getBufferSize

public int getBufferSize()

Specified by:

getBufferSize in interface ClientConsumerInternal

acknowledge

public void acknowledge(ClientMessage message)
                 throws HornetQException

Specified by:

acknowledge in interface ClientConsumerInternal

Throws:

HornetQException

flushAcks

public void flushAcks()
               throws HornetQException

Specified by:

flushAcks in interface ClientConsumerInternal

Throws:

HornetQException

flowControl

public void flowControl(int messageBytes,
               boolean discountSlowConsumer)
                 throws HornetQException

LargeMessageBuffer will call flowcontrol here, while other handleMessage will also be calling flowControl. So, this operation needs to be atomic.

Specified by:

flowControl in interface ClientConsumerInternal

Parameters:

discountSlowConsumer - When dealing with slowConsumers, we need to discount one credit that was pre-sent when the first receive was called. For largeMessage that is only done at the latest packet

Throws:

HornetQException