/* * * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. * */ package org.apache.qpid.client.protocol; 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.ThreadFactory; import java.util.concurrent.TimeUnit; import org.apache.mina.filter.codec.ProtocolCodecException; 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.failover.FailoverException; import org.apache.qpid.client.failover.FailoverHandler; import org.apache.qpid.client.failover.FailoverState; import org.apache.qpid.client.state.AMQState; 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.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.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.protocol.ProtocolEngine; import org.apache.qpid.thread.Threading; import org.apache.qpid.transport.NetworkDriver; import org.apache.qpid.transport.network.io.IoTransport; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * 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 * specific event model of AMQP, by revealing the type of the received events (from decoded data), and passing the * event on to more specific handlers for the type. In this sense, it channels the richer event model of AMQP, * expressed in terms of methods and so on, through the cruder, general purpose event model of MINA, expressed in * terms of "message received" and so on. * *

There is a 1:1 mapping between an AMQProtocolHandler and an {@link AMQConnection}. The connection class is * exposed to the end user of the AMQP client API, and also implements the JMS Connection API, so provides the public * API calls through which an individual connection can be manipulated. This protocol handler talks to the network * through MINA, in a behind the scenes role; it is not an exposed part of the client API. * *

There is a 1:many mapping between an AMQProtocolHandler and a set of {@link AMQSession}s. At the MINA level, * there is one session per connection. At the AMQP level there can be many channels which are also called sessions in * JMS parlance. The {@link AMQSession}s are managed through an {@link AMQProtocolSession} instance. The protocol * session is similar to the MINA per-connection session, except that it can span the lifecycle of multiple MINA sessions * in the event of failover. See below for more information about this. * *

Mina provides a session container that can be used to store/retrieve arbitrary objects as String named * attributes. A more convenient, type-safe, container for session data is provided in the form of * {@link AMQProtocolSession}. * *

A common way to use MINA is to have a single instance of the event handler, and for MINA to pass in its session * object with every event, and for per-connection data to be held in the MINA session (perhaps using a type-safe wrapper * as described above). This event handler is different, because dealing with failover complicates things. To the * end client of an AMQConnection, a failed over connection is still handled through the same connection instance, but * behind the scenes a new transport connection, and MINA session will have been created. The MINA session object cannot * be used to track the state of the fail-over process, because it is destroyed and a new one is created, as the old * connection is shutdown and a new one created. For this reason, an AMQProtocolHandler is created per AMQConnection * and the protocol session data is held outside of the MINA IOSession. * *

This handler is responsible for setting up the filter chain to filter all events for this handler through. * The filter chain is set up as a stack of event handers that perform the following functions (working upwards from * the network traffic at the bottom), handing off incoming events to an asynchronous thread pool to do the work, * optionally handling secure sockets encoding/decoding, encoding/decoding the AMQP format itself. * *

*
CRC Card
Responsibilities Collaborations *
Maintain fail-over state. *
*
* * @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 * be merged, although there is sense in keeping the session model separate. Will clarify things by having data * held per protocol handler, per protocol session, per network connection, per channel, in separate classes, so * that lifecycles of the fields match lifecycles of their containing objects. */ 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. */ private AMQConnection _connection; /** Our wrapper for a protocol session that provides access to session values in a typesafe manner. */ private volatile AMQProtocolSession _protocolSession; /** Holds the state of the protocol session. */ private AMQStateManager _stateManager; /** Holds the method listeners, */ private final CopyOnWriteArraySet _frameListeners = new CopyOnWriteArraySet(); /** * We create the failover handler when the session is created since it needs a reference to the IoSession in order * to be able to send errors during failover back to the client application. The session won't be available in the * case where we failing over due to a Connection.Redirect message from the broker. */ private FailoverHandler _failoverHandler; /** * This flag is used to track whether failover is being attempted. It is used to prevent the application constantly * attempting failover where it is failing. */ private FailoverState _failoverState = FailoverState.NOT_STARTED; /** Used to provide a condition to wait upon for operations that are required to wait for failover to complete. */ private CountDownLatch _failoverLatch; /** The last failover exception that occurred */ private FailoverException _lastFailoverException; /** Defines the default timeout to use for synchronous protocol commands. */ 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. * * @param con The client connection that this is the event handler for. */ public AMQProtocolHandler(AMQConnection con) { _connection = con; _protocolSession = new AMQProtocolSession(this, _connection); _stateManager = new AMQStateManager(_protocolSession); _codecFactory = new AMQCodecFactory(false, _protocolSession); _poolReference.setThreadFactory(new ThreadFactory() { public Thread newThread(final Runnable runnable) { try { return Threading.getThreadFactory().createThread(runnable); } catch (Exception e) { throw new RuntimeException("Failed to create thread", e); } } }); _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 */ public void createIoTransportSession(BrokerDetails brokerDetail) { _protocolSession = new AMQProtocolSession(this, _connection); _stateManager.setProtocolSession(_protocolSession); IoTransport.connect_0_9(getProtocolSession(), brokerDetail.getHost(), brokerDetail.getPort(), brokerDetail.getBooleanProperty(BrokerDetails.OPTIONS_SSL)); _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 * where the connection died, an attempt to failover automatically to a new connection may be started. The failover * 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. * * @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 closed() { if (_connection.isClosed()) { _logger.debug("Session closed called by client"); } else { _logger.debug("Session closed called with failover state currently " + _failoverState); // reconnetablility was introduced here so as not to disturb the client as they have made their intentions // known through the policy settings. if ((_failoverState != FailoverState.IN_PROGRESS) && _connection.failoverAllowed()) { _logger.debug("FAILOVER STARTING"); if (_failoverState == FailoverState.NOT_STARTED) { _failoverState = FailoverState.IN_PROGRESS; startFailoverThread(); } else { _logger.debug("Not starting failover as state currently " + _failoverState); } } else { _logger.debug("Failover not allowed by policy."); // or already in progress? if (_logger.isDebugEnabled()) { _logger.debug(_connection.getFailoverPolicy().toString()); } if (_failoverState != FailoverState.IN_PROGRESS) { _logger.debug("sessionClose() not allowed to failover"); _connection.exceptionReceived(new AMQDisconnectedException( "Server closed connection and reconnection " + "not permitted.", _stateManager.getLastException())); } else { _logger.debug("sessionClose() failover in progress"); } } } _logger.debug("Protocol Session [" + this + "] closed"); } /** See {@link FailoverHandler} to see rationale for separate thread. */ private void startFailoverThread() { if(!_connection.isClosed()) { final Thread failoverThread; try { failoverThread = Threading.getThreadFactory().createThread(_failoverHandler); } catch (Exception e) { throw new RuntimeException("Failed to create thread", e); } 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 readerIdle() { _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 the NetworkDriver */ public void exception(Throwable cause) { if (_failoverState == FailoverState.NOT_STARTED) { // if (!(cause instanceof AMQUndeliveredException) && (!(cause instanceof AMQAuthenticationException))) if ((cause instanceof AMQConnectionClosedException) || cause instanceof IOException) { _logger.info("Exception caught therefore going to attempt failover: " + cause, cause); // this will attempt failover _networkDriver.close(); closed(); } else { if (cause instanceof ProtocolCodecException) { _logger.info("Protocol Exception caught NOT going to attempt failover as " + "cause isn't AMQConnectionClosedException: " + cause, cause); AMQException amqe = new AMQException("Protocol handler error: " + cause, cause); propagateExceptionToAllWaiters(amqe); } _connection.exceptionReceived(cause); } // FIXME Need to correctly handle other exceptions. Things like ... // if (cause instanceof AMQChannelClosedException) // which will cause the JMSSession to end due to a channel close and so that Session needs // to be removed from the map so we can correctly still call close without an exception when trying to close // the server closed session. See also CloseChannelMethodHandler as the sessionClose is never called on exception } // we reach this point if failover was attempted and failed therefore we need to let the calling app // know since we cannot recover the situation else if (_failoverState == FailoverState.FAILED) { _logger.error("Exception caught by protocol handler: " + cause, cause); // we notify the state manager of the error in case we have any clients waiting on a state // change. Those "waiters" will be interrupted and can handle the exception AMQException amqe = new AMQException("Protocol handler error: " + cause, cause); propagateExceptionToAllWaiters(amqe); _connection.exceptionReceived(cause); } } /** * There are two cases where we have other threads potentially blocking for events to be handled by this class. * These are for the state manager (waiting for a state change) or a frame listener (waiting for a particular type * of frame to arrive). When an error occurs we need to notify these waiters so that they can react appropriately. * * This should be called only when the exception is fatal for the connection. * * @param e the exception to propagate * * @see #propagateExceptionToFrameListeners */ public void propagateExceptionToAllWaiters(Exception e) { getStateManager().error(e); propagateExceptionToFrameListeners(e); } /** * This caters for the case where we only need to propagate an exception to the the frame listeners to interupt any * protocol level waits. * * This will would normally be used to notify all Frame Listeners that Failover is about to occur and they should * stop waiting and relinquish the Failover lock {@see FailoverHandler}. * * Once the {@link FailoverHandler} has re-established the connection then the listeners will be able to re-attempt * their protocol request and so listen again for the correct frame. * * @param e the exception to propagate */ public void propagateExceptionToFrameListeners(Exception e) { synchronized (_frameListeners) { if (!_frameListeners.isEmpty()) { final Iterator it = _frameListeners.iterator(); while (it.hasNext()) { final AMQMethodListener ml = (AMQMethodListener) it.next(); ml.error(e); } } } } public void notifyFailoverStarting() { // Set the last exception in the sync block to ensure the ordering with add. // either this gets done and the add does the ml.error // or the add completes first and the iterator below will do ml.error synchronized (_frameListeners) { _lastFailoverException = new FailoverException("Failing over about to start"); } //Only notify the Frame listeners that failover is going to occur as the State listeners shouldn't be // interrupted unless failover cannot restore the state. propagateExceptionToFrameListeners(_lastFailoverException); } public void failoverInProgress() { _lastFailoverException = null; } private static int _messageReceivedCount; public void received(ByteBuffer msg) { try { _readBytes += msg.remaining(); final ArrayList dataBlocks = _codecFactory.getDecoder().decodeBuffer(msg); Job.fireAsynchEvent(_poolReference.getPool(), _readJob, new Runnable() { 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(); _logger.info("Broker suggested using protocol version:" + _suggestedProtocolVersion); // 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); } } } }); } catch (Exception e) { propagateExceptionToFrameListeners(e); exception(e); } } public void methodBodyReceived(final int channelId, final AMQBody bodyFrame) throws AMQException { if (_logger.isDebugEnabled()) { _logger.debug("(" + System.identityHashCode(this) + ")Method frame received: " + bodyFrame); } final AMQMethodEvent evt = new AMQMethodEvent(channelId, (AMQMethodBody) bodyFrame); try { boolean wasAnyoneInterested = getStateManager().methodReceived(evt); synchronized (_frameListeners) { if (!_frameListeners.isEmpty()) { //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:" + _frameListeners, null); } } catch (AMQException e) { propagateExceptionToFrameListeners(e); exception(e); } } private static int _messagesOut; public StateWaiter createWaiter(Set states) throws AMQException { return getStateManager().createWaiter(states); } /** * Convenience method that writes a frame to the protocol session. Equivalent to calling * getProtocolSession().write(). * * @param frame the frame to write */ public void writeFrame(AMQDataBlock frame) { writeFrame(frame, false); } public void writeFrame(AMQDataBlock frame, boolean 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(); } } /** * Convenience method that writes a frame to the protocol session and waits for a particular response. Equivalent to * calling getProtocolSession().write() then waiting for the response. * * @param frame * @param listener the blocking listener. Note the calling thread will block. */ public AMQMethodEvent writeCommandFrameAndWaitForReply(AMQFrame frame, BlockingMethodFrameListener listener) throws AMQException, FailoverException { return writeCommandFrameAndWaitForReply(frame, listener, DEFAULT_SYNC_TIMEOUT); } /** * Convenience method that writes a frame to the protocol session and waits for a particular response. Equivalent to * calling getProtocolSession().write() then waiting for the response. * * @param frame * @param listener the blocking listener. Note the calling thread will block. */ public AMQMethodEvent writeCommandFrameAndWaitForReply(AMQFrame frame, BlockingMethodFrameListener listener, long timeout) throws AMQException, FailoverException { try { synchronized (_frameListeners) { if (_lastFailoverException != null) { 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 } writeFrame(frame); return listener.blockForFrame(timeout); // When control resumes before this line, a reply will have been received // that matches the criteria defined in the blocking listener } finally { // If we don't removeKey the listener then no-one will _frameListeners.remove(listener); } } /** More convenient method to write a frame and wait for it's response. */ public AMQMethodEvent syncWrite(AMQFrame frame, Class responseClass) throws AMQException, FailoverException { return syncWrite(frame, responseClass, DEFAULT_SYNC_TIMEOUT); } /** More convenient method to write a frame and wait for it's response. */ public AMQMethodEvent syncWrite(AMQFrame frame, Class responseClass, long timeout) throws AMQException, FailoverException { return writeCommandFrameAndWaitForReply(frame, new SpecificMethodFrameListener(frame.getChannel(), responseClass), timeout); } public void closeSession(AMQSession session) throws AMQException { _protocolSession.closeSession(session); } /** * Closes the connection. * *

If a failover exception occurs whilst closing the connection it is ignored, as the connection is closed * anyway. * * @param timeout The timeout to wait for an acknowledgement to the close request. * * @throws AMQException If the close fails for any reason. */ public void closeConnection(long timeout) throws AMQException { 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); //If the connection is already closed then don't do a syncWrite if (!getStateManager().getCurrentState().equals(AMQState.CONNECTION_CLOSED)) { 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 _readBytes; } /** @return the number of bytes written to this protocol session */ public long getWrittenBytes() { return _writtenBytes; } public void failover(String host, int port) { _failoverHandler.setHost(host); _failoverHandler.setPort(port); // see javadoc for FailoverHandler to see rationale for separate thread startFailoverThread(); } public void blockUntilNotFailingOver() throws InterruptedException { synchronized(_failoverLatchChange) { if (_failoverLatch != null) { if(!_failoverLatch.await(MAXIMUM_STATE_WAIT_TIME, TimeUnit.MILLISECONDS)) { } } } } public AMQShortString generateQueueName() { return _protocolSession.generateQueueName(); } public CountDownLatch getFailoverLatch() { return _failoverLatch; } public void setFailoverLatch(CountDownLatch failoverLatch) { synchronized (_failoverLatchChange) { _failoverLatch = failoverLatch; } } public AMQConnection getConnection() { return _connection; } public AMQStateManager getStateManager() { return _stateManager; } public void setStateManager(AMQStateManager stateManager) { _stateManager = stateManager; _stateManager.setProtocolSession(_protocolSession); } public AMQProtocolSession getProtocolSession() { return _protocolSession; } FailoverState getFailoverState() { return _failoverState; } public void setFailoverState(FailoverState failoverState) { _failoverState = failoverState; } public byte getProtocolMajorVersion() { return _protocolSession.getProtocolMajorVersion(); } public byte getProtocolMinorVersion() { return _protocolSession.getProtocolMinorVersion(); } public MethodRegistry getMethodRegistry() { return _protocolSession.getMethodRegistry(); } public ProtocolVersion getProtocolVersion() { 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; } }