Package org.apache.activemq.broker.region

Class Topic

java.lang.Object

org.apache.activemq.broker.region.BaseDestination

org.apache.activemq.broker.region.Topic

All Implemented Interfaces:

Destination, Message.MessageDestination, Service, Task

Direct Known Subclasses:

TempTopic

public class Topic
extends BaseDestination
implements Task

The Topic is a destination that sends a copy of a message to every active Subscription registered.

Field Summary

Fields

Modifier and Type	Field	Description
protected CopyOnWriteArrayList<Subscription>	consumers
protected static org.slf4j.Logger	LOG

Fields inherited from class org.apache.activemq.broker.region.BaseDestination

blockedProducerWarningInterval, broker, brokerService, cursorMemoryHighWaterMark, deadLetterStrategy, DEFAULT_INACTIVE_TIMEOUT_BEFORE_GC, destination, destinationStatistics, DUPLICATE_FROM_STORE_MSG_PREFIX, EXPIRE_MESSAGE_PERIOD, expireMessagesPeriod, lastBlockedProducerWarnTime, MAX_AUDIT_DEPTH, MAX_BROWSE_PAGE_SIZE, MAX_PAGE_SIZE, MAX_PRODUCERS_TO_AUDIT, memoryUsage, regionBroker, scheduler, started, store, storeUsageHighWaterMark, systemUsage

Fields inherited from interface org.apache.activemq.broker.region.Destination

DEFAULT_BLOCKED_PRODUCER_WARNING_INTERVAL, DEFAULT_DEAD_LETTER_STRATEGY

Constructor Summary

Constructors

Constructor	Description
Topic(BrokerService brokerService, ActiveMQDestination destination, TopicMessageStore store, DestinationStatistics parentStats, TaskRunnerFactory taskFactory)

Method Summary

Modifier and Type	Method	Description
void	acknowledge(ConnectionContext context, Subscription sub, MessageAck ack, MessageReference node)
void	activate(ConnectionContext context, DurableTopicSubscription subscription)
void	addSubscription(ConnectionContext context, Subscription sub)
Message[]	browse()
void	clearPendingMessages(int pendingAdditionsCount)	force a reread of the store - after transaction recovery completion
void	deactivate(ConnectionContext context, DurableTopicSubscription sub, List<MessageReference> dispatched)
void	deleteSubscription(ConnectionContext context, SubscriptionKey key)
protected void	dispatch(ConnectionContext context, Message message)
void	doBrowse(List<Message> browseList, int max)
void	gc()
List<Subscription>	getConsumers()
DispatchPolicy	getDispatchPolicy()
Map<SubscriptionKey,DurableTopicSubscription>	getDurableTopicSubs()
protected org.slf4j.Logger	getLog()
SubscriptionRecoveryPolicy	getSubscriptionRecoveryPolicy()
void	initialize()	initialize the destination
protected boolean	isOptimizeStorage()
boolean	iterate()
Message	loadMessage(MessageId messageId)
boolean	lock(MessageReference node, LockOwner sub)
void	messageExpired(ConnectionContext context, Subscription subs, MessageReference reference)	Inform the Destination a message has expired
void	recoverRetroactiveMessages(ConnectionContext context, Subscription subscription)
void	removeSubscription(ConnectionContext context, Subscription sub, long lastDeliveredSequenceId)
void	send(ProducerBrokerExchange producerExchange, Message message)
void	setDispatchPolicy(DispatchPolicy dispatchPolicy)
void	setSubscriptionRecoveryPolicy(SubscriptionRecoveryPolicy recoveryPolicy)
void	start()
void	stop()
String	toString()
void	wakeup()	optionally called by a Subscriber - to inform the Destination its ready for more messages

Methods inherited from class org.apache.activemq.broker.region.BaseDestination

addProducer, canGC, convertToNonRangedAck, createConnectionContext, dispose, duplicateFromStore, fastProducer, getActiveMQDestination, getBlockedProducerWarningInterval, getCursorMemoryHighWaterMark, getDeadLetterStrategy, getDestinationSequenceId, getDestinationStatistics, getExpireMessagesPeriod, getInactiveTimeoutBeforeGC, getMaxAuditDepth, getMaxBrowsePageSize, getMaxExpirePageSize, getMaxPageSize, getMaxProducersToAudit, getMemoryUsage, getMessageStore, getMinimumMessageSize, getName, getOptimizeMessageStoreInFlightLimit, getSlowConsumerStrategy, getStoreUsageHighWaterMark, getSystemUsage, getTempUsage, hasRegularConsumers, isActive, isAdvisoryForConsumed, isAdvisoryForDelivery, isAdvisoryForDiscardingMessages, isAdvisoryForFastProducers, isAdvisoryForSlowConsumers, isAdvisoryWhenFull, isAlwaysRetroactive, isDisposed, isDLQ, isDoOptimzeMessageStorage, isEnableAudit, isFlowControlLogRequired, isFull, isGcIfInactive, isGcWithNetworkConsumers, isIncludeBodyForAdvisory, isLazyDispatch, isPersistJMSRedelivered, isPrioritizedMessages, isProducerFlowControl, isReduceMemoryFootprint, isSendAdvisoryIfNoConsumers, isSendDuplicateFromStoreToDLQ, isUseCache, markForGC, messageConsumed, messageDelivered, messageDiscarded, onMessageWithNoConsumers, processDispatchNotification, removeProducer, setAdvisoryForConsumed, setAdvisoryForDelivery, setAdvisoryForDiscardingMessages, setAdvisoryForFastProducers, setAdvisoryForSlowConsumers, setAdvisoryWhenFull, setAlwaysRetroactive, setBlockedProducerWarningInterval, setCursorMemoryHighWaterMark, setDeadLetterStrategy, setDoOptimzeMessageStorage, setEnableAudit, setExpireMessagesPeriod, setGcIfInactive, setGcWithNetworkConsumers, setInactiveTimeoutBeforeGC, setIncludeBodyForAdvisory, setLazyDispatch, setMaxAuditDepth, setMaxBrowsePageSize, setMaxExpirePageSize, setMaxPageSize, setMaxProducersToAudit, setMemoryUsage, setMinimumMessageSize, setOptimizeMessageStoreInFlightLimit, setPersistJMSRedelivered, setPrioritizedMessages, setProducerFlowControl, setReduceMemoryFootprint, setSendAdvisoryIfNoConsumers, setSendDuplicateFromStoreToDLQ, setSlowConsumerStrategy, setStoreUsageHighWaterMark, setUseCache, slowConsumer, waitForSpace, waitForSpace

Methods inherited from class java.lang.Object

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

Field Detail

LOG

protected static final org.slf4j.Logger LOG

consumers

protected final CopyOnWriteArrayList<Subscription> consumers

Constructor Detail

Topic

public Topic(BrokerService brokerService,
             ActiveMQDestination destination,
             TopicMessageStore store,
             DestinationStatistics parentStats,
             TaskRunnerFactory taskFactory)
      throws Exception

Throws:

Exception

Method Detail

initialize

public void initialize()
                throws Exception

Description copied from class: BaseDestination

initialize the destination

Overrides:

initialize in class BaseDestination

Throws:

Exception

getConsumers

public List<Subscription> getConsumers()

Specified by:

getConsumers in interface Destination

Specified by:

getConsumers in class BaseDestination

lock

public boolean lock(MessageReference node,
                    LockOwner sub)

addSubscription

public void addSubscription(ConnectionContext context,
                            Subscription sub)
                     throws Exception

Specified by:

addSubscription in interface Destination

Overrides:

addSubscription in class BaseDestination

Throws:

Exception

removeSubscription

public void removeSubscription(ConnectionContext context,
                               Subscription sub,
                               long lastDeliveredSequenceId)
                        throws Exception

Specified by:

removeSubscription in interface Destination

Overrides:

removeSubscription in class BaseDestination

Throws:

Exception

deleteSubscription

public void deleteSubscription(ConnectionContext context,
                               SubscriptionKey key)
                        throws Exception

Throws:

Exception

activate

public void activate(ConnectionContext context,
                     DurableTopicSubscription subscription)
              throws Exception

Throws:

Exception

deactivate

public void deactivate(ConnectionContext context,
                       DurableTopicSubscription sub,
                       List<MessageReference> dispatched)
                throws Exception

Throws:

Exception

recoverRetroactiveMessages

public void recoverRetroactiveMessages(ConnectionContext context,
                                       Subscription subscription)
                                throws Exception

Throws:

Exception

send

public void send(ProducerBrokerExchange producerExchange,
                 Message message)
          throws Exception

Specified by:

send in interface Destination

Throws:

Exception

toString

public String toString()

Overrides:

toString in class Object

acknowledge

public void acknowledge(ConnectionContext context,
                        Subscription sub,
                        MessageAck ack,
                        MessageReference node)
                 throws IOException

Specified by:

acknowledge in interface Destination

Throws:

IOException

gc

public void gc()

Specified by:

gc in interface Destination

loadMessage

public Message loadMessage(MessageId messageId)
                    throws IOException

Throws:

IOException

start

public void start()
           throws Exception

Specified by:

start in interface Service

Throws:

Exception

stop

public void stop()
          throws Exception

Specified by:

stop in interface Service

Throws:

Exception

browse

public Message[] browse()

Specified by:

browse in interface Destination

doBrowse

public void doBrowse(List<Message> browseList,
                     int max)

iterate

public boolean iterate()

Specified by:

iterate in interface Task

getDispatchPolicy

public DispatchPolicy getDispatchPolicy()

setDispatchPolicy

public void setDispatchPolicy(DispatchPolicy dispatchPolicy)

getSubscriptionRecoveryPolicy

public SubscriptionRecoveryPolicy getSubscriptionRecoveryPolicy()

setSubscriptionRecoveryPolicy

public void setSubscriptionRecoveryPolicy(SubscriptionRecoveryPolicy recoveryPolicy)

wakeup

public final void wakeup()

Description copied from interface: Destination

optionally called by a Subscriber - to inform the Destination its ready for more messages

Specified by:

wakeup in interface Destination

dispatch

protected void dispatch(ConnectionContext context,
                        Message message)
                 throws Exception

Throws:

Exception

messageExpired

public void messageExpired(ConnectionContext context,
                           Subscription subs,
                           MessageReference reference)

Description copied from interface: Destination

Inform the Destination a message has expired

Specified by:

messageExpired in interface Destination

getLog

protected org.slf4j.Logger getLog()

Specified by:

getLog in class BaseDestination

isOptimizeStorage

protected boolean isOptimizeStorage()

clearPendingMessages

public void clearPendingMessages(int pendingAdditionsCount)

force a reread of the store - after transaction recovery completion

Specified by:

clearPendingMessages in interface Destination

Parameters:

pendingAdditionsCount -

getDurableTopicSubs

public Map<SubscriptionKey,DurableTopicSubscription> getDurableTopicSubs()