001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.activemq.broker; 018 019import java.io.EOFException; 020import java.io.IOException; 021import java.net.SocketException; 022import java.net.URI; 023import java.util.Collection; 024import java.util.HashMap; 025import java.util.Iterator; 026import java.util.LinkedList; 027import java.util.List; 028import java.util.Map; 029import java.util.Properties; 030import java.util.concurrent.ConcurrentHashMap; 031import java.util.concurrent.CopyOnWriteArrayList; 032import java.util.concurrent.CountDownLatch; 033import java.util.concurrent.TimeUnit; 034import java.util.concurrent.atomic.AtomicBoolean; 035import java.util.concurrent.atomic.AtomicInteger; 036import java.util.concurrent.atomic.AtomicReference; 037import java.util.concurrent.locks.ReentrantReadWriteLock; 038 039import javax.transaction.xa.XAResource; 040 041import org.apache.activemq.advisory.AdvisorySupport; 042import org.apache.activemq.broker.region.ConnectionStatistics; 043import org.apache.activemq.broker.region.RegionBroker; 044import org.apache.activemq.command.ActiveMQDestination; 045import org.apache.activemq.command.BrokerInfo; 046import org.apache.activemq.command.Command; 047import org.apache.activemq.command.CommandTypes; 048import org.apache.activemq.command.ConnectionControl; 049import org.apache.activemq.command.ConnectionError; 050import org.apache.activemq.command.ConnectionId; 051import org.apache.activemq.command.ConnectionInfo; 052import org.apache.activemq.command.ConsumerControl; 053import org.apache.activemq.command.ConsumerId; 054import org.apache.activemq.command.ConsumerInfo; 055import org.apache.activemq.command.ControlCommand; 056import org.apache.activemq.command.DataArrayResponse; 057import org.apache.activemq.command.DestinationInfo; 058import org.apache.activemq.command.ExceptionResponse; 059import org.apache.activemq.command.FlushCommand; 060import org.apache.activemq.command.IntegerResponse; 061import org.apache.activemq.command.KeepAliveInfo; 062import org.apache.activemq.command.Message; 063import org.apache.activemq.command.MessageAck; 064import org.apache.activemq.command.MessageDispatch; 065import org.apache.activemq.command.MessageDispatchNotification; 066import org.apache.activemq.command.MessagePull; 067import org.apache.activemq.command.ProducerAck; 068import org.apache.activemq.command.ProducerId; 069import org.apache.activemq.command.ProducerInfo; 070import org.apache.activemq.command.RemoveInfo; 071import org.apache.activemq.command.RemoveSubscriptionInfo; 072import org.apache.activemq.command.Response; 073import org.apache.activemq.command.SessionId; 074import org.apache.activemq.command.SessionInfo; 075import org.apache.activemq.command.ShutdownInfo; 076import org.apache.activemq.command.TransactionId; 077import org.apache.activemq.command.TransactionInfo; 078import org.apache.activemq.command.WireFormatInfo; 079import org.apache.activemq.network.DemandForwardingBridge; 080import org.apache.activemq.network.MBeanNetworkListener; 081import org.apache.activemq.network.NetworkBridgeConfiguration; 082import org.apache.activemq.network.NetworkBridgeFactory; 083import org.apache.activemq.security.MessageAuthorizationPolicy; 084import org.apache.activemq.state.CommandVisitor; 085import org.apache.activemq.state.ConnectionState; 086import org.apache.activemq.state.ConsumerState; 087import org.apache.activemq.state.ProducerState; 088import org.apache.activemq.state.SessionState; 089import org.apache.activemq.state.TransactionState; 090import org.apache.activemq.thread.Task; 091import org.apache.activemq.thread.TaskRunner; 092import org.apache.activemq.thread.TaskRunnerFactory; 093import org.apache.activemq.transaction.Transaction; 094import org.apache.activemq.transport.DefaultTransportListener; 095import org.apache.activemq.transport.ResponseCorrelator; 096import org.apache.activemq.transport.TransmitCallback; 097import org.apache.activemq.transport.Transport; 098import org.apache.activemq.transport.TransportDisposedIOException; 099import org.apache.activemq.util.IntrospectionSupport; 100import org.apache.activemq.util.MarshallingSupport; 101import org.slf4j.Logger; 102import org.slf4j.LoggerFactory; 103import org.slf4j.MDC; 104 105public class TransportConnection implements Connection, Task, CommandVisitor { 106 private static final Logger LOG = LoggerFactory.getLogger(TransportConnection.class); 107 private static final Logger TRANSPORTLOG = LoggerFactory.getLogger(TransportConnection.class.getName() + ".Transport"); 108 private static final Logger SERVICELOG = LoggerFactory.getLogger(TransportConnection.class.getName() + ".Service"); 109 // Keeps track of the broker and connector that created this connection. 110 protected final Broker broker; 111 protected final BrokerService brokerService; 112 protected final TransportConnector connector; 113 // Keeps track of the state of the connections. 114 // protected final ConcurrentHashMap localConnectionStates=new 115 // ConcurrentHashMap(); 116 protected final Map<ConnectionId, ConnectionState> brokerConnectionStates; 117 // The broker and wireformat info that was exchanged. 118 protected BrokerInfo brokerInfo; 119 protected final List<Command> dispatchQueue = new LinkedList<Command>(); 120 protected TaskRunner taskRunner; 121 protected final AtomicReference<Throwable> transportException = new AtomicReference<Throwable>(); 122 protected AtomicBoolean dispatchStopped = new AtomicBoolean(false); 123 private final Transport transport; 124 private MessageAuthorizationPolicy messageAuthorizationPolicy; 125 private WireFormatInfo wireFormatInfo; 126 // Used to do async dispatch.. this should perhaps be pushed down into the 127 // transport layer.. 128 private boolean inServiceException; 129 private final ConnectionStatistics statistics = new ConnectionStatistics(); 130 private boolean manageable; 131 private boolean slow; 132 private boolean markedCandidate; 133 private boolean blockedCandidate; 134 private boolean blocked; 135 private boolean connected; 136 private boolean active; 137 138 // state management around pending stop 139 private static final int NEW = 0; 140 private static final int STARTING = 1; 141 private static final int STARTED = 2; 142 private static final int PENDING_STOP = 3; 143 private final AtomicInteger status = new AtomicInteger(NEW); 144 145 private long timeStamp; 146 private final AtomicBoolean stopping = new AtomicBoolean(false); 147 private final CountDownLatch stopped = new CountDownLatch(1); 148 private final AtomicBoolean asyncException = new AtomicBoolean(false); 149 private final Map<ProducerId, ProducerBrokerExchange> producerExchanges = new HashMap<ProducerId, ProducerBrokerExchange>(); 150 private final Map<ConsumerId, ConsumerBrokerExchange> consumerExchanges = new HashMap<ConsumerId, ConsumerBrokerExchange>(); 151 private final CountDownLatch dispatchStoppedLatch = new CountDownLatch(1); 152 private ConnectionContext context; 153 private boolean networkConnection; 154 private boolean faultTolerantConnection; 155 private final AtomicInteger protocolVersion = new AtomicInteger(CommandTypes.PROTOCOL_VERSION); 156 private DemandForwardingBridge duplexBridge; 157 private final TaskRunnerFactory taskRunnerFactory; 158 private final TaskRunnerFactory stopTaskRunnerFactory; 159 private TransportConnectionStateRegister connectionStateRegister = new SingleTransportConnectionStateRegister(); 160 private final ReentrantReadWriteLock serviceLock = new ReentrantReadWriteLock(); 161 private String duplexNetworkConnectorId; 162 163 /** 164 * @param taskRunnerFactory - can be null if you want direct dispatch to the transport 165 * else commands are sent async. 166 * @param stopTaskRunnerFactory - can <b>not</b> be null, used for stopping this connection. 167 */ 168 public TransportConnection(TransportConnector connector, final Transport transport, Broker broker, 169 TaskRunnerFactory taskRunnerFactory, TaskRunnerFactory stopTaskRunnerFactory) { 170 this.connector = connector; 171 this.broker = broker; 172 this.brokerService = broker.getBrokerService(); 173 174 RegionBroker rb = (RegionBroker) broker.getAdaptor(RegionBroker.class); 175 brokerConnectionStates = rb.getConnectionStates(); 176 if (connector != null) { 177 this.statistics.setParent(connector.getStatistics()); 178 this.messageAuthorizationPolicy = connector.getMessageAuthorizationPolicy(); 179 } 180 this.taskRunnerFactory = taskRunnerFactory; 181 this.stopTaskRunnerFactory = stopTaskRunnerFactory; 182 this.transport = transport; 183 if( this.transport instanceof BrokerServiceAware ) { 184 ((BrokerServiceAware)this.transport).setBrokerService(brokerService); 185 } 186 this.transport.setTransportListener(new DefaultTransportListener() { 187 @Override 188 public void onCommand(Object o) { 189 serviceLock.readLock().lock(); 190 try { 191 if (!(o instanceof Command)) { 192 throw new RuntimeException("Protocol violation - Command corrupted: " + o.toString()); 193 } 194 Command command = (Command) o; 195 if (!brokerService.isStopping()) { 196 Response response = service(command); 197 if (response != null && !brokerService.isStopping()) { 198 dispatchSync(response); 199 } 200 } else { 201 throw new BrokerStoppedException("Broker " + brokerService + " is being stopped"); 202 } 203 } finally { 204 serviceLock.readLock().unlock(); 205 } 206 } 207 208 @Override 209 public void onException(IOException exception) { 210 serviceLock.readLock().lock(); 211 try { 212 serviceTransportException(exception); 213 } finally { 214 serviceLock.readLock().unlock(); 215 } 216 } 217 }); 218 connected = true; 219 } 220 221 /** 222 * Returns the number of messages to be dispatched to this connection 223 * 224 * @return size of dispatch queue 225 */ 226 @Override 227 public int getDispatchQueueSize() { 228 synchronized (dispatchQueue) { 229 return dispatchQueue.size(); 230 } 231 } 232 233 public void serviceTransportException(IOException e) { 234 if (!stopping.get() && status.get() != PENDING_STOP) { 235 transportException.set(e); 236 if (TRANSPORTLOG.isDebugEnabled()) { 237 TRANSPORTLOG.debug(this + " failed: " + e, e); 238 } else if (TRANSPORTLOG.isWarnEnabled() && !suppressed(e)) { 239 TRANSPORTLOG.warn(this + " failed: " + e); 240 } 241 stopAsync(e); 242 } 243 } 244 245 private boolean suppressed(IOException e) { 246 return !connector.isWarnOnRemoteClose() && ((e instanceof SocketException && e.getMessage().indexOf("reset") != -1) || e instanceof EOFException); 247 } 248 249 /** 250 * Calls the serviceException method in an async thread. Since handling a 251 * service exception closes a socket, we should not tie up broker threads 252 * since client sockets may hang or cause deadlocks. 253 */ 254 @Override 255 public void serviceExceptionAsync(final IOException e) { 256 if (asyncException.compareAndSet(false, true)) { 257 new Thread("Async Exception Handler") { 258 @Override 259 public void run() { 260 serviceException(e); 261 } 262 }.start(); 263 } 264 } 265 266 /** 267 * Closes a clients connection due to a detected error. Errors are ignored 268 * if: the client is closing or broker is closing. Otherwise, the connection 269 * error transmitted to the client before stopping it's transport. 270 */ 271 @Override 272 public void serviceException(Throwable e) { 273 // are we a transport exception such as not being able to dispatch 274 // synchronously to a transport 275 if (e instanceof IOException) { 276 serviceTransportException((IOException) e); 277 } else if (e.getClass() == BrokerStoppedException.class) { 278 // Handle the case where the broker is stopped 279 // But the client is still connected. 280 if (!stopping.get()) { 281 SERVICELOG.debug("Broker has been stopped. Notifying client and closing his connection."); 282 ConnectionError ce = new ConnectionError(); 283 ce.setException(e); 284 dispatchSync(ce); 285 // Record the error that caused the transport to stop 286 transportException.set(e); 287 // Wait a little bit to try to get the output buffer to flush 288 // the exception notification to the client. 289 try { 290 Thread.sleep(500); 291 } catch (InterruptedException ie) { 292 Thread.currentThread().interrupt(); 293 } 294 // Worst case is we just kill the connection before the 295 // notification gets to him. 296 stopAsync(); 297 } 298 } else if (!stopping.get() && !inServiceException) { 299 inServiceException = true; 300 try { 301 if (SERVICELOG.isDebugEnabled()) { 302 SERVICELOG.debug("Async error occurred: " + e, e); 303 } else { 304 SERVICELOG.warn("Async error occurred: " + e); 305 } 306 ConnectionError ce = new ConnectionError(); 307 ce.setException(e); 308 if (status.get() == PENDING_STOP) { 309 dispatchSync(ce); 310 } else { 311 dispatchAsync(ce); 312 } 313 } finally { 314 inServiceException = false; 315 } 316 } 317 } 318 319 @Override 320 public Response service(Command command) { 321 MDC.put("activemq.connector", connector.getUri().toString()); 322 Response response = null; 323 boolean responseRequired = command.isResponseRequired(); 324 int commandId = command.getCommandId(); 325 try { 326 if (status.get() != PENDING_STOP) { 327 response = command.visit(this); 328 } else { 329 response = new ExceptionResponse(transportException.get()); 330 } 331 } catch (Throwable e) { 332 if (SERVICELOG.isDebugEnabled() && e.getClass() != BrokerStoppedException.class) { 333 SERVICELOG.debug("Error occured while processing " + (responseRequired ? "sync" : "async") 334 + " command: " + command + ", exception: " + e, e); 335 } 336 337 if (e instanceof SuppressReplyException || (e.getCause() instanceof SuppressReplyException)) { 338 LOG.info("Suppressing reply to: " + command + " on: " + e + ", cause: " + e.getCause()); 339 responseRequired = false; 340 } 341 342 if (responseRequired) { 343 if (e instanceof SecurityException || e.getCause() instanceof SecurityException) { 344 SERVICELOG.warn("Security Error occurred on connection to: {}, {}", 345 transport.getRemoteAddress(), e.getMessage()); 346 } 347 response = new ExceptionResponse(e); 348 } else { 349 forceRollbackOnlyOnFailedAsyncTransactionOp(e, command); 350 serviceException(e); 351 } 352 } 353 if (responseRequired) { 354 if (response == null) { 355 response = new Response(); 356 } 357 response.setCorrelationId(commandId); 358 } 359 // The context may have been flagged so that the response is not 360 // sent. 361 if (context != null) { 362 if (context.isDontSendReponse()) { 363 context.setDontSendReponse(false); 364 response = null; 365 } 366 context = null; 367 } 368 MDC.remove("activemq.connector"); 369 return response; 370 } 371 372 private void forceRollbackOnlyOnFailedAsyncTransactionOp(Throwable e, Command command) { 373 if (brokerService.isRollbackOnlyOnAsyncException() && !(e instanceof IOException) && isInTransaction(command)) { 374 Transaction transaction = getActiveTransaction(command); 375 if (transaction != null && !transaction.isRollbackOnly()) { 376 LOG.debug("on async exception, force rollback of transaction for: " + command, e); 377 transaction.setRollbackOnly(e); 378 } 379 } 380 } 381 382 private Transaction getActiveTransaction(Command command) { 383 Transaction transaction = null; 384 try { 385 if (command instanceof Message) { 386 Message messageSend = (Message) command; 387 ProducerId producerId = messageSend.getProducerId(); 388 ProducerBrokerExchange producerExchange = getProducerBrokerExchange(producerId); 389 transaction = producerExchange.getConnectionContext().getTransactions().get(messageSend.getTransactionId()); 390 } else if (command instanceof MessageAck) { 391 MessageAck messageAck = (MessageAck) command; 392 ConsumerBrokerExchange consumerExchange = getConsumerBrokerExchange(messageAck.getConsumerId()); 393 if (consumerExchange != null) { 394 transaction = consumerExchange.getConnectionContext().getTransactions().get(messageAck.getTransactionId()); 395 } 396 } 397 } catch(Exception ignored){ 398 LOG.trace("failed to find active transaction for command: " + command, ignored); 399 } 400 return transaction; 401 } 402 403 private boolean isInTransaction(Command command) { 404 return command instanceof Message && ((Message)command).isInTransaction() 405 || command instanceof MessageAck && ((MessageAck)command).isInTransaction(); 406 } 407 408 @Override 409 public Response processKeepAlive(KeepAliveInfo info) throws Exception { 410 return null; 411 } 412 413 @Override 414 public Response processRemoveSubscription(RemoveSubscriptionInfo info) throws Exception { 415 broker.removeSubscription(lookupConnectionState(info.getConnectionId()).getContext(), info); 416 return null; 417 } 418 419 @Override 420 public Response processWireFormat(WireFormatInfo info) throws Exception { 421 wireFormatInfo = info; 422 protocolVersion.set(info.getVersion()); 423 return null; 424 } 425 426 @Override 427 public Response processShutdown(ShutdownInfo info) throws Exception { 428 stopAsync(); 429 return null; 430 } 431 432 @Override 433 public Response processFlush(FlushCommand command) throws Exception { 434 return null; 435 } 436 437 @Override 438 public Response processBeginTransaction(TransactionInfo info) throws Exception { 439 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 440 context = null; 441 if (cs != null) { 442 context = cs.getContext(); 443 } 444 if (cs == null) { 445 throw new NullPointerException("Context is null"); 446 } 447 // Avoid replaying dup commands 448 if (cs.getTransactionState(info.getTransactionId()) == null) { 449 cs.addTransactionState(info.getTransactionId()); 450 broker.beginTransaction(context, info.getTransactionId()); 451 } 452 return null; 453 } 454 455 @Override 456 public int getActiveTransactionCount() { 457 int rc = 0; 458 for (TransportConnectionState cs : connectionStateRegister.listConnectionStates()) { 459 Collection<TransactionState> transactions = cs.getTransactionStates(); 460 for (TransactionState transaction : transactions) { 461 rc++; 462 } 463 } 464 return rc; 465 } 466 467 @Override 468 public Long getOldestActiveTransactionDuration() { 469 TransactionState oldestTX = null; 470 for (TransportConnectionState cs : connectionStateRegister.listConnectionStates()) { 471 Collection<TransactionState> transactions = cs.getTransactionStates(); 472 for (TransactionState transaction : transactions) { 473 if( oldestTX ==null || oldestTX.getCreatedAt() < transaction.getCreatedAt() ) { 474 oldestTX = transaction; 475 } 476 } 477 } 478 if( oldestTX == null ) { 479 return null; 480 } 481 return System.currentTimeMillis() - oldestTX.getCreatedAt(); 482 } 483 484 @Override 485 public Response processEndTransaction(TransactionInfo info) throws Exception { 486 // No need to do anything. This packet is just sent by the client 487 // make sure he is synced with the server as commit command could 488 // come from a different connection. 489 return null; 490 } 491 492 @Override 493 public Response processPrepareTransaction(TransactionInfo info) throws Exception { 494 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 495 context = null; 496 if (cs != null) { 497 context = cs.getContext(); 498 } 499 if (cs == null) { 500 throw new NullPointerException("Context is null"); 501 } 502 TransactionState transactionState = cs.getTransactionState(info.getTransactionId()); 503 if (transactionState == null) { 504 throw new IllegalStateException("Cannot prepare a transaction that had not been started or previously returned XA_RDONLY: " 505 + info.getTransactionId()); 506 } 507 // Avoid dups. 508 if (!transactionState.isPrepared()) { 509 transactionState.setPrepared(true); 510 int result = broker.prepareTransaction(context, info.getTransactionId()); 511 transactionState.setPreparedResult(result); 512 if (result == XAResource.XA_RDONLY) { 513 // we are done, no further rollback or commit from TM 514 cs.removeTransactionState(info.getTransactionId()); 515 } 516 IntegerResponse response = new IntegerResponse(result); 517 return response; 518 } else { 519 IntegerResponse response = new IntegerResponse(transactionState.getPreparedResult()); 520 return response; 521 } 522 } 523 524 @Override 525 public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception { 526 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 527 context = cs.getContext(); 528 cs.removeTransactionState(info.getTransactionId()); 529 broker.commitTransaction(context, info.getTransactionId(), true); 530 return null; 531 } 532 533 @Override 534 public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception { 535 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 536 context = cs.getContext(); 537 cs.removeTransactionState(info.getTransactionId()); 538 broker.commitTransaction(context, info.getTransactionId(), false); 539 return null; 540 } 541 542 @Override 543 public Response processRollbackTransaction(TransactionInfo info) throws Exception { 544 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 545 context = cs.getContext(); 546 cs.removeTransactionState(info.getTransactionId()); 547 broker.rollbackTransaction(context, info.getTransactionId()); 548 return null; 549 } 550 551 @Override 552 public Response processForgetTransaction(TransactionInfo info) throws Exception { 553 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 554 context = cs.getContext(); 555 broker.forgetTransaction(context, info.getTransactionId()); 556 return null; 557 } 558 559 @Override 560 public Response processRecoverTransactions(TransactionInfo info) throws Exception { 561 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 562 context = cs.getContext(); 563 TransactionId[] preparedTransactions = broker.getPreparedTransactions(context); 564 return new DataArrayResponse(preparedTransactions); 565 } 566 567 @Override 568 public Response processMessage(Message messageSend) throws Exception { 569 ProducerId producerId = messageSend.getProducerId(); 570 ProducerBrokerExchange producerExchange = getProducerBrokerExchange(producerId); 571 if (producerExchange.canDispatch(messageSend)) { 572 broker.send(producerExchange, messageSend); 573 } 574 return null; 575 } 576 577 @Override 578 public Response processMessageAck(MessageAck ack) throws Exception { 579 ConsumerBrokerExchange consumerExchange = getConsumerBrokerExchange(ack.getConsumerId()); 580 if (consumerExchange != null) { 581 broker.acknowledge(consumerExchange, ack); 582 } else if (ack.isInTransaction()) { 583 LOG.warn("no matching consumer, ignoring ack {}", consumerExchange, ack); 584 } 585 return null; 586 } 587 588 @Override 589 public Response processMessagePull(MessagePull pull) throws Exception { 590 return broker.messagePull(lookupConnectionState(pull.getConsumerId()).getContext(), pull); 591 } 592 593 @Override 594 public Response processMessageDispatchNotification(MessageDispatchNotification notification) throws Exception { 595 broker.processDispatchNotification(notification); 596 return null; 597 } 598 599 @Override 600 public Response processAddDestination(DestinationInfo info) throws Exception { 601 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 602 broker.addDestinationInfo(cs.getContext(), info); 603 if (info.getDestination().isTemporary()) { 604 cs.addTempDestination(info); 605 } 606 return null; 607 } 608 609 @Override 610 public Response processRemoveDestination(DestinationInfo info) throws Exception { 611 TransportConnectionState cs = lookupConnectionState(info.getConnectionId()); 612 broker.removeDestinationInfo(cs.getContext(), info); 613 if (info.getDestination().isTemporary()) { 614 cs.removeTempDestination(info.getDestination()); 615 } 616 return null; 617 } 618 619 @Override 620 public Response processAddProducer(ProducerInfo info) throws Exception { 621 SessionId sessionId = info.getProducerId().getParentId(); 622 ConnectionId connectionId = sessionId.getParentId(); 623 TransportConnectionState cs = lookupConnectionState(connectionId); 624 if (cs == null) { 625 throw new IllegalStateException("Cannot add a producer to a connection that had not been registered: " 626 + connectionId); 627 } 628 SessionState ss = cs.getSessionState(sessionId); 629 if (ss == null) { 630 throw new IllegalStateException("Cannot add a producer to a session that had not been registered: " 631 + sessionId); 632 } 633 // Avoid replaying dup commands 634 if (!ss.getProducerIds().contains(info.getProducerId())) { 635 ActiveMQDestination destination = info.getDestination(); 636 // Do not check for null here as it would cause the count of max producers to exclude 637 // anonymous producers. The isAdvisoryTopic method checks for null so it is safe to 638 // call it from here with a null Destination value. 639 if (!AdvisorySupport.isAdvisoryTopic(destination)) { 640 if (getProducerCount(connectionId) >= connector.getMaximumProducersAllowedPerConnection()){ 641 throw new IllegalStateException("Can't add producer on connection " + connectionId + ": at maximum limit: " + connector.getMaximumProducersAllowedPerConnection()); 642 } 643 } 644 broker.addProducer(cs.getContext(), info); 645 try { 646 ss.addProducer(info); 647 } catch (IllegalStateException e) { 648 broker.removeProducer(cs.getContext(), info); 649 } 650 651 } 652 return null; 653 } 654 655 @Override 656 public Response processRemoveProducer(ProducerId id) throws Exception { 657 SessionId sessionId = id.getParentId(); 658 ConnectionId connectionId = sessionId.getParentId(); 659 TransportConnectionState cs = lookupConnectionState(connectionId); 660 SessionState ss = cs.getSessionState(sessionId); 661 if (ss == null) { 662 throw new IllegalStateException("Cannot remove a producer from a session that had not been registered: " 663 + sessionId); 664 } 665 ProducerState ps = ss.removeProducer(id); 666 if (ps == null) { 667 throw new IllegalStateException("Cannot remove a producer that had not been registered: " + id); 668 } 669 removeProducerBrokerExchange(id); 670 broker.removeProducer(cs.getContext(), ps.getInfo()); 671 return null; 672 } 673 674 @Override 675 public Response processAddConsumer(ConsumerInfo info) throws Exception { 676 SessionId sessionId = info.getConsumerId().getParentId(); 677 ConnectionId connectionId = sessionId.getParentId(); 678 TransportConnectionState cs = lookupConnectionState(connectionId); 679 if (cs == null) { 680 throw new IllegalStateException("Cannot add a consumer to a connection that had not been registered: " 681 + connectionId); 682 } 683 SessionState ss = cs.getSessionState(sessionId); 684 if (ss == null) { 685 throw new IllegalStateException(broker.getBrokerName() 686 + " Cannot add a consumer to a session that had not been registered: " + sessionId); 687 } 688 // Avoid replaying dup commands 689 if (!ss.getConsumerIds().contains(info.getConsumerId())) { 690 ActiveMQDestination destination = info.getDestination(); 691 if (destination != null && !AdvisorySupport.isAdvisoryTopic(destination)) { 692 if (getConsumerCount(connectionId) >= connector.getMaximumConsumersAllowedPerConnection()){ 693 throw new IllegalStateException("Can't add consumer on connection " + connectionId + ": at maximum limit: " + connector.getMaximumConsumersAllowedPerConnection()); 694 } 695 } 696 697 broker.addConsumer(cs.getContext(), info); 698 try { 699 ss.addConsumer(info); 700 addConsumerBrokerExchange(cs, info.getConsumerId()); 701 } catch (IllegalStateException e) { 702 broker.removeConsumer(cs.getContext(), info); 703 } 704 705 } 706 return null; 707 } 708 709 @Override 710 public Response processRemoveConsumer(ConsumerId id, long lastDeliveredSequenceId) throws Exception { 711 SessionId sessionId = id.getParentId(); 712 ConnectionId connectionId = sessionId.getParentId(); 713 TransportConnectionState cs = lookupConnectionState(connectionId); 714 if (cs == null) { 715 throw new IllegalStateException("Cannot remove a consumer from a connection that had not been registered: " 716 + connectionId); 717 } 718 SessionState ss = cs.getSessionState(sessionId); 719 if (ss == null) { 720 throw new IllegalStateException("Cannot remove a consumer from a session that had not been registered: " 721 + sessionId); 722 } 723 ConsumerState consumerState = ss.removeConsumer(id); 724 if (consumerState == null) { 725 throw new IllegalStateException("Cannot remove a consumer that had not been registered: " + id); 726 } 727 ConsumerInfo info = consumerState.getInfo(); 728 info.setLastDeliveredSequenceId(lastDeliveredSequenceId); 729 broker.removeConsumer(cs.getContext(), consumerState.getInfo()); 730 removeConsumerBrokerExchange(id); 731 return null; 732 } 733 734 @Override 735 public Response processAddSession(SessionInfo info) throws Exception { 736 ConnectionId connectionId = info.getSessionId().getParentId(); 737 TransportConnectionState cs = lookupConnectionState(connectionId); 738 // Avoid replaying dup commands 739 if (cs != null && !cs.getSessionIds().contains(info.getSessionId())) { 740 broker.addSession(cs.getContext(), info); 741 try { 742 cs.addSession(info); 743 } catch (IllegalStateException e) { 744 LOG.warn("Failed to add session: {}", info.getSessionId(), e); 745 broker.removeSession(cs.getContext(), info); 746 } 747 } 748 return null; 749 } 750 751 @Override 752 public Response processRemoveSession(SessionId id, long lastDeliveredSequenceId) throws Exception { 753 ConnectionId connectionId = id.getParentId(); 754 TransportConnectionState cs = lookupConnectionState(connectionId); 755 if (cs == null) { 756 throw new IllegalStateException("Cannot remove session from connection that had not been registered: " + connectionId); 757 } 758 SessionState session = cs.getSessionState(id); 759 if (session == null) { 760 throw new IllegalStateException("Cannot remove session that had not been registered: " + id); 761 } 762 // Don't let new consumers or producers get added while we are closing 763 // this down. 764 session.shutdown(); 765 // Cascade the connection stop to the consumers and producers. 766 for (ConsumerId consumerId : session.getConsumerIds()) { 767 try { 768 processRemoveConsumer(consumerId, lastDeliveredSequenceId); 769 } catch (Throwable e) { 770 LOG.warn("Failed to remove consumer: {}", consumerId, e); 771 } 772 } 773 for (ProducerId producerId : session.getProducerIds()) { 774 try { 775 processRemoveProducer(producerId); 776 } catch (Throwable e) { 777 LOG.warn("Failed to remove producer: {}", producerId, e); 778 } 779 } 780 cs.removeSession(id); 781 broker.removeSession(cs.getContext(), session.getInfo()); 782 return null; 783 } 784 785 @Override 786 public Response processAddConnection(ConnectionInfo info) throws Exception { 787 // Older clients should have been defaulting this field to true.. but 788 // they were not. 789 if (wireFormatInfo != null && wireFormatInfo.getVersion() <= 2) { 790 info.setClientMaster(true); 791 } 792 TransportConnectionState state; 793 // Make sure 2 concurrent connections by the same ID only generate 1 794 // TransportConnectionState object. 795 synchronized (brokerConnectionStates) { 796 state = (TransportConnectionState) brokerConnectionStates.get(info.getConnectionId()); 797 if (state == null) { 798 state = new TransportConnectionState(info, this); 799 brokerConnectionStates.put(info.getConnectionId(), state); 800 } 801 state.incrementReference(); 802 } 803 // If there are 2 concurrent connections for the same connection id, 804 // then last one in wins, we need to sync here 805 // to figure out the winner. 806 synchronized (state.getConnectionMutex()) { 807 if (state.getConnection() != this) { 808 LOG.debug("Killing previous stale connection: {}", state.getConnection().getRemoteAddress()); 809 state.getConnection().stop(); 810 LOG.debug("Connection {} taking over previous connection: {}", getRemoteAddress(), state.getConnection().getRemoteAddress()); 811 state.setConnection(this); 812 state.reset(info); 813 } 814 } 815 registerConnectionState(info.getConnectionId(), state); 816 LOG.debug("Setting up new connection id: {}, address: {}, info: {}", new Object[]{ info.getConnectionId(), getRemoteAddress(), info }); 817 this.faultTolerantConnection = info.isFaultTolerant(); 818 // Setup the context. 819 String clientId = info.getClientId(); 820 context = new ConnectionContext(); 821 context.setBroker(broker); 822 context.setClientId(clientId); 823 context.setClientMaster(info.isClientMaster()); 824 context.setConnection(this); 825 context.setConnectionId(info.getConnectionId()); 826 context.setConnector(connector); 827 context.setMessageAuthorizationPolicy(getMessageAuthorizationPolicy()); 828 context.setNetworkConnection(networkConnection); 829 context.setFaultTolerant(faultTolerantConnection); 830 context.setTransactions(new ConcurrentHashMap<TransactionId, Transaction>()); 831 context.setUserName(info.getUserName()); 832 context.setWireFormatInfo(wireFormatInfo); 833 context.setReconnect(info.isFailoverReconnect()); 834 this.manageable = info.isManageable(); 835 context.setConnectionState(state); 836 state.setContext(context); 837 state.setConnection(this); 838 if (info.getClientIp() == null) { 839 info.setClientIp(getRemoteAddress()); 840 } 841 842 try { 843 broker.addConnection(context, info); 844 } catch (Exception e) { 845 synchronized (brokerConnectionStates) { 846 brokerConnectionStates.remove(info.getConnectionId()); 847 } 848 unregisterConnectionState(info.getConnectionId()); 849 LOG.warn("Failed to add Connection {} due to {}", info.getConnectionId(), e); 850 if (e instanceof SecurityException) { 851 // close this down - in case the peer of this transport doesn't play nice 852 delayedStop(2000, "Failed with SecurityException: " + e.getLocalizedMessage(), e); 853 } 854 throw e; 855 } 856 if (info.isManageable()) { 857 // send ConnectionCommand 858 ConnectionControl command = this.connector.getConnectionControl(); 859 command.setFaultTolerant(broker.isFaultTolerantConfiguration()); 860 if (info.isFailoverReconnect()) { 861 command.setRebalanceConnection(false); 862 } 863 dispatchAsync(command); 864 } 865 return null; 866 } 867 868 @Override 869 public synchronized Response processRemoveConnection(ConnectionId id, long lastDeliveredSequenceId) 870 throws InterruptedException { 871 LOG.debug("remove connection id: {}", id); 872 TransportConnectionState cs = lookupConnectionState(id); 873 if (cs != null) { 874 // Don't allow things to be added to the connection state while we 875 // are shutting down. 876 cs.shutdown(); 877 // Cascade the connection stop to the sessions. 878 for (SessionId sessionId : cs.getSessionIds()) { 879 try { 880 processRemoveSession(sessionId, lastDeliveredSequenceId); 881 } catch (Throwable e) { 882 SERVICELOG.warn("Failed to remove session {}", sessionId, e); 883 } 884 } 885 // Cascade the connection stop to temp destinations. 886 for (Iterator<DestinationInfo> iter = cs.getTempDestinations().iterator(); iter.hasNext(); ) { 887 DestinationInfo di = iter.next(); 888 try { 889 broker.removeDestination(cs.getContext(), di.getDestination(), 0); 890 } catch (Throwable e) { 891 SERVICELOG.warn("Failed to remove tmp destination {}", di.getDestination(), e); 892 } 893 iter.remove(); 894 } 895 try { 896 broker.removeConnection(cs.getContext(), cs.getInfo(), transportException.get()); 897 } catch (Throwable e) { 898 SERVICELOG.warn("Failed to remove connection {}", cs.getInfo(), e); 899 } 900 TransportConnectionState state = unregisterConnectionState(id); 901 if (state != null) { 902 synchronized (brokerConnectionStates) { 903 // If we are the last reference, we should remove the state 904 // from the broker. 905 if (state.decrementReference() == 0) { 906 brokerConnectionStates.remove(id); 907 } 908 } 909 } 910 } 911 return null; 912 } 913 914 @Override 915 public Response processProducerAck(ProducerAck ack) throws Exception { 916 // A broker should not get ProducerAck messages. 917 return null; 918 } 919 920 @Override 921 public Connector getConnector() { 922 return connector; 923 } 924 925 @Override 926 public void dispatchSync(Command message) { 927 try { 928 processDispatch(message); 929 } catch (IOException e) { 930 serviceExceptionAsync(e); 931 } 932 } 933 934 @Override 935 public void dispatchAsync(Command message) { 936 if (!stopping.get()) { 937 if (taskRunner == null) { 938 dispatchSync(message); 939 } else { 940 synchronized (dispatchQueue) { 941 dispatchQueue.add(message); 942 } 943 try { 944 taskRunner.wakeup(); 945 } catch (InterruptedException e) { 946 Thread.currentThread().interrupt(); 947 } 948 } 949 } else { 950 if (message.isMessageDispatch()) { 951 MessageDispatch md = (MessageDispatch) message; 952 TransmitCallback sub = md.getTransmitCallback(); 953 broker.postProcessDispatch(md); 954 if (sub != null) { 955 sub.onFailure(); 956 } 957 } 958 } 959 } 960 961 protected void processDispatch(Command command) throws IOException { 962 MessageDispatch messageDispatch = (MessageDispatch) (command.isMessageDispatch() ? command : null); 963 try { 964 if (!stopping.get()) { 965 if (messageDispatch != null) { 966 try { 967 broker.preProcessDispatch(messageDispatch); 968 } catch (RuntimeException convertToIO) { 969 throw new IOException(convertToIO); 970 } 971 } 972 dispatch(command); 973 } 974 } catch (IOException e) { 975 if (messageDispatch != null) { 976 TransmitCallback sub = messageDispatch.getTransmitCallback(); 977 broker.postProcessDispatch(messageDispatch); 978 if (sub != null) { 979 sub.onFailure(); 980 } 981 messageDispatch = null; 982 throw e; 983 } else { 984 if (TRANSPORTLOG.isDebugEnabled()) { 985 TRANSPORTLOG.debug("Unexpected exception on asyncDispatch, command of type: " + command.getDataStructureType(), e); 986 } 987 } 988 } finally { 989 if (messageDispatch != null) { 990 TransmitCallback sub = messageDispatch.getTransmitCallback(); 991 broker.postProcessDispatch(messageDispatch); 992 if (sub != null) { 993 sub.onSuccess(); 994 } 995 } 996 } 997 } 998 999 @Override 1000 public boolean iterate() { 1001 try { 1002 if (status.get() == PENDING_STOP || stopping.get()) { 1003 if (dispatchStopped.compareAndSet(false, true)) { 1004 if (transportException.get() == null) { 1005 try { 1006 dispatch(new ShutdownInfo()); 1007 } catch (Throwable ignore) { 1008 } 1009 } 1010 dispatchStoppedLatch.countDown(); 1011 } 1012 return false; 1013 } 1014 if (!dispatchStopped.get()) { 1015 Command command = null; 1016 synchronized (dispatchQueue) { 1017 if (dispatchQueue.isEmpty()) { 1018 return false; 1019 } 1020 command = dispatchQueue.remove(0); 1021 } 1022 processDispatch(command); 1023 return true; 1024 } 1025 return false; 1026 } catch (IOException e) { 1027 if (dispatchStopped.compareAndSet(false, true)) { 1028 dispatchStoppedLatch.countDown(); 1029 } 1030 serviceExceptionAsync(e); 1031 return false; 1032 } 1033 } 1034 1035 /** 1036 * Returns the statistics for this connection 1037 */ 1038 @Override 1039 public ConnectionStatistics getStatistics() { 1040 return statistics; 1041 } 1042 1043 public MessageAuthorizationPolicy getMessageAuthorizationPolicy() { 1044 return messageAuthorizationPolicy; 1045 } 1046 1047 public void setMessageAuthorizationPolicy(MessageAuthorizationPolicy messageAuthorizationPolicy) { 1048 this.messageAuthorizationPolicy = messageAuthorizationPolicy; 1049 } 1050 1051 @Override 1052 public boolean isManageable() { 1053 return manageable; 1054 } 1055 1056 @Override 1057 public void start() throws Exception { 1058 if (status.compareAndSet(NEW, STARTING)) { 1059 try { 1060 synchronized (this) { 1061 if (taskRunnerFactory != null) { 1062 taskRunner = taskRunnerFactory.createTaskRunner(this, "ActiveMQ Connection Dispatcher: " 1063 + getRemoteAddress()); 1064 } else { 1065 taskRunner = null; 1066 } 1067 transport.start(); 1068 active = true; 1069 BrokerInfo info = connector.getBrokerInfo().copy(); 1070 if (connector.isUpdateClusterClients()) { 1071 info.setPeerBrokerInfos(this.broker.getPeerBrokerInfos()); 1072 } else { 1073 info.setPeerBrokerInfos(null); 1074 } 1075 dispatchAsync(info); 1076 1077 connector.onStarted(this); 1078 } 1079 } catch (Exception e) { 1080 // Force clean up on an error starting up. 1081 status.set(PENDING_STOP); 1082 throw e; 1083 } finally { 1084 // stop() can be called from within the above block, 1085 // but we want to be sure start() completes before 1086 // stop() runs, so queue the stop until right now: 1087 if (!status.compareAndSet(STARTING, STARTED)) { 1088 LOG.debug("Calling the delayed stop() after start() {}", this); 1089 stop(); 1090 } 1091 } 1092 } 1093 } 1094 1095 @Override 1096 public void stop() throws Exception { 1097 // do not stop task the task runner factories (taskRunnerFactory, stopTaskRunnerFactory) 1098 // as their lifecycle is handled elsewhere 1099 1100 stopAsync(); 1101 while (!stopped.await(5, TimeUnit.SECONDS)) { 1102 LOG.info("The connection to '{}' is taking a long time to shutdown.", transport.getRemoteAddress()); 1103 } 1104 } 1105 1106 public void delayedStop(final int waitTime, final String reason, Throwable cause) { 1107 if (waitTime > 0) { 1108 status.compareAndSet(STARTING, PENDING_STOP); 1109 transportException.set(cause); 1110 try { 1111 stopTaskRunnerFactory.execute(new Runnable() { 1112 @Override 1113 public void run() { 1114 try { 1115 Thread.sleep(waitTime); 1116 stopAsync(); 1117 LOG.info("Stopping {} because {}", transport.getRemoteAddress(), reason); 1118 } catch (InterruptedException e) { 1119 } 1120 } 1121 }); 1122 } catch (Throwable t) { 1123 LOG.warn("Cannot create stopAsync. This exception will be ignored.", t); 1124 } 1125 } 1126 } 1127 1128 public void stopAsync(Throwable cause) { 1129 transportException.set(cause); 1130 stopAsync(); 1131 } 1132 1133 public void stopAsync() { 1134 // If we're in the middle of starting then go no further... for now. 1135 if (status.compareAndSet(STARTING, PENDING_STOP)) { 1136 LOG.debug("stopAsync() called in the middle of start(). Delaying till start completes.."); 1137 return; 1138 } 1139 if (stopping.compareAndSet(false, true)) { 1140 // Let all the connection contexts know we are shutting down 1141 // so that in progress operations can notice and unblock. 1142 List<TransportConnectionState> connectionStates = listConnectionStates(); 1143 for (TransportConnectionState cs : connectionStates) { 1144 ConnectionContext connectionContext = cs.getContext(); 1145 if (connectionContext != null) { 1146 connectionContext.getStopping().set(true); 1147 } 1148 } 1149 try { 1150 stopTaskRunnerFactory.execute(new Runnable() { 1151 @Override 1152 public void run() { 1153 serviceLock.writeLock().lock(); 1154 try { 1155 doStop(); 1156 } catch (Throwable e) { 1157 LOG.debug("Error occurred while shutting down a connection {}", this, e); 1158 } finally { 1159 stopped.countDown(); 1160 serviceLock.writeLock().unlock(); 1161 } 1162 } 1163 }); 1164 } catch (Throwable t) { 1165 LOG.warn("Cannot create async transport stopper thread. This exception is ignored. Not waiting for stop to complete", t); 1166 stopped.countDown(); 1167 } 1168 } 1169 } 1170 1171 @Override 1172 public String toString() { 1173 return "Transport Connection to: " + transport.getRemoteAddress(); 1174 } 1175 1176 protected void doStop() throws Exception { 1177 LOG.debug("Stopping connection: {}", transport.getRemoteAddress()); 1178 connector.onStopped(this); 1179 try { 1180 synchronized (this) { 1181 if (duplexBridge != null) { 1182 duplexBridge.stop(); 1183 } 1184 } 1185 } catch (Exception ignore) { 1186 LOG.trace("Exception caught stopping. This exception is ignored.", ignore); 1187 } 1188 try { 1189 transport.stop(); 1190 LOG.debug("Stopped transport: {}", transport.getRemoteAddress()); 1191 } catch (Exception e) { 1192 LOG.debug("Could not stop transport to {}. This exception is ignored.", transport.getRemoteAddress(), e); 1193 } 1194 if (taskRunner != null) { 1195 taskRunner.shutdown(1); 1196 taskRunner = null; 1197 } 1198 active = false; 1199 // Run the MessageDispatch callbacks so that message references get 1200 // cleaned up. 1201 synchronized (dispatchQueue) { 1202 for (Iterator<Command> iter = dispatchQueue.iterator(); iter.hasNext(); ) { 1203 Command command = iter.next(); 1204 if (command.isMessageDispatch()) { 1205 MessageDispatch md = (MessageDispatch) command; 1206 TransmitCallback sub = md.getTransmitCallback(); 1207 broker.postProcessDispatch(md); 1208 if (sub != null) { 1209 sub.onFailure(); 1210 } 1211 } 1212 } 1213 dispatchQueue.clear(); 1214 } 1215 // 1216 // Remove all logical connection associated with this connection 1217 // from the broker. 1218 if (!broker.isStopped()) { 1219 List<TransportConnectionState> connectionStates = listConnectionStates(); 1220 connectionStates = listConnectionStates(); 1221 for (TransportConnectionState cs : connectionStates) { 1222 cs.getContext().getStopping().set(true); 1223 try { 1224 LOG.debug("Cleaning up connection resources: {}", getRemoteAddress()); 1225 processRemoveConnection(cs.getInfo().getConnectionId(), RemoveInfo.LAST_DELIVERED_UNKNOWN); 1226 } catch (Throwable ignore) { 1227 LOG.debug("Exception caught removing connection {}. This exception is ignored.", cs.getInfo().getConnectionId(), ignore); 1228 } 1229 } 1230 } 1231 LOG.debug("Connection Stopped: {}", getRemoteAddress()); 1232 } 1233 1234 /** 1235 * @return Returns the blockedCandidate. 1236 */ 1237 public boolean isBlockedCandidate() { 1238 return blockedCandidate; 1239 } 1240 1241 /** 1242 * @param blockedCandidate The blockedCandidate to set. 1243 */ 1244 public void setBlockedCandidate(boolean blockedCandidate) { 1245 this.blockedCandidate = blockedCandidate; 1246 } 1247 1248 /** 1249 * @return Returns the markedCandidate. 1250 */ 1251 public boolean isMarkedCandidate() { 1252 return markedCandidate; 1253 } 1254 1255 /** 1256 * @param markedCandidate The markedCandidate to set. 1257 */ 1258 public void setMarkedCandidate(boolean markedCandidate) { 1259 this.markedCandidate = markedCandidate; 1260 if (!markedCandidate) { 1261 timeStamp = 0; 1262 blockedCandidate = false; 1263 } 1264 } 1265 1266 /** 1267 * @param slow The slow to set. 1268 */ 1269 public void setSlow(boolean slow) { 1270 this.slow = slow; 1271 } 1272 1273 /** 1274 * @return true if the Connection is slow 1275 */ 1276 @Override 1277 public boolean isSlow() { 1278 return slow; 1279 } 1280 1281 /** 1282 * @return true if the Connection is potentially blocked 1283 */ 1284 public boolean isMarkedBlockedCandidate() { 1285 return markedCandidate; 1286 } 1287 1288 /** 1289 * Mark the Connection, so we can deem if it's collectable on the next sweep 1290 */ 1291 public void doMark() { 1292 if (timeStamp == 0) { 1293 timeStamp = System.currentTimeMillis(); 1294 } 1295 } 1296 1297 /** 1298 * @return if after being marked, the Connection is still writing 1299 */ 1300 @Override 1301 public boolean isBlocked() { 1302 return blocked; 1303 } 1304 1305 /** 1306 * @return true if the Connection is connected 1307 */ 1308 @Override 1309 public boolean isConnected() { 1310 return connected; 1311 } 1312 1313 /** 1314 * @param blocked The blocked to set. 1315 */ 1316 public void setBlocked(boolean blocked) { 1317 this.blocked = blocked; 1318 } 1319 1320 /** 1321 * @param connected The connected to set. 1322 */ 1323 public void setConnected(boolean connected) { 1324 this.connected = connected; 1325 } 1326 1327 /** 1328 * @return true if the Connection is active 1329 */ 1330 @Override 1331 public boolean isActive() { 1332 return active; 1333 } 1334 1335 /** 1336 * @param active The active to set. 1337 */ 1338 public void setActive(boolean active) { 1339 this.active = active; 1340 } 1341 1342 /** 1343 * @return true if the Connection is starting 1344 */ 1345 public boolean isStarting() { 1346 return status.get() == STARTING; 1347 } 1348 1349 @Override 1350 public synchronized boolean isNetworkConnection() { 1351 return networkConnection; 1352 } 1353 1354 @Override 1355 public boolean isFaultTolerantConnection() { 1356 return this.faultTolerantConnection; 1357 } 1358 1359 /** 1360 * @return true if the Connection needs to stop 1361 */ 1362 public boolean isPendingStop() { 1363 return status.get() == PENDING_STOP; 1364 } 1365 1366 @Override 1367 public Response processBrokerInfo(BrokerInfo info) { 1368 if (info.isSlaveBroker()) { 1369 LOG.error(" Slave Brokers are no longer supported - slave trying to attach is: {}", info.getBrokerName()); 1370 } else if (info.isNetworkConnection() && info.isDuplexConnection()) { 1371 // so this TransportConnection is the rear end of a network bridge 1372 // We have been requested to create a two way pipe ... 1373 try { 1374 Properties properties = MarshallingSupport.stringToProperties(info.getNetworkProperties()); 1375 Map<String, String> props = createMap(properties); 1376 NetworkBridgeConfiguration config = new NetworkBridgeConfiguration(); 1377 IntrospectionSupport.setProperties(config, props, ""); 1378 config.setBrokerName(broker.getBrokerName()); 1379 1380 // check for existing duplex connection hanging about 1381 1382 // We first look if existing network connection already exists for the same broker Id and network connector name 1383 // It's possible in case of brief network fault to have this transport connector side of the connection always active 1384 // and the duplex network connector side wanting to open a new one 1385 // In this case, the old connection must be broken 1386 String duplexNetworkConnectorId = config.getName() + "@" + info.getBrokerId(); 1387 CopyOnWriteArrayList<TransportConnection> connections = this.connector.getConnections(); 1388 synchronized (connections) { 1389 for (Iterator<TransportConnection> iter = connections.iterator(); iter.hasNext(); ) { 1390 TransportConnection c = iter.next(); 1391 if ((c != this) && (duplexNetworkConnectorId.equals(c.getDuplexNetworkConnectorId()))) { 1392 LOG.warn("Stopping an existing active duplex connection [{}] for network connector ({}).", c, duplexNetworkConnectorId); 1393 c.stopAsync(); 1394 // better to wait for a bit rather than get connection id already in use and failure to start new bridge 1395 c.getStopped().await(1, TimeUnit.SECONDS); 1396 } 1397 } 1398 setDuplexNetworkConnectorId(duplexNetworkConnectorId); 1399 } 1400 Transport localTransport = NetworkBridgeFactory.createLocalTransport(config, broker.getVmConnectorURI()); 1401 Transport remoteBridgeTransport = transport; 1402 if (! (remoteBridgeTransport instanceof ResponseCorrelator)) { 1403 // the vm transport case is already wrapped 1404 remoteBridgeTransport = new ResponseCorrelator(remoteBridgeTransport); 1405 } 1406 String duplexName = localTransport.toString(); 1407 if (duplexName.contains("#")) { 1408 duplexName = duplexName.substring(duplexName.lastIndexOf("#")); 1409 } 1410 MBeanNetworkListener listener = new MBeanNetworkListener(brokerService, config, brokerService.createDuplexNetworkConnectorObjectName(duplexName)); 1411 listener.setCreatedByDuplex(true); 1412 duplexBridge = NetworkBridgeFactory.createBridge(config, localTransport, remoteBridgeTransport, listener); 1413 duplexBridge.setBrokerService(brokerService); 1414 // now turn duplex off this side 1415 info.setDuplexConnection(false); 1416 duplexBridge.setCreatedByDuplex(true); 1417 duplexBridge.duplexStart(this, brokerInfo, info); 1418 LOG.info("Started responder end of duplex bridge {}", duplexNetworkConnectorId); 1419 return null; 1420 } catch (TransportDisposedIOException e) { 1421 LOG.warn("Duplex bridge {} was stopped before it was correctly started.", duplexNetworkConnectorId); 1422 return null; 1423 } catch (Exception e) { 1424 LOG.error("Failed to create responder end of duplex network bridge {}", duplexNetworkConnectorId, e); 1425 return null; 1426 } 1427 } 1428 // We only expect to get one broker info command per connection 1429 if (this.brokerInfo != null) { 1430 LOG.warn("Unexpected extra broker info command received: {}", info); 1431 } 1432 this.brokerInfo = info; 1433 networkConnection = true; 1434 List<TransportConnectionState> connectionStates = listConnectionStates(); 1435 for (TransportConnectionState cs : connectionStates) { 1436 cs.getContext().setNetworkConnection(true); 1437 } 1438 return null; 1439 } 1440 1441 @SuppressWarnings({"unchecked", "rawtypes"}) 1442 private HashMap<String, String> createMap(Properties properties) { 1443 return new HashMap(properties); 1444 } 1445 1446 protected void dispatch(Command command) throws IOException { 1447 try { 1448 setMarkedCandidate(true); 1449 transport.oneway(command); 1450 } finally { 1451 setMarkedCandidate(false); 1452 } 1453 } 1454 1455 @Override 1456 public String getRemoteAddress() { 1457 return transport.getRemoteAddress(); 1458 } 1459 1460 public Transport getTransport() { 1461 return transport; 1462 } 1463 1464 @Override 1465 public String getConnectionId() { 1466 List<TransportConnectionState> connectionStates = listConnectionStates(); 1467 for (TransportConnectionState cs : connectionStates) { 1468 if (cs.getInfo().getClientId() != null) { 1469 return cs.getInfo().getClientId(); 1470 } 1471 return cs.getInfo().getConnectionId().toString(); 1472 } 1473 return null; 1474 } 1475 1476 @Override 1477 public void updateClient(ConnectionControl control) { 1478 if (isActive() && isBlocked() == false && isFaultTolerantConnection() && this.wireFormatInfo != null 1479 && this.wireFormatInfo.getVersion() >= 6) { 1480 dispatchAsync(control); 1481 } 1482 } 1483 1484 public ProducerBrokerExchange getProducerBrokerExchangeIfExists(ProducerInfo producerInfo){ 1485 ProducerBrokerExchange result = null; 1486 if (producerInfo != null && producerInfo.getProducerId() != null){ 1487 synchronized (producerExchanges){ 1488 result = producerExchanges.get(producerInfo.getProducerId()); 1489 } 1490 } 1491 return result; 1492 } 1493 1494 private ProducerBrokerExchange getProducerBrokerExchange(ProducerId id) throws IOException { 1495 ProducerBrokerExchange result = producerExchanges.get(id); 1496 if (result == null) { 1497 synchronized (producerExchanges) { 1498 result = new ProducerBrokerExchange(); 1499 TransportConnectionState state = lookupConnectionState(id); 1500 context = state.getContext(); 1501 result.setConnectionContext(context); 1502 if (context.isReconnect() || (context.isNetworkConnection() && connector.isAuditNetworkProducers())) { 1503 result.setLastStoredSequenceId(brokerService.getPersistenceAdapter().getLastProducerSequenceId(id)); 1504 } 1505 SessionState ss = state.getSessionState(id.getParentId()); 1506 if (ss != null) { 1507 result.setProducerState(ss.getProducerState(id)); 1508 ProducerState producerState = ss.getProducerState(id); 1509 if (producerState != null && producerState.getInfo() != null) { 1510 ProducerInfo info = producerState.getInfo(); 1511 result.setMutable(info.getDestination() == null || info.getDestination().isComposite()); 1512 } 1513 } 1514 producerExchanges.put(id, result); 1515 } 1516 } else { 1517 context = result.getConnectionContext(); 1518 } 1519 return result; 1520 } 1521 1522 private void removeProducerBrokerExchange(ProducerId id) { 1523 synchronized (producerExchanges) { 1524 producerExchanges.remove(id); 1525 } 1526 } 1527 1528 private ConsumerBrokerExchange getConsumerBrokerExchange(ConsumerId id) { 1529 ConsumerBrokerExchange result = consumerExchanges.get(id); 1530 return result; 1531 } 1532 1533 private ConsumerBrokerExchange addConsumerBrokerExchange(TransportConnectionState connectionState, ConsumerId id) { 1534 ConsumerBrokerExchange result = consumerExchanges.get(id); 1535 if (result == null) { 1536 synchronized (consumerExchanges) { 1537 result = new ConsumerBrokerExchange(); 1538 context = connectionState.getContext(); 1539 result.setConnectionContext(context); 1540 SessionState ss = connectionState.getSessionState(id.getParentId()); 1541 if (ss != null) { 1542 ConsumerState cs = ss.getConsumerState(id); 1543 if (cs != null) { 1544 ConsumerInfo info = cs.getInfo(); 1545 if (info != null) { 1546 if (info.getDestination() != null && info.getDestination().isPattern()) { 1547 result.setWildcard(true); 1548 } 1549 } 1550 } 1551 } 1552 consumerExchanges.put(id, result); 1553 } 1554 } 1555 return result; 1556 } 1557 1558 private void removeConsumerBrokerExchange(ConsumerId id) { 1559 synchronized (consumerExchanges) { 1560 consumerExchanges.remove(id); 1561 } 1562 } 1563 1564 public int getProtocolVersion() { 1565 return protocolVersion.get(); 1566 } 1567 1568 @Override 1569 public Response processControlCommand(ControlCommand command) throws Exception { 1570 return null; 1571 } 1572 1573 @Override 1574 public Response processMessageDispatch(MessageDispatch dispatch) throws Exception { 1575 return null; 1576 } 1577 1578 @Override 1579 public Response processConnectionControl(ConnectionControl control) throws Exception { 1580 if (control != null) { 1581 faultTolerantConnection = control.isFaultTolerant(); 1582 } 1583 return null; 1584 } 1585 1586 @Override 1587 public Response processConnectionError(ConnectionError error) throws Exception { 1588 return null; 1589 } 1590 1591 @Override 1592 public Response processConsumerControl(ConsumerControl control) throws Exception { 1593 ConsumerBrokerExchange consumerExchange = getConsumerBrokerExchange(control.getConsumerId()); 1594 broker.processConsumerControl(consumerExchange, control); 1595 return null; 1596 } 1597 1598 protected synchronized TransportConnectionState registerConnectionState(ConnectionId connectionId, 1599 TransportConnectionState state) { 1600 TransportConnectionState cs = null; 1601 if (!connectionStateRegister.isEmpty() && !connectionStateRegister.doesHandleMultipleConnectionStates()) { 1602 // swap implementations 1603 TransportConnectionStateRegister newRegister = new MapTransportConnectionStateRegister(); 1604 newRegister.intialize(connectionStateRegister); 1605 connectionStateRegister = newRegister; 1606 } 1607 cs = connectionStateRegister.registerConnectionState(connectionId, state); 1608 return cs; 1609 } 1610 1611 protected synchronized TransportConnectionState unregisterConnectionState(ConnectionId connectionId) { 1612 return connectionStateRegister.unregisterConnectionState(connectionId); 1613 } 1614 1615 protected synchronized List<TransportConnectionState> listConnectionStates() { 1616 return connectionStateRegister.listConnectionStates(); 1617 } 1618 1619 protected synchronized TransportConnectionState lookupConnectionState(String connectionId) { 1620 return connectionStateRegister.lookupConnectionState(connectionId); 1621 } 1622 1623 protected synchronized TransportConnectionState lookupConnectionState(ConsumerId id) { 1624 return connectionStateRegister.lookupConnectionState(id); 1625 } 1626 1627 protected synchronized TransportConnectionState lookupConnectionState(ProducerId id) { 1628 return connectionStateRegister.lookupConnectionState(id); 1629 } 1630 1631 protected synchronized TransportConnectionState lookupConnectionState(SessionId id) { 1632 return connectionStateRegister.lookupConnectionState(id); 1633 } 1634 1635 // public only for testing 1636 public synchronized TransportConnectionState lookupConnectionState(ConnectionId connectionId) { 1637 return connectionStateRegister.lookupConnectionState(connectionId); 1638 } 1639 1640 protected synchronized void setDuplexNetworkConnectorId(String duplexNetworkConnectorId) { 1641 this.duplexNetworkConnectorId = duplexNetworkConnectorId; 1642 } 1643 1644 protected synchronized String getDuplexNetworkConnectorId() { 1645 return this.duplexNetworkConnectorId; 1646 } 1647 1648 public boolean isStopping() { 1649 return stopping.get(); 1650 } 1651 1652 protected CountDownLatch getStopped() { 1653 return stopped; 1654 } 1655 1656 private int getProducerCount(ConnectionId connectionId) { 1657 int result = 0; 1658 TransportConnectionState cs = lookupConnectionState(connectionId); 1659 if (cs != null) { 1660 for (SessionId sessionId : cs.getSessionIds()) { 1661 SessionState sessionState = cs.getSessionState(sessionId); 1662 if (sessionState != null) { 1663 result += sessionState.getProducerIds().size(); 1664 } 1665 } 1666 } 1667 return result; 1668 } 1669 1670 private int getConsumerCount(ConnectionId connectionId) { 1671 int result = 0; 1672 TransportConnectionState cs = lookupConnectionState(connectionId); 1673 if (cs != null) { 1674 for (SessionId sessionId : cs.getSessionIds()) { 1675 SessionState sessionState = cs.getSessionState(sessionId); 1676 if (sessionState != null) { 1677 result += sessionState.getConsumerIds().size(); 1678 } 1679 } 1680 } 1681 return result; 1682 } 1683 1684 public WireFormatInfo getRemoteWireFormatInfo() { 1685 return wireFormatInfo; 1686 } 1687}