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; 018 019import java.io.IOException; 020import java.net.URI; 021import java.net.URISyntaxException; 022import java.util.*; 023import java.util.concurrent.ConcurrentHashMap; 024import java.util.concurrent.ConcurrentMap; 025import java.util.concurrent.CopyOnWriteArrayList; 026import java.util.concurrent.CountDownLatch; 027import java.util.concurrent.LinkedBlockingQueue; 028import java.util.concurrent.RejectedExecutionHandler; 029import java.util.concurrent.ThreadFactory; 030import java.util.concurrent.ThreadPoolExecutor; 031import java.util.concurrent.TimeUnit; 032import java.util.concurrent.atomic.AtomicBoolean; 033import java.util.concurrent.atomic.AtomicInteger; 034 035import javax.jms.Connection; 036import javax.jms.ConnectionConsumer; 037import javax.jms.ConnectionMetaData; 038import javax.jms.Destination; 039import javax.jms.ExceptionListener; 040import javax.jms.IllegalStateException; 041import javax.jms.InvalidDestinationException; 042import javax.jms.JMSException; 043import javax.jms.Queue; 044import javax.jms.QueueConnection; 045import javax.jms.QueueSession; 046import javax.jms.ServerSessionPool; 047import javax.jms.Session; 048import javax.jms.Topic; 049import javax.jms.TopicConnection; 050import javax.jms.TopicSession; 051import javax.jms.XAConnection; 052 053import org.apache.activemq.advisory.DestinationSource; 054import org.apache.activemq.blob.BlobTransferPolicy; 055import org.apache.activemq.broker.region.policy.RedeliveryPolicyMap; 056import org.apache.activemq.command.ActiveMQDestination; 057import org.apache.activemq.command.ActiveMQMessage; 058import org.apache.activemq.command.ActiveMQTempDestination; 059import org.apache.activemq.command.ActiveMQTempQueue; 060import org.apache.activemq.command.ActiveMQTempTopic; 061import org.apache.activemq.command.BrokerInfo; 062import org.apache.activemq.command.Command; 063import org.apache.activemq.command.CommandTypes; 064import org.apache.activemq.command.ConnectionControl; 065import org.apache.activemq.command.ConnectionError; 066import org.apache.activemq.command.ConnectionId; 067import org.apache.activemq.command.ConnectionInfo; 068import org.apache.activemq.command.ConsumerControl; 069import org.apache.activemq.command.ConsumerId; 070import org.apache.activemq.command.ConsumerInfo; 071import org.apache.activemq.command.ControlCommand; 072import org.apache.activemq.command.DestinationInfo; 073import org.apache.activemq.command.ExceptionResponse; 074import org.apache.activemq.command.Message; 075import org.apache.activemq.command.MessageDispatch; 076import org.apache.activemq.command.MessageId; 077import org.apache.activemq.command.ProducerAck; 078import org.apache.activemq.command.ProducerId; 079import org.apache.activemq.command.RemoveInfo; 080import org.apache.activemq.command.RemoveSubscriptionInfo; 081import org.apache.activemq.command.Response; 082import org.apache.activemq.command.SessionId; 083import org.apache.activemq.command.ShutdownInfo; 084import org.apache.activemq.command.WireFormatInfo; 085import org.apache.activemq.management.JMSConnectionStatsImpl; 086import org.apache.activemq.management.JMSStatsImpl; 087import org.apache.activemq.management.StatsCapable; 088import org.apache.activemq.management.StatsImpl; 089import org.apache.activemq.state.CommandVisitorAdapter; 090import org.apache.activemq.thread.Scheduler; 091import org.apache.activemq.thread.TaskRunnerFactory; 092import org.apache.activemq.transport.FutureResponse; 093import org.apache.activemq.transport.RequestTimedOutIOException; 094import org.apache.activemq.transport.ResponseCallback; 095import org.apache.activemq.transport.Transport; 096import org.apache.activemq.transport.TransportListener; 097import org.apache.activemq.transport.failover.FailoverTransport; 098import org.apache.activemq.util.IdGenerator; 099import org.apache.activemq.util.IntrospectionSupport; 100import org.apache.activemq.util.JMSExceptionSupport; 101import org.apache.activemq.util.LongSequenceGenerator; 102import org.apache.activemq.util.ServiceSupport; 103import org.apache.activemq.util.ThreadPoolUtils; 104import org.slf4j.Logger; 105import org.slf4j.LoggerFactory; 106 107public class ActiveMQConnection implements Connection, TopicConnection, QueueConnection, StatsCapable, Closeable, TransportListener, EnhancedConnection { 108 109 public static final String DEFAULT_USER = ActiveMQConnectionFactory.DEFAULT_USER; 110 public static final String DEFAULT_PASSWORD = ActiveMQConnectionFactory.DEFAULT_PASSWORD; 111 public static final String DEFAULT_BROKER_URL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL; 112 public static int DEFAULT_THREAD_POOL_SIZE = 1000; 113 114 private static final Logger LOG = LoggerFactory.getLogger(ActiveMQConnection.class); 115 116 public final ConcurrentMap<ActiveMQTempDestination, ActiveMQTempDestination> activeTempDestinations = new ConcurrentHashMap<ActiveMQTempDestination, ActiveMQTempDestination>(); 117 118 protected boolean dispatchAsync=true; 119 protected boolean alwaysSessionAsync = true; 120 121 private TaskRunnerFactory sessionTaskRunner; 122 private final ThreadPoolExecutor executor; 123 124 // Connection state variables 125 private final ConnectionInfo info; 126 private ExceptionListener exceptionListener; 127 private ClientInternalExceptionListener clientInternalExceptionListener; 128 private boolean clientIDSet; 129 private boolean isConnectionInfoSentToBroker; 130 private boolean userSpecifiedClientID; 131 132 // Configuration options variables 133 private ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy(); 134 private BlobTransferPolicy blobTransferPolicy; 135 private RedeliveryPolicyMap redeliveryPolicyMap; 136 private MessageTransformer transformer; 137 138 private boolean disableTimeStampsByDefault; 139 private boolean optimizedMessageDispatch = true; 140 private boolean copyMessageOnSend = true; 141 private boolean useCompression; 142 private boolean objectMessageSerializationDefered; 143 private boolean useAsyncSend; 144 private boolean optimizeAcknowledge; 145 private long optimizeAcknowledgeTimeOut = 0; 146 private long optimizedAckScheduledAckInterval = 0; 147 private boolean nestedMapAndListEnabled = true; 148 private boolean useRetroactiveConsumer; 149 private boolean exclusiveConsumer; 150 private boolean alwaysSyncSend; 151 private int closeTimeout = 15000; 152 private boolean watchTopicAdvisories = true; 153 private long warnAboutUnstartedConnectionTimeout = 500L; 154 private int sendTimeout =0; 155 private boolean sendAcksAsync=true; 156 private boolean checkForDuplicates = true; 157 private boolean queueOnlyConnection = false; 158 private boolean consumerExpiryCheckEnabled = true; 159 160 private final Transport transport; 161 private final IdGenerator clientIdGenerator; 162 private final JMSStatsImpl factoryStats; 163 private final JMSConnectionStatsImpl stats; 164 165 private final AtomicBoolean started = new AtomicBoolean(false); 166 private final AtomicBoolean closing = new AtomicBoolean(false); 167 private final AtomicBoolean closed = new AtomicBoolean(false); 168 private final AtomicBoolean transportFailed = new AtomicBoolean(false); 169 private final CopyOnWriteArrayList<ActiveMQSession> sessions = new CopyOnWriteArrayList<ActiveMQSession>(); 170 private final CopyOnWriteArrayList<ActiveMQConnectionConsumer> connectionConsumers = new CopyOnWriteArrayList<ActiveMQConnectionConsumer>(); 171 private final CopyOnWriteArrayList<TransportListener> transportListeners = new CopyOnWriteArrayList<TransportListener>(); 172 173 // Maps ConsumerIds to ActiveMQConsumer objects 174 private final ConcurrentMap<ConsumerId, ActiveMQDispatcher> dispatchers = new ConcurrentHashMap<ConsumerId, ActiveMQDispatcher>(); 175 private final ConcurrentMap<ProducerId, ActiveMQMessageProducer> producers = new ConcurrentHashMap<ProducerId, ActiveMQMessageProducer>(); 176 private final LongSequenceGenerator sessionIdGenerator = new LongSequenceGenerator(); 177 private final SessionId connectionSessionId; 178 private final LongSequenceGenerator consumerIdGenerator = new LongSequenceGenerator(); 179 private final LongSequenceGenerator tempDestinationIdGenerator = new LongSequenceGenerator(); 180 private final LongSequenceGenerator localTransactionIdGenerator = new LongSequenceGenerator(); 181 182 private AdvisoryConsumer advisoryConsumer; 183 private final CountDownLatch brokerInfoReceived = new CountDownLatch(1); 184 private BrokerInfo brokerInfo; 185 private IOException firstFailureError; 186 private int producerWindowSize = ActiveMQConnectionFactory.DEFAULT_PRODUCER_WINDOW_SIZE; 187 188 // Assume that protocol is the latest. Change to the actual protocol 189 // version when a WireFormatInfo is received. 190 private final AtomicInteger protocolVersion = new AtomicInteger(CommandTypes.PROTOCOL_VERSION); 191 private final long timeCreated; 192 private final ConnectionAudit connectionAudit = new ConnectionAudit(); 193 private DestinationSource destinationSource; 194 private final Object ensureConnectionInfoSentMutex = new Object(); 195 private boolean useDedicatedTaskRunner; 196 protected AtomicInteger transportInterruptionProcessingComplete = new AtomicInteger(0); 197 private long consumerFailoverRedeliveryWaitPeriod; 198 private Scheduler scheduler; 199 private boolean messagePrioritySupported = false; 200 private boolean transactedIndividualAck = false; 201 private boolean nonBlockingRedelivery = false; 202 private boolean rmIdFromConnectionId = false; 203 204 private int maxThreadPoolSize = DEFAULT_THREAD_POOL_SIZE; 205 private RejectedExecutionHandler rejectedTaskHandler = null; 206 207 private List<String> trustedPackages = new ArrayList<String>(); 208 private boolean trustAllPackages = false; 209 private int connectResponseTimeout; 210 211 /** 212 * Construct an <code>ActiveMQConnection</code> 213 * 214 * @param transport 215 * @param factoryStats 216 * @throws Exception 217 */ 218 protected ActiveMQConnection(final Transport transport, IdGenerator clientIdGenerator, IdGenerator connectionIdGenerator, JMSStatsImpl factoryStats) throws Exception { 219 220 this.transport = transport; 221 this.clientIdGenerator = clientIdGenerator; 222 this.factoryStats = factoryStats; 223 224 // Configure a single threaded executor who's core thread can timeout if 225 // idle 226 executor = new ThreadPoolExecutor(1, 1, 5, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(), new ThreadFactory() { 227 @Override 228 public Thread newThread(Runnable r) { 229 Thread thread = new Thread(r, "ActiveMQ Connection Executor: " + transport); 230 //Don't make these daemon threads - see https://issues.apache.org/jira/browse/AMQ-796 231 //thread.setDaemon(true); 232 return thread; 233 } 234 }); 235 // asyncConnectionThread.allowCoreThreadTimeOut(true); 236 String uniqueId = connectionIdGenerator.generateId(); 237 this.info = new ConnectionInfo(new ConnectionId(uniqueId)); 238 this.info.setManageable(true); 239 this.info.setFaultTolerant(transport.isFaultTolerant()); 240 this.connectionSessionId = new SessionId(info.getConnectionId(), -1); 241 242 this.transport.setTransportListener(this); 243 244 this.stats = new JMSConnectionStatsImpl(sessions, this instanceof XAConnection); 245 this.factoryStats.addConnection(this); 246 this.timeCreated = System.currentTimeMillis(); 247 this.connectionAudit.setCheckForDuplicates(transport.isFaultTolerant()); 248 } 249 250 protected void setUserName(String userName) { 251 this.info.setUserName(userName); 252 } 253 254 protected void setPassword(String password) { 255 this.info.setPassword(password); 256 } 257 258 /** 259 * A static helper method to create a new connection 260 * 261 * @return an ActiveMQConnection 262 * @throws JMSException 263 */ 264 public static ActiveMQConnection makeConnection() throws JMSException { 265 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(); 266 return (ActiveMQConnection)factory.createConnection(); 267 } 268 269 /** 270 * A static helper method to create a new connection 271 * 272 * @param uri 273 * @return and ActiveMQConnection 274 * @throws JMSException 275 */ 276 public static ActiveMQConnection makeConnection(String uri) throws JMSException, URISyntaxException { 277 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(uri); 278 return (ActiveMQConnection)factory.createConnection(); 279 } 280 281 /** 282 * A static helper method to create a new connection 283 * 284 * @param user 285 * @param password 286 * @param uri 287 * @return an ActiveMQConnection 288 * @throws JMSException 289 */ 290 public static ActiveMQConnection makeConnection(String user, String password, String uri) throws JMSException, URISyntaxException { 291 ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(user, password, new URI(uri)); 292 return (ActiveMQConnection)factory.createConnection(); 293 } 294 295 /** 296 * @return a number unique for this connection 297 */ 298 public JMSConnectionStatsImpl getConnectionStats() { 299 return stats; 300 } 301 302 /** 303 * Creates a <CODE>Session</CODE> object. 304 * 305 * @param transacted indicates whether the session is transacted 306 * @param acknowledgeMode indicates whether the consumer or the client will 307 * acknowledge any messages it receives; ignored if the 308 * session is transacted. Legal values are 309 * <code>Session.AUTO_ACKNOWLEDGE</code>, 310 * <code>Session.CLIENT_ACKNOWLEDGE</code>, and 311 * <code>Session.DUPS_OK_ACKNOWLEDGE</code>. 312 * @return a newly created session 313 * @throws JMSException if the <CODE>Connection</CODE> object fails to 314 * create a session due to some internal error or lack of 315 * support for the specific transaction and acknowledgement 316 * mode. 317 * @see Session#AUTO_ACKNOWLEDGE 318 * @see Session#CLIENT_ACKNOWLEDGE 319 * @see Session#DUPS_OK_ACKNOWLEDGE 320 * @since 1.1 321 */ 322 @Override 323 public Session createSession(boolean transacted, int acknowledgeMode) throws JMSException { 324 checkClosedOrFailed(); 325 ensureConnectionInfoSent(); 326 if(!transacted) { 327 if (acknowledgeMode==Session.SESSION_TRANSACTED) { 328 throw new JMSException("acknowledgeMode SESSION_TRANSACTED cannot be used for an non-transacted Session"); 329 } else if (acknowledgeMode < Session.SESSION_TRANSACTED || acknowledgeMode > ActiveMQSession.MAX_ACK_CONSTANT) { 330 throw new JMSException("invalid acknowledgeMode: " + acknowledgeMode + ". Valid values are Session.AUTO_ACKNOWLEDGE (1), " + 331 "Session.CLIENT_ACKNOWLEDGE (2), Session.DUPS_OK_ACKNOWLEDGE (3), ActiveMQSession.INDIVIDUAL_ACKNOWLEDGE (4) or for transacted sessions Session.SESSION_TRANSACTED (0)"); 332 } 333 } 334 return new ActiveMQSession(this, getNextSessionId(), transacted ? Session.SESSION_TRANSACTED : (acknowledgeMode == Session.SESSION_TRANSACTED 335 ? Session.AUTO_ACKNOWLEDGE : acknowledgeMode), isDispatchAsync(), isAlwaysSessionAsync()); 336 } 337 338 /** 339 * @return sessionId 340 */ 341 protected SessionId getNextSessionId() { 342 return new SessionId(info.getConnectionId(), sessionIdGenerator.getNextSequenceId()); 343 } 344 345 /** 346 * Gets the client identifier for this connection. 347 * <P> 348 * This value is specific to the JMS provider. It is either preconfigured by 349 * an administrator in a <CODE> ConnectionFactory</CODE> object or assigned 350 * dynamically by the application by calling the <code>setClientID</code> 351 * method. 352 * 353 * @return the unique client identifier 354 * @throws JMSException if the JMS provider fails to return the client ID 355 * for this connection due to some internal error. 356 */ 357 @Override 358 public String getClientID() throws JMSException { 359 checkClosedOrFailed(); 360 return this.info.getClientId(); 361 } 362 363 /** 364 * Sets the client identifier for this connection. 365 * <P> 366 * The preferred way to assign a JMS client's client identifier is for it to 367 * be configured in a client-specific <CODE>ConnectionFactory</CODE> 368 * object and transparently assigned to the <CODE>Connection</CODE> object 369 * it creates. 370 * <P> 371 * Alternatively, a client can set a connection's client identifier using a 372 * provider-specific value. The facility to set a connection's client 373 * identifier explicitly is not a mechanism for overriding the identifier 374 * that has been administratively configured. It is provided for the case 375 * where no administratively specified identifier exists. If one does exist, 376 * an attempt to change it by setting it must throw an 377 * <CODE>IllegalStateException</CODE>. If a client sets the client 378 * identifier explicitly, it must do so immediately after it creates the 379 * connection and before any other action on the connection is taken. After 380 * this point, setting the client identifier is a programming error that 381 * should throw an <CODE>IllegalStateException</CODE>. 382 * <P> 383 * The purpose of the client identifier is to associate a connection and its 384 * objects with a state maintained on behalf of the client by a provider. 385 * The only such state identified by the JMS API is that required to support 386 * durable subscriptions. 387 * <P> 388 * If another connection with the same <code>clientID</code> is already 389 * running when this method is called, the JMS provider should detect the 390 * duplicate ID and throw an <CODE>InvalidClientIDException</CODE>. 391 * 392 * @param newClientID the unique client identifier 393 * @throws JMSException if the JMS provider fails to set the client ID for 394 * this connection due to some internal error. 395 * @throws javax.jms.InvalidClientIDException if the JMS client specifies an 396 * invalid or duplicate client ID. 397 * @throws javax.jms.IllegalStateException if the JMS client attempts to set 398 * a connection's client ID at the wrong time or when it has 399 * been administratively configured. 400 */ 401 @Override 402 public void setClientID(String newClientID) throws JMSException { 403 checkClosedOrFailed(); 404 405 if (this.clientIDSet) { 406 throw new IllegalStateException("The clientID has already been set"); 407 } 408 409 if (this.isConnectionInfoSentToBroker) { 410 throw new IllegalStateException("Setting clientID on a used Connection is not allowed"); 411 } 412 413 this.info.setClientId(newClientID); 414 this.userSpecifiedClientID = true; 415 ensureConnectionInfoSent(); 416 } 417 418 /** 419 * Sets the default client id that the connection will use if explicitly not 420 * set with the setClientId() call. 421 */ 422 public void setDefaultClientID(String clientID) throws JMSException { 423 this.info.setClientId(clientID); 424 this.userSpecifiedClientID = true; 425 } 426 427 /** 428 * Gets the metadata for this connection. 429 * 430 * @return the connection metadata 431 * @throws JMSException if the JMS provider fails to get the connection 432 * metadata for this connection. 433 * @see javax.jms.ConnectionMetaData 434 */ 435 @Override 436 public ConnectionMetaData getMetaData() throws JMSException { 437 checkClosedOrFailed(); 438 return ActiveMQConnectionMetaData.INSTANCE; 439 } 440 441 /** 442 * Gets the <CODE>ExceptionListener</CODE> object for this connection. Not 443 * every <CODE>Connection</CODE> has an <CODE>ExceptionListener</CODE> 444 * associated with it. 445 * 446 * @return the <CODE>ExceptionListener</CODE> for this connection, or 447 * null, if no <CODE>ExceptionListener</CODE> is associated with 448 * this connection. 449 * @throws JMSException if the JMS provider fails to get the 450 * <CODE>ExceptionListener</CODE> for this connection. 451 * @see javax.jms.Connection#setExceptionListener(ExceptionListener) 452 */ 453 @Override 454 public ExceptionListener getExceptionListener() throws JMSException { 455 checkClosedOrFailed(); 456 return this.exceptionListener; 457 } 458 459 /** 460 * Sets an exception listener for this connection. 461 * <P> 462 * If a JMS provider detects a serious problem with a connection, it informs 463 * the connection's <CODE> ExceptionListener</CODE>, if one has been 464 * registered. It does this by calling the listener's <CODE>onException 465 * </CODE> 466 * method, passing it a <CODE>JMSException</CODE> object describing the 467 * problem. 468 * <P> 469 * An exception listener allows a client to be notified of a problem 470 * asynchronously. Some connections only consume messages, so they would 471 * have no other way to learn their connection has failed. 472 * <P> 473 * A connection serializes execution of its <CODE>ExceptionListener</CODE>. 474 * <P> 475 * A JMS provider should attempt to resolve connection problems itself 476 * before it notifies the client of them. 477 * 478 * @param listener the exception listener 479 * @throws JMSException if the JMS provider fails to set the exception 480 * listener for this connection. 481 */ 482 @Override 483 public void setExceptionListener(ExceptionListener listener) throws JMSException { 484 checkClosedOrFailed(); 485 this.exceptionListener = listener; 486 } 487 488 /** 489 * Gets the <code>ClientInternalExceptionListener</code> object for this connection. 490 * Not every <CODE>ActiveMQConnectionn</CODE> has a <CODE>ClientInternalExceptionListener</CODE> 491 * associated with it. 492 * 493 * @return the listener or <code>null</code> if no listener is registered with the connection. 494 */ 495 public ClientInternalExceptionListener getClientInternalExceptionListener() { 496 return clientInternalExceptionListener; 497 } 498 499 /** 500 * Sets a client internal exception listener for this connection. 501 * The connection will notify the listener, if one has been registered, of exceptions thrown by container components 502 * (e.g. an EJB container in case of Message Driven Beans) during asynchronous processing of a message. 503 * It does this by calling the listener's <code>onException()</code> method passing it a <code>Throwable</code> 504 * describing the problem. 505 * 506 * @param listener the exception listener 507 */ 508 public void setClientInternalExceptionListener(ClientInternalExceptionListener listener) { 509 this.clientInternalExceptionListener = listener; 510 } 511 512 /** 513 * Starts (or restarts) a connection's delivery of incoming messages. A call 514 * to <CODE>start</CODE> on a connection that has already been started is 515 * ignored. 516 * 517 * @throws JMSException if the JMS provider fails to start message delivery 518 * due to some internal error. 519 * @see javax.jms.Connection#stop() 520 */ 521 @Override 522 public void start() throws JMSException { 523 checkClosedOrFailed(); 524 ensureConnectionInfoSent(); 525 if (started.compareAndSet(false, true)) { 526 for (Iterator<ActiveMQSession> i = sessions.iterator(); i.hasNext();) { 527 ActiveMQSession session = i.next(); 528 session.start(); 529 } 530 } 531 } 532 533 /** 534 * Temporarily stops a connection's delivery of incoming messages. Delivery 535 * can be restarted using the connection's <CODE>start</CODE> method. When 536 * the connection is stopped, delivery to all the connection's message 537 * consumers is inhibited: synchronous receives block, and messages are not 538 * delivered to message listeners. 539 * <P> 540 * This call blocks until receives and/or message listeners in progress have 541 * completed. 542 * <P> 543 * Stopping a connection has no effect on its ability to send messages. A 544 * call to <CODE>stop</CODE> on a connection that has already been stopped 545 * is ignored. 546 * <P> 547 * A call to <CODE>stop</CODE> must not return until delivery of messages 548 * has paused. This means that a client can rely on the fact that none of 549 * its message listeners will be called and that all threads of control 550 * waiting for <CODE>receive</CODE> calls to return will not return with a 551 * message until the connection is restarted. The receive timers for a 552 * stopped connection continue to advance, so receives may time out while 553 * the connection is stopped. 554 * <P> 555 * If message listeners are running when <CODE>stop</CODE> is invoked, the 556 * <CODE>stop</CODE> call must wait until all of them have returned before 557 * it may return. While these message listeners are completing, they must 558 * have the full services of the connection available to them. 559 * 560 * @throws JMSException if the JMS provider fails to stop message delivery 561 * due to some internal error. 562 * @see javax.jms.Connection#start() 563 */ 564 @Override 565 public void stop() throws JMSException { 566 doStop(true); 567 } 568 569 /** 570 * @see #stop() 571 * @param checkClosed <tt>true</tt> to check for already closed and throw {@link java.lang.IllegalStateException} if already closed, 572 * <tt>false</tt> to skip this check 573 * @throws JMSException if the JMS provider fails to stop message delivery due to some internal error. 574 */ 575 void doStop(boolean checkClosed) throws JMSException { 576 if (checkClosed) { 577 checkClosedOrFailed(); 578 } 579 if (started.compareAndSet(true, false)) { 580 synchronized(sessions) { 581 for (Iterator<ActiveMQSession> i = sessions.iterator(); i.hasNext();) { 582 ActiveMQSession s = i.next(); 583 s.stop(); 584 } 585 } 586 } 587 } 588 589 /** 590 * Closes the connection. 591 * <P> 592 * Since a provider typically allocates significant resources outside the 593 * JVM on behalf of a connection, clients should close these resources when 594 * they are not needed. Relying on garbage collection to eventually reclaim 595 * these resources may not be timely enough. 596 * <P> 597 * There is no need to close the sessions, producers, and consumers of a 598 * closed connection. 599 * <P> 600 * Closing a connection causes all temporary destinations to be deleted. 601 * <P> 602 * When this method is invoked, it should not return until message 603 * processing has been shut down in an orderly fashion. This means that all 604 * message listeners that may have been running have returned, and that all 605 * pending receives have returned. A close terminates all pending message 606 * receives on the connection's sessions' consumers. The receives may return 607 * with a message or with null, depending on whether there was a message 608 * available at the time of the close. If one or more of the connection's 609 * sessions' message listeners is processing a message at the time when 610 * connection <CODE>close</CODE> is invoked, all the facilities of the 611 * connection and its sessions must remain available to those listeners 612 * until they return control to the JMS provider. 613 * <P> 614 * Closing a connection causes any of its sessions' transactions in progress 615 * to be rolled back. In the case where a session's work is coordinated by 616 * an external transaction manager, a session's <CODE>commit</CODE> and 617 * <CODE> rollback</CODE> methods are not used and the result of a closed 618 * session's work is determined later by the transaction manager. Closing a 619 * connection does NOT force an acknowledgment of client-acknowledged 620 * sessions. 621 * <P> 622 * Invoking the <CODE>acknowledge</CODE> method of a received message from 623 * a closed connection's session must throw an 624 * <CODE>IllegalStateException</CODE>. Closing a closed connection must 625 * NOT throw an exception. 626 * 627 * @throws JMSException if the JMS provider fails to close the connection 628 * due to some internal error. For example, a failure to 629 * release resources or to close a socket connection can 630 * cause this exception to be thrown. 631 */ 632 @Override 633 public void close() throws JMSException { 634 try { 635 // If we were running, lets stop first. 636 if (!closed.get() && !transportFailed.get()) { 637 // do not fail if already closed as according to JMS spec we must not 638 // throw exception if already closed 639 doStop(false); 640 } 641 642 synchronized (this) { 643 if (!closed.get()) { 644 closing.set(true); 645 646 if (destinationSource != null) { 647 destinationSource.stop(); 648 destinationSource = null; 649 } 650 if (advisoryConsumer != null) { 651 advisoryConsumer.dispose(); 652 advisoryConsumer = null; 653 } 654 655 Scheduler scheduler = this.scheduler; 656 if (scheduler != null) { 657 try { 658 scheduler.stop(); 659 } catch (Exception e) { 660 JMSException ex = JMSExceptionSupport.create(e); 661 throw ex; 662 } 663 } 664 665 long lastDeliveredSequenceId = -1; 666 for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();) { 667 ActiveMQSession s = i.next(); 668 s.dispose(); 669 lastDeliveredSequenceId = Math.max(lastDeliveredSequenceId, s.getLastDeliveredSequenceId()); 670 } 671 for (Iterator<ActiveMQConnectionConsumer> i = this.connectionConsumers.iterator(); i.hasNext();) { 672 ActiveMQConnectionConsumer c = i.next(); 673 c.dispose(); 674 } 675 676 this.activeTempDestinations.clear(); 677 678 try { 679 if (isConnectionInfoSentToBroker) { 680 // If we announced ourselves to the broker.. Try to let the broker 681 // know that the connection is being shutdown. 682 RemoveInfo removeCommand = info.createRemoveCommand(); 683 removeCommand.setLastDeliveredSequenceId(lastDeliveredSequenceId); 684 try { 685 syncSendPacket(removeCommand, closeTimeout); 686 } catch (JMSException e) { 687 if (e.getCause() instanceof RequestTimedOutIOException) { 688 // expected 689 } else { 690 throw e; 691 } 692 } 693 doAsyncSendPacket(new ShutdownInfo()); 694 } 695 } finally { // release anyway even if previous communication fails 696 started.set(false); 697 698 // TODO if we move the TaskRunnerFactory to the connection 699 // factory 700 // then we may need to call 701 // factory.onConnectionClose(this); 702 if (sessionTaskRunner != null) { 703 sessionTaskRunner.shutdown(); 704 } 705 closed.set(true); 706 closing.set(false); 707 } 708 } 709 } 710 } finally { 711 try { 712 if (executor != null) { 713 ThreadPoolUtils.shutdown(executor); 714 } 715 } catch (Throwable e) { 716 LOG.warn("Error shutting down thread pool: " + executor + ". This exception will be ignored.", e); 717 } 718 719 ServiceSupport.dispose(this.transport); 720 721 factoryStats.removeConnection(this); 722 } 723 } 724 725 /** 726 * Tells the broker to terminate its VM. This can be used to cleanly 727 * terminate a broker running in a standalone java process. Server must have 728 * property enable.vm.shutdown=true defined to allow this to work. 729 */ 730 // TODO : org.apache.activemq.message.BrokerAdminCommand not yet 731 // implemented. 732 /* 733 * public void terminateBrokerVM() throws JMSException { BrokerAdminCommand 734 * command = new BrokerAdminCommand(); 735 * command.setCommand(BrokerAdminCommand.SHUTDOWN_SERVER_VM); 736 * asyncSendPacket(command); } 737 */ 738 739 /** 740 * Create a durable connection consumer for this connection (optional 741 * operation). This is an expert facility not used by regular JMS clients. 742 * 743 * @param topic topic to access 744 * @param subscriptionName durable subscription name 745 * @param messageSelector only messages with properties matching the message 746 * selector expression are delivered. A value of null or an 747 * empty string indicates that there is no message selector 748 * for the message consumer. 749 * @param sessionPool the server session pool to associate with this durable 750 * connection consumer 751 * @param maxMessages the maximum number of messages that can be assigned to 752 * a server session at one time 753 * @return the durable connection consumer 754 * @throws JMSException if the <CODE>Connection</CODE> object fails to 755 * create a connection consumer due to some internal error 756 * or invalid arguments for <CODE>sessionPool</CODE> and 757 * <CODE>messageSelector</CODE>. 758 * @throws javax.jms.InvalidDestinationException if an invalid destination 759 * is specified. 760 * @throws javax.jms.InvalidSelectorException if the message selector is 761 * invalid. 762 * @see javax.jms.ConnectionConsumer 763 * @since 1.1 764 */ 765 @Override 766 public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName, String messageSelector, ServerSessionPool sessionPool, int maxMessages) 767 throws JMSException { 768 return this.createDurableConnectionConsumer(topic, subscriptionName, messageSelector, sessionPool, maxMessages, false); 769 } 770 771 /** 772 * Create a durable connection consumer for this connection (optional 773 * operation). This is an expert facility not used by regular JMS clients. 774 * 775 * @param topic topic to access 776 * @param subscriptionName durable subscription name 777 * @param messageSelector only messages with properties matching the message 778 * selector expression are delivered. A value of null or an 779 * empty string indicates that there is no message selector 780 * for the message consumer. 781 * @param sessionPool the server session pool to associate with this durable 782 * connection consumer 783 * @param maxMessages the maximum number of messages that can be assigned to 784 * a server session at one time 785 * @param noLocal set true if you want to filter out messages published 786 * locally 787 * @return the durable connection consumer 788 * @throws JMSException if the <CODE>Connection</CODE> object fails to 789 * create a connection consumer due to some internal error 790 * or invalid arguments for <CODE>sessionPool</CODE> and 791 * <CODE>messageSelector</CODE>. 792 * @throws javax.jms.InvalidDestinationException if an invalid destination 793 * is specified. 794 * @throws javax.jms.InvalidSelectorException if the message selector is 795 * invalid. 796 * @see javax.jms.ConnectionConsumer 797 * @since 1.1 798 */ 799 public ConnectionConsumer createDurableConnectionConsumer(Topic topic, String subscriptionName, String messageSelector, ServerSessionPool sessionPool, int maxMessages, 800 boolean noLocal) throws JMSException { 801 checkClosedOrFailed(); 802 803 if (queueOnlyConnection) { 804 throw new IllegalStateException("QueueConnection cannot be used to create Pub/Sub based resources."); 805 } 806 807 ensureConnectionInfoSent(); 808 SessionId sessionId = new SessionId(info.getConnectionId(), -1); 809 ConsumerInfo info = new ConsumerInfo(new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId())); 810 info.setDestination(ActiveMQMessageTransformation.transformDestination(topic)); 811 info.setSubscriptionName(subscriptionName); 812 info.setSelector(messageSelector); 813 info.setPrefetchSize(maxMessages); 814 info.setDispatchAsync(isDispatchAsync()); 815 816 // Allows the options on the destination to configure the consumerInfo 817 if (info.getDestination().getOptions() != null) { 818 Map<String, String> options = new HashMap<String, String>(info.getDestination().getOptions()); 819 IntrospectionSupport.setProperties(this.info, options, "consumer."); 820 } 821 822 return new ActiveMQConnectionConsumer(this, sessionPool, info); 823 } 824 825 // Properties 826 // ------------------------------------------------------------------------- 827 828 /** 829 * Returns true if this connection has been started 830 * 831 * @return true if this Connection is started 832 */ 833 public boolean isStarted() { 834 return started.get(); 835 } 836 837 /** 838 * Returns true if the connection is closed 839 */ 840 public boolean isClosed() { 841 return closed.get(); 842 } 843 844 /** 845 * Returns true if the connection is in the process of being closed 846 */ 847 public boolean isClosing() { 848 return closing.get(); 849 } 850 851 /** 852 * Returns true if the underlying transport has failed 853 */ 854 public boolean isTransportFailed() { 855 return transportFailed.get(); 856 } 857 858 /** 859 * @return Returns the prefetchPolicy. 860 */ 861 public ActiveMQPrefetchPolicy getPrefetchPolicy() { 862 return prefetchPolicy; 863 } 864 865 /** 866 * Sets the <a 867 * href="http://activemq.apache.org/what-is-the-prefetch-limit-for.html">prefetch 868 * policy</a> for consumers created by this connection. 869 */ 870 public void setPrefetchPolicy(ActiveMQPrefetchPolicy prefetchPolicy) { 871 this.prefetchPolicy = prefetchPolicy; 872 } 873 874 /** 875 */ 876 public Transport getTransportChannel() { 877 return transport; 878 } 879 880 /** 881 * @return Returns the clientID of the connection, forcing one to be 882 * generated if one has not yet been configured. 883 */ 884 public String getInitializedClientID() throws JMSException { 885 ensureConnectionInfoSent(); 886 return info.getClientId(); 887 } 888 889 /** 890 * @return Returns the timeStampsDisableByDefault. 891 */ 892 public boolean isDisableTimeStampsByDefault() { 893 return disableTimeStampsByDefault; 894 } 895 896 /** 897 * Sets whether or not timestamps on messages should be disabled or not. If 898 * you disable them it adds a small performance boost. 899 */ 900 public void setDisableTimeStampsByDefault(boolean timeStampsDisableByDefault) { 901 this.disableTimeStampsByDefault = timeStampsDisableByDefault; 902 } 903 904 /** 905 * @return Returns the dispatchOptimizedMessage. 906 */ 907 public boolean isOptimizedMessageDispatch() { 908 return optimizedMessageDispatch; 909 } 910 911 /** 912 * If this flag is set then an larger prefetch limit is used - only 913 * applicable for durable topic subscribers. 914 */ 915 public void setOptimizedMessageDispatch(boolean dispatchOptimizedMessage) { 916 this.optimizedMessageDispatch = dispatchOptimizedMessage; 917 } 918 919 /** 920 * @return Returns the closeTimeout. 921 */ 922 public int getCloseTimeout() { 923 return closeTimeout; 924 } 925 926 /** 927 * Sets the timeout before a close is considered complete. Normally a 928 * close() on a connection waits for confirmation from the broker; this 929 * allows that operation to timeout to save the client hanging if there is 930 * no broker 931 */ 932 public void setCloseTimeout(int closeTimeout) { 933 this.closeTimeout = closeTimeout; 934 } 935 936 /** 937 * @return ConnectionInfo 938 */ 939 public ConnectionInfo getConnectionInfo() { 940 return this.info; 941 } 942 943 public boolean isUseRetroactiveConsumer() { 944 return useRetroactiveConsumer; 945 } 946 947 /** 948 * Sets whether or not retroactive consumers are enabled. Retroactive 949 * consumers allow non-durable topic subscribers to receive old messages 950 * that were published before the non-durable subscriber started. 951 */ 952 public void setUseRetroactiveConsumer(boolean useRetroactiveConsumer) { 953 this.useRetroactiveConsumer = useRetroactiveConsumer; 954 } 955 956 public boolean isNestedMapAndListEnabled() { 957 return nestedMapAndListEnabled; 958 } 959 960 /** 961 * Enables/disables whether or not Message properties and MapMessage entries 962 * support <a 963 * href="http://activemq.apache.org/structured-message-properties-and-mapmessages.html">Nested 964 * Structures</a> of Map and List objects 965 */ 966 public void setNestedMapAndListEnabled(boolean structuredMapsEnabled) { 967 this.nestedMapAndListEnabled = structuredMapsEnabled; 968 } 969 970 public boolean isExclusiveConsumer() { 971 return exclusiveConsumer; 972 } 973 974 /** 975 * Enables or disables whether or not queue consumers should be exclusive or 976 * not for example to preserve ordering when not using <a 977 * href="http://activemq.apache.org/message-groups.html">Message Groups</a> 978 * 979 * @param exclusiveConsumer 980 */ 981 public void setExclusiveConsumer(boolean exclusiveConsumer) { 982 this.exclusiveConsumer = exclusiveConsumer; 983 } 984 985 /** 986 * Adds a transport listener so that a client can be notified of events in 987 * the underlying transport 988 */ 989 public void addTransportListener(TransportListener transportListener) { 990 transportListeners.add(transportListener); 991 } 992 993 public void removeTransportListener(TransportListener transportListener) { 994 transportListeners.remove(transportListener); 995 } 996 997 public boolean isUseDedicatedTaskRunner() { 998 return useDedicatedTaskRunner; 999 } 1000 1001 public void setUseDedicatedTaskRunner(boolean useDedicatedTaskRunner) { 1002 this.useDedicatedTaskRunner = useDedicatedTaskRunner; 1003 } 1004 1005 public TaskRunnerFactory getSessionTaskRunner() { 1006 synchronized (this) { 1007 if (sessionTaskRunner == null) { 1008 sessionTaskRunner = new TaskRunnerFactory("ActiveMQ Session Task", ThreadPriorities.INBOUND_CLIENT_SESSION, false, 1000, isUseDedicatedTaskRunner(), maxThreadPoolSize); 1009 sessionTaskRunner.setRejectedTaskHandler(rejectedTaskHandler); 1010 } 1011 } 1012 return sessionTaskRunner; 1013 } 1014 1015 public void setSessionTaskRunner(TaskRunnerFactory sessionTaskRunner) { 1016 this.sessionTaskRunner = sessionTaskRunner; 1017 } 1018 1019 public MessageTransformer getTransformer() { 1020 return transformer; 1021 } 1022 1023 /** 1024 * Sets the transformer used to transform messages before they are sent on 1025 * to the JMS bus or when they are received from the bus but before they are 1026 * delivered to the JMS client 1027 */ 1028 public void setTransformer(MessageTransformer transformer) { 1029 this.transformer = transformer; 1030 } 1031 1032 /** 1033 * @return the statsEnabled 1034 */ 1035 public boolean isStatsEnabled() { 1036 return this.stats.isEnabled(); 1037 } 1038 1039 /** 1040 * @param statsEnabled the statsEnabled to set 1041 */ 1042 public void setStatsEnabled(boolean statsEnabled) { 1043 this.stats.setEnabled(statsEnabled); 1044 } 1045 1046 /** 1047 * Returns the {@link DestinationSource} object which can be used to listen to destinations 1048 * being created or destroyed or to enquire about the current destinations available on the broker 1049 * 1050 * @return a lazily created destination source 1051 * @throws JMSException 1052 */ 1053 @Override 1054 public DestinationSource getDestinationSource() throws JMSException { 1055 if (destinationSource == null) { 1056 destinationSource = new DestinationSource(this); 1057 destinationSource.start(); 1058 } 1059 return destinationSource; 1060 } 1061 1062 // Implementation methods 1063 // ------------------------------------------------------------------------- 1064 1065 /** 1066 * Used internally for adding Sessions to the Connection 1067 * 1068 * @param session 1069 * @throws JMSException 1070 * @throws JMSException 1071 */ 1072 protected void addSession(ActiveMQSession session) throws JMSException { 1073 this.sessions.add(session); 1074 if (sessions.size() > 1 || session.isTransacted()) { 1075 optimizedMessageDispatch = false; 1076 } 1077 } 1078 1079 /** 1080 * Used interanlly for removing Sessions from a Connection 1081 * 1082 * @param session 1083 */ 1084 protected void removeSession(ActiveMQSession session) { 1085 this.sessions.remove(session); 1086 this.removeDispatcher(session); 1087 } 1088 1089 /** 1090 * Add a ConnectionConsumer 1091 * 1092 * @param connectionConsumer 1093 * @throws JMSException 1094 */ 1095 protected void addConnectionConsumer(ActiveMQConnectionConsumer connectionConsumer) throws JMSException { 1096 this.connectionConsumers.add(connectionConsumer); 1097 } 1098 1099 /** 1100 * Remove a ConnectionConsumer 1101 * 1102 * @param connectionConsumer 1103 */ 1104 protected void removeConnectionConsumer(ActiveMQConnectionConsumer connectionConsumer) { 1105 this.connectionConsumers.remove(connectionConsumer); 1106 this.removeDispatcher(connectionConsumer); 1107 } 1108 1109 /** 1110 * Creates a <CODE>TopicSession</CODE> object. 1111 * 1112 * @param transacted indicates whether the session is transacted 1113 * @param acknowledgeMode indicates whether the consumer or the client will 1114 * acknowledge any messages it receives; ignored if the 1115 * session is transacted. Legal values are 1116 * <code>Session.AUTO_ACKNOWLEDGE</code>, 1117 * <code>Session.CLIENT_ACKNOWLEDGE</code>, and 1118 * <code>Session.DUPS_OK_ACKNOWLEDGE</code>. 1119 * @return a newly created topic session 1120 * @throws JMSException if the <CODE>TopicConnection</CODE> object fails 1121 * to create a session due to some internal error or lack of 1122 * support for the specific transaction and acknowledgement 1123 * mode. 1124 * @see Session#AUTO_ACKNOWLEDGE 1125 * @see Session#CLIENT_ACKNOWLEDGE 1126 * @see Session#DUPS_OK_ACKNOWLEDGE 1127 */ 1128 @Override 1129 public TopicSession createTopicSession(boolean transacted, int acknowledgeMode) throws JMSException { 1130 return new ActiveMQTopicSession((ActiveMQSession)createSession(transacted, acknowledgeMode)); 1131 } 1132 1133 /** 1134 * Creates a connection consumer for this connection (optional operation). 1135 * This is an expert facility not used by regular JMS clients. 1136 * 1137 * @param topic the topic to access 1138 * @param messageSelector only messages with properties matching the message 1139 * selector expression are delivered. A value of null or an 1140 * empty string indicates that there is no message selector 1141 * for the message consumer. 1142 * @param sessionPool the server session pool to associate with this 1143 * connection consumer 1144 * @param maxMessages the maximum number of messages that can be assigned to 1145 * a server session at one time 1146 * @return the connection consumer 1147 * @throws JMSException if the <CODE>TopicConnection</CODE> object fails 1148 * to create a connection consumer due to some internal 1149 * error or invalid arguments for <CODE>sessionPool</CODE> 1150 * and <CODE>messageSelector</CODE>. 1151 * @throws javax.jms.InvalidDestinationException if an invalid topic is 1152 * specified. 1153 * @throws javax.jms.InvalidSelectorException if the message selector is 1154 * invalid. 1155 * @see javax.jms.ConnectionConsumer 1156 */ 1157 @Override 1158 public ConnectionConsumer createConnectionConsumer(Topic topic, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException { 1159 return createConnectionConsumer(topic, messageSelector, sessionPool, maxMessages, false); 1160 } 1161 1162 /** 1163 * Creates a connection consumer for this connection (optional operation). 1164 * This is an expert facility not used by regular JMS clients. 1165 * 1166 * @param queue the queue to access 1167 * @param messageSelector only messages with properties matching the message 1168 * selector expression are delivered. A value of null or an 1169 * empty string indicates that there is no message selector 1170 * for the message consumer. 1171 * @param sessionPool the server session pool to associate with this 1172 * connection consumer 1173 * @param maxMessages the maximum number of messages that can be assigned to 1174 * a server session at one time 1175 * @return the connection consumer 1176 * @throws JMSException if the <CODE>QueueConnection</CODE> object fails 1177 * to create a connection consumer due to some internal 1178 * error or invalid arguments for <CODE>sessionPool</CODE> 1179 * and <CODE>messageSelector</CODE>. 1180 * @throws javax.jms.InvalidDestinationException if an invalid queue is 1181 * specified. 1182 * @throws javax.jms.InvalidSelectorException if the message selector is 1183 * invalid. 1184 * @see javax.jms.ConnectionConsumer 1185 */ 1186 @Override 1187 public ConnectionConsumer createConnectionConsumer(Queue queue, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException { 1188 return createConnectionConsumer(queue, messageSelector, sessionPool, maxMessages, false); 1189 } 1190 1191 /** 1192 * Creates a connection consumer for this connection (optional operation). 1193 * This is an expert facility not used by regular JMS clients. 1194 * 1195 * @param destination the destination to access 1196 * @param messageSelector only messages with properties matching the message 1197 * selector expression are delivered. A value of null or an 1198 * empty string indicates that there is no message selector 1199 * for the message consumer. 1200 * @param sessionPool the server session pool to associate with this 1201 * connection consumer 1202 * @param maxMessages the maximum number of messages that can be assigned to 1203 * a server session at one time 1204 * @return the connection consumer 1205 * @throws JMSException if the <CODE>Connection</CODE> object fails to 1206 * create a connection consumer due to some internal error 1207 * or invalid arguments for <CODE>sessionPool</CODE> and 1208 * <CODE>messageSelector</CODE>. 1209 * @throws javax.jms.InvalidDestinationException if an invalid destination 1210 * is specified. 1211 * @throws javax.jms.InvalidSelectorException if the message selector is 1212 * invalid. 1213 * @see javax.jms.ConnectionConsumer 1214 * @since 1.1 1215 */ 1216 @Override 1217 public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector, ServerSessionPool sessionPool, int maxMessages) throws JMSException { 1218 return createConnectionConsumer(destination, messageSelector, sessionPool, maxMessages, false); 1219 } 1220 1221 public ConnectionConsumer createConnectionConsumer(Destination destination, String messageSelector, ServerSessionPool sessionPool, int maxMessages, boolean noLocal) 1222 throws JMSException { 1223 1224 checkClosedOrFailed(); 1225 ensureConnectionInfoSent(); 1226 1227 ConsumerId consumerId = createConsumerId(); 1228 ConsumerInfo consumerInfo = new ConsumerInfo(consumerId); 1229 consumerInfo.setDestination(ActiveMQMessageTransformation.transformDestination(destination)); 1230 consumerInfo.setSelector(messageSelector); 1231 consumerInfo.setPrefetchSize(maxMessages); 1232 consumerInfo.setNoLocal(noLocal); 1233 consumerInfo.setDispatchAsync(isDispatchAsync()); 1234 1235 // Allows the options on the destination to configure the consumerInfo 1236 if (consumerInfo.getDestination().getOptions() != null) { 1237 Map<String, String> options = new HashMap<String, String>(consumerInfo.getDestination().getOptions()); 1238 IntrospectionSupport.setProperties(consumerInfo, options, "consumer."); 1239 } 1240 1241 return new ActiveMQConnectionConsumer(this, sessionPool, consumerInfo); 1242 } 1243 1244 /** 1245 * @return 1246 */ 1247 private ConsumerId createConsumerId() { 1248 return new ConsumerId(connectionSessionId, consumerIdGenerator.getNextSequenceId()); 1249 } 1250 1251 /** 1252 * Creates a <CODE>QueueSession</CODE> object. 1253 * 1254 * @param transacted indicates whether the session is transacted 1255 * @param acknowledgeMode indicates whether the consumer or the client will 1256 * acknowledge any messages it receives; ignored if the 1257 * session is transacted. Legal values are 1258 * <code>Session.AUTO_ACKNOWLEDGE</code>, 1259 * <code>Session.CLIENT_ACKNOWLEDGE</code>, and 1260 * <code>Session.DUPS_OK_ACKNOWLEDGE</code>. 1261 * @return a newly created queue session 1262 * @throws JMSException if the <CODE>QueueConnection</CODE> object fails 1263 * to create a session due to some internal error or lack of 1264 * support for the specific transaction and acknowledgement 1265 * mode. 1266 * @see Session#AUTO_ACKNOWLEDGE 1267 * @see Session#CLIENT_ACKNOWLEDGE 1268 * @see Session#DUPS_OK_ACKNOWLEDGE 1269 */ 1270 @Override 1271 public QueueSession createQueueSession(boolean transacted, int acknowledgeMode) throws JMSException { 1272 return new ActiveMQQueueSession((ActiveMQSession)createSession(transacted, acknowledgeMode)); 1273 } 1274 1275 /** 1276 * Ensures that the clientID was manually specified and not auto-generated. 1277 * If the clientID was not specified this method will throw an exception. 1278 * This method is used to ensure that the clientID + durableSubscriber name 1279 * are used correctly. 1280 * 1281 * @throws JMSException 1282 */ 1283 public void checkClientIDWasManuallySpecified() throws JMSException { 1284 if (!userSpecifiedClientID) { 1285 throw new JMSException("You cannot create a durable subscriber without specifying a unique clientID on a Connection"); 1286 } 1287 } 1288 1289 /** 1290 * send a Packet through the Connection - for internal use only 1291 * 1292 * @param command 1293 * @throws JMSException 1294 */ 1295 public void asyncSendPacket(Command command) throws JMSException { 1296 if (isClosed()) { 1297 throw new ConnectionClosedException(); 1298 } else { 1299 doAsyncSendPacket(command); 1300 } 1301 } 1302 1303 private void doAsyncSendPacket(Command command) throws JMSException { 1304 try { 1305 this.transport.oneway(command); 1306 } catch (IOException e) { 1307 throw JMSExceptionSupport.create(e); 1308 } 1309 } 1310 1311 /** 1312 * Send a packet through a Connection - for internal use only 1313 * 1314 * @param command 1315 * @return 1316 * @throws JMSException 1317 */ 1318 public void syncSendPacket(final Command command, final AsyncCallback onComplete) throws JMSException { 1319 if(onComplete==null) { 1320 syncSendPacket(command); 1321 } else { 1322 if (isClosed()) { 1323 throw new ConnectionClosedException(); 1324 } 1325 try { 1326 this.transport.asyncRequest(command, new ResponseCallback() { 1327 @Override 1328 public void onCompletion(FutureResponse resp) { 1329 Response response; 1330 Throwable exception = null; 1331 try { 1332 response = resp.getResult(); 1333 if (response.isException()) { 1334 ExceptionResponse er = (ExceptionResponse)response; 1335 exception = er.getException(); 1336 } 1337 } catch (Exception e) { 1338 exception = e; 1339 } 1340 if(exception!=null) { 1341 if ( exception instanceof JMSException) { 1342 onComplete.onException((JMSException) exception); 1343 } else { 1344 if (isClosed()||closing.get()) { 1345 LOG.debug("Received an exception but connection is closing"); 1346 } 1347 JMSException jmsEx = null; 1348 try { 1349 jmsEx = JMSExceptionSupport.create(exception); 1350 } catch(Throwable e) { 1351 LOG.error("Caught an exception trying to create a JMSException for " +exception,e); 1352 } 1353 // dispose of transport for security exceptions on connection initiation 1354 if (exception instanceof SecurityException && command instanceof ConnectionInfo){ 1355 forceCloseOnSecurityException(exception); 1356 } 1357 if (jmsEx !=null) { 1358 onComplete.onException(jmsEx); 1359 } 1360 } 1361 } else { 1362 onComplete.onSuccess(); 1363 } 1364 } 1365 }); 1366 } catch (IOException e) { 1367 throw JMSExceptionSupport.create(e); 1368 } 1369 } 1370 } 1371 1372 private void forceCloseOnSecurityException(Throwable exception) { 1373 LOG.trace("force close on security exception:" + this + ", transport=" + transport, exception); 1374 onException(new IOException("Force close due to SecurityException on connect", exception)); 1375 } 1376 1377 public Response syncSendPacket(Command command, int timeout) throws JMSException { 1378 if (isClosed()) { 1379 throw new ConnectionClosedException(); 1380 } else { 1381 1382 try { 1383 Response response = (Response)(timeout > 0 1384 ? this.transport.request(command, timeout) 1385 : this.transport.request(command)); 1386 if (response.isException()) { 1387 ExceptionResponse er = (ExceptionResponse)response; 1388 if (er.getException() instanceof JMSException) { 1389 throw (JMSException)er.getException(); 1390 } else { 1391 if (isClosed()||closing.get()) { 1392 LOG.debug("Received an exception but connection is closing"); 1393 } 1394 JMSException jmsEx = null; 1395 try { 1396 jmsEx = JMSExceptionSupport.create(er.getException()); 1397 } catch(Throwable e) { 1398 LOG.error("Caught an exception trying to create a JMSException for " +er.getException(),e); 1399 } 1400 if (er.getException() instanceof SecurityException && command instanceof ConnectionInfo){ 1401 forceCloseOnSecurityException(er.getException()); 1402 } 1403 if (jmsEx !=null) { 1404 throw jmsEx; 1405 } 1406 } 1407 } 1408 return response; 1409 } catch (IOException e) { 1410 throw JMSExceptionSupport.create(e); 1411 } 1412 } 1413 } 1414 1415 /** 1416 * Send a packet through a Connection - for internal use only 1417 * 1418 * @param command 1419 * @return 1420 * @throws JMSException 1421 */ 1422 public Response syncSendPacket(Command command) throws JMSException { 1423 return syncSendPacket(command, 0); 1424 } 1425 1426 /** 1427 * @return statistics for this Connection 1428 */ 1429 @Override 1430 public StatsImpl getStats() { 1431 return stats; 1432 } 1433 1434 /** 1435 * simply throws an exception if the Connection is already closed or the 1436 * Transport has failed 1437 * 1438 * @throws JMSException 1439 */ 1440 protected synchronized void checkClosedOrFailed() throws JMSException { 1441 checkClosed(); 1442 if (transportFailed.get()) { 1443 throw new ConnectionFailedException(firstFailureError); 1444 } 1445 } 1446 1447 /** 1448 * simply throws an exception if the Connection is already closed 1449 * 1450 * @throws JMSException 1451 */ 1452 protected synchronized void checkClosed() throws JMSException { 1453 if (closed.get()) { 1454 throw new ConnectionClosedException(); 1455 } 1456 } 1457 1458 /** 1459 * Send the ConnectionInfo to the Broker 1460 * 1461 * @throws JMSException 1462 */ 1463 protected void ensureConnectionInfoSent() throws JMSException { 1464 synchronized(this.ensureConnectionInfoSentMutex) { 1465 // Can we skip sending the ConnectionInfo packet?? 1466 if (isConnectionInfoSentToBroker || closed.get()) { 1467 return; 1468 } 1469 //TODO shouldn't this check be on userSpecifiedClientID rather than the value of clientID? 1470 if (info.getClientId() == null || info.getClientId().trim().length() == 0) { 1471 info.setClientId(clientIdGenerator.generateId()); 1472 } 1473 syncSendPacket(info.copy(), getConnectResponseTimeout()); 1474 1475 this.isConnectionInfoSentToBroker = true; 1476 // Add a temp destination advisory consumer so that 1477 // We know what the valid temporary destinations are on the 1478 // broker without having to do an RPC to the broker. 1479 1480 ConsumerId consumerId = new ConsumerId(new SessionId(info.getConnectionId(), -1), consumerIdGenerator.getNextSequenceId()); 1481 if (watchTopicAdvisories) { 1482 advisoryConsumer = new AdvisoryConsumer(this, consumerId); 1483 } 1484 } 1485 } 1486 1487 public synchronized boolean isWatchTopicAdvisories() { 1488 return watchTopicAdvisories; 1489 } 1490 1491 public synchronized void setWatchTopicAdvisories(boolean watchTopicAdvisories) { 1492 this.watchTopicAdvisories = watchTopicAdvisories; 1493 } 1494 1495 /** 1496 * @return Returns the useAsyncSend. 1497 */ 1498 public boolean isUseAsyncSend() { 1499 return useAsyncSend; 1500 } 1501 1502 /** 1503 * Forces the use of <a 1504 * href="http://activemq.apache.org/async-sends.html">Async Sends</a> which 1505 * adds a massive performance boost; but means that the send() method will 1506 * return immediately whether the message has been sent or not which could 1507 * lead to message loss. 1508 */ 1509 public void setUseAsyncSend(boolean useAsyncSend) { 1510 this.useAsyncSend = useAsyncSend; 1511 } 1512 1513 /** 1514 * @return true if always sync send messages 1515 */ 1516 public boolean isAlwaysSyncSend() { 1517 return this.alwaysSyncSend; 1518 } 1519 1520 /** 1521 * Set true if always require messages to be sync sent 1522 * 1523 * @param alwaysSyncSend 1524 */ 1525 public void setAlwaysSyncSend(boolean alwaysSyncSend) { 1526 this.alwaysSyncSend = alwaysSyncSend; 1527 } 1528 1529 /** 1530 * @return the messagePrioritySupported 1531 */ 1532 public boolean isMessagePrioritySupported() { 1533 return this.messagePrioritySupported; 1534 } 1535 1536 /** 1537 * @param messagePrioritySupported the messagePrioritySupported to set 1538 */ 1539 public void setMessagePrioritySupported(boolean messagePrioritySupported) { 1540 this.messagePrioritySupported = messagePrioritySupported; 1541 } 1542 1543 /** 1544 * Cleans up this connection so that it's state is as if the connection was 1545 * just created. This allows the Resource Adapter to clean up a connection 1546 * so that it can be reused without having to close and recreate the 1547 * connection. 1548 */ 1549 public void cleanup() throws JMSException { 1550 doCleanup(false); 1551 } 1552 1553 public boolean isUserSpecifiedClientID() { 1554 return userSpecifiedClientID; 1555 } 1556 1557 public void doCleanup(boolean removeConnection) throws JMSException { 1558 if (advisoryConsumer != null && !isTransportFailed()) { 1559 advisoryConsumer.dispose(); 1560 advisoryConsumer = null; 1561 } 1562 1563 for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();) { 1564 ActiveMQSession s = i.next(); 1565 s.dispose(); 1566 } 1567 for (Iterator<ActiveMQConnectionConsumer> i = this.connectionConsumers.iterator(); i.hasNext();) { 1568 ActiveMQConnectionConsumer c = i.next(); 1569 c.dispose(); 1570 } 1571 1572 if (removeConnection) { 1573 if (isConnectionInfoSentToBroker) { 1574 if (!transportFailed.get() && !closing.get()) { 1575 syncSendPacket(info.createRemoveCommand()); 1576 } 1577 isConnectionInfoSentToBroker = false; 1578 } 1579 if (userSpecifiedClientID) { 1580 info.setClientId(null); 1581 userSpecifiedClientID = false; 1582 } 1583 clientIDSet = false; 1584 } 1585 1586 started.set(false); 1587 } 1588 1589 /** 1590 * Changes the associated username/password that is associated with this 1591 * connection. If the connection has been used, you must called cleanup() 1592 * before calling this method. 1593 * 1594 * @throws IllegalStateException if the connection is in used. 1595 */ 1596 public void changeUserInfo(String userName, String password) throws JMSException { 1597 if (isConnectionInfoSentToBroker) { 1598 throw new IllegalStateException("changeUserInfo used Connection is not allowed"); 1599 } 1600 this.info.setUserName(userName); 1601 this.info.setPassword(password); 1602 } 1603 1604 /** 1605 * @return Returns the resourceManagerId. 1606 * @throws JMSException 1607 */ 1608 public String getResourceManagerId() throws JMSException { 1609 if (isRmIdFromConnectionId()) { 1610 return info.getConnectionId().getValue(); 1611 } 1612 waitForBrokerInfo(); 1613 if (brokerInfo == null) { 1614 throw new JMSException("Connection failed before Broker info was received."); 1615 } 1616 return brokerInfo.getBrokerId().getValue(); 1617 } 1618 1619 /** 1620 * Returns the broker name if one is available or null if one is not 1621 * available yet. 1622 */ 1623 public String getBrokerName() { 1624 try { 1625 brokerInfoReceived.await(5, TimeUnit.SECONDS); 1626 if (brokerInfo == null) { 1627 return null; 1628 } 1629 return brokerInfo.getBrokerName(); 1630 } catch (InterruptedException e) { 1631 Thread.currentThread().interrupt(); 1632 return null; 1633 } 1634 } 1635 1636 /** 1637 * Returns the broker information if it is available or null if it is not 1638 * available yet. 1639 */ 1640 public BrokerInfo getBrokerInfo() { 1641 return brokerInfo; 1642 } 1643 1644 /** 1645 * @return Returns the RedeliveryPolicy. 1646 * @throws JMSException 1647 */ 1648 public RedeliveryPolicy getRedeliveryPolicy() throws JMSException { 1649 return redeliveryPolicyMap.getDefaultEntry(); 1650 } 1651 1652 /** 1653 * Sets the redelivery policy to be used when messages are rolled back 1654 */ 1655 public void setRedeliveryPolicy(RedeliveryPolicy redeliveryPolicy) { 1656 this.redeliveryPolicyMap.setDefaultEntry(redeliveryPolicy); 1657 } 1658 1659 public BlobTransferPolicy getBlobTransferPolicy() { 1660 if (blobTransferPolicy == null) { 1661 blobTransferPolicy = createBlobTransferPolicy(); 1662 } 1663 return blobTransferPolicy; 1664 } 1665 1666 /** 1667 * Sets the policy used to describe how out-of-band BLOBs (Binary Large 1668 * OBjects) are transferred from producers to brokers to consumers 1669 */ 1670 public void setBlobTransferPolicy(BlobTransferPolicy blobTransferPolicy) { 1671 this.blobTransferPolicy = blobTransferPolicy; 1672 } 1673 1674 /** 1675 * @return Returns the alwaysSessionAsync. 1676 */ 1677 public boolean isAlwaysSessionAsync() { 1678 return alwaysSessionAsync; 1679 } 1680 1681 /** 1682 * If this flag is not set then a separate thread is not used for dispatching messages for each Session in 1683 * the Connection. However, a separate thread is always used if there is more than one session, or the session 1684 * isn't in auto acknowledge or duplicates ok mode. By default this value is set to true and session dispatch 1685 * happens asynchronously. 1686 */ 1687 public void setAlwaysSessionAsync(boolean alwaysSessionAsync) { 1688 this.alwaysSessionAsync = alwaysSessionAsync; 1689 } 1690 1691 /** 1692 * @return Returns the optimizeAcknowledge. 1693 */ 1694 public boolean isOptimizeAcknowledge() { 1695 return optimizeAcknowledge; 1696 } 1697 1698 /** 1699 * Enables an optimised acknowledgement mode where messages are acknowledged 1700 * in batches rather than individually 1701 * 1702 * @param optimizeAcknowledge The optimizeAcknowledge to set. 1703 */ 1704 public void setOptimizeAcknowledge(boolean optimizeAcknowledge) { 1705 this.optimizeAcknowledge = optimizeAcknowledge; 1706 } 1707 1708 /** 1709 * The max time in milliseconds between optimized ack batches 1710 * @param optimizeAcknowledgeTimeOut 1711 */ 1712 public void setOptimizeAcknowledgeTimeOut(long optimizeAcknowledgeTimeOut) { 1713 this.optimizeAcknowledgeTimeOut = optimizeAcknowledgeTimeOut; 1714 } 1715 1716 public long getOptimizeAcknowledgeTimeOut() { 1717 return optimizeAcknowledgeTimeOut; 1718 } 1719 1720 public long getWarnAboutUnstartedConnectionTimeout() { 1721 return warnAboutUnstartedConnectionTimeout; 1722 } 1723 1724 /** 1725 * Enables the timeout from a connection creation to when a warning is 1726 * generated if the connection is not properly started via {@link #start()} 1727 * and a message is received by a consumer. It is a very common gotcha to 1728 * forget to <a 1729 * href="http://activemq.apache.org/i-am-not-receiving-any-messages-what-is-wrong.html">start 1730 * the connection</a> so this option makes the default case to create a 1731 * warning if the user forgets. To disable the warning just set the value to < 1732 * 0 (say -1). 1733 */ 1734 public void setWarnAboutUnstartedConnectionTimeout(long warnAboutUnstartedConnectionTimeout) { 1735 this.warnAboutUnstartedConnectionTimeout = warnAboutUnstartedConnectionTimeout; 1736 } 1737 1738 /** 1739 * @return the sendTimeout (in milliseconds) 1740 */ 1741 public int getSendTimeout() { 1742 return sendTimeout; 1743 } 1744 1745 /** 1746 * @param sendTimeout the sendTimeout to set (in milliseconds) 1747 */ 1748 public void setSendTimeout(int sendTimeout) { 1749 this.sendTimeout = sendTimeout; 1750 } 1751 1752 /** 1753 * @return the sendAcksAsync 1754 */ 1755 public boolean isSendAcksAsync() { 1756 return sendAcksAsync; 1757 } 1758 1759 /** 1760 * @param sendAcksAsync the sendAcksAsync to set 1761 */ 1762 public void setSendAcksAsync(boolean sendAcksAsync) { 1763 this.sendAcksAsync = sendAcksAsync; 1764 } 1765 1766 /** 1767 * Returns the time this connection was created 1768 */ 1769 public long getTimeCreated() { 1770 return timeCreated; 1771 } 1772 1773 private void waitForBrokerInfo() throws JMSException { 1774 try { 1775 brokerInfoReceived.await(); 1776 } catch (InterruptedException e) { 1777 Thread.currentThread().interrupt(); 1778 throw JMSExceptionSupport.create(e); 1779 } 1780 } 1781 1782 // Package protected so that it can be used in unit tests 1783 public Transport getTransport() { 1784 return transport; 1785 } 1786 1787 public void addProducer(ProducerId producerId, ActiveMQMessageProducer producer) { 1788 producers.put(producerId, producer); 1789 } 1790 1791 public void removeProducer(ProducerId producerId) { 1792 producers.remove(producerId); 1793 } 1794 1795 public void addDispatcher(ConsumerId consumerId, ActiveMQDispatcher dispatcher) { 1796 dispatchers.put(consumerId, dispatcher); 1797 } 1798 1799 public void removeDispatcher(ConsumerId consumerId) { 1800 dispatchers.remove(consumerId); 1801 } 1802 1803 public boolean hasDispatcher(ConsumerId consumerId) { 1804 return dispatchers.containsKey(consumerId); 1805 } 1806 1807 /** 1808 * @param o - the command to consume 1809 */ 1810 @Override 1811 public void onCommand(final Object o) { 1812 final Command command = (Command)o; 1813 if (!closed.get() && command != null) { 1814 try { 1815 command.visit(new CommandVisitorAdapter() { 1816 @Override 1817 public Response processMessageDispatch(MessageDispatch md) throws Exception { 1818 waitForTransportInterruptionProcessingToComplete(); 1819 ActiveMQDispatcher dispatcher = dispatchers.get(md.getConsumerId()); 1820 if (dispatcher != null) { 1821 // Copy in case a embedded broker is dispatching via 1822 // vm:// 1823 // md.getMessage() == null to signal end of queue 1824 // browse. 1825 Message msg = md.getMessage(); 1826 if (msg != null) { 1827 msg = msg.copy(); 1828 msg.setReadOnlyBody(true); 1829 msg.setReadOnlyProperties(true); 1830 msg.setRedeliveryCounter(md.getRedeliveryCounter()); 1831 msg.setConnection(ActiveMQConnection.this); 1832 msg.setMemoryUsage(null); 1833 md.setMessage(msg); 1834 } 1835 dispatcher.dispatch(md); 1836 } else { 1837 LOG.debug("{} no dispatcher for {} in {}", this, md, dispatchers); 1838 } 1839 return null; 1840 } 1841 1842 @Override 1843 public Response processProducerAck(ProducerAck pa) throws Exception { 1844 if (pa != null && pa.getProducerId() != null) { 1845 ActiveMQMessageProducer producer = producers.get(pa.getProducerId()); 1846 if (producer != null) { 1847 producer.onProducerAck(pa); 1848 } 1849 } 1850 return null; 1851 } 1852 1853 @Override 1854 public Response processBrokerInfo(BrokerInfo info) throws Exception { 1855 brokerInfo = info; 1856 brokerInfoReceived.countDown(); 1857 optimizeAcknowledge &= !brokerInfo.isFaultTolerantConfiguration(); 1858 getBlobTransferPolicy().setBrokerUploadUrl(info.getBrokerUploadUrl()); 1859 return null; 1860 } 1861 1862 @Override 1863 public Response processConnectionError(final ConnectionError error) throws Exception { 1864 executor.execute(new Runnable() { 1865 @Override 1866 public void run() { 1867 onAsyncException(error.getException()); 1868 } 1869 }); 1870 return null; 1871 } 1872 1873 @Override 1874 public Response processControlCommand(ControlCommand command) throws Exception { 1875 return null; 1876 } 1877 1878 @Override 1879 public Response processConnectionControl(ConnectionControl control) throws Exception { 1880 onConnectionControl((ConnectionControl)command); 1881 return null; 1882 } 1883 1884 @Override 1885 public Response processConsumerControl(ConsumerControl control) throws Exception { 1886 onConsumerControl((ConsumerControl)command); 1887 return null; 1888 } 1889 1890 @Override 1891 public Response processWireFormat(WireFormatInfo info) throws Exception { 1892 onWireFormatInfo((WireFormatInfo)command); 1893 return null; 1894 } 1895 }); 1896 } catch (Exception e) { 1897 onClientInternalException(e); 1898 } 1899 } 1900 1901 for (Iterator<TransportListener> iter = transportListeners.iterator(); iter.hasNext();) { 1902 TransportListener listener = iter.next(); 1903 listener.onCommand(command); 1904 } 1905 } 1906 1907 protected void onWireFormatInfo(WireFormatInfo info) { 1908 protocolVersion.set(info.getVersion()); 1909 } 1910 1911 /** 1912 * Handles async client internal exceptions. 1913 * A client internal exception is usually one that has been thrown 1914 * by a container runtime component during asynchronous processing of a 1915 * message that does not affect the connection itself. 1916 * This method notifies the <code>ClientInternalExceptionListener</code> by invoking 1917 * its <code>onException</code> method, if one has been registered with this connection. 1918 * 1919 * @param error the exception that the problem 1920 */ 1921 public void onClientInternalException(final Throwable error) { 1922 if ( !closed.get() && !closing.get() ) { 1923 if ( this.clientInternalExceptionListener != null ) { 1924 executor.execute(new Runnable() { 1925 @Override 1926 public void run() { 1927 ActiveMQConnection.this.clientInternalExceptionListener.onException(error); 1928 } 1929 }); 1930 } else { 1931 LOG.debug("Async client internal exception occurred with no exception listener registered: " 1932 + error, error); 1933 } 1934 } 1935 } 1936 1937 /** 1938 * Used for handling async exceptions 1939 * 1940 * @param error 1941 */ 1942 public void onAsyncException(Throwable error) { 1943 if (!closed.get() && !closing.get()) { 1944 if (this.exceptionListener != null) { 1945 1946 if (!(error instanceof JMSException)) { 1947 error = JMSExceptionSupport.create(error); 1948 } 1949 final JMSException e = (JMSException)error; 1950 1951 executor.execute(new Runnable() { 1952 @Override 1953 public void run() { 1954 ActiveMQConnection.this.exceptionListener.onException(e); 1955 } 1956 }); 1957 1958 } else { 1959 LOG.debug("Async exception with no exception listener: " + error, error); 1960 } 1961 } 1962 } 1963 1964 @Override 1965 public void onException(final IOException error) { 1966 onAsyncException(error); 1967 if (!closed.get() && !closing.get()) { 1968 executor.execute(new Runnable() { 1969 @Override 1970 public void run() { 1971 transportFailed(error); 1972 ServiceSupport.dispose(ActiveMQConnection.this.transport); 1973 brokerInfoReceived.countDown(); 1974 try { 1975 doCleanup(true); 1976 } catch (JMSException e) { 1977 LOG.warn("Exception during connection cleanup, " + e, e); 1978 } 1979 for (Iterator<TransportListener> iter = transportListeners.iterator(); iter.hasNext();) { 1980 TransportListener listener = iter.next(); 1981 listener.onException(error); 1982 } 1983 } 1984 }); 1985 } 1986 } 1987 1988 @Override 1989 public void transportInterupted() { 1990 transportInterruptionProcessingComplete.set(1); 1991 for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();) { 1992 ActiveMQSession s = i.next(); 1993 s.clearMessagesInProgress(transportInterruptionProcessingComplete); 1994 } 1995 1996 for (ActiveMQConnectionConsumer connectionConsumer : this.connectionConsumers) { 1997 connectionConsumer.clearMessagesInProgress(transportInterruptionProcessingComplete); 1998 } 1999 2000 if (transportInterruptionProcessingComplete.decrementAndGet() > 0) { 2001 if (LOG.isDebugEnabled()) { 2002 LOG.debug("transport interrupted - processing required, dispatchers: " + transportInterruptionProcessingComplete.get()); 2003 } 2004 signalInterruptionProcessingNeeded(); 2005 } 2006 2007 for (Iterator<TransportListener> iter = transportListeners.iterator(); iter.hasNext();) { 2008 TransportListener listener = iter.next(); 2009 listener.transportInterupted(); 2010 } 2011 } 2012 2013 @Override 2014 public void transportResumed() { 2015 for (Iterator<TransportListener> iter = transportListeners.iterator(); iter.hasNext();) { 2016 TransportListener listener = iter.next(); 2017 listener.transportResumed(); 2018 } 2019 } 2020 2021 /** 2022 * Create the DestinationInfo object for the temporary destination. 2023 * 2024 * @param topic - if its true topic, else queue. 2025 * @return DestinationInfo 2026 * @throws JMSException 2027 */ 2028 protected ActiveMQTempDestination createTempDestination(boolean topic) throws JMSException { 2029 2030 // Check if Destination info is of temporary type. 2031 ActiveMQTempDestination dest; 2032 if (topic) { 2033 dest = new ActiveMQTempTopic(info.getConnectionId(), tempDestinationIdGenerator.getNextSequenceId()); 2034 } else { 2035 dest = new ActiveMQTempQueue(info.getConnectionId(), tempDestinationIdGenerator.getNextSequenceId()); 2036 } 2037 2038 DestinationInfo info = new DestinationInfo(); 2039 info.setConnectionId(this.info.getConnectionId()); 2040 info.setOperationType(DestinationInfo.ADD_OPERATION_TYPE); 2041 info.setDestination(dest); 2042 syncSendPacket(info); 2043 2044 dest.setConnection(this); 2045 activeTempDestinations.put(dest, dest); 2046 return dest; 2047 } 2048 2049 /** 2050 * @param destination 2051 * @throws JMSException 2052 */ 2053 public void deleteTempDestination(ActiveMQTempDestination destination) throws JMSException { 2054 2055 checkClosedOrFailed(); 2056 2057 for (ActiveMQSession session : this.sessions) { 2058 if (session.isInUse(destination)) { 2059 throw new JMSException("A consumer is consuming from the temporary destination"); 2060 } 2061 } 2062 2063 activeTempDestinations.remove(destination); 2064 2065 DestinationInfo destInfo = new DestinationInfo(); 2066 destInfo.setConnectionId(this.info.getConnectionId()); 2067 destInfo.setOperationType(DestinationInfo.REMOVE_OPERATION_TYPE); 2068 destInfo.setDestination(destination); 2069 destInfo.setTimeout(0); 2070 syncSendPacket(destInfo); 2071 } 2072 2073 public boolean isDeleted(ActiveMQDestination dest) { 2074 2075 // If we are not watching the advisories.. then 2076 // we will assume that the temp destination does exist. 2077 if (advisoryConsumer == null) { 2078 return false; 2079 } 2080 2081 return !activeTempDestinations.containsValue(dest); 2082 } 2083 2084 public boolean isCopyMessageOnSend() { 2085 return copyMessageOnSend; 2086 } 2087 2088 public LongSequenceGenerator getLocalTransactionIdGenerator() { 2089 return localTransactionIdGenerator; 2090 } 2091 2092 public boolean isUseCompression() { 2093 return useCompression; 2094 } 2095 2096 /** 2097 * Enables the use of compression of the message bodies 2098 */ 2099 public void setUseCompression(boolean useCompression) { 2100 this.useCompression = useCompression; 2101 } 2102 2103 public void destroyDestination(ActiveMQDestination destination) throws JMSException { 2104 2105 checkClosedOrFailed(); 2106 ensureConnectionInfoSent(); 2107 2108 DestinationInfo info = new DestinationInfo(); 2109 info.setConnectionId(this.info.getConnectionId()); 2110 info.setOperationType(DestinationInfo.REMOVE_OPERATION_TYPE); 2111 info.setDestination(destination); 2112 info.setTimeout(0); 2113 syncSendPacket(info); 2114 } 2115 2116 public boolean isDispatchAsync() { 2117 return dispatchAsync; 2118 } 2119 2120 /** 2121 * Enables or disables the default setting of whether or not consumers have 2122 * their messages <a 2123 * href="http://activemq.apache.org/consumer-dispatch-async.html">dispatched 2124 * synchronously or asynchronously by the broker</a>. For non-durable 2125 * topics for example we typically dispatch synchronously by default to 2126 * minimize context switches which boost performance. However sometimes its 2127 * better to go slower to ensure that a single blocked consumer socket does 2128 * not block delivery to other consumers. 2129 * 2130 * @param asyncDispatch If true then consumers created on this connection 2131 * will default to having their messages dispatched 2132 * asynchronously. The default value is true. 2133 */ 2134 public void setDispatchAsync(boolean asyncDispatch) { 2135 this.dispatchAsync = asyncDispatch; 2136 } 2137 2138 public boolean isObjectMessageSerializationDefered() { 2139 return objectMessageSerializationDefered; 2140 } 2141 2142 /** 2143 * When an object is set on an ObjectMessage, the JMS spec requires the 2144 * object to be serialized by that set method. Enabling this flag causes the 2145 * object to not get serialized. The object may subsequently get serialized 2146 * if the message needs to be sent over a socket or stored to disk. 2147 */ 2148 public void setObjectMessageSerializationDefered(boolean objectMessageSerializationDefered) { 2149 this.objectMessageSerializationDefered = objectMessageSerializationDefered; 2150 } 2151 2152 /** 2153 * Unsubscribes a durable subscription that has been created by a client. 2154 * <P> 2155 * This method deletes the state being maintained on behalf of the 2156 * subscriber by its provider. 2157 * <P> 2158 * It is erroneous for a client to delete a durable subscription while there 2159 * is an active <CODE>MessageConsumer </CODE> or 2160 * <CODE>TopicSubscriber</CODE> for the subscription, or while a consumed 2161 * message is part of a pending transaction or has not been acknowledged in 2162 * the session. 2163 * 2164 * @param name the name used to identify this subscription 2165 * @throws JMSException if the session fails to unsubscribe to the durable 2166 * subscription due to some internal error. 2167 * @throws InvalidDestinationException if an invalid subscription name is 2168 * specified. 2169 * @since 1.1 2170 */ 2171 public void unsubscribe(String name) throws InvalidDestinationException, JMSException { 2172 checkClosedOrFailed(); 2173 RemoveSubscriptionInfo rsi = new RemoveSubscriptionInfo(); 2174 rsi.setConnectionId(getConnectionInfo().getConnectionId()); 2175 rsi.setSubscriptionName(name); 2176 rsi.setClientId(getConnectionInfo().getClientId()); 2177 syncSendPacket(rsi); 2178 } 2179 2180 /** 2181 * Internal send method optimized: - It does not copy the message - It can 2182 * only handle ActiveMQ messages. - You can specify if the send is async or 2183 * sync - Does not allow you to send /w a transaction. 2184 */ 2185 void send(ActiveMQDestination destination, ActiveMQMessage msg, MessageId messageId, int deliveryMode, int priority, long timeToLive, boolean async) throws JMSException { 2186 checkClosedOrFailed(); 2187 2188 if (destination.isTemporary() && isDeleted(destination)) { 2189 throw new JMSException("Cannot publish to a deleted Destination: " + destination); 2190 } 2191 2192 msg.setJMSDestination(destination); 2193 msg.setJMSDeliveryMode(deliveryMode); 2194 long expiration = 0L; 2195 2196 if (!isDisableTimeStampsByDefault()) { 2197 long timeStamp = System.currentTimeMillis(); 2198 msg.setJMSTimestamp(timeStamp); 2199 if (timeToLive > 0) { 2200 expiration = timeToLive + timeStamp; 2201 } 2202 } 2203 2204 msg.setJMSExpiration(expiration); 2205 msg.setJMSPriority(priority); 2206 msg.setJMSRedelivered(false); 2207 msg.setMessageId(messageId); 2208 msg.onSend(); 2209 msg.setProducerId(msg.getMessageId().getProducerId()); 2210 2211 if (LOG.isDebugEnabled()) { 2212 LOG.debug("Sending message: " + msg); 2213 } 2214 2215 if (async) { 2216 asyncSendPacket(msg); 2217 } else { 2218 syncSendPacket(msg); 2219 } 2220 } 2221 2222 protected void onConnectionControl(ConnectionControl command) { 2223 if (command.isFaultTolerant()) { 2224 this.optimizeAcknowledge = false; 2225 for (Iterator<ActiveMQSession> i = this.sessions.iterator(); i.hasNext();) { 2226 ActiveMQSession s = i.next(); 2227 s.setOptimizeAcknowledge(false); 2228 } 2229 } 2230 } 2231 2232 protected void onConsumerControl(ConsumerControl command) { 2233 if (command.isClose()) { 2234 for (ActiveMQSession session : this.sessions) { 2235 session.close(command.getConsumerId()); 2236 } 2237 } else { 2238 for (ActiveMQSession session : this.sessions) { 2239 session.setPrefetchSize(command.getConsumerId(), command.getPrefetch()); 2240 } 2241 for (ActiveMQConnectionConsumer connectionConsumer: connectionConsumers) { 2242 ConsumerInfo consumerInfo = connectionConsumer.getConsumerInfo(); 2243 if (consumerInfo.getConsumerId().equals(command.getConsumerId())) { 2244 consumerInfo.setPrefetchSize(command.getPrefetch()); 2245 } 2246 } 2247 } 2248 } 2249 2250 protected void transportFailed(IOException error) { 2251 transportFailed.set(true); 2252 if (firstFailureError == null) { 2253 firstFailureError = error; 2254 } 2255 } 2256 2257 /** 2258 * Should a JMS message be copied to a new JMS Message object as part of the 2259 * send() method in JMS. This is enabled by default to be compliant with the 2260 * JMS specification. You can disable it if you do not mutate JMS messages 2261 * after they are sent for a performance boost 2262 */ 2263 public void setCopyMessageOnSend(boolean copyMessageOnSend) { 2264 this.copyMessageOnSend = copyMessageOnSend; 2265 } 2266 2267 @Override 2268 public String toString() { 2269 return "ActiveMQConnection {id=" + info.getConnectionId() + ",clientId=" + info.getClientId() + ",started=" + started.get() + "}"; 2270 } 2271 2272 protected BlobTransferPolicy createBlobTransferPolicy() { 2273 return new BlobTransferPolicy(); 2274 } 2275 2276 public int getProtocolVersion() { 2277 return protocolVersion.get(); 2278 } 2279 2280 public int getProducerWindowSize() { 2281 return producerWindowSize; 2282 } 2283 2284 public void setProducerWindowSize(int producerWindowSize) { 2285 this.producerWindowSize = producerWindowSize; 2286 } 2287 2288 public void setAuditDepth(int auditDepth) { 2289 connectionAudit.setAuditDepth(auditDepth); 2290 } 2291 2292 public void setAuditMaximumProducerNumber(int auditMaximumProducerNumber) { 2293 connectionAudit.setAuditMaximumProducerNumber(auditMaximumProducerNumber); 2294 } 2295 2296 protected void removeDispatcher(ActiveMQDispatcher dispatcher) { 2297 connectionAudit.removeDispatcher(dispatcher); 2298 } 2299 2300 protected boolean isDuplicate(ActiveMQDispatcher dispatcher, Message message) { 2301 return checkForDuplicates && connectionAudit.isDuplicate(dispatcher, message); 2302 } 2303 2304 protected void rollbackDuplicate(ActiveMQDispatcher dispatcher, Message message) { 2305 connectionAudit.rollbackDuplicate(dispatcher, message); 2306 } 2307 2308 public IOException getFirstFailureError() { 2309 return firstFailureError; 2310 } 2311 2312 protected void waitForTransportInterruptionProcessingToComplete() throws InterruptedException { 2313 if (!closed.get() && !transportFailed.get() && transportInterruptionProcessingComplete.get()>0) { 2314 LOG.warn("dispatch with outstanding dispatch interruption processing count " + transportInterruptionProcessingComplete.get()); 2315 signalInterruptionProcessingComplete(); 2316 } 2317 } 2318 2319 protected void transportInterruptionProcessingComplete() { 2320 if (transportInterruptionProcessingComplete.decrementAndGet() == 0) { 2321 signalInterruptionProcessingComplete(); 2322 } 2323 } 2324 2325 private void signalInterruptionProcessingComplete() { 2326 if (LOG.isDebugEnabled()) { 2327 LOG.debug("transportInterruptionProcessingComplete: " + transportInterruptionProcessingComplete.get() 2328 + " for:" + this.getConnectionInfo().getConnectionId()); 2329 } 2330 2331 FailoverTransport failoverTransport = transport.narrow(FailoverTransport.class); 2332 if (failoverTransport != null) { 2333 failoverTransport.connectionInterruptProcessingComplete(this.getConnectionInfo().getConnectionId()); 2334 if (LOG.isDebugEnabled()) { 2335 LOG.debug("notified failover transport (" + failoverTransport 2336 + ") of interruption completion for: " + this.getConnectionInfo().getConnectionId()); 2337 } 2338 } 2339 transportInterruptionProcessingComplete.set(0); 2340 } 2341 2342 private void signalInterruptionProcessingNeeded() { 2343 FailoverTransport failoverTransport = transport.narrow(FailoverTransport.class); 2344 if (failoverTransport != null) { 2345 failoverTransport.getStateTracker().transportInterrupted(this.getConnectionInfo().getConnectionId()); 2346 if (LOG.isDebugEnabled()) { 2347 LOG.debug("notified failover transport (" + failoverTransport 2348 + ") of pending interruption processing for: " + this.getConnectionInfo().getConnectionId()); 2349 } 2350 } 2351 } 2352 2353 /* 2354 * specify the amount of time in milliseconds that a consumer with a transaction pending recovery 2355 * will wait to receive re dispatched messages. 2356 * default value is 0 so there is no wait by default. 2357 */ 2358 public void setConsumerFailoverRedeliveryWaitPeriod(long consumerFailoverRedeliveryWaitPeriod) { 2359 this.consumerFailoverRedeliveryWaitPeriod = consumerFailoverRedeliveryWaitPeriod; 2360 } 2361 2362 public long getConsumerFailoverRedeliveryWaitPeriod() { 2363 return consumerFailoverRedeliveryWaitPeriod; 2364 } 2365 2366 protected Scheduler getScheduler() throws JMSException { 2367 Scheduler result = scheduler; 2368 if (result == null) { 2369 if (isClosing() || isClosed()) { 2370 // without lock contention report the closing state 2371 throw new ConnectionClosedException(); 2372 } 2373 synchronized (this) { 2374 result = scheduler; 2375 if (result == null) { 2376 checkClosed(); 2377 try { 2378 result = new Scheduler("ActiveMQConnection["+info.getConnectionId().getValue()+"] Scheduler"); 2379 result.start(); 2380 scheduler = result; 2381 } catch(Exception e) { 2382 throw JMSExceptionSupport.create(e); 2383 } 2384 } 2385 } 2386 } 2387 return result; 2388 } 2389 2390 protected ThreadPoolExecutor getExecutor() { 2391 return this.executor; 2392 } 2393 2394 protected CopyOnWriteArrayList<ActiveMQSession> getSessions() { 2395 return sessions; 2396 } 2397 2398 /** 2399 * @return the checkForDuplicates 2400 */ 2401 public boolean isCheckForDuplicates() { 2402 return this.checkForDuplicates; 2403 } 2404 2405 /** 2406 * @param checkForDuplicates the checkForDuplicates to set 2407 */ 2408 public void setCheckForDuplicates(boolean checkForDuplicates) { 2409 this.checkForDuplicates = checkForDuplicates; 2410 } 2411 2412 public boolean isTransactedIndividualAck() { 2413 return transactedIndividualAck; 2414 } 2415 2416 public void setTransactedIndividualAck(boolean transactedIndividualAck) { 2417 this.transactedIndividualAck = transactedIndividualAck; 2418 } 2419 2420 public boolean isNonBlockingRedelivery() { 2421 return nonBlockingRedelivery; 2422 } 2423 2424 public void setNonBlockingRedelivery(boolean nonBlockingRedelivery) { 2425 this.nonBlockingRedelivery = nonBlockingRedelivery; 2426 } 2427 2428 public boolean isRmIdFromConnectionId() { 2429 return rmIdFromConnectionId; 2430 } 2431 2432 public void setRmIdFromConnectionId(boolean rmIdFromConnectionId) { 2433 this.rmIdFromConnectionId = rmIdFromConnectionId; 2434 } 2435 2436 /** 2437 * Removes any TempDestinations that this connection has cached, ignoring 2438 * any exceptions generated because the destination is in use as they should 2439 * not be removed. 2440 * Used from a pooled connection, b/c it will not be explicitly closed. 2441 */ 2442 public void cleanUpTempDestinations() { 2443 2444 if (this.activeTempDestinations == null || this.activeTempDestinations.isEmpty()) { 2445 return; 2446 } 2447 2448 Iterator<ConcurrentMap.Entry<ActiveMQTempDestination, ActiveMQTempDestination>> entries 2449 = this.activeTempDestinations.entrySet().iterator(); 2450 while(entries.hasNext()) { 2451 ConcurrentMap.Entry<ActiveMQTempDestination, ActiveMQTempDestination> entry = entries.next(); 2452 try { 2453 // Only delete this temp destination if it was created from this connection. The connection used 2454 // for the advisory consumer may also have a reference to this temp destination. 2455 ActiveMQTempDestination dest = entry.getValue(); 2456 String thisConnectionId = (info.getConnectionId() == null) ? "" : info.getConnectionId().toString(); 2457 if (dest.getConnectionId() != null && dest.getConnectionId().equals(thisConnectionId)) { 2458 this.deleteTempDestination(entry.getValue()); 2459 } 2460 } catch (Exception ex) { 2461 // the temp dest is in use so it can not be deleted. 2462 // it is ok to leave it to connection tear down phase 2463 } 2464 } 2465 } 2466 2467 /** 2468 * Sets the Connection wide RedeliveryPolicyMap for handling messages that are being rolled back. 2469 * @param redeliveryPolicyMap the redeliveryPolicyMap to set 2470 */ 2471 public void setRedeliveryPolicyMap(RedeliveryPolicyMap redeliveryPolicyMap) { 2472 this.redeliveryPolicyMap = redeliveryPolicyMap; 2473 } 2474 2475 /** 2476 * Gets the Connection's configured RedeliveryPolicyMap which will be used by all the 2477 * Consumers when dealing with transaction messages that have been rolled back. 2478 * 2479 * @return the redeliveryPolicyMap 2480 */ 2481 public RedeliveryPolicyMap getRedeliveryPolicyMap() { 2482 return redeliveryPolicyMap; 2483 } 2484 2485 public int getMaxThreadPoolSize() { 2486 return maxThreadPoolSize; 2487 } 2488 2489 public void setMaxThreadPoolSize(int maxThreadPoolSize) { 2490 this.maxThreadPoolSize = maxThreadPoolSize; 2491 } 2492 2493 /** 2494 * Enable enforcement of QueueConnection semantics. 2495 * 2496 * @return this object, useful for chaining 2497 */ 2498 ActiveMQConnection enforceQueueOnlyConnection() { 2499 this.queueOnlyConnection = true; 2500 return this; 2501 } 2502 2503 public RejectedExecutionHandler getRejectedTaskHandler() { 2504 return rejectedTaskHandler; 2505 } 2506 2507 public void setRejectedTaskHandler(RejectedExecutionHandler rejectedTaskHandler) { 2508 this.rejectedTaskHandler = rejectedTaskHandler; 2509 } 2510 2511 /** 2512 * Gets the configured time interval that is used to force all MessageConsumers that have optimizedAcknowledge enabled 2513 * to send an ack for any outstanding Message Acks. By default this value is set to zero meaning that the consumers 2514 * will not do any background Message acknowledgment. 2515 * 2516 * @return the scheduledOptimizedAckInterval 2517 */ 2518 public long getOptimizedAckScheduledAckInterval() { 2519 return optimizedAckScheduledAckInterval; 2520 } 2521 2522 /** 2523 * Sets the amount of time between scheduled sends of any outstanding Message Acks for consumers that 2524 * have been configured with optimizeAcknowledge enabled. 2525 * 2526 * @param optimizedAckScheduledAckInterval the scheduledOptimizedAckInterval to set 2527 */ 2528 public void setOptimizedAckScheduledAckInterval(long optimizedAckScheduledAckInterval) { 2529 this.optimizedAckScheduledAckInterval = optimizedAckScheduledAckInterval; 2530 } 2531 2532 /** 2533 * @return true if MessageConsumer instance will check for expired messages before dispatch. 2534 */ 2535 public boolean isConsumerExpiryCheckEnabled() { 2536 return consumerExpiryCheckEnabled; 2537 } 2538 2539 /** 2540 * Controls whether message expiration checking is done in each MessageConsumer 2541 * prior to dispatching a message. Disabling this check can lead to consumption 2542 * of expired messages. 2543 * 2544 * @param consumerExpiryCheckEnabled 2545 * controls whether expiration checking is done prior to dispatch. 2546 */ 2547 public void setConsumerExpiryCheckEnabled(boolean consumerExpiryCheckEnabled) { 2548 this.consumerExpiryCheckEnabled = consumerExpiryCheckEnabled; 2549 } 2550 2551 public List<String> getTrustedPackages() { 2552 return trustedPackages; 2553 } 2554 2555 public void setTrustedPackages(List<String> trustedPackages) { 2556 this.trustedPackages = trustedPackages; 2557 } 2558 2559 public boolean isTrustAllPackages() { 2560 return trustAllPackages; 2561 } 2562 2563 public void setTrustAllPackages(boolean trustAllPackages) { 2564 this.trustAllPackages = trustAllPackages; 2565 } 2566 2567 public int getConnectResponseTimeout() { 2568 return connectResponseTimeout; 2569 } 2570 2571 public void setConnectResponseTimeout(int connectResponseTimeout) { 2572 this.connectResponseTimeout = connectResponseTimeout; 2573 } 2574}