ConnectionContext connectionContext = entry.getValue().context;
if (connectionContext == null) {
continue;
}
Connection connection = connectionContext.getConnection();
if (connection == null) {
LOG.debug("slowConsumer abort ignored, no connection in context:" + connectionContext);
}
if (!abortMap.containsKey(connection)) {
abortMap.put(connection, new ArrayList<Subscription>());
}
abortMap.get(connection).add(entry.getKey());
}
for (Entry<Connection, List<Subscription>> entry : abortMap.entrySet()) {
final Connection connection = entry.getKey();
final List<Subscription> subscriptions = entry.getValue();
if (abortSubscriberConnection) {
LOG.info("aborting connection:{} with {} slow consumers",
connection.getConnectionId(), subscriptions.size());
if (LOG.isTraceEnabled()) {
for (Subscription subscription : subscriptions) {
LOG.trace("Connection {} being aborted because of slow consumer: {} on destination: {}",
new Object[] { connection.getConnectionId(),
subscription.getConsumerInfo().getConsumerId(),
subscription.getActiveMQDestination() });
}
}
try {
scheduler.executeAfterDelay(new Runnable() {
@Override
public void run() {
connection.serviceException(new InactivityIOException(
subscriptions.size() + " Consumers was slow too often (>"
+ maxSlowCount + ") or too long (>"
+ maxSlowDuration + "): "));
}}, 0l);
} catch (Exception e) {
LOG.info("exception on aborting connection {} with {} slow consumers",
connection.getConnectionId(), subscriptions.size());
}
} else {
// just abort each consumer
for (Subscription subscription : subscriptions) {
final Subscription subToClose = subscription;
LOG.info("aborting slow consumer: {} for destination:{}",
subscription.getConsumerInfo().getConsumerId(),
subscription.getActiveMQDestination());
// tell the remote consumer to close
try {
ConsumerControl stopConsumer = new ConsumerControl();
stopConsumer.setConsumerId(subscription.getConsumerInfo().getConsumerId());
stopConsumer.setClose(true);
connection.dispatchAsync(stopConsumer);
} catch (Exception e) {
LOG.info("exception on aborting slow consumer: {}", subscription.getConsumerInfo().getConsumerId(), e);
}
// force a local remove in case remote is unresponsive
try {
scheduler.executeAfterDelay(new Runnable() {
@Override
public void run() {
try {
RemoveInfo removeCommand = subToClose.getConsumerInfo().createRemoveCommand();
if (connection instanceof CommandVisitor) {
// avoid service exception handling and logging
removeCommand.visit((CommandVisitor) connection);
} else {
connection.service(removeCommand);
}
} catch (IllegalStateException ignoredAsRemoteHasDoneTheJob) {
} catch (Exception e) {
LOG.info("exception on local remove of slow consumer: {}", subToClose.getConsumerInfo().getConsumerId(), e);
}