Package com.rabbitmq.client.impl
Class AMQConnection
java.lang.Object
com.rabbitmq.client.impl.ShutdownNotifierComponent
com.rabbitmq.client.impl.AMQConnection
- All Implemented Interfaces:
Connection
,NetworkConnection
,ShutdownNotifier
,Closeable
,AutoCloseable
- Direct Known Subclasses:
RecoveryAwareAMQConnection
public class AMQConnection
extends ShutdownNotifierComponent
implements Connection, NetworkConnection
Concrete class representing and managing an AMQP connection to a broker.
To create a broker connection, use ConnectionFactory
. See Connection
for an example.
-
Nested Class Summary
Nested ClassesModifier and TypeClassDescriptionprivate static class
private class
private class
-
Field Summary
FieldsModifier and TypeFieldDescriptionprivate final BlockingCell
<Object> Object used for blocking main application thread when doing all the necessary connection shutdown operationsprivate boolean
Flag indicating whether the client received Connection.Close message from the brokerprivate final AMQChannel
The special channel 0 (not managed by the_channelManager
)private ChannelManager
Object that manages a set of channelsprivate final ExceptionHandler
Handler for (uncaught) exceptions that crop up in theAMQConnection.MainLoop
.private final FrameHandler
Frame source/sinkprivate int
Maximum frame length, or zero if no limit is setprivate int
Currently-configured heart-beat interval, in seconds.private HeartbeatSender
Manages heart-beat sending for this connectionprivate boolean
Flag indicating we are still negotiating the connection in startprivate int
Count of socket-timeouts that have happened without any incoming framesprivate boolean
Flag controlling the main driver loop's terminationSaved server properties field from connection.startprivate final String
protected ConsumerWorkService
private final Collection
<BlockedListener> static final double
private final int
private final boolean
private static final Version
private final ExecutorService
private final CredentialsProvider
private final CredentialsRefreshService
private final ErrorOnWriteListener
private final AtomicBoolean
private final int
private final ScheduledExecutorService
private String
private Thread
private final AtomicBoolean
private static final org.slf4j.Logger
private Thread
private static final int
private final int
protected final MetricsCollector
protected final ObservationCollector
private final List
<RecoveryCanBeginListener> private final int
private final int
private final int
private final SaslConfig
private final ExecutorService
private final int
private static long
private ThreadFactory
private final TrafficListener
private final int
-
Constructor Summary
ConstructorsConstructorDescriptionAMQConnection
(ConnectionParams params, FrameHandler frameHandler) AMQConnection
(ConnectionParams params, FrameHandler frameHandler, MetricsCollector metricsCollector, ObservationCollector observationCollector) Construct a new connection -
Method Summary
Modifier and TypeMethodDescriptionvoid
abort()
Public API - Abort this connection and all its channels with theAMQP.REPLY_SUCCESS
close code and message 'OK'.void
abort
(int timeout) Public API - Abort this connection and all its channels with theAMQP.REPLY_SUCCESS
close code and message 'OK'.void
Public API - Abort this connection and all its channels.void
Public API - Abort this connection and all its channels.addBlockedListener
(BlockedCallback blockedCallback, UnblockedCallback unblockedCallback) Add a lambda-basedBlockedListener
.void
addBlockedListener
(BlockedListener listener) Add aBlockedListener
.void
private static void
Private API - check required preconditions and protocol invariantsprivate static boolean
checkUnsignedShort
(int value) void
Remove allBlockedListener
s.void
close()
Public API - Close this connection and all its channels with theAMQP.REPLY_SUCCESS
close code and message 'OK'.void
close
(int timeout) Public API - Close this connection and all its channels with theAMQP.REPLY_SUCCESS
close code and message 'OK'.void
Public API - Close this connection and all its channels.void
Protected API - Delegates tothe six-argument close method
, passing -1 for the timeout, and false for the abort flag.void
close
(int closeCode, String closeMessage, boolean initiatedByApplication, Throwable cause, int timeout, boolean abort) Protected API - Close this connection with the given code, message, source and timeout value for all the close operations to complete.void
Public API - Close this connection and all its channels.private void
protected void
configureChannelManager
(ChannelManager channelManager) (package private) ObservationCollector.ConnectionInfo
Public API - Create a new channel, using an internally allocated channel number.createChannel
(int channelNumber) Public API - Create a new channel, using the specified channel number if possible.(package private) AMQChannel
Retrieve a copy of the default table of client properties that will be sent to the server during connection startup.final void
disconnectChannel
(ChannelN channel) Protected API - respond, in the driver thread, to a ShutdownSignal.void
private APIprivate void
private void
void
flush()
Public API - flush the output buffersRetrieve the host.int
Get the negotiated maximum channel number.int
Get a copy of the map of client properties sent to the serverReturns client-provided connection name, if any.Protected API - retrieve the current ExceptionHandlerint
Get the negotiated maximum frame size.int
Get the negotiated heartbeat interval.private String
getId()
Public API - Returns a unique ID for this connection.Retrieve the local host.int
Retrieve the local port number.(package private) int
int
getPort()
Retrieve the port number.Retrieve the server properties.void
handleConnectionClose
(Command closeCommand) private void
void
private APIvoid
private APIboolean
handleReadFrame
(Frame frame) private APIprivate void
Called when a frame-read operation times outboolean
private void
private void
protected ChannelManager
instantiateChannelManager
(int channelMax, ThreadFactory threadFactory) void
ioLoopThread
(Thread thread) boolean
private boolean
protected int
negotiateChannelMax
(int requestedChannelMax, int serverMax) Private API, allows for easier simulation of bogus clients.private static int
negotiatedMaxValue
(int clientValue, int serverValue) private void
private boolean
boolean
Handles incoming control commands on channel zero.private void
boolean
removeBlockedListener
(BlockedListener listener) Remove aBlockedListener
.void
void
setHeartbeat
(int heartbeat) Protected API - set the heartbeat timeout.void
Public API - Sets a unique ID for this connection.void
setThreadFactory
(ThreadFactory threadFactory) Makes it possible to override thread factory that is used to instantiate connection network I/O loop.Protected API - causes all attached channels to terminate (shutdown) with a ShutdownSignal built from the argument, and stops this connection from accepting further work from the application.void
start()
Start up the connection, including the MainLoop thread.void
Package private API, allows for easier testing.private ShutdownSignalException
startShutdown
(Method reason, boolean initiatedByApplication, Throwable cause, boolean notifyRpc) toString()
boolean
boolean
Public APIvoid
writeFrame
(Frame f) Public API - sends a frame directly to the broker.Methods inherited from class com.rabbitmq.client.impl.ShutdownNotifierComponent
addShutdownListener, getCloseReason, isOpen, notifyListeners, removeShutdownListener, setShutdownCauseIfOpen
Methods inherited from class java.lang.Object
clone, equals, finalize, getClass, hashCode, notify, notifyAll, wait, wait, wait
Methods inherited from interface com.rabbitmq.client.Connection
openChannel, openChannel
Methods inherited from interface com.rabbitmq.client.ShutdownNotifier
addShutdownListener, getCloseReason, isOpen, notifyListeners, removeShutdownListener
-
Field Details
-
MAX_UNSIGNED_SHORT
private static final int MAX_UNSIGNED_SHORT- See Also:
-
LOGGER
private static final org.slf4j.Logger LOGGER -
CHANNEL_SHUTDOWN_TIMEOUT_MULTIPLIER
public static final double CHANNEL_SHUTDOWN_TIMEOUT_MULTIPLIER- See Also:
-
consumerWorkServiceExecutor
-
heartbeatExecutor
-
shutdownExecutor
-
mainLoopThread
-
ioLoopThreadSet
-
ioLoopThread
-
threadFactory
-
id
-
recoveryCanBeginListeners
-
errorOnWriteListener
-
workPoolTimeout
private final int workPoolTimeout -
finalShutdownStarted
-
connectionInfo
-
clientVersion
-
_channel0
The special channel 0 (not managed by the_channelManager
) -
_workService
-
_frameHandler
Frame source/sink -
_running
private volatile boolean _runningFlag controlling the main driver loop's termination -
_exceptionHandler
Handler for (uncaught) exceptions that crop up in theAMQConnection.MainLoop
. -
_appContinuation
Object used for blocking main application thread when doing all the necessary connection shutdown operations -
_brokerInitiatedShutdown
private volatile boolean _brokerInitiatedShutdownFlag indicating whether the client received Connection.Close message from the broker -
_inConnectionNegotiation
private volatile boolean _inConnectionNegotiationFlag indicating we are still negotiating the connection in start -
_heartbeatSender
Manages heart-beat sending for this connection -
_virtualHost
-
_clientProperties
-
saslConfig
-
requestedHeartbeat
private final int requestedHeartbeat -
requestedChannelMax
private final int requestedChannelMax -
requestedFrameMax
private final int requestedFrameMax -
handshakeTimeout
private final int handshakeTimeout -
shutdownTimeout
private final int shutdownTimeout -
credentialsProvider
-
blockedListeners
-
metricsCollector
-
observationCollector
-
channelRpcTimeout
private final int channelRpcTimeout -
channelShouldCheckRpcResponseType
private final boolean channelShouldCheckRpcResponseType -
trafficListener
-
credentialsRefreshService
-
_frameMax
private volatile int _frameMaxMaximum frame length, or zero if no limit is set -
_missedHeartbeats
private volatile int _missedHeartbeatsCount of socket-timeouts that have happened without any incoming frames -
_heartbeat
private volatile int _heartbeatCurrently-configured heart-beat interval, in seconds. 0 meaning none. -
_channelManager
Object that manages a set of channels -
_serverProperties
Saved server properties field from connection.start -
maxInboundMessageBodySize
private final int maxInboundMessageBodySize -
SOCKET_CLOSE_TIMEOUT
private static long SOCKET_CLOSE_TIMEOUT
-
-
Constructor Details
-
AMQConnection
-
AMQConnection
public AMQConnection(ConnectionParams params, FrameHandler frameHandler, MetricsCollector metricsCollector, ObservationCollector observationCollector) Construct a new connection- Parameters:
params
- parameters for it
-
-
Method Details
-
defaultClientProperties
Retrieve a copy of the default table of client properties that will be sent to the server during connection startup. This method is called when each new ConnectionFactory instance is constructed.- Returns:
- a map of client properties
- See Also:
-
disconnectChannel
Protected API - respond, in the driver thread, to a ShutdownSignal.- Parameters:
channel
- the channel to disconnect
-
ensureIsOpen
- Throws:
AlreadyClosedException
-
getAddress
Retrieve the host.- Specified by:
getAddress
in interfaceConnection
- Specified by:
getAddress
in interfaceNetworkConnection
- Returns:
- the hostname of the peer we're connected to.
-
getLocalAddress
Description copied from interface:NetworkConnection
Retrieve the local host.- Specified by:
getLocalAddress
in interfaceNetworkConnection
- Returns:
- the client socket address.
-
getPort
public int getPort()Retrieve the port number.- Specified by:
getPort
in interfaceConnection
- Specified by:
getPort
in interfaceNetworkConnection
- Returns:
- the port number of the peer we're connected to.
-
getLocalPort
public int getLocalPort()Description copied from interface:NetworkConnection
Retrieve the local port number.- Specified by:
getLocalPort
in interfaceNetworkConnection
- Returns:
- the client socket port number
-
getFrameHandler
-
getServerProperties
Retrieve the server properties.- Specified by:
getServerProperties
in interfaceConnection
- Returns:
- a map of the server properties. This typically includes the product name and version of the server.
-
createChannel0
AMQChannel createChannel0() -
initializeConsumerWorkService
private void initializeConsumerWorkService() -
initializeHeartbeatSender
private void initializeHeartbeatSender() -
start
Start up the connection, including the MainLoop thread. Sends the protocol version negotiation header, and runs through Connection.Start/.StartOk, Connection.Tune/.TuneOk, and then calls Connection.Open and waits for the OpenOk. Sets heart-beat and frame max values after tuning has taken place.- Throws:
IOException
- if an error is encountered either before, or during, protocol negotiation; sub-classesProtocolVersionMismatchException
andPossibleAuthenticationFailureException
will be thrown in the corresponding circumstances.AuthenticationFailureException
will be thrown if the broker closes the connection with ACCESS_REFUSED. If an exception is thrown, connection resources allocated can all be garbage collected when the connection object is no longer referenced.TimeoutException
-
instantiateChannelManager
-
configureChannelManager
-
startMainLoop
public void startMainLoop()Package private API, allows for easier testing. -
negotiateChannelMax
protected int negotiateChannelMax(int requestedChannelMax, int serverMax) Private API, allows for easier simulation of bogus clients. -
checkPreconditions
private static void checkPreconditions()Private API - check required preconditions and protocol invariants -
getChannelMax
public int getChannelMax()Get the negotiated maximum channel number. Usable channel numbers range from 1 to this number, inclusive.- Specified by:
getChannelMax
in interfaceConnection
- Returns:
- the maximum channel number permitted for this connection.
-
getFrameMax
public int getFrameMax()Get the negotiated maximum frame size.- Specified by:
getFrameMax
in interfaceConnection
- Returns:
- the maximum frame size, in octets; zero if unlimited
-
getHeartbeat
public int getHeartbeat()Get the negotiated heartbeat interval.- Specified by:
getHeartbeat
in interfaceConnection
- Returns:
- the heartbeat interval, in seconds; zero if none
-
setHeartbeat
public void setHeartbeat(int heartbeat) Protected API - set the heartbeat timeout. Should only be called during tuning. -
setThreadFactory
Makes it possible to override thread factory that is used to instantiate connection network I/O loop. Only necessary in the environments with restricted- Parameters:
threadFactory
- thread factory to use
-
getThreadFactory
- Returns:
- Thread factory used by this connection.
-
getClientProperties
Description copied from interface:Connection
Get a copy of the map of client properties sent to the server- Specified by:
getClientProperties
in interfaceConnection
- Returns:
- a copy of the map of client properties
-
getClientProvidedName
Description copied from interface:Connection
Returns client-provided connection name, if any. Note that the value returned does not uniquely identify a connection and cannot be used as a connection identifier in HTTP API requests.- Specified by:
getClientProvidedName
in interfaceConnection
- Returns:
- client-provided connection name, if any
- See Also:
-
getExceptionHandler
Protected API - retrieve the current ExceptionHandler- Specified by:
getExceptionHandler
in interfaceConnection
- See Also:
-
willShutDownConsumerExecutor
public boolean willShutDownConsumerExecutor()Public API- Returns:
- true if this work service instance uses its own consumerWorkServiceExecutor (as opposed to a shared one)
-
createChannel
Public API - Create a new channel, using the specified channel number if possible.Use
Connection.openChannel(int)
if you want to use anOptional
to deal with a value.- Specified by:
createChannel
in interfaceConnection
- Parameters:
channelNumber
- the channel number to allocate- Returns:
- a new channel descriptor, or null if this channel number is already in use
- Throws:
IOException
- if an I/O problem is encountered
-
createChannel
Public API - Create a new channel, using an internally allocated channel number. If automatic connection recovery is enabled, the channel returned by this method will beRecoverable
.Use
Connection.openChannel()
if you want to use anOptional
to deal with a value.- Specified by:
createChannel
in interfaceConnection
- Returns:
- a new channel descriptor, or null if none is available
- Throws:
IOException
- if an I/O problem is encountered
-
writeFrame
Public API - sends a frame directly to the broker.- Throws:
IOException
-
flush
Public API - flush the output buffers- Throws:
IOException
-
negotiatedMaxValue
private static int negotiatedMaxValue(int clientValue, int serverValue) -
checkUnsignedShort
private static boolean checkUnsignedShort(int value) -
handleReadFrame
private API -
isRunning
public boolean isRunning() -
hasBrokerInitiatedShutdown
public boolean hasBrokerInitiatedShutdown() -
readFrame
- Throws:
IOException
-
handleHeartbeatFailure
public void handleHeartbeatFailure()private API -
handleIoError
private API -
handleFailure
-
doFinalShutdown
public void doFinalShutdown()private API -
closeMainLoopThreadIfNecessary
private void closeMainLoopThreadIfNecessary() -
notInMainLoopThread
private boolean notInMainLoopThread() -
mainLoopReadThreadNotNull
private boolean mainLoopReadThreadNotNull() -
notifyRecoveryCanBeginListeners
private void notifyRecoveryCanBeginListeners() -
addRecoveryCanBeginListener
-
removeRecoveryCanBeginListener
-
handleSocketTimeout
Called when a frame-read operation times out- Throws:
MissedHeartbeatException
- if heart-beats have been missedSocketTimeoutException
-
processControlCommand
Handles incoming control commands on channel zero.- Throws:
IOException
- See Also:
-
handleConnectionClose
-
shutdown
public ShutdownSignalException shutdown(Method reason, boolean initiatedByApplication, Throwable cause, boolean notifyRpc) Protected API - causes all attached channels to terminate (shutdown) with a ShutdownSignal built from the argument, and stops this connection from accepting further work from the application.ShutdownListener
s for the connection are notified when the main loop terminates.- Parameters:
reason
- description of reason for the exceptioninitiatedByApplication
- true if caused by a client commandcause
- trigger exception which caused shutdownnotifyRpc
- true if outstanding rpc should be informed of shutdown- Returns:
- a shutdown signal built using the given arguments
-
startShutdown
private ShutdownSignalException startShutdown(Method reason, boolean initiatedByApplication, Throwable cause, boolean notifyRpc) -
finishShutdown
-
close
Public API - Close this connection and all its channels with theAMQP.REPLY_SUCCESS
close code and message 'OK'. Waits for all the close operations to complete.- Specified by:
close
in interfaceAutoCloseable
- Specified by:
close
in interfaceCloseable
- Specified by:
close
in interfaceConnection
- Throws:
IOException
- if an I/O problem is encountered
-
close
Public API - Close this connection and all its channels with theAMQP.REPLY_SUCCESS
close code and message 'OK'. This method behaves in a similar way asConnection.close()
, with the only difference that it waits with a provided timeout for all the close operations to complete. When timeout is reached the socket is forced to close.- Specified by:
close
in interfaceConnection
- Parameters:
timeout
- timeout (in milliseconds) for completing all the close-related operations, use -1 for infinity- Throws:
IOException
- if an I/O problem is encountered
-
close
Public API - Close this connection and all its channels. Waits for all the close operations to complete.- Specified by:
close
in interfaceConnection
- Parameters:
closeCode
- the close code (See under "Reply Codes" in the AMQP specification)closeMessage
- a message indicating the reason for closing the connection- Throws:
IOException
- if an I/O problem is encountered
-
close
Public API - Close this connection and all its channels. Waits with the given timeout for all the close operations to complete. When timeout is reached the socket is forced to close.- Specified by:
close
in interfaceConnection
- Parameters:
closeCode
- the close code (See under "Reply Codes" in the AMQP specification)closeMessage
- a message indicating the reason for closing the connectiontimeout
- timeout (in milliseconds) for completing all the close-related operations, use -1 for infinity- Throws:
IOException
- if an I/O problem is encountered
-
abort
public void abort()Public API - Abort this connection and all its channels with theAMQP.REPLY_SUCCESS
close code and message 'OK'. Forces the connection to close. Any encountered exceptions in the close operations are silently discarded.- Specified by:
abort
in interfaceConnection
-
abort
Public API - Abort this connection and all its channels. Forces the connection to close and waits for all the close operations to complete. Any encountered exceptions in the close operations are silently discarded.- Specified by:
abort
in interfaceConnection
- Parameters:
closeCode
- the close code (See under "Reply Codes" in the AMQP specification)closeMessage
- a message indicating the reason for closing the connection
-
abort
public void abort(int timeout) Public API - Abort this connection and all its channels with theAMQP.REPLY_SUCCESS
close code and message 'OK'. This method behaves in a similar way asConnection.abort()
, with the only difference that it waits with a provided timeout for all the close operations to complete. When timeout is reached the socket is forced to close.- Specified by:
abort
in interfaceConnection
- Parameters:
timeout
- timeout (in milliseconds) for completing all the close-related operations, use -1 for infinity
-
abort
Public API - Abort this connection and all its channels. Forces the connection to close and waits with the given timeout for all the close operations to complete. When timeout is reached the socket is forced to close. Any encountered exceptions in the close operations are silently discarded.- Specified by:
abort
in interfaceConnection
- Parameters:
closeCode
- the close code (See under "Reply Codes" in the AMQP specification)closeMessage
- a message indicating the reason for closing the connectiontimeout
- timeout (in milliseconds) for completing all the close-related operations, use -1 for infinity
-
close
public void close(int closeCode, String closeMessage, boolean initiatedByApplication, Throwable cause) throws IOException Protected API - Delegates tothe six-argument close method
, passing -1 for the timeout, and false for the abort flag.- Throws:
IOException
-
close
public void close(int closeCode, String closeMessage, boolean initiatedByApplication, Throwable cause, int timeout, boolean abort) throws IOException Protected API - Close this connection with the given code, message, source and timeout value for all the close operations to complete. Specifies if any encountered exceptions should be ignored.- Throws:
IOException
-
toString
-
getHostAddress
-
addBlockedListener
Description copied from interface:Connection
Add aBlockedListener
.- Specified by:
addBlockedListener
in interfaceConnection
- Parameters:
listener
- the listener to add
-
addBlockedListener
public BlockedListener addBlockedListener(BlockedCallback blockedCallback, UnblockedCallback unblockedCallback) Description copied from interface:Connection
Add a lambda-basedBlockedListener
.- Specified by:
addBlockedListener
in interfaceConnection
- Parameters:
blockedCallback
- the callback when the connection is blockedunblockedCallback
- the callback when the connection is unblocked- Returns:
- the listener that wraps the callback
- See Also:
-
removeBlockedListener
Description copied from interface:Connection
Remove aBlockedListener
.- Specified by:
removeBlockedListener
in interfaceConnection
- Parameters:
listener
- the listener to remove- Returns:
true
if the listener was found and removed,false
otherwise
-
clearBlockedListeners
public void clearBlockedListeners()Description copied from interface:Connection
Remove allBlockedListener
s.- Specified by:
clearBlockedListeners
in interfaceConnection
-
getId
Public API - Returns a unique ID for this connection. This ID must be unique, otherwise some services like the metrics collector won't work properly. This ID doesn't have to be provided by the client, services that require it will be assigned automatically if not set.- Specified by:
getId
in interfaceConnection
- Returns:
- unique ID for this connection.
-
setId
Public API - Sets a unique ID for this connection. This ID must be unique, otherwise some services like the metrics collector won't work properly. This ID doesn't have to be provided by the client, services that require it will be assigned automatically if not set.- Specified by:
setId
in interfaceConnection
-
ioLoopThread
-
getChannelRpcTimeout
public int getChannelRpcTimeout() -
willCheckRpcResponseType
public boolean willCheckRpcResponseType() -
getTrafficListener
-
getMaxInboundMessageBodySize
int getMaxInboundMessageBodySize() -
connectionInfo
ObservationCollector.ConnectionInfo connectionInfo()
-