http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java index 5401142..70d3bbd 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java @@ -362,6 +362,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return manager; } + @Override public final synchronized void start() throws Exception { if (state != SERVER_STATE.STOPPED) { ActiveMQServerLogger.LOGGER.debug("Server already started!"); @@ -467,6 +468,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { nodeManager = createNodeManager(configuration.getJournalLocation(), true); } + @Override public Activation getActivation() { return activation; } @@ -515,10 +517,12 @@ public class ActiveMQServerImpl implements ActiveMQServer { }); } + @Override public final void stop() throws Exception { stop(false); } + @Override public void addActivationParam(String key, Object val) { activationParams.put(key, val); } @@ -528,6 +532,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return postOffice.isAddressBound(SimpleString.toSimpleString(address)); } + @Override public void threadDump() { StringWriter str = new StringWriter(); PrintWriter out = new PrintWriter(str); @@ -553,6 +558,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { ActiveMQServerLogger.LOGGER.threadDump(str.toString()); } + @Override public final void stop(boolean failoverOnServerShutdown) throws Exception { stop(failoverOnServerShutdown, false, false); } @@ -848,6 +854,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { // ActiveMQServer implementation // ----------------------------------------------------------- + @Override public String describe() { StringWriter str = new StringWriter(); PrintWriter out = new PrintWriter(str); @@ -857,6 +864,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return str.toString(); } + @Override public String destroyConnectionWithSessionMetadata(String metaKey, String parameterValue) throws Exception { StringBuffer operationsExecuted = new StringBuffer(); @@ -901,66 +909,82 @@ public class ActiveMQServerImpl implements ActiveMQServer { } + @Override public void setIdentity(String identity) { this.identity = identity; } + @Override public String getIdentity() { return identity; } + @Override public ScheduledExecutorService getScheduledPool() { return scheduledPool; } + @Override public Configuration getConfiguration() { return configuration; } + @Override public PagingManager getPagingManager() { return pagingManager; } + @Override public RemotingService getRemotingService() { return remotingService; } + @Override public StorageManager getStorageManager() { return storageManager; } + @Override public ActiveMQSecurityManager getSecurityManager() { return securityManager; } + @Override public ManagementService getManagementService() { return managementService; } + @Override public HierarchicalRepository<Set<Role>> getSecurityRepository() { return securityRepository; } + @Override public NodeManager getNodeManager() { return nodeManager; } + @Override public HierarchicalRepository<AddressSettings> getAddressSettingsRepository() { return addressSettingsRepository; } + @Override public ResourceManager getResourceManager() { return resourceManager; } + @Override public Version getVersion() { return version; } + @Override public boolean isStarted() { return state == SERVER_STATE.STARTED; } + @Override public ClusterManager getClusterManager() { return clusterManager; } @@ -1027,6 +1051,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return sessionCount; } + @Override public void checkQueueCreationLimit(String username) throws Exception { if (configuration.getResourceLimitSettings() != null && configuration.getResourceLimitSettings().containsKey(username)) { ResourceLimitSettings limits = configuration.getResourceLimitSettings().get(username); @@ -1082,10 +1107,12 @@ public class ActiveMQServerImpl implements ActiveMQServer { return securityStore; } + @Override public void removeSession(final String name) throws Exception { sessions.remove(name); } + @Override public ServerSession lookupSession(String key, String value) { // getSessions is called here in a try to minimize locking the Server while this check is being done Set<ServerSession> allSessions = getSessions(); @@ -1100,6 +1127,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return null; } + @Override public synchronized List<ServerSession> getSessions(final String connectionID) { Set<Entry<String, ServerSession>> sessionEntries = sessions.entrySet(); List<ServerSession> matchingSessions = new ArrayList<ServerSession>(); @@ -1112,6 +1140,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return matchingSessions; } + @Override public synchronized Set<ServerSession> getSessions() { return new HashSet<ServerSession>(sessions.values()); } @@ -1126,26 +1155,32 @@ public class ActiveMQServerImpl implements ActiveMQServer { return activationLatch.await(timeout, unit); } + @Override public ActiveMQServerControlImpl getActiveMQServerControl() { return messagingServerControl; } + @Override public int getConnectionCount() { return remotingService.getConnections().size(); } + @Override public PostOffice getPostOffice() { return postOffice; } + @Override public QueueFactory getQueueFactory() { return queueFactory; } + @Override public SimpleString getNodeID() { return nodeManager == null ? null : nodeManager.getNodeId(); } + @Override public Queue createQueue(final SimpleString address, final SimpleString queueName, final SimpleString filterString, @@ -1154,6 +1189,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return createQueue(address, queueName, filterString, null, durable, temporary, false, false, false); } + @Override public Queue createQueue(final SimpleString address, final SimpleString queueName, final SimpleString filterString, @@ -1163,6 +1199,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return createQueue(address, queueName, filterString, user, durable, temporary, false, false, false); } + @Override public Queue createQueue(final SimpleString address, final SimpleString queueName, final SimpleString filterString, @@ -1185,6 +1222,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { * @param durable * @throws Exception */ + @Override public void createSharedQueue(final SimpleString address, final SimpleString name, final SimpleString filterString, @@ -1207,6 +1245,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { } + @Override public Queue locateQueue(SimpleString queueName) { Binding binding = postOffice.getBinding(queueName); @@ -1223,6 +1262,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return (Queue) binding.getBindable(); } + @Override public Queue deployQueue(final SimpleString address, final SimpleString queueName, final SimpleString filterString, @@ -1233,6 +1273,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { return createQueue(address, queueName, filterString, null, durable, temporary, true, false, false); } + @Override public void destroyQueue(final SimpleString queueName) throws Exception { // The session is passed as an argument to verify if the user has authorization to delete the queue // in some cases (such as temporary queues) this should happen regardless of the authorization @@ -1240,16 +1281,19 @@ public class ActiveMQServerImpl implements ActiveMQServer { destroyQueue(queueName, null, true); } + @Override public void destroyQueue(final SimpleString queueName, final SecurityAuth session) throws Exception { destroyQueue(queueName, session, true); } + @Override public void destroyQueue(final SimpleString queueName, final SecurityAuth session, final boolean checkConsumerCount) throws Exception { destroyQueue(queueName, session, checkConsumerCount, false); } + @Override public void destroyQueue(final SimpleString queueName, final SecurityAuth session, final boolean checkConsumerCount, @@ -1283,18 +1327,22 @@ public class ActiveMQServerImpl implements ActiveMQServer { queue.deleteQueue(removeConsumers); } + @Override public void registerActivateCallback(final ActivateCallback callback) { activateCallbacks.add(callback); } + @Override public void unregisterActivateCallback(final ActivateCallback callback) { activateCallbacks.remove(callback); } + @Override public ExecutorFactory getExecutorFactory() { return executorFactory; } + @Override public void setGroupingHandler(final GroupingHandler groupingHandler) { if (this.groupingHandler != null && managementService != null) { // Removing old groupNotification @@ -1307,18 +1355,22 @@ public class ActiveMQServerImpl implements ActiveMQServer { } + @Override public GroupingHandler getGroupingHandler() { return groupingHandler; } + @Override public ReplicationManager getReplicationManager() { return activation.getReplicationManager(); } + @Override public ConnectorsService getConnectorsService() { return connectorsService; } + @Override public void deployDivert(DivertConfiguration config) throws Exception { if (config.getName() == null) { ActiveMQServerLogger.LOGGER.divertWithNoName(); @@ -1361,6 +1413,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { managementService.registerDivert(divert, config); } + @Override public void destroyDivert(SimpleString name) throws Exception { Binding binding = postOffice.getBinding(name); if (binding == null) { @@ -1373,18 +1426,21 @@ public class ActiveMQServerImpl implements ActiveMQServer { postOffice.removeBinding(name, null); } + @Override public void deployBridge(BridgeConfiguration config) throws Exception { if (clusterManager != null) { clusterManager.deployBridge(config); } } + @Override public void destroyBridge(String name) throws Exception { if (clusterManager != null) { clusterManager.destroyBridge(name); } } + @Override public ServerSession getSessionByID(String sessionName) { return sessions.get(sessionName); } @@ -1499,6 +1555,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { } } + @Override public ServiceRegistry getServiceRegistry() { return serviceRegistry; } @@ -1607,6 +1664,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { if (dumpInfoInterval > 0) { scheduledPool.scheduleWithFixedDelay(new Runnable() { + @Override public void run() { ActiveMQServerLogger.LOGGER.dumpServerInfo(dumper.dump()); } @@ -1890,6 +1948,7 @@ public class ActiveMQServerImpl implements ActiveMQServer { boolean failedAlready = false; + @Override public synchronized void onIOException(Exception cause, String message, SequentialFile file) { if (!failedAlready) { failedAlready = true; @@ -1901,10 +1960,12 @@ public class ActiveMQServerImpl implements ActiveMQServer { } } + @Override public void addProtocolManagerFactory(ProtocolManagerFactory factory) { protocolManagerFactories.add(factory); } + @Override public void removeProtocolManagerFactory(ProtocolManagerFactory factory) { protocolManagerFactories.remove(factory); }
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AutoCreatedQueueManagerImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AutoCreatedQueueManagerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AutoCreatedQueueManagerImpl.java index f8fa60f..b6b5508 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AutoCreatedQueueManagerImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/AutoCreatedQueueManagerImpl.java @@ -30,6 +30,7 @@ public class AutoCreatedQueueManagerImpl implements AutoCreatedQueueManager { private final ActiveMQServer server; private final Runnable runnable = new Runnable() { + @Override public void run() { try { Queue queue = server.locateQueue(queueName); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ConnectorsService.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ConnectorsService.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ConnectorsService.java index ea4768d..789b9b6 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ConnectorsService.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ConnectorsService.java @@ -68,6 +68,7 @@ public final class ConnectorsService implements ActiveMQComponent { this.serviceRegistry = serviceRegistry; } + @Override public void start() throws Exception { Collection<Pair<ConnectorServiceFactory, ConnectorServiceConfiguration>> connectorServiceFactories = serviceRegistry.getConnectorServices(configuration.getConnectorServiceConfigurations()); @@ -104,6 +105,7 @@ public final class ConnectorsService implements ActiveMQComponent { connectors.add(connectorService); } + @Override public void stop() throws Exception { if (!isStarted) { return; @@ -120,6 +122,7 @@ public final class ConnectorsService implements ActiveMQComponent { isStarted = false; } + @Override public boolean isStarted() { return isStarted; } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java index ee6d0fe..c2f8b90 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/DivertImpl.java @@ -72,6 +72,7 @@ public class DivertImpl implements Divert { this.storageManager = storageManager; } + @Override public void route(final ServerMessage message, final RoutingContext context) throws Exception { // We must make a copy of the message, otherwise things like returning credits to the page won't work // properly on ack, since the original address will be overwritten @@ -112,22 +113,27 @@ public class DivertImpl implements Divert { route(message, context); } + @Override public SimpleString getRoutingName() { return routingName; } + @Override public SimpleString getUniqueName() { return uniqueName; } + @Override public boolean isExclusive() { return exclusive; } + @Override public Filter getFilter() { return filter; } + @Override public Transformer getTransformer() { return transformer; } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java index 39becf0..c474a45 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LastValueQueue.java @@ -162,6 +162,7 @@ public class LastValueQueue extends QueueImpl { return ref; } + @Override public void handled() { ref.handled(); // We need to remove the entry from the map just before it gets delivered @@ -182,50 +183,62 @@ public class LastValueQueue extends QueueImpl { this.ref = ref; } + @Override public MessageReference copy(final Queue queue) { return ref.copy(queue); } + @Override public void decrementDeliveryCount() { ref.decrementDeliveryCount(); } + @Override public int getDeliveryCount() { return ref.getDeliveryCount(); } + @Override public ServerMessage getMessage() { return ref.getMessage(); } + @Override public Queue getQueue() { return ref.getQueue(); } + @Override public long getScheduledDeliveryTime() { return ref.getScheduledDeliveryTime(); } + @Override public void incrementDeliveryCount() { ref.incrementDeliveryCount(); } + @Override public void setDeliveryCount(final int deliveryCount) { ref.setDeliveryCount(deliveryCount); } + @Override public void setScheduledDeliveryTime(final long scheduledDeliveryTime) { ref.setScheduledDeliveryTime(scheduledDeliveryTime); } + @Override public void setPersistedCount(int count) { ref.setPersistedCount(count); } + @Override public int getPersistedCount() { return ref.getPersistedCount(); } + @Override public boolean isPaged() { return false; } @@ -241,6 +254,7 @@ public class LastValueQueue extends QueueImpl { /* (non-Javadoc) * @see org.apache.activemq.artemis.core.server.MessageReference#getMessageMemoryEstimate() */ + @Override public int getMessageMemoryEstimate() { return ref.getMessage().getMemoryEstimate(); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LiveOnlyActivation.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LiveOnlyActivation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LiveOnlyActivation.java index 59ffd6a..d8224dc 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LiveOnlyActivation.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/LiveOnlyActivation.java @@ -53,6 +53,7 @@ public class LiveOnlyActivation extends Activation { this.liveOnlyPolicy = liveOnlyPolicy; } + @Override public void run() { try { activeMQServer.initialisePart1(false); @@ -81,6 +82,7 @@ public class LiveOnlyActivation extends Activation { } } + @Override public void freezeConnections(RemotingService remotingService) { // connect to the scale-down target first so that when we freeze/disconnect the clients we can tell them where // we're sending the messages http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/MessageReferenceImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/MessageReferenceImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/MessageReferenceImpl.java index ca1c0a2..96413f7 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/MessageReferenceImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/MessageReferenceImpl.java @@ -89,6 +89,7 @@ public class MessageReferenceImpl implements MessageReference { /** * @return the persistedCount */ + @Override public int getPersistedCount() { return persistedCount; } @@ -96,10 +97,12 @@ public class MessageReferenceImpl implements MessageReference { /** * @param persistedCount the persistedCount to set */ + @Override public void setPersistedCount(int persistedCount) { this.persistedCount = persistedCount; } + @Override public MessageReference copy(final Queue queue) { return new MessageReferenceImpl(this, queue); } @@ -108,39 +111,48 @@ public class MessageReferenceImpl implements MessageReference { return MessageReferenceImpl.memoryOffset; } + @Override public int getDeliveryCount() { return deliveryCount.get(); } + @Override public void setDeliveryCount(final int deliveryCount) { this.deliveryCount.set(deliveryCount); this.persistedCount = this.deliveryCount.get(); } + @Override public void incrementDeliveryCount() { deliveryCount.incrementAndGet(); } + @Override public void decrementDeliveryCount() { deliveryCount.decrementAndGet(); } + @Override public long getScheduledDeliveryTime() { return scheduledDeliveryTime; } + @Override public void setScheduledDeliveryTime(final long scheduledDeliveryTime) { this.scheduledDeliveryTime = scheduledDeliveryTime; } + @Override public ServerMessage getMessage() { return message; } + @Override public Queue getQueue() { return queue; } + @Override public void handled() { queue.referenceHandled(); } @@ -155,10 +167,12 @@ public class MessageReferenceImpl implements MessageReference { return alreadyAcked; } + @Override public boolean isPaged() { return false; } + @Override public void acknowledge() throws Exception { queue.acknowledge(this); } @@ -173,6 +187,7 @@ public class MessageReferenceImpl implements MessageReference { return this.consumerID; } + @Override public int getMessageMemoryEstimate() { return message.getMemoryEstimate(); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java index 47793c1..cd6908e 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java @@ -158,6 +158,7 @@ public class PostOfficeJournalLoader implements JournalLoader { } } + @Override public void handleAddMessage(Map<Long, Map<Long, AddMessageRecord>> queueMap) throws Exception { for (Map.Entry<Long, Map<Long, AddMessageRecord>> entry : queueMap.entrySet()) { long queueID = entry.getKey(); @@ -206,6 +207,7 @@ public class PostOfficeJournalLoader implements JournalLoader { } } + @Override public void handleNoMessageReferences(Map<Long, ServerMessage> messages) { for (ServerMessage msg : messages.values()) { if (msg.getRefCount() == 0) { @@ -306,6 +308,7 @@ public class PostOfficeJournalLoader implements JournalLoader { * @param pendingNonTXPageCounter * @throws Exception */ + @Override public void recoverPendingPageCounters(List<PageCountPending> pendingNonTXPageCounter) throws Exception { // We need a structure of the following // Address -> PageID -> QueueID -> List<PageCountPending> http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java index d3089af..280bb13 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueFactoryImpl.java @@ -60,10 +60,12 @@ public class QueueFactoryImpl implements QueueFactory { this.executorFactory = executorFactory; } + @Override public void setPostOffice(final PostOffice postOffice) { this.postOffice = postOffice; } + @Override public Queue createQueue(final long persistenceID, final SimpleString address, final SimpleString name, http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java index a5b3622..bea9f6d 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java @@ -240,6 +240,7 @@ public class QueueImpl implements Queue { public List<SimpleString> getGroupsUsed() { final CountDownLatch flush = new CountDownLatch(1); executor.execute(new Runnable() { + @Override public void run() { flush.countDown(); } @@ -381,6 +382,7 @@ public class QueueImpl implements Queue { return name; } + @Override public SimpleString getUser() { return user; } @@ -389,6 +391,7 @@ public class QueueImpl implements Queue { return false; } + @Override public void route(final ServerMessage message, final RoutingContext context) throws Exception { context.addQueue(address, this); } @@ -399,48 +402,59 @@ public class QueueImpl implements Queue { } // Queue implementation ---------------------------------------------------------------------------------------- + @Override public synchronized void setConsumersRefCount(final ReferenceCounter referenceCounter) { if (refCountForConsumers == null) { this.refCountForConsumers = referenceCounter; } } + @Override public ReferenceCounter getConsumersRefCount() { return refCountForConsumers; } + @Override public boolean isDurable() { return durable; } + @Override public boolean isTemporary() { return temporary; } + @Override public boolean isAutoCreated() { return autoCreated; } + @Override public SimpleString getName() { return name; } + @Override public SimpleString getAddress() { return address; } + @Override public long getID() { return id; } + @Override public PageSubscription getPageSubscription() { return pageSubscription; } + @Override public Filter getFilter() { return filter; } + @Override public void unproposed(final SimpleString groupID) { if (groupID.toString().endsWith("." + this.getName())) { // this means this unproposed belongs to this routing, so we will @@ -451,6 +465,7 @@ public class QueueImpl implements Queue { final SimpleString groupIDToRemove = (SimpleString) groupID.subSequence(0, groupID.length() - getName().length() - 1); // using an executor so we don't want to hold anyone just because of this getExecutor().execute(new Runnable() { + @Override public void run() { synchronized (QueueImpl.this) { if (groups.remove(groupIDToRemove) != null) { @@ -466,6 +481,7 @@ public class QueueImpl implements Queue { } /* Called when a message is cancelled back into the queue */ + @Override public synchronized void addHead(final MessageReference ref) { flushDeliveriesInTransit(); if (scheduledDeliveryHandler.checkAndSchedule(ref, false)) { @@ -478,6 +494,7 @@ public class QueueImpl implements Queue { } /* Called when a message is cancelled back into the queue */ + @Override public synchronized void addHead(final List<MessageReference> refs) { flushDeliveriesInTransit(); for (MessageReference ref : refs) { @@ -489,6 +506,7 @@ public class QueueImpl implements Queue { deliverAsync(); } + @Override public synchronized void reload(final MessageReference ref) { queueMemorySize.addAndGet(ref.getMessageMemoryEstimate()); if (!scheduledDeliveryHandler.checkAndSchedule(ref, true)) { @@ -500,10 +518,12 @@ public class QueueImpl implements Queue { messagesAdded++; } + @Override public void addTail(final MessageReference ref) { addTail(ref, false); } + @Override public void addTail(final MessageReference ref, final boolean direct) { if (scheduledDeliveryHandler.checkAndSchedule(ref, true)) { synchronized (this) { @@ -571,6 +591,7 @@ public class QueueImpl implements Queue { } } + @Override public void forceDelivery() { if (pageSubscription != null && pageSubscription.isPaging()) { if (isTrace) { @@ -586,6 +607,7 @@ public class QueueImpl implements Queue { deliverAsync(); } + @Override public void deliverAsync() { if (scheduledRunners.get() < MAX_SCHEDULED_RUNNERS) { scheduledRunners.incrementAndGet(); @@ -602,12 +624,14 @@ public class QueueImpl implements Queue { } + @Override public void close() throws Exception { if (checkQueueSizeFuture != null) { checkQueueSizeFuture.cancel(false); } getExecutor().execute(new Runnable() { + @Override public void run() { try { cancelRedistributor(); @@ -624,6 +648,7 @@ public class QueueImpl implements Queue { } } + @Override public Executor getExecutor() { if (pageSubscription != null && pageSubscription.isPaging()) { // When in page mode, we don't want to have concurrent IO on the same PageStore @@ -641,6 +666,7 @@ public class QueueImpl implements Queue { flushExecutor(); } + @Override public boolean flushExecutor() { boolean ok = internalFlushExecutor(10000); @@ -664,6 +690,7 @@ public class QueueImpl implements Queue { return result; } + @Override public void addConsumer(final Consumer consumer) throws Exception { if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) { ActiveMQServerLogger.LOGGER.debug(this + " adding consumer " + consumer); @@ -687,6 +714,7 @@ public class QueueImpl implements Queue { } + @Override public void removeConsumer(final Consumer consumer) { synchronized (this) { consumersChanged = true; @@ -733,6 +761,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized void addRedistributor(final long delay) { if (redistributorFuture != null) { redistributorFuture.cancel(false); @@ -759,6 +788,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized void cancelRedistributor() throws Exception { if (redistributor != null) { redistributor.stop(); @@ -786,14 +816,17 @@ public class QueueImpl implements Queue { super.finalize(); } + @Override public synchronized int getConsumerCount() { return consumerSet.size(); } + @Override public synchronized Set<Consumer> getConsumers() { return new HashSet<Consumer>(consumerSet); } + @Override public boolean hasMatchingConsumer(final ServerMessage message) { for (ConsumerHolder holder : consumerList) { Consumer consumer = holder.consumer; @@ -816,14 +849,17 @@ public class QueueImpl implements Queue { return false; } + @Override public LinkedListIterator<MessageReference> iterator() { return new SynchronizedIterator(messageReferences.iterator()); } + @Override public TotalQueueIterator totalIterator() { return new TotalQueueIterator(); } + @Override public synchronized MessageReference removeReferenceWithID(final long id1) throws Exception { LinkedListIterator<MessageReference> iterator = iterator(); @@ -856,6 +892,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized MessageReference getReference(final long id1) { LinkedListIterator<MessageReference> iterator = iterator(); @@ -876,6 +913,7 @@ public class QueueImpl implements Queue { } } + @Override public long getMessageCount() { synchronized (this) { if (pageSubscription != null) { @@ -891,14 +929,17 @@ public class QueueImpl implements Queue { } } + @Override public synchronized int getScheduledCount() { return scheduledDeliveryHandler.getScheduledCount(); } + @Override public synchronized List<MessageReference> getScheduledMessages() { return scheduledDeliveryHandler.getScheduledReferences(); } + @Override public Map<String, List<MessageReference>> getDeliveringMessages() { List<ConsumerHolder> consumerListClone = cloneConsumersList(); @@ -915,10 +956,12 @@ public class QueueImpl implements Queue { return mapReturn; } + @Override public int getDeliveringCount() { return deliveringCount.get(); } + @Override public void acknowledge(final MessageReference ref) throws Exception { if (ref.isPaged()) { pageSubscription.ack((PagedReference) ref); @@ -939,6 +982,7 @@ public class QueueImpl implements Queue { } + @Override public void acknowledge(final Transaction tx, final MessageReference ref) throws Exception { if (ref.isPaged()) { pageSubscription.ackTx(tx, (PagedReference) ref); @@ -962,6 +1006,7 @@ public class QueueImpl implements Queue { messagesAcknowledged++; } + @Override public void reacknowledge(final Transaction tx, final MessageReference ref) throws Exception { ServerMessage message = ref.getMessage(); @@ -1001,14 +1046,17 @@ public class QueueImpl implements Queue { } } + @Override public void cancel(final Transaction tx, final MessageReference reference) { cancel(tx, reference, false); } + @Override public void cancel(final Transaction tx, final MessageReference reference, boolean ignoreRedeliveryCheck) { getRefsOperation(tx, ignoreRedeliveryCheck).addAck(reference); } + @Override public synchronized void cancel(final MessageReference reference, final long timeBase) throws Exception { if (checkRedelivery(reference, timeBase, false)) { if (!scheduledDeliveryHandler.checkAndSchedule(reference, false)) { @@ -1022,6 +1070,7 @@ public class QueueImpl implements Queue { } } + @Override public void expire(final MessageReference ref) throws Exception { if (expiryAddress != null) { if (isTrace) { @@ -1037,14 +1086,17 @@ public class QueueImpl implements Queue { } } + @Override public SimpleString getExpiryAddress() { return this.expiryAddress; } + @Override public void referenceHandled() { incDelivering(); } + @Override public void incrementMesssagesAdded() { messagesAdded++; } @@ -1061,6 +1113,7 @@ public class QueueImpl implements Queue { } } + @Override public long getMessagesAdded() { if (pageSubscription != null) { return messagesAdded + pageSubscription.getCounter().getValue() - pagedReferences.get(); @@ -1070,22 +1123,27 @@ public class QueueImpl implements Queue { } } + @Override public long getMessagesAcknowledged() { return messagesAcknowledged; } + @Override public int deleteAllReferences() throws Exception { return deleteAllReferences(DEFAULT_FLUSH_LIMIT); } + @Override public int deleteAllReferences(final int flushLimit) throws Exception { return deleteMatchingReferences(flushLimit, null); } + @Override public int deleteMatchingReferences(Filter filter) throws Exception { return deleteMatchingReferences(DEFAULT_FLUSH_LIMIT, filter); } + @Override public synchronized int deleteMatchingReferences(final int flushLimit, final Filter filter1) throws Exception { return iterQueue(flushLimit, filter1, new QueueIterateAction() { @Override @@ -1195,6 +1253,7 @@ public class QueueImpl implements Queue { } } + @Override public void destroyPaging() throws Exception { // it could be null on embedded or certain unit tests if (pageSubscription != null) { @@ -1203,6 +1262,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized boolean deleteReference(final long messageID) throws Exception { boolean deleted = false; @@ -1237,10 +1297,12 @@ public class QueueImpl implements Queue { } } + @Override public void deleteQueue() throws Exception { deleteQueue(false); } + @Override public void deleteQueue(boolean removeConsumers) throws Exception { synchronized (this) { this.queueDestroyed = true; @@ -1279,6 +1341,7 @@ public class QueueImpl implements Queue { } + @Override public synchronized boolean expireReference(final long messageID) throws Exception { if (expiryAddress != null && expiryAddress.equals(this.address)) { // check expire with itself would be silly (waste of time) @@ -1307,6 +1370,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized int expireReferences(final Filter filter) throws Exception { if (expiryAddress != null && expiryAddress.equals(this.address)) { // check expire with itself would be silly (waste of time) @@ -1342,6 +1406,7 @@ public class QueueImpl implements Queue { } } + @Override public void expireReferences() { if (expiryAddress != null && expiryAddress.equals(this.address)) { // check expire with itself would be silly (waste of time) @@ -1360,6 +1425,7 @@ public class QueueImpl implements Queue { public AtomicInteger scannerRunning = new AtomicInteger(0); + @Override public void run() { synchronized (QueueImpl.this) { if (queueDestroyed) { @@ -1406,6 +1472,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized boolean sendMessageToDeadLetterAddress(final long messageID) throws Exception { LinkedListIterator<MessageReference> iter = iterator(); @@ -1427,6 +1494,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized int sendMessagesToDeadLetterAddress(Filter filter) throws Exception { int count = 0; LinkedListIterator<MessageReference> iter = iterator(); @@ -1449,10 +1517,12 @@ public class QueueImpl implements Queue { } } + @Override public boolean moveReference(final long messageID, final SimpleString toAddress) throws Exception { return moveReference(messageID, toAddress, false); } + @Override public synchronized boolean moveReference(final long messageID, final SimpleString toAddress, final boolean rejectDuplicate) throws Exception { @@ -1482,10 +1552,12 @@ public class QueueImpl implements Queue { } } + @Override public int moveReferences(final Filter filter, final SimpleString toAddress) throws Exception { return moveReferences(DEFAULT_FLUSH_LIMIT, filter, toAddress, false); } + @Override public synchronized int moveReferences(final int flushLimit, final Filter filter, final SimpleString toAddress, @@ -1526,6 +1598,7 @@ public class QueueImpl implements Queue { }); } + @Override public int retryMessages(Filter filter) throws Exception { final HashMap<SimpleString, Long> queues = new HashMap<>(); @@ -1568,6 +1641,7 @@ public class QueueImpl implements Queue { } + @Override public synchronized boolean changeReferencePriority(final long messageID, final byte newPriority) throws Exception { LinkedListIterator<MessageReference> iter = iterator(); @@ -1591,6 +1665,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized int changeReferencesPriority(final Filter filter, final byte newPriority) throws Exception { LinkedListIterator<MessageReference> iter = iterator(); @@ -1613,6 +1688,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized void resetAllIterators() { for (ConsumerHolder holder : this.consumerList) { if (holder.iter != null) { @@ -1622,6 +1698,7 @@ public class QueueImpl implements Queue { } } + @Override public synchronized void pause() { try { this.flushDeliveriesInTransit(); @@ -1632,16 +1709,19 @@ public class QueueImpl implements Queue { paused = true; } + @Override public synchronized void resume() { paused = false; deliverAsync(); } + @Override public synchronized boolean isPaused() { return paused; } + @Override public boolean isDirectDeliver() { return directDeliver; } @@ -1649,6 +1729,7 @@ public class QueueImpl implements Queue { /** * @return the internalQueue */ + @Override public boolean isInternalQueue() { return internalQueue; } @@ -1656,6 +1737,7 @@ public class QueueImpl implements Queue { /** * @param internalQueue the internalQueue to set */ + @Override public void setInternalQueue(boolean internalQueue) { this.internalQueue = internalQueue; } @@ -2038,6 +2120,7 @@ public class QueueImpl implements Queue { } } + @Override public boolean checkRedelivery(final MessageReference reference, final long timeBase, final boolean ignoreRedeliveryDelay) throws Exception { @@ -2177,10 +2260,12 @@ public class QueueImpl implements Queue { storageManager.afterCompleteOperations(new IOCallback() { + @Override public void onError(final int errorCode, final String errorMessage) { ActiveMQServerLogger.LOGGER.ioErrorRedistributing(errorCode, errorMessage); } + @Override public void done() { deliverAsync(); } @@ -2471,6 +2556,7 @@ public class QueueImpl implements Queue { return consumerListClone; } + @Override public void postAcknowledge(final MessageReference ref) { QueueImpl queue = (QueueImpl) ref.getQueue(); @@ -2539,14 +2625,17 @@ public class QueueImpl implements Queue { return delay; } + @Override public synchronized void resetMessagesAdded() { messagesAdded = 0; } + @Override public synchronized void resetMessagesAcknowledged() { messagesAcknowledged = 0; } + @Override public float getRate() { float timeSlice = ((System.currentTimeMillis() - queueRateCheckTime.getAndSet(System.currentTimeMillis())) / 1000.0f); if (timeSlice == 0) { @@ -2579,6 +2668,7 @@ public class QueueImpl implements Queue { this.executor1 = executor; } + @Override public void run() { synchronized (QueueImpl.this) { internalAddRedistributor(executor1); @@ -2595,6 +2685,7 @@ public class QueueImpl implements Queue { */ private final class DeliverRunner implements Runnable { + @Override public void run() { try { // during the transition between paging and nonpaging, we could have this using a different executor @@ -2623,6 +2714,7 @@ public class QueueImpl implements Queue { this.scheduleExpiry = scheduleExpiry; } + @Override public void run() { try { depage(scheduleExpiry); @@ -2650,30 +2742,35 @@ public class QueueImpl implements Queue { this.iter = iter; } + @Override public void close() { synchronized (QueueImpl.this) { iter.close(); } } + @Override public void repeat() { synchronized (QueueImpl.this) { iter.repeat(); } } + @Override public boolean hasNext() { synchronized (QueueImpl.this) { return iter.hasNext(); } } + @Override public MessageReference next() { synchronized (QueueImpl.this) { return iter.next(); } } + @Override public void remove() { synchronized (QueueImpl.this) { iter.remove(); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java index cbd3236..1a5bcbd 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java @@ -40,6 +40,7 @@ public final class RoutingContextImpl implements RoutingContext { this.transaction = transaction; } + @Override public void clear() { transaction = null; @@ -48,6 +49,7 @@ public final class RoutingContextImpl implements RoutingContext { queueCount = 0; } + @Override public void addQueue(final SimpleString address, final Queue queue) { RouteContextList listing = getContextListing(address); @@ -75,6 +77,7 @@ public final class RoutingContextImpl implements RoutingContext { return listing == null ? false : listing.isAlreadyAcked(queue); } + @Override public RouteContextList getContextListing(SimpleString address) { RouteContextList listing = map.get(address); if (listing == null) { @@ -84,26 +87,32 @@ public final class RoutingContextImpl implements RoutingContext { return listing; } + @Override public Transaction getTransaction() { return transaction; } + @Override public void setTransaction(final Transaction tx) { transaction = tx; } + @Override public List<Queue> getNonDurableQueues(SimpleString address) { return getContextListing(address).getNonDurableQueues(); } + @Override public List<Queue> getDurableQueues(SimpleString address) { return getContextListing(address).getDurableQueues(); } + @Override public int getQueueCount() { return queueCount; } + @Override public Map<SimpleString, RouteContextList> getContexListing() { return this.map; } @@ -116,10 +125,12 @@ public final class RoutingContextImpl implements RoutingContext { private final List<Queue> ackedQueues = new ArrayList<>(); + @Override public int getNumberOfDurableQueues() { return durableQueue.size(); } + @Override public int getNumberOfNonDurableQueues() { return nonDurableQueue.size(); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerImpl.java index 97ce13e..9fc5bab 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerImpl.java @@ -53,6 +53,7 @@ public class ScheduledDeliveryHandlerImpl implements ScheduledDeliveryHandler { this.scheduledExecutor = scheduledExecutor; } + @Override public boolean checkAndSchedule(final MessageReference ref, final boolean tail) { long deliveryTime = ref.getScheduledDeliveryTime(); @@ -76,12 +77,14 @@ public class ScheduledDeliveryHandlerImpl implements ScheduledDeliveryHandler { } } + @Override public int getScheduledCount() { synchronized (scheduledReferences) { return scheduledReferences.size(); } } + @Override public List<MessageReference> getScheduledReferences() { List<MessageReference> refs = new LinkedList<MessageReference>(); @@ -93,6 +96,7 @@ public class ScheduledDeliveryHandlerImpl implements ScheduledDeliveryHandler { return refs; } + @Override public List<MessageReference> cancel(final Filter filter) { List<MessageReference> refs = new ArrayList<MessageReference>(); @@ -110,6 +114,7 @@ public class ScheduledDeliveryHandlerImpl implements ScheduledDeliveryHandler { return refs; } + @Override public MessageReference removeReferenceWithID(final long id) { synchronized (scheduledReferences) { Iterator<RefScheduled> iter = scheduledReferences.iterator(); @@ -163,6 +168,7 @@ public class ScheduledDeliveryHandlerImpl implements ScheduledDeliveryHandler { this.deliveryTime = deliveryTime; } + @Override public void run() { HashMap<Queue, LinkedList<MessageReference>> refs = new HashMap<Queue, LinkedList<MessageReference>>(); @@ -260,6 +266,7 @@ public class ScheduledDeliveryHandlerImpl implements ScheduledDeliveryHandler { static class MessageReferenceComparator implements Comparator<RefScheduled> { + @Override public int compare(RefScheduled ref1, RefScheduled ref2) { long diff = ref1.getRef().getScheduledDeliveryTime() - ref2.getRef().getScheduledDeliveryTime(); http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java index ae30549..e74b60e 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java @@ -100,6 +100,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { private volatile LargeMessageDeliverer largeMessageDeliverer = null; + @Override public String debug() { return toString() + "::Delivering " + this.deliveringRefs.size(); } @@ -222,34 +223,42 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { // ServerConsumer implementation // ---------------------------------------------------------------------- + @Override public Object getProtocolContext() { return protocolContext; } + @Override public void setProtocolContext(Object protocolContext) { this.protocolContext = protocolContext; } + @Override public long getID() { return id; } + @Override public boolean isBrowseOnly() { return browseOnly; } + @Override public long getCreationTime() { return creationTime; } + @Override public String getConnectionID() { return this.session.getConnectionID().toString(); } + @Override public String getSessionID() { return this.session.getName(); } + @Override public List<MessageReference> getDeliveringMessages() { List<MessageReference> refs = new LinkedList<MessageReference>(); synchronized (lock) { @@ -263,6 +272,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { return refs; } + @Override public HandleStatus handle(final MessageReference ref) throws Exception { if (callback != null && !callback.hasCredits(this) || availableCredits != null && availableCredits.get() <= 0) { if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) { @@ -349,6 +359,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { } } + @Override public void proceedDeliver(MessageReference reference) throws Exception { try { ServerMessage message = reference.getMessage(); @@ -373,6 +384,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { } } + @Override public Filter getFilter() { return filter; } @@ -459,11 +471,13 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { * When the consumer receives such a "forced delivery" message, it discards it and knows that * there are no other messages to be delivered. */ + @Override public synchronized void forceDelivery(final long sequence) { promptDelivery(); // JBPAPP-6030 - Using the executor to avoid distributed dead locks messageQueue.getExecutor().execute(new Runnable() { + @Override public void run() { try { // We execute this on the same executor to make sure the force delivery message is written after @@ -473,6 +487,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { if (transferring) { // Case it's transferring (reattach), we will retry later messageQueue.getExecutor().execute(new Runnable() { + @Override public void run() { forceDelivery(sequence); } @@ -496,6 +511,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { } + @Override public LinkedList<MessageReference> cancelRefs(final boolean failed, final boolean lastConsumedAsDelivered, final Transaction tx) throws Exception { @@ -545,6 +561,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { return refs; } + @Override public void setStarted(final boolean started) { synchronized (lock) { // This is to make sure that the delivery process has finished any pending delivery @@ -564,6 +581,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { } } + @Override public void setTransferring(final boolean transferring) { synchronized (lock) { // This is to make sure that the delivery process has finished any pending delivery @@ -599,6 +617,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { } } + @Override public void receiveCredits(final int credits) { if (credits == -1) { if (ActiveMQServerLogger.LOGGER.isDebugEnabled()) { @@ -636,10 +655,12 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { } } + @Override public Queue getQueue() { return messageQueue; } + @Override public void acknowledge(Transaction tx, final long messageID) throws Exception { if (browseOnly) { return; @@ -717,6 +738,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { } } + @Override public void individualAcknowledge(final Transaction tx, final long messageID) throws Exception { if (browseOnly) { return; @@ -737,6 +759,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { acks++; } + @Override public void individualCancel(final long messageID, boolean failed) throws Exception { if (browseOnly) { return; @@ -755,6 +778,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { ref.getQueue().cancel(ref, System.currentTimeMillis()); } + @Override public MessageReference removeReferenceByID(final long messageID) throws Exception { if (browseOnly) { return null; @@ -787,6 +811,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { } } + @Override public void readyForWriting(final boolean ready) { if (ready) { writeReady.set(true); @@ -813,6 +838,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { return "ServerConsumerImpl [id=" + id + ", filter=" + filter + ", binding=" + binding + "]"; } + @Override public String toManagementString() { return "ServerConsumer [id=" + getConnectionID() + ":" + getSessionID() + ":" + id + ", filter=" + filter + ", binding=" + binding.toManagementString() + "]"; } @@ -833,6 +859,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { // Private -------------------------------------------------------------------------------------- + @Override public void promptDelivery() { // largeMessageDeliverer is always set inside a lock // if we don't acquire a lock, we will have NPE eventually @@ -880,6 +907,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { // ------------------------------------------------------------------------ private final Runnable resumeLargeMessageRunnable = new Runnable() { + @Override public void run() { synchronized (lock) { try { @@ -1068,6 +1096,7 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener { iterator.close(); } + @Override public synchronized void run() { // if the reference was busy during the previous iteration, handle it now if (current != null) { http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java index 8485b51..645d326 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerMessageImpl.java @@ -85,25 +85,30 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { super(other, properties); } + @Override public boolean isServerMessage() { return true; } + @Override public ServerMessageImpl setMessageID(final long id) { messageID = id; return this; } + @Override public MessageReference createReference(final Queue queue) { MessageReference ref = new MessageReferenceImpl(this, queue); return ref; } + @Override public boolean hasInternalProperties() { return properties.hasInternalProperties(); } + @Override public int incrementRefCount() throws Exception { int count = refCount.incrementAndGet(); @@ -119,6 +124,7 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { return count; } + @Override public int decrementRefCount() throws Exception { int count = refCount.decrementAndGet(); @@ -139,24 +145,29 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { return count; } + @Override public int incrementDurableRefCount() { return durableRefCount.incrementAndGet(); } + @Override public int decrementDurableRefCount() { return durableRefCount.decrementAndGet(); } + @Override public int getRefCount() { return refCount.get(); } + @Override public boolean isLargeMessage() { return false; } private volatile int memoryEstimate = -1; + @Override public int getMemoryEstimate() { if (memoryEstimate == -1) { memoryEstimate = ServerMessageImpl.memoryOffset + buffer.capacity() + properties.getMemoryOffset(); @@ -165,6 +176,7 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { return memoryEstimate; } + @Override public ServerMessage copy(final long newID) { ServerMessage m = new ServerMessageImpl(this); @@ -173,9 +185,11 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { return m; } + @Override public void finishCopy() throws Exception { } + @Override public ServerMessage copy() { // This is a simple copy, used only to avoid changing original properties return new ServerMessageImpl(this); @@ -187,6 +201,7 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { return makeCopyForExpiryOrDLA(newID, originalReference, expiry, true); } + @Override public ServerMessage makeCopyForExpiryOrDLA(final long newID, MessageReference originalReference, final boolean expiry, @@ -246,6 +261,7 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { bufferValid = false; } + @Override public void setPagingStore(final PagingStore pagingStore) { this.pagingStore = pagingStore; @@ -254,15 +270,18 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { address = pagingStore.getAddress(); } + @Override public synchronized void forceAddress(final SimpleString address) { this.address = address; bufferValid = false; } + @Override public PagingStore getPagingStore() { return pagingStore; } + @Override public boolean storeIsPaging() { if (pagingStore != null) { return pagingStore.isPaging(); @@ -289,12 +308,14 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { } + @Override public InputStream getBodyInputStream() { return null; } // Encoding stuff + @Override public void encodeMessageIDToBuffer() { // We first set the message id - this needs to be set on the buffer since this buffer will be re-used @@ -318,6 +339,7 @@ public class ServerMessageImpl extends MessageImpl implements ServerMessage { } } + @Override public Object getDuplicateProperty() { return getObjectProperty(Message.HDR_DUPLICATE_DETECTION_ID); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java index 16ec608..77345e3 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java @@ -279,35 +279,43 @@ public class ServerSessionImpl implements ServerSession, FailureListener { /** * @return the sessionContext */ + @Override public OperationContext getSessionContext() { return context; } + @Override public String getUsername() { return username; } + @Override public String getPassword() { return password; } + @Override public int getMinLargeMessageSize() { return minLargeMessageSize; } + @Override public String getName() { return name; } + @Override public Object getConnectionID() { return remotingConnection.getID(); } + @Override public Set<ServerConsumer> getServerConsumers() { Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values()); return Collections.unmodifiableSet(consumersClone); } + @Override public boolean removeConsumer(final long consumerID) throws Exception { return consumers.remove(consumerID) != null; } @@ -370,10 +378,12 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public QueueCreator getQueueCreator() { return queueCreator; } + @Override public ServerConsumer createConsumer(final long consumerID, final SimpleString queueName, final SimpleString filterString, @@ -457,6 +467,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return new ServerConsumerImpl(consumerID, this, (QueueBinding) binding, filter, started, browseOnly, storageManager, callback, preAcknowledge, strictUpdateDeliveryCount, managementService, supportLargeMessage, credits); } + @Override public Queue createQueue(final SimpleString address, final SimpleString name, final SimpleString filterString, @@ -519,6 +530,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { server.createSharedQueue(address, name, filterString, SimpleString.toSimpleString(getUsername()), durable); } + @Override public RemotingConnection getRemotingConnection() { return remotingConnection; } @@ -553,6 +565,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void connectionFailed(ActiveMQException exception, boolean failedOver) { run(); } @@ -562,6 +575,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { connectionFailed(me, failedOver); } + @Override public void connectionClosed() { run(); } @@ -573,6 +587,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } + @Override public void deleteQueue(final SimpleString queueToDelete) throws Exception { Binding binding = postOffice.getBinding(queueToDelete); @@ -591,6 +606,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public QueueQueryResult executeQueueQuery(final SimpleString name) throws Exception { boolean autoCreateJmsQueues = name.toString().startsWith(ResourceNames.JMS_QUEUE) && server.getAddressSettingsRepository().getMatch(name.toString()).isAutoCreateJmsQueues(); @@ -625,6 +641,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return response; } + @Override public BindingQueryResult executeBindingQuery(final SimpleString address) throws Exception { boolean autoCreateJmsQueues = address.toString().startsWith(ResourceNames.JMS_QUEUE) && server.getAddressSettingsRepository().getMatch(address.toString()).isAutoCreateJmsQueues(); @@ -650,6 +667,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return new BindingQueryResult(!names.isEmpty(), names, autoCreateJmsQueues); } + @Override public void forceConsumerDelivery(final long consumerID, final long sequence) throws Exception { ServerConsumer consumer = consumers.get(consumerID); @@ -668,6 +686,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void acknowledge(final long consumerID, final long messageID) throws Exception { ServerConsumer consumer = findConsumer(consumerID); @@ -708,6 +727,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return consumer; } + @Override public void individualAcknowledge(final long consumerID, final long messageID) throws Exception { ServerConsumer consumer = findConsumer(consumerID); @@ -725,6 +745,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } + @Override public void individualCancel(final long consumerID, final long messageID, boolean failed) throws Exception { ServerConsumer consumer = consumers.get(consumerID); @@ -734,6 +755,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } + @Override public void expire(final long consumerID, final long messageID) throws Exception { MessageReference ref = consumers.get(consumerID).removeReferenceByID(messageID); @@ -742,6 +764,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void commit() throws Exception { if (isTrace) { ActiveMQServerLogger.LOGGER.trace("Calling commit"); @@ -761,6 +784,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void rollback(final boolean considerLastMessageAsDelivered) throws Exception { rollback(false, considerLastMessageAsDelivered); } @@ -803,6 +827,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return transactionFactory.newTransaction(xid, storageManager, timeoutSeconds); } + @Override public synchronized void xaCommit(final Xid xid, final boolean onePhase) throws Exception { if (tx != null && tx.getXid().equals(xid)) { @@ -848,6 +873,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaEnd(final Xid xid) throws Exception { if (tx != null && tx.getXid().equals(xid)) { if (tx.getState() == Transaction.State.SUSPENDED) { @@ -891,6 +917,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaForget(final Xid xid) throws Exception { long id = resourceManager.removeHeuristicCompletion(xid); @@ -909,6 +936,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaJoin(final Xid xid) throws Exception { Transaction theTx = resourceManager.getTransaction(xid); @@ -927,6 +955,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaResume(final Xid xid) throws Exception { if (tx != null) { final String msg = "Cannot resume, session is currently doing work in a transaction " + tx.getXid(); @@ -954,6 +983,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaRollback(final Xid xid) throws Exception { if (tx != null && tx.getXid().equals(xid)) { final String msg = "Cannot roll back, session is currently doing work in a transaction " + tx.getXid(); @@ -1011,6 +1041,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaStart(final Xid xid) throws Exception { if (tx != null) { ActiveMQServerLogger.LOGGER.xidReplacedOnXStart(tx.getXid().toString(), xid.toString()); @@ -1044,6 +1075,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaFailed(final Xid xid) throws Exception { Transaction theTX = resourceManager.getTransaction(xid); @@ -1066,6 +1098,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaSuspend() throws Exception { if (isTrace) { @@ -1091,6 +1124,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public synchronized void xaPrepare(final Xid xid) throws Exception { if (tx != null && tx.getXid().equals(xid)) { final String msg = "Cannot commit, session is currently doing work in a transaction " + tx.getXid(); @@ -1123,6 +1157,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public List<Xid> xaGetInDoubtXids() { List<Xid> xids = new ArrayList<Xid>(); @@ -1133,10 +1168,12 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return xids; } + @Override public int xaGetTimeout() { return resourceManager.getTimeoutSeconds(); } + @Override public void xaSetTimeout(final int timeout) { timeoutSeconds = timeout; if (tx != null) { @@ -1144,14 +1181,17 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void start() { setStarted(true); } + @Override public void stop() { setStarted(false); } + @Override public void waitContextCompletion() { try { if (!context.waitCompletion(10000)) { @@ -1163,13 +1203,16 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void close(final boolean failed) { if (closed) return; context.executeOnCompletion(new IOCallback() { + @Override public void onError(int errorCode, String errorMessage) { } + @Override public void done() { try { doClose(failed); @@ -1181,6 +1224,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { }); } + @Override public void closeConsumer(final long consumerID) throws Exception { final ServerConsumer consumer = consumers.get(consumerID); @@ -1192,6 +1236,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void receiveConsumerCredits(final long consumerID, final int credits) throws Exception { ServerConsumer consumer = consumers.get(consumerID); @@ -1212,6 +1257,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return tx; } + @Override public void sendLarge(final MessageInternal message) throws Exception { // need to create the LargeMessage before continue long id = storageManager.generateID(); @@ -1229,6 +1275,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { currentLargeMessage = largeMsg; } + @Override public void send(final ServerMessage message, final boolean direct) throws Exception { //large message may come from StompSession directly, in which //case the id header already generated. @@ -1276,6 +1323,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void sendContinuations(final int packetSize, final long messageBodySize, final byte[] body, @@ -1302,10 +1350,12 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void requestProducerCredits(final SimpleString address, final int credits) throws Exception { PagingStore store = server.getPagingManager().getPageStore(address); if (!store.checkMemory(new Runnable() { + @Override public void run() { callback.sendProducerCreditsMessage(credits, address); } @@ -1314,6 +1364,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void setTransferring(final boolean transferring) { Set<ServerConsumer> consumersClone = new HashSet<ServerConsumer>(consumers.values()); @@ -1322,6 +1373,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public void addMetaData(String key, String data) { if (metaData == null) { metaData = new HashMap<String, String>(); @@ -1329,6 +1381,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { metaData.put(key, data); } + @Override public boolean addUniqueMetaData(String key, String data) { ServerSession sessionWithMetaData = server.lookupSession(key, data); if (sessionWithMetaData != null && sessionWithMetaData != this) { @@ -1341,6 +1394,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public String getMetaData(String key) { String data = null; if (metaData != null) { @@ -1354,6 +1408,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return data; } + @Override public String[] getTargetAddresses() { Map<SimpleString, Pair<UUID, AtomicLong>> copy = cloneTargetAddresses(); Iterator<SimpleString> iter = copy.keySet().iterator(); @@ -1367,6 +1422,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { return addresses; } + @Override public String getLastSentMessageID(String address) { Pair<UUID, AtomicLong> value = targetAddressInfos.get(SimpleString.toSimpleString(address)); if (value != null) { @@ -1377,6 +1433,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener { } } + @Override public long getCreationTime() { return this.creationTime; } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServiceRegistryImpl.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServiceRegistryImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServiceRegistryImpl.java index d00e15e..46dbd58 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServiceRegistryImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServiceRegistryImpl.java @@ -104,6 +104,7 @@ public class ServiceRegistryImpl implements ServiceRegistry { for (final ConnectorServiceConfiguration config : configs) { if (connectorServices.get(config.getConnectorName()) == null) { ConnectorServiceFactory factory = AccessController.doPrivileged(new PrivilegedAction<ConnectorServiceFactory>() { + @Override public ConnectorServiceFactory run() { return (ConnectorServiceFactory) ClassloadingUtil.newInstanceFromClassLoader(config.getFactoryClassName()); } @@ -184,6 +185,7 @@ public class ServiceRegistryImpl implements ServiceRegistry { if (factory == null && className != null) { factory = AccessController.doPrivileged(new PrivilegedAction<AcceptorFactory>() { + @Override public AcceptorFactory run() { return (AcceptorFactory) ClassloadingUtil.newInstanceFromClassLoader(className); } @@ -206,6 +208,7 @@ public class ServiceRegistryImpl implements ServiceRegistry { if (className != null) { try { transformer = AccessController.doPrivileged(new PrivilegedAction<Transformer>() { + @Override public Transformer run() { return (Transformer) ClassloadingUtil.newInstanceFromClassLoader(className); } @@ -222,6 +225,7 @@ public class ServiceRegistryImpl implements ServiceRegistry { if (classNames != null) { for (final String className : classNames) { BaseInterceptor interceptor = AccessController.doPrivileged(new PrivilegedAction<BaseInterceptor>() { + @Override public BaseInterceptor run() { return (BaseInterceptor) ClassloadingUtil.newInstanceFromClassLoader(className); } http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java index fa8be06..12e9298 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingBackupActivation.java @@ -91,6 +91,7 @@ public final class SharedNothingBackupActivation extends Activation { replicationEndpoint = new ReplicationEndpoint(activeMQServer, shutdownOnCriticalIO, attemptFailBack, this); } + @Override public void run() { try { synchronized (activeMQServer) { @@ -269,6 +270,7 @@ public final class SharedNothingBackupActivation extends Activation { } } + @Override public void close(final boolean permanently, boolean restarting) throws Exception { synchronized (this) { if (backupQuorum != null) http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/25ae4724/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java ---------------------------------------------------------------------- diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java index 52d6260..8803e22 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/SharedNothingLiveActivation.java @@ -82,6 +82,7 @@ public class SharedNothingLiveActivation extends LiveActivation { } } + @Override public void run() { try { if (replicatedPolicy.isCheckForLiveServer() && isNodeIdUsed()) { @@ -155,6 +156,7 @@ public class SharedNothingLiveActivation extends LiveActivation { replicationManager = new ReplicationManager(rc, activeMQServer.getExecutorFactory()); replicationManager.start(); Thread t = new Thread(new Runnable() { + @Override public void run() { try { activeMQServer.getStorageManager().startReplication(replicationManager, activeMQServer.getPagingManager(), activeMQServer.getNodeID().toString(), isFailBackRequest && replicatedPolicy.isAllowAutoFailBack(), replicatedPolicy.getInitialReplicationSyncTimeout()); @@ -229,6 +231,7 @@ public class SharedNothingLiveActivation extends LiveActivation { @Override public void connectionClosed() { activeMQServer.getThreadPool().execute(new Runnable() { + @Override public void run() { synchronized (replicationLock) { if (replicationManager != null) { @@ -300,6 +303,7 @@ public class SharedNothingLiveActivation extends LiveActivation { } } + @Override public void close(boolean permanently, boolean restarting) throws Exception { replicationManager = null; @@ -334,6 +338,7 @@ public class SharedNothingLiveActivation extends LiveActivation { } } + @Override public ReplicationManager getReplicationManager() { synchronized (replicationLock) { return replicationManager;
