| /** |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.activemq.state; |
| |
| import java.io.IOException; |
| import java.util.Iterator; |
| import java.util.LinkedHashMap; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Vector; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| |
| import javax.jms.TransactionRolledBackException; |
| import javax.transaction.xa.XAResource; |
| |
| import org.apache.activemq.command.Command; |
| import org.apache.activemq.command.ConnectionId; |
| import org.apache.activemq.command.ConnectionInfo; |
| import org.apache.activemq.command.ConsumerControl; |
| import org.apache.activemq.command.ConsumerId; |
| import org.apache.activemq.command.ConsumerInfo; |
| import org.apache.activemq.command.DestinationInfo; |
| import org.apache.activemq.command.ExceptionResponse; |
| import org.apache.activemq.command.IntegerResponse; |
| import org.apache.activemq.command.Message; |
| import org.apache.activemq.command.MessagePull; |
| import org.apache.activemq.command.ProducerId; |
| import org.apache.activemq.command.ProducerInfo; |
| import org.apache.activemq.command.Response; |
| import org.apache.activemq.command.SessionId; |
| import org.apache.activemq.command.SessionInfo; |
| import org.apache.activemq.command.TransactionInfo; |
| import org.apache.activemq.transport.Transport; |
| import org.apache.activemq.util.IOExceptionSupport; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| /** |
| * Tracks the state of a connection so a newly established transport can be |
| * re-initialized to the state that was tracked. |
| * |
| * |
| */ |
| public class ConnectionStateTracker extends CommandVisitorAdapter { |
| private static final Logger LOG = LoggerFactory.getLogger(ConnectionStateTracker.class); |
| |
| private static final Tracked TRACKED_RESPONSE_MARKER = new Tracked(null); |
| private static final int MESSAGE_PULL_SIZE = 400; |
| protected final ConcurrentMap<ConnectionId, ConnectionState> connectionStates = new ConcurrentHashMap<>(); |
| |
| private boolean trackTransactions; |
| private boolean restoreSessions = true; |
| private boolean restoreConsumers = true; |
| private boolean restoreProducers = true; |
| private boolean restoreTransaction = true; |
| private boolean trackMessages = true; |
| private boolean trackTransactionProducers = true; |
| private int maxCacheSize = 128 * 1024; |
| private long currentCacheSize; // use long to prevent overflow for folks who set high max. |
| |
| private final Map<Object,Command> messageCache = new LinkedHashMap<Object,Command>(){ |
| @Override |
| protected boolean removeEldestEntry(Map.Entry<Object,Command> eldest) { |
| boolean result = currentCacheSize > maxCacheSize; |
| if (result) { |
| if (eldest.getValue() instanceof Message) { |
| currentCacheSize -= ((Message)eldest.getValue()).getSize(); |
| } else if (eldest.getValue() instanceof MessagePull) { |
| currentCacheSize -= MESSAGE_PULL_SIZE; |
| } |
| if (LOG.isTraceEnabled()) { |
| LOG.trace("removing tracked message: " + eldest.getKey()); |
| } |
| } |
| return result; |
| } |
| }; |
| |
| private class RemoveTransactionAction implements ResponseHandler { |
| private final TransactionInfo info; |
| |
| public RemoveTransactionAction(TransactionInfo info) { |
| this.info = info; |
| } |
| |
| @Override |
| public void onResponse(Command response) { |
| ConnectionId connectionId = info.getConnectionId(); |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| cs.removeTransactionState(info.getTransactionId()); |
| } |
| } |
| } |
| |
| private final class ExceptionResponseCheckAction implements ResponseHandler { |
| private final Command tracked; |
| |
| public ExceptionResponseCheckAction(Command tracked) { |
| this.tracked = tracked; |
| } |
| |
| @Override |
| public void onResponse(Command response) { |
| if (ExceptionResponse.DATA_STRUCTURE_TYPE == response.getDataStructureType()) { |
| if (tracked.getDataStructureType() == ConsumerInfo.DATA_STRUCTURE_TYPE) { |
| processRemoveConsumer(((ConsumerInfo) tracked).getConsumerId(), 0l); |
| } else if (tracked.getDataStructureType() == ProducerInfo.DATA_STRUCTURE_TYPE) { |
| processRemoveProducer(((ProducerInfo) tracked).getProducerId()); |
| } |
| } |
| } |
| } |
| |
| private class PrepareReadonlyTransactionAction extends RemoveTransactionAction { |
| public PrepareReadonlyTransactionAction(TransactionInfo info) { |
| super(info); |
| } |
| |
| @Override |
| public void onResponse(Command command) { |
| if (command instanceof IntegerResponse) { |
| IntegerResponse response = (IntegerResponse) command; |
| if (XAResource.XA_RDONLY == response.getResult()) { |
| // all done, no commit or rollback from TM |
| super.onResponse(command); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Entry point for all tracked commands in the tracker. Commands should be tracked before |
| * there is an attempt to send them on the wire. Upon a successful send of a command it is |
| * necessary to call the trackBack method to complete the tracking of the given command. |
| * |
| * @param command |
| * The command that is to be tracked by this tracker. |
| * |
| * @return null if the command is not state tracked. |
| * |
| * @throws IOException if an error occurs during setup of the tracking operation. |
| */ |
| public Tracked track(Command command) throws IOException { |
| try { |
| return (Tracked)command.visit(this); |
| } catch (IOException e) { |
| throw e; |
| } catch (Throwable e) { |
| throw IOExceptionSupport.create(e); |
| } |
| } |
| |
| /** |
| * Completes the two phase tracking operation for a command that is sent on the wire. Once |
| * the command is sent successfully to complete the tracking operation or otherwise update |
| * the state of the tracker. |
| * |
| * @param command |
| * The command that was previously provided to the track method. |
| */ |
| public void trackBack(Command command) { |
| if (command != null) { |
| if (trackMessages && command.isMessage()) { |
| Message message = (Message) command; |
| if (message.getTransactionId()==null) { |
| currentCacheSize = currentCacheSize + message.getSize(); |
| } |
| } else if (command instanceof MessagePull) { |
| // We only track one MessagePull per consumer so only add to cache size |
| // when the command has been marked as tracked. |
| if (((MessagePull)command).isTracked()) { |
| // just needs to be a rough estimate of size, ~4 identifiers |
| currentCacheSize += MESSAGE_PULL_SIZE; |
| } |
| } |
| } |
| } |
| |
| public void restore(Transport transport) throws IOException { |
| // Restore the connections. |
| for (Iterator<ConnectionState> iter = connectionStates.values().iterator(); iter.hasNext();) { |
| ConnectionState connectionState = iter.next(); |
| connectionState.getInfo().setFailoverReconnect(true); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("conn: " + connectionState.getInfo().getConnectionId()); |
| } |
| transport.oneway(connectionState.getInfo()); |
| restoreTempDestinations(transport, connectionState); |
| |
| if (restoreSessions) { |
| restoreSessions(transport, connectionState); |
| } |
| |
| if (restoreTransaction) { |
| restoreTransactions(transport, connectionState); |
| } |
| } |
| |
| // now flush messages and MessagePull commands. |
| for (Command msg : messageCache.values()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("command: " + (msg.isMessage() ? ((Message) msg).getMessageId() : msg)); |
| } |
| transport.oneway(msg); |
| } |
| } |
| |
| private void restoreTransactions(Transport transport, ConnectionState connectionState) throws IOException { |
| Vector<TransactionInfo> toRollback = new Vector<>(); |
| for (TransactionState transactionState : connectionState.getTransactionStates()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("tx: " + transactionState.getId()); |
| } |
| |
| // rollback any completed transactions - no way to know if commit got there |
| // or if reply went missing |
| // |
| if (!transactionState.getCommands().isEmpty()) { |
| Command lastCommand = transactionState.getCommands().get(transactionState.getCommands().size() - 1); |
| if (lastCommand instanceof TransactionInfo) { |
| TransactionInfo transactionInfo = (TransactionInfo) lastCommand; |
| if (transactionInfo.getType() == TransactionInfo.COMMIT_ONE_PHASE) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("rolling back potentially completed tx: " + transactionState.getId()); |
| } |
| toRollback.add(transactionInfo); |
| continue; |
| } |
| } |
| } |
| |
| // replay short lived producers that may have been involved in the transaction |
| for (ProducerState producerState : transactionState.getProducerStates().values()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("tx replay producer :" + producerState.getInfo()); |
| } |
| transport.oneway(producerState.getInfo()); |
| } |
| |
| for (Command command : transactionState.getCommands()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("tx replay: " + command); |
| } |
| transport.oneway(command); |
| } |
| |
| for (ProducerState producerState : transactionState.getProducerStates().values()) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("tx remove replayed producer :" + producerState.getInfo()); |
| } |
| transport.oneway(producerState.getInfo().createRemoveCommand()); |
| } |
| } |
| |
| for (TransactionInfo command: toRollback) { |
| // respond to the outstanding commit |
| ExceptionResponse response = new ExceptionResponse(); |
| response.setException(new TransactionRolledBackException("Transaction completion in doubt due to failover. Forcing rollback of " + command.getTransactionId())); |
| response.setCorrelationId(command.getCommandId()); |
| transport.getTransportListener().onCommand(response); |
| } |
| } |
| |
| /** |
| * @param transport |
| * @param connectionState |
| * @throws IOException |
| */ |
| protected void restoreSessions(Transport transport, ConnectionState connectionState) throws IOException { |
| // Restore the connection's sessions |
| for (Iterator iter2 = connectionState.getSessionStates().iterator(); iter2.hasNext();) { |
| SessionState sessionState = (SessionState)iter2.next(); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("session: " + sessionState.getInfo().getSessionId()); |
| } |
| transport.oneway(sessionState.getInfo()); |
| |
| if (restoreProducers) { |
| restoreProducers(transport, sessionState); |
| } |
| |
| if (restoreConsumers) { |
| restoreConsumers(transport, sessionState); |
| } |
| } |
| } |
| |
| /** |
| * @param transport |
| * @param sessionState |
| * @throws IOException |
| */ |
| protected void restoreConsumers(Transport transport, SessionState sessionState) throws IOException { |
| // Restore the session's consumers but possibly in pull only (prefetch 0 state) till recovery complete |
| final ConnectionState connectionState = connectionStates.get(sessionState.getInfo().getSessionId().getParentId()); |
| final boolean connectionInterruptionProcessingComplete = connectionState.isConnectionInterruptProcessingComplete(); |
| for (ConsumerState consumerState : sessionState.getConsumerStates()) { |
| ConsumerInfo infoToSend = consumerState.getInfo(); |
| if (!connectionInterruptionProcessingComplete && infoToSend.getPrefetchSize() > 0) { |
| infoToSend = consumerState.getInfo().copy(); |
| connectionState.getRecoveringPullConsumers().put(infoToSend.getConsumerId(), consumerState.getInfo()); |
| infoToSend.setPrefetchSize(0); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("restore consumer: " + infoToSend.getConsumerId() + " in pull mode pending recovery, overriding prefetch: " + consumerState.getInfo().getPrefetchSize()); |
| } |
| } |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("consumer: " + infoToSend.getConsumerId()); |
| } |
| transport.oneway(infoToSend); |
| } |
| } |
| |
| /** |
| * @param transport |
| * @param sessionState |
| * @throws IOException |
| */ |
| protected void restoreProducers(Transport transport, SessionState sessionState) throws IOException { |
| // Restore the session's producers |
| for (Iterator iter3 = sessionState.getProducerStates().iterator(); iter3.hasNext();) { |
| ProducerState producerState = (ProducerState)iter3.next(); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("producer: " + producerState.getInfo().getProducerId()); |
| } |
| transport.oneway(producerState.getInfo()); |
| } |
| } |
| |
| /** |
| * @param transport |
| * @param connectionState |
| * @throws IOException |
| */ |
| protected void restoreTempDestinations(Transport transport, ConnectionState connectionState) |
| throws IOException { |
| // Restore the connection's temp destinations. |
| for (Iterator iter2 = connectionState.getTempDestinations().iterator(); iter2.hasNext();) { |
| DestinationInfo info = (DestinationInfo)iter2.next(); |
| transport.oneway(info); |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("tempDest: " + info.getDestination()); |
| } |
| } |
| } |
| |
| @Override |
| public Response processAddDestination(DestinationInfo info) { |
| if (info != null) { |
| ConnectionState cs = connectionStates.get(info.getConnectionId()); |
| if (cs != null && info.getDestination().isTemporary()) { |
| cs.addTempDestination(info); |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processRemoveDestination(DestinationInfo info) { |
| if (info != null) { |
| ConnectionState cs = connectionStates.get(info.getConnectionId()); |
| if (cs != null && info.getDestination().isTemporary()) { |
| cs.removeTempDestination(info.getDestination()); |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processAddProducer(ProducerInfo info) { |
| if (info != null && info.getProducerId() != null) { |
| SessionId sessionId = info.getProducerId().getParentId(); |
| if (sessionId != null) { |
| ConnectionId connectionId = sessionId.getParentId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| SessionState ss = cs.getSessionState(sessionId); |
| if (ss != null) { |
| ss.addProducer(info); |
| if (info.isResponseRequired()) { |
| return new Tracked(new ExceptionResponseCheckAction(info)); |
| } |
| } |
| } |
| } |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processRemoveProducer(ProducerId id) { |
| if (id != null) { |
| SessionId sessionId = id.getParentId(); |
| if (sessionId != null) { |
| ConnectionId connectionId = sessionId.getParentId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| SessionState ss = cs.getSessionState(sessionId); |
| if (ss != null) { |
| ss.removeProducer(id); |
| } |
| } |
| } |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processAddConsumer(ConsumerInfo info) { |
| if (info != null) { |
| SessionId sessionId = info.getConsumerId().getParentId(); |
| if (sessionId != null) { |
| ConnectionId connectionId = sessionId.getParentId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| SessionState ss = cs.getSessionState(sessionId); |
| if (ss != null) { |
| ss.addConsumer(info); |
| if (info.isResponseRequired()) { |
| return new Tracked(new ExceptionResponseCheckAction(info)); |
| } |
| } |
| } |
| } |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processRemoveConsumer(ConsumerId id, long lastDeliveredSequenceId) { |
| if (id != null) { |
| SessionId sessionId = id.getParentId(); |
| if (sessionId != null) { |
| ConnectionId connectionId = sessionId.getParentId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| SessionState ss = cs.getSessionState(sessionId); |
| if (ss != null) { |
| ss.removeConsumer(id); |
| } |
| cs.getRecoveringPullConsumers().remove(id); |
| } |
| } |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processAddSession(SessionInfo info) { |
| if (info != null) { |
| ConnectionId connectionId = info.getSessionId().getParentId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| cs.addSession(info); |
| } |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processRemoveSession(SessionId id, long lastDeliveredSequenceId) { |
| if (id != null) { |
| ConnectionId connectionId = id.getParentId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| cs.removeSession(id); |
| } |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processAddConnection(ConnectionInfo info) { |
| if (info != null) { |
| connectionStates.put(info.getConnectionId(), new ConnectionState(info)); |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processRemoveConnection(ConnectionId id, long lastDeliveredSequenceId) throws Exception { |
| if (id != null) { |
| connectionStates.remove(id); |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| |
| @Override |
| public Response processMessage(Message send) throws Exception { |
| if (send != null) { |
| if (trackTransactions && send.getTransactionId() != null) { |
| ProducerId producerId = send.getProducerId(); |
| ConnectionId connectionId = producerId.getParentId().getParentId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| TransactionState transactionState = cs.getTransactionState(send.getTransactionId()); |
| if (transactionState != null) { |
| transactionState.addCommand(send); |
| |
| if (trackTransactionProducers) { |
| // for jmstemplate, track the producer in case it is closed before commit |
| // and needs to be replayed |
| SessionState ss = cs.getSessionState(producerId.getParentId()); |
| ProducerState producerState = ss.getProducerState(producerId); |
| producerState.setTransactionState(transactionState); |
| } |
| } |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| }else if (trackMessages) { |
| messageCache.put(send.getMessageId(), send); |
| } |
| } |
| return null; |
| } |
| |
| @Override |
| public Response processBeginTransaction(TransactionInfo info) { |
| if (trackTransactions && info != null && info.getTransactionId() != null) { |
| ConnectionId connectionId = info.getConnectionId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| cs.addTransactionState(info.getTransactionId()); |
| TransactionState state = cs.getTransactionState(info.getTransactionId()); |
| state.addCommand(info); |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| return null; |
| } |
| |
| @Override |
| public Response processPrepareTransaction(TransactionInfo info) throws Exception { |
| if (trackTransactions && info != null && info.getTransactionId() != null) { |
| ConnectionId connectionId = info.getConnectionId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| TransactionState transactionState = cs.getTransactionState(info.getTransactionId()); |
| if (transactionState != null) { |
| transactionState.addCommand(info); |
| return new Tracked(new PrepareReadonlyTransactionAction(info)); |
| } |
| } |
| } |
| } |
| return null; |
| } |
| |
| @Override |
| public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception { |
| if (trackTransactions && info != null && info.getTransactionId() != null) { |
| ConnectionId connectionId = info.getConnectionId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| TransactionState transactionState = cs.getTransactionState(info.getTransactionId()); |
| if (transactionState != null) { |
| transactionState.addCommand(info); |
| return new Tracked(new RemoveTransactionAction(info)); |
| } |
| } |
| } |
| } |
| return null; |
| } |
| |
| @Override |
| public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception { |
| if (trackTransactions && info != null && info.getTransactionId() != null) { |
| ConnectionId connectionId = info.getConnectionId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| TransactionState transactionState = cs.getTransactionState(info.getTransactionId()); |
| if (transactionState != null) { |
| transactionState.addCommand(info); |
| return new Tracked(new RemoveTransactionAction(info)); |
| } |
| } |
| } |
| } |
| return null; |
| } |
| |
| @Override |
| public Response processRollbackTransaction(TransactionInfo info) throws Exception { |
| if (trackTransactions && info != null && info.getTransactionId() != null) { |
| ConnectionId connectionId = info.getConnectionId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| TransactionState transactionState = cs.getTransactionState(info.getTransactionId()); |
| if (transactionState != null) { |
| transactionState.addCommand(info); |
| return new Tracked(new RemoveTransactionAction(info)); |
| } |
| } |
| } |
| } |
| return null; |
| } |
| |
| @Override |
| public Response processEndTransaction(TransactionInfo info) throws Exception { |
| if (trackTransactions && info != null && info.getTransactionId() != null) { |
| ConnectionId connectionId = info.getConnectionId(); |
| if (connectionId != null) { |
| ConnectionState cs = connectionStates.get(connectionId); |
| if (cs != null) { |
| TransactionState transactionState = cs.getTransactionState(info.getTransactionId()); |
| if (transactionState != null) { |
| transactionState.addCommand(info); |
| } |
| } |
| } |
| return TRACKED_RESPONSE_MARKER; |
| } |
| return null; |
| } |
| |
| @Override |
| public Response processMessagePull(MessagePull pull) throws Exception { |
| if (pull != null) { |
| // leave a single instance in the cache |
| final String id = pull.getDestination() + "::" + pull.getConsumerId(); |
| if (messageCache.put(id.intern(), pull) == null) { |
| // Only marked as tracked if this is the first request we've seen. |
| pull.setTracked(true); |
| } |
| } |
| return null; |
| } |
| |
| public boolean isRestoreConsumers() { |
| return restoreConsumers; |
| } |
| |
| public void setRestoreConsumers(boolean restoreConsumers) { |
| this.restoreConsumers = restoreConsumers; |
| } |
| |
| public boolean isRestoreProducers() { |
| return restoreProducers; |
| } |
| |
| public void setRestoreProducers(boolean restoreProducers) { |
| this.restoreProducers = restoreProducers; |
| } |
| |
| public boolean isRestoreSessions() { |
| return restoreSessions; |
| } |
| |
| public void setRestoreSessions(boolean restoreSessions) { |
| this.restoreSessions = restoreSessions; |
| } |
| |
| public boolean isTrackTransactions() { |
| return trackTransactions; |
| } |
| |
| public void setTrackTransactions(boolean trackTransactions) { |
| this.trackTransactions = trackTransactions; |
| } |
| |
| public boolean isTrackTransactionProducers() { |
| return this.trackTransactionProducers; |
| } |
| |
| public void setTrackTransactionProducers(boolean trackTransactionProducers) { |
| this.trackTransactionProducers = trackTransactionProducers; |
| } |
| |
| public boolean isRestoreTransaction() { |
| return restoreTransaction; |
| } |
| |
| public void setRestoreTransaction(boolean restoreTransaction) { |
| this.restoreTransaction = restoreTransaction; |
| } |
| |
| public boolean isTrackMessages() { |
| return trackMessages; |
| } |
| |
| public void setTrackMessages(boolean trackMessages) { |
| this.trackMessages = trackMessages; |
| } |
| |
| public int getMaxCacheSize() { |
| return maxCacheSize; |
| } |
| |
| public void setMaxCacheSize(int maxCacheSize) { |
| this.maxCacheSize = maxCacheSize; |
| } |
| |
| /** |
| * @return the current cache size for the Message and MessagePull Command cache. |
| */ |
| public long getCurrentCacheSize() { |
| return this.currentCacheSize; |
| } |
| |
| public void connectionInterruptProcessingComplete(Transport transport, ConnectionId connectionId) { |
| ConnectionState connectionState = connectionStates.get(connectionId); |
| if (connectionState != null) { |
| connectionState.setConnectionInterruptProcessingComplete(true); |
| Map<ConsumerId, ConsumerInfo> stalledConsumers = connectionState.getRecoveringPullConsumers(); |
| for (Entry<ConsumerId, ConsumerInfo> entry: stalledConsumers.entrySet()) { |
| ConsumerControl control = new ConsumerControl(); |
| control.setConsumerId(entry.getKey()); |
| control.setPrefetch(entry.getValue().getPrefetchSize()); |
| control.setDestination(entry.getValue().getDestination()); |
| try { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("restored recovering consumer: " + control.getConsumerId() + " with: " + control.getPrefetch()); |
| } |
| transport.oneway(control); |
| } catch (Exception ex) { |
| if (LOG.isDebugEnabled()) { |
| LOG.debug("Failed to submit control for consumer: " + control.getConsumerId() |
| + " with: " + control.getPrefetch(), ex); |
| } |
| } |
| } |
| stalledConsumers.clear(); |
| } |
| } |
| |
| public void transportInterrupted(ConnectionId connectionId) { |
| ConnectionState connectionState = connectionStates.get(connectionId); |
| if (connectionState != null) { |
| connectionState.setConnectionInterruptProcessingComplete(false); |
| } |
| } |
| } |
