diff options
Diffstat (limited to 'java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java')
| -rw-r--r-- | java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java | 547 |
1 files changed, 291 insertions, 256 deletions
diff --git a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java index e92817f713..a567c2c215 100644 --- a/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java +++ b/java/client/src/main/java/org/apache/qpid/client/protocol/AMQProtocolHandler.java @@ -20,24 +20,23 @@ */ package org.apache.qpid.client.protocol; -import org.apache.mina.common.IdleStatus; -import org.apache.mina.common.IoFilterChain; -import org.apache.mina.common.IoHandlerAdapter; -import org.apache.mina.common.IoSession; -import org.apache.mina.filter.ReadThrottleFilterBuilder; -import org.apache.mina.filter.SSLFilter; -import org.apache.mina.filter.WriteBufferLimitFilterBuilder; +import java.io.IOException; +import java.net.SocketAddress; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + import org.apache.mina.filter.codec.ProtocolCodecException; -import org.apache.mina.filter.codec.ProtocolCodecFilter; -import org.apache.mina.filter.executor.ExecutorFilter; import org.apache.qpid.AMQConnectionClosedException; import org.apache.qpid.AMQDisconnectedException; import org.apache.qpid.AMQException; import org.apache.qpid.AMQTimeoutException; import org.apache.qpid.client.AMQConnection; import org.apache.qpid.client.AMQSession; -import org.apache.qpid.client.SSLConfiguration; -import org.apache.qpid.client.configuration.ClientProperties; import org.apache.qpid.client.failover.FailoverException; import org.apache.qpid.client.failover.FailoverHandler; import org.apache.qpid.client.failover.FailoverState; @@ -46,23 +45,29 @@ import org.apache.qpid.client.state.AMQStateManager; import org.apache.qpid.client.state.StateWaiter; import org.apache.qpid.client.state.listener.SpecificMethodFrameListener; import org.apache.qpid.codec.AMQCodecFactory; -import org.apache.qpid.framing.*; +import org.apache.qpid.framing.AMQBody; +import org.apache.qpid.framing.AMQDataBlock; +import org.apache.qpid.framing.AMQFrame; +import org.apache.qpid.framing.AMQMethodBody; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.framing.ConnectionCloseBody; +import org.apache.qpid.framing.ConnectionCloseOkBody; +import org.apache.qpid.framing.HeartbeatBody; +import org.apache.qpid.framing.MethodRegistry; +import org.apache.qpid.framing.ProtocolInitiation; +import org.apache.qpid.framing.ProtocolVersion; import org.apache.qpid.jms.BrokerDetails; -import org.apache.qpid.pool.ReadWriteThreadModel; +import org.apache.qpid.pool.Job; +import org.apache.qpid.pool.ReferenceCountingExecutorService; import org.apache.qpid.protocol.AMQConstant; import org.apache.qpid.protocol.AMQMethodEvent; import org.apache.qpid.protocol.AMQMethodListener; -import org.apache.qpid.ssl.SSLContextFactory; +import org.apache.qpid.protocol.ProtocolEngine; +import org.apache.qpid.transport.NetworkDriver; import org.apache.qpid.transport.network.io.IoTransport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.util.Iterator; -import java.util.Set; -import java.util.concurrent.CopyOnWriteArraySet; -import java.util.concurrent.CountDownLatch; - /** * AMQProtocolHandler is the client side protocol handler for AMQP, it handles all protocol events received from the * network by MINA. The primary purpose of AMQProtocolHandler is to translate the generic event model of MINA into the @@ -102,17 +107,10 @@ import java.util.concurrent.CountDownLatch; * * <p/><table id="crc"><caption>CRC Card</caption> * <tr><th> Responsibilities <th> Collaborations - * <tr><td> Create the filter chain to filter this handlers events. - * <td> {@link ProtocolCodecFilter}, {@link SSLContextFactory}, {@link SSLFilter}, {@link ReadWriteThreadModel}. - * * <tr><td> Maintain fail-over state. * <tr><td> * </table> * - * @todo Explain the system property: amqj.shared_read_write_pool. How does putting the protocol codec filter before the - * async write filter make it a shared pool? The pooling filter uses the same thread pool for reading and writing - * anyway, see {@link org.apache.qpid.pool.PoolingFilter}, docs for comments. Will putting the protocol codec - * filter before it mean not doing the read/write asynchronously but in the main filter thread? * @todo Use a single handler instance, by shifting everything to do with the 'protocol session' state, including * failover state, into AMQProtocolSession, and tracking that from AMQConnection? The lifecycles of * AMQProtocolSesssion and AMQConnection will be the same, so if there is high cohesion between them, they could @@ -120,13 +118,15 @@ import java.util.concurrent.CountDownLatch; * held per protocol handler, per protocol session, per network connection, per channel, in seperate classes, so * that lifecycles of the fields match lifecycles of their containing objects. */ -public class AMQProtocolHandler extends IoHandlerAdapter +public class AMQProtocolHandler implements ProtocolEngine { /** Used for debugging. */ private static final Logger _logger = LoggerFactory.getLogger(AMQProtocolHandler.class); private static final Logger _protocolLogger = LoggerFactory.getLogger("qpid.protocol"); private static final boolean PROTOCOL_DEBUG = (System.getProperty("amqj.protocol.logging.level") != null); + private static final long MAXIMUM_STATE_WAIT_TIME = Long.parseLong(System.getProperty("amqj.MaximumStateWait", "30000")); + /** * The connection that this protocol handler is associated with. There is a 1-1 mapping between connection * instances and protocol handler instances. @@ -137,7 +137,7 @@ public class AMQProtocolHandler extends IoHandlerAdapter private volatile AMQProtocolSession _protocolSession; /** Holds the state of the protocol session. */ - private AMQStateManager _stateManager = new AMQStateManager(); + private AMQStateManager _stateManager; /** Holds the method listeners, */ private final CopyOnWriteArraySet<AMQMethodListener> _frameListeners = new CopyOnWriteArraySet<AMQMethodListener>(); @@ -162,7 +162,19 @@ public class AMQProtocolHandler extends IoHandlerAdapter private FailoverException _lastFailoverException; /** Defines the default timeout to use for synchronous protocol commands. */ - private final long DEFAULT_SYNC_TIMEOUT = 1000 * 30; + private final long DEFAULT_SYNC_TIMEOUT = Long.getLong("amqj.default_syncwrite_timeout", 1000 * 30); + + /** Object to lock on when changing the latch */ + private Object _failoverLatchChange = new Object(); + private AMQCodecFactory _codecFactory; + private Job _readJob; + private Job _writeJob; + private ReferenceCountingExecutorService _poolReference = ReferenceCountingExecutorService.getInstance(); + private NetworkDriver _networkDriver; + private ProtocolVersion _suggestedProtocolVersion; + + private long _writtenBytes; + private long _readBytes; /** * Creates a new protocol handler, associated with the specified client connection instance. @@ -172,101 +184,30 @@ public class AMQProtocolHandler extends IoHandlerAdapter public AMQProtocolHandler(AMQConnection con) { _connection = con; - } - - /** - * Invoked by MINA when a MINA session for a new connection is created. This method sets up the filter chain on the - * session, which filters the events handled by this handler. The filter chain consists of, handing off events - * to an asynchronous thread pool, optionally encoding/decoding ssl, encoding/decoding AMQP. - * - * @param session The MINA session. - * - * @throws Exception Any underlying exceptions are allowed to fall through to MINA. - */ - public void sessionCreated(IoSession session) throws Exception - { - _logger.debug("Protocol session created for session " + System.identityHashCode(session)); - _failoverHandler = new FailoverHandler(this, session); - - final ProtocolCodecFilter pcf = new ProtocolCodecFilter(new AMQCodecFactory(false)); - - if (Boolean.getBoolean("amqj.shared_read_write_pool")) - { - session.getFilterChain().addBefore("AsynchronousWriteFilter", "protocolFilter", pcf); - } - else - { - session.getFilterChain().addLast("protocolFilter", pcf); - } - // we only add the SSL filter where we have an SSL connection - if (_connection.getSSLConfiguration() != null) - { - SSLConfiguration sslConfig = _connection.getSSLConfiguration(); - SSLContextFactory sslFactory = - new SSLContextFactory(sslConfig.getKeystorePath(), sslConfig.getKeystorePassword(), sslConfig.getCertType()); - SSLFilter sslFilter = new SSLFilter(sslFactory.buildClientContext()); - sslFilter.setUseClientMode(true); - session.getFilterChain().addBefore("protocolFilter", "ssl", sslFilter); - } - - try - { - ReadWriteThreadModel threadModel = ReadWriteThreadModel.getInstance(); - threadModel.getAsynchronousReadFilter().createNewJobForSession(session); - threadModel.getAsynchronousWriteFilter().createNewJobForSession(session); - } - catch (RuntimeException e) - { - _logger.error(e.getMessage(), e); - } - - if (Boolean.getBoolean(ClientProperties.PROTECTIO_PROP_NAME)) - { - try - { - //Add IO Protection Filters - IoFilterChain chain = session.getFilterChain(); - - session.getFilterChain().addLast("tempExecutorFilterForFilterBuilder", new ExecutorFilter()); - - ReadThrottleFilterBuilder readfilter = new ReadThrottleFilterBuilder(); - readfilter.setMaximumConnectionBufferSize(Integer.parseInt(System.getProperty( - ClientProperties.READ_BUFFER_LIMIT_PROP_NAME, ClientProperties.READ_BUFFER_LIMIT_DEFAULT))); - readfilter.attach(chain); - - WriteBufferLimitFilterBuilder writefilter = new WriteBufferLimitFilterBuilder(); - writefilter.setMaximumConnectionBufferSize(Integer.parseInt(System.getProperty( - ClientProperties.WRITE_BUFFER_LIMIT_PROP_NAME, ClientProperties.WRITE_BUFFER_LIMIT_DEFAULT))); - writefilter.attach(chain); - session.getFilterChain().remove("tempExecutorFilterForFilterBuilder"); - - _logger.info("Using IO Read/Write Filter Protection"); - } - catch (Exception e) - { - _logger.error("Unable to attach IO Read/Write Filter Protection :" + e.getMessage()); - } - } - _protocolSession = new AMQProtocolSession(this, session, _connection); - - _stateManager.setProtocolSession(_protocolSession); - - _protocolSession.init(); + _protocolSession = new AMQProtocolSession(this, _connection); + _stateManager = new AMQStateManager(_protocolSession); + _codecFactory = new AMQCodecFactory(false, _protocolSession); + _readJob = new Job(_poolReference, Job.MAX_JOB_EVENTS, true); + _writeJob = new Job(_poolReference, Job.MAX_JOB_EVENTS, false); + _poolReference.acquireExecutorService(); + _failoverHandler = new FailoverHandler(this); } /** * Called when we want to create a new IoTransport session - * @param brokerDetail + * @param brokerDetail */ public void createIoTransportSession(BrokerDetails brokerDetail) { _protocolSession = new AMQProtocolSession(this, _connection); _stateManager.setProtocolSession(_protocolSession); IoTransport.connect_0_9(getProtocolSession(), - brokerDetail.getHost(), brokerDetail.getPort()); + brokerDetail.getHost(), + brokerDetail.getPort(), + brokerDetail.useSSL()); _protocolSession.init(); } - + /** * Called when the network connection is closed. This can happen, either because the client explicitly requested * that the connection be closed, in which case nothing is done, or because the connection died. In the case @@ -274,16 +215,10 @@ public class AMQProtocolHandler extends IoHandlerAdapter * process will be started, provided that it is the clients policy to allow failover, and provided that a failover * has not already been started or failed. * - * <p/>It is important to note that when the connection dies this method may be called or {@link #exceptionCaught} - * may be called first followed by this method. This depends on whether the client was trying to send data at the - * time of the failure. - * - * @param session The MINA session. - * * @todo Clarify: presumably exceptionCaught is called when the client is sending during a connection failure and * not otherwise? The above comment doesn't make that clear. */ - public void sessionClosed(IoSession session) + public void closed() { if (_connection.isClosed()) { @@ -322,7 +257,8 @@ public class AMQProtocolHandler extends IoHandlerAdapter { _logger.debug("sessionClose() not allowed to failover"); _connection.exceptionReceived(new AMQDisconnectedException( - "Server closed connection and reconnection " + "not permitted.", null)); + "Server closed connection and reconnection " + "not permitted.", + _stateManager.getLastException())); } else { @@ -337,41 +273,37 @@ public class AMQProtocolHandler extends IoHandlerAdapter /** See {@link FailoverHandler} to see rationale for separate thread. */ private void startFailoverThread() { - Thread failoverThread = new Thread(_failoverHandler); - failoverThread.setName("Failover"); - // Do not inherit daemon-ness from current thread as this can be a daemon - // thread such as a AnonymousIoService thread. - failoverThread.setDaemon(false); - failoverThread.start(); + if(!_connection.isClosed()) + { + Thread failoverThread = new Thread(_failoverHandler); + failoverThread.setName("Failover"); + // Do not inherit daemon-ness from current thread as this can be a daemon + // thread such as a AnonymousIoService thread. + failoverThread.setDaemon(false); + failoverThread.start(); + } } - public void sessionIdle(IoSession session, IdleStatus status) throws Exception + public void readerIdle() { - _logger.debug("Protocol Session [" + this + ":" + session + "] idle: " + status); - if (IdleStatus.WRITER_IDLE.equals(status)) - { - // write heartbeat frame: - _logger.debug("Sent heartbeat"); - session.write(HeartbeatBody.FRAME); - HeartbeatDiagnostics.sent(); - } - else if (IdleStatus.READER_IDLE.equals(status)) - { - // failover: - HeartbeatDiagnostics.timeout(); - _logger.warn("Timed out while waiting for heartbeat from peer."); - session.close(); - } + _logger.debug("Protocol Session [" + this + "] idle: reader"); + // failover: + HeartbeatDiagnostics.timeout(); + _logger.warn("Timed out while waiting for heartbeat from peer."); + _networkDriver.close(); + } + + public void writerIdle() + { + _logger.debug("Protocol Session [" + this + "] idle: reader"); + writeFrame(HeartbeatBody.FRAME); + HeartbeatDiagnostics.sent(); } /** - * Invoked when any exception is thrown by a user IoHandler implementation or by MINA. If the cause is an - * IOException, MINA will close the connection automatically. - * - * @param session The MINA session. - * @param cause The exception that triggered this event. + * Invoked when any exception is thrown by the NetworkDriver */ - public void exceptionCaught(IoSession session, Throwable cause) + public void exception(Throwable cause) { if (_failoverState == FailoverState.NOT_STARTED) { @@ -379,9 +311,9 @@ public class AMQProtocolHandler extends IoHandlerAdapter if ((cause instanceof AMQConnectionClosedException) || cause instanceof IOException) { _logger.info("Exception caught therefore going to attempt failover: " + cause, cause); - // this will attemp failover - - sessionClosed(session); + // this will attempt failover + _networkDriver.close(); + closed(); } else { @@ -428,12 +360,12 @@ public class AMQProtocolHandler extends IoHandlerAdapter * @param e the exception to propagate * * @see #propagateExceptionToFrameListeners - * @see #propagateExceptionToStateWaiters */ public void propagateExceptionToAllWaiters(Exception e) { + getStateManager().error(e); + propagateExceptionToFrameListeners(e); - propagateExceptionToStateWaiters(e); } /** @@ -450,33 +382,20 @@ public class AMQProtocolHandler extends IoHandlerAdapter */ public void propagateExceptionToFrameListeners(Exception e) { - if (!_frameListeners.isEmpty()) + synchronized (_frameListeners) { - final Iterator it = _frameListeners.iterator(); - while (it.hasNext()) + if (!_frameListeners.isEmpty()) { - final AMQMethodListener ml = (AMQMethodListener) it.next(); - ml.error(e); + final Iterator it = _frameListeners.iterator(); + while (it.hasNext()) + { + final AMQMethodListener ml = (AMQMethodListener) it.next(); + ml.error(e); + } } } } - /** - * This caters for the case where we only need to propogate an exception to the the state manager to interupt any - * thing waiting for a state change. - * - * Currently (2008-07-15) the state manager is only used during 0-8/0-9 Connection establishement. - * - * Normally the state manager would not need to be notified without notifiying the frame listeners so in normal - * cases {@link #propagateExceptionToAllWaiters} would be the correct choice. - * - * @param e the exception to propagate - */ - public void propagateExceptionToStateWaiters(Exception e) - { - getStateManager().error(e); - } - public void notifyFailoverStarting() { // Set the last exception in the sync block to ensure the ordering with add. @@ -499,48 +418,81 @@ public class AMQProtocolHandler extends IoHandlerAdapter private static int _messageReceivedCount; - public void messageReceived(IoSession session, Object message) throws Exception - { - if (PROTOCOL_DEBUG) - { - _protocolLogger.info(String.format("RECV: [%s] %s", this, message)); - } - if(message instanceof AMQFrame) + public void received(ByteBuffer msg) + { + try { - final boolean debug = _logger.isDebugEnabled(); - final long msgNumber = ++_messageReceivedCount; + _readBytes += msg.remaining(); + final ArrayList<AMQDataBlock> dataBlocks = _codecFactory.getDecoder().decodeBuffer(msg); - if (debug && ((msgNumber % 1000) == 0)) + Job.fireAsynchEvent(_poolReference.getPool(), _readJob, new Runnable() { - _logger.debug("Received " + _messageReceivedCount + " protocol messages"); - } - - AMQFrame frame = (AMQFrame) message; - - final AMQBody bodyFrame = frame.getBodyFrame(); - HeartbeatDiagnostics.received(bodyFrame instanceof HeartbeatBody); - - bodyFrame.handle(frame.getChannel(), _protocolSession); - - _connection.bytesReceived(_protocolSession.getIoSession().getReadBytes()); + public void run() + { + // Decode buffer + + for (AMQDataBlock message : dataBlocks) + { + + try + { + if (PROTOCOL_DEBUG) + { + _protocolLogger.info(String.format("RECV: [%s] %s", this, message)); + } + + if(message instanceof AMQFrame) + { + final boolean debug = _logger.isDebugEnabled(); + final long msgNumber = ++_messageReceivedCount; + + if (debug && ((msgNumber % 1000) == 0)) + { + _logger.debug("Received " + _messageReceivedCount + " protocol messages"); + } + + AMQFrame frame = (AMQFrame) message; + + final AMQBody bodyFrame = frame.getBodyFrame(); + + HeartbeatDiagnostics.received(bodyFrame instanceof HeartbeatBody); + + bodyFrame.handle(frame.getChannel(), _protocolSession); + + _connection.bytesReceived(_readBytes); + } + else if (message instanceof ProtocolInitiation) + { + // We get here if the server sends a response to our initial protocol header + // suggesting an alternate ProtocolVersion; the server will then close the + // connection. + ProtocolInitiation protocolInit = (ProtocolInitiation) message; + _suggestedProtocolVersion = protocolInit.checkVersion(); + + // get round a bug in old versions of qpid whereby the connection is not closed + _stateManager.changeState(AMQState.CONNECTION_CLOSED); + } + } + catch (Exception e) + { + _logger.error("Exception processing frame", e); + propagateExceptionToFrameListeners(e); + exception(e); + } + } + } + }); } - else if (message instanceof ProtocolInitiation) + catch (Exception e) { - // We get here if the server sends a response to our initial protocol header - // suggesting an alternate ProtocolVersion; the server will then close the - // connection. - ProtocolInitiation protocolInit = (ProtocolInitiation) message; - ProtocolVersion pv = protocolInit.checkVersion(); - getConnection().setProtocolVersion(pv); - - // get round a bug in old versions of qpid whereby the connection is not closed - _stateManager.changeState(AMQState.CONNECTION_CLOSED); + propagateExceptionToFrameListeners(e); + exception(e); } } - public void methodBodyReceived(final int channelId, final AMQBody bodyFrame, IoSession session)//, final IoSession session) + public void methodBodyReceived(final int channelId, final AMQBody bodyFrame) throws AMQException { @@ -556,18 +508,20 @@ public class AMQProtocolHandler extends IoHandlerAdapter { boolean wasAnyoneInterested = getStateManager().methodReceived(evt); - if (!_frameListeners.isEmpty()) + synchronized (_frameListeners) { - //This iterator is safe from the error state as the frame listeners always add before they send so their - // will be ready and waiting for this response. - Iterator it = _frameListeners.iterator(); - while (it.hasNext()) + if (!_frameListeners.isEmpty()) { - final AMQMethodListener listener = (AMQMethodListener) it.next(); - wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested; + //This iterator is safe from the error state as the frame listeners always add before they send so their + // will be ready and waiting for this response. + Iterator it = _frameListeners.iterator(); + while (it.hasNext()) + { + final AMQMethodListener listener = (AMQMethodListener) it.next(); + wasAnyoneInterested = listener.methodReceived(evt) || wasAnyoneInterested; + } } } - if (!wasAnyoneInterested) { throw new AMQException(null, "AMQMethodEvent " + evt + " was not processed by any listener. Listeners:" @@ -578,32 +532,13 @@ public class AMQProtocolHandler extends IoHandlerAdapter { propagateExceptionToFrameListeners(e); - exceptionCaught(session, e); + exception(e); } } private static int _messagesOut; - public void messageSent(IoSession session, Object message) throws Exception - { - if (PROTOCOL_DEBUG) - { - _protocolLogger.debug(String.format("SEND: [%s] %s", this, message)); - } - - final long sentMessages = _messagesOut++; - - final boolean debug = _logger.isDebugEnabled(); - - if (debug && ((sentMessages % 1000) == 0)) - { - _logger.debug("Sent " + _messagesOut + " protocol messages"); - } - - _connection.bytesSent(session.getWrittenBytes()); - } - public StateWaiter createWaiter(Set<AMQState> states) throws AMQException { return getStateManager().createWaiter(states); @@ -617,12 +552,40 @@ public class AMQProtocolHandler extends IoHandlerAdapter */ public void writeFrame(AMQDataBlock frame) { - _protocolSession.writeFrame(frame); + writeFrame(frame, false); } public void writeFrame(AMQDataBlock frame, boolean wait) { - _protocolSession.writeFrame(frame, wait); + final ByteBuffer buf = frame.toNioByteBuffer(); + _writtenBytes += buf.remaining(); + Job.fireAsynchEvent(_poolReference.getPool(), _writeJob, new Runnable() + { + public void run() + { + _networkDriver.send(buf); + } + }); + if (PROTOCOL_DEBUG) + { + _protocolLogger.debug(String.format("SEND: [%s] %s", this, frame)); + } + + final long sentMessages = _messagesOut++; + + final boolean debug = _logger.isDebugEnabled(); + + if (debug && ((sentMessages % 1000) == 0)) + { + _logger.debug("Sent " + _messagesOut + " protocol messages"); + } + + _connection.bytesSent(_writtenBytes); + + if (wait) + { + _networkDriver.flush(); + } } /** @@ -657,9 +620,30 @@ public class AMQProtocolHandler extends IoHandlerAdapter throw _lastFailoverException; } + if(_stateManager.getCurrentState() == AMQState.CONNECTION_CLOSED || + _stateManager.getCurrentState() == AMQState.CONNECTION_CLOSING) + { + Exception e = _stateManager.getLastException(); + if (e != null) + { + if (e instanceof AMQException) + { + AMQException amqe = (AMQException) e; + + throw amqe.cloneForCurrentThread(); + } + else + { + throw new AMQException(AMQConstant.INTERNAL_ERROR, e.getMessage(), e); + } + } + } + _frameListeners.add(listener); + //FIXME: At this point here we should check or before add we should check _stateManager is in an open + // state so as we don't check we are likely just to time out here as I believe is being seen in QPID-1255 } - _protocolSession.writeFrame(frame); + writeFrame(frame); return listener.blockForFrame(timeout); // When control resumes before this line, a reply will have been received @@ -703,38 +687,42 @@ public class AMQProtocolHandler extends IoHandlerAdapter */ public void closeConnection(long timeout) throws AMQException { - getStateManager().changeState(AMQState.CONNECTION_CLOSING); - ConnectionCloseBody body = _protocolSession.getMethodRegistry().createConnectionCloseBody(AMQConstant.REPLY_SUCCESS.getCode(), // replyCode new AMQShortString("JMS client is closing the connection."), 0, 0); final AMQFrame frame = body.generateFrame(0); - try - { - syncWrite(frame, ConnectionCloseOkBody.class, timeout); - _protocolSession.closeProtocolSession(); - } - catch (AMQTimeoutException e) - { - _protocolSession.closeProtocolSession(false); - } - catch (FailoverException e) + //If the connection is already closed then don't do a syncWrite + if (!getStateManager().getCurrentState().equals(AMQState.CONNECTION_CLOSED)) { - _logger.debug("FailoverException interrupted connection close, ignoring as connection close anyway."); + try + { + syncWrite(frame, ConnectionCloseOkBody.class, timeout); + _networkDriver.close(); + closed(); + } + catch (AMQTimeoutException e) + { + closed(); + } + catch (FailoverException e) + { + _logger.debug("FailoverException interrupted connection close, ignoring as connection close anyway."); + } } + _poolReference.releaseExecutorService(); } /** @return the number of bytes read from this protocol session */ public long getReadBytes() { - return _protocolSession.getIoSession().getReadBytes(); + return _readBytes; } /** @return the number of bytes written to this protocol session */ public long getWrittenBytes() { - return _protocolSession.getIoSession().getWrittenBytes(); + return _writtenBytes; } public void failover(String host, int port) @@ -747,9 +735,15 @@ public class AMQProtocolHandler extends IoHandlerAdapter public void blockUntilNotFailingOver() throws InterruptedException { - if (_failoverLatch != null) + synchronized(_failoverLatchChange) { - _failoverLatch.await(); + if (_failoverLatch != null) + { + if(!_failoverLatch.await(MAXIMUM_STATE_WAIT_TIME, TimeUnit.MILLISECONDS)) + { + + } + } } } @@ -765,7 +759,10 @@ public class AMQProtocolHandler extends IoHandlerAdapter public void setFailoverLatch(CountDownLatch failoverLatch) { - _failoverLatch = failoverLatch; + synchronized (_failoverLatchChange) + { + _failoverLatch = failoverLatch; + } } public AMQConnection getConnection() @@ -781,6 +778,7 @@ public class AMQProtocolHandler extends IoHandlerAdapter public void setStateManager(AMQStateManager stateManager) { _stateManager = stateManager; + _stateManager.setProtocolSession(_protocolSession); } public AMQProtocolSession getProtocolSession() @@ -817,4 +815,41 @@ public class AMQProtocolHandler extends IoHandlerAdapter { return _protocolSession.getProtocolVersion(); } + + public SocketAddress getRemoteAddress() + { + return _networkDriver.getRemoteAddress(); + } + + public SocketAddress getLocalAddress() + { + return _networkDriver.getLocalAddress(); + } + + public void setNetworkDriver(NetworkDriver driver) + { + _networkDriver = driver; + } + + /** @param delay delay in seconds (not ms) */ + void initHeartbeats(int delay) + { + if (delay > 0) + { + getNetworkDriver().setMaxWriteIdle(delay); + getNetworkDriver().setMaxReadIdle(HeartbeatConfig.CONFIG.getTimeout(delay)); + HeartbeatDiagnostics.init(delay, HeartbeatConfig.CONFIG.getTimeout(delay)); + } + } + + public NetworkDriver getNetworkDriver() + { + return _networkDriver; + } + + public ProtocolVersion getSuggestedProtocolVersion() + { + return _suggestedProtocolVersion; + } + } |