connection.py 37 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011
  1. """Client (Connection)."""
  2. from __future__ import absolute_import, unicode_literals
  3. import os
  4. import socket
  5. import sys
  6. from collections import OrderedDict
  7. from contextlib import contextmanager
  8. from itertools import count, cycle
  9. from operator import itemgetter
  10. try:
  11. from ssl import CERT_NONE
  12. ssl_available = True
  13. except ImportError: # pragma: no cover
  14. CERT_NONE = None
  15. ssl_available = False
  16. # jython breaks on relative import for .exceptions for some reason
  17. # (Issue #112)
  18. from kombu import exceptions
  19. from .five import (
  20. bytes_if_py2, python_2_unicode_compatible, reraise, string_t, text_t,
  21. )
  22. from .log import get_logger
  23. from .resource import Resource
  24. from .transport import get_transport_cls, supports_librabbitmq
  25. from .utils.collections import HashedSeq
  26. from .utils.functional import dictfilter, lazy, retry_over_time, shufflecycle
  27. from .utils.objects import cached_property
  28. from .utils.url import as_url, parse_url, quote, urlparse
  29. __all__ = ('Connection', 'ConnectionPool', 'ChannelPool')
  30. logger = get_logger(__name__)
  31. roundrobin_failover = cycle
  32. resolve_aliases = {
  33. 'pyamqp': 'amqp',
  34. 'librabbitmq': 'amqp',
  35. }
  36. failover_strategies = {
  37. 'round-robin': roundrobin_failover,
  38. 'shuffle': shufflecycle,
  39. }
  40. _log_connection = os.environ.get('KOMBU_LOG_CONNECTION', False)
  41. _log_channel = os.environ.get('KOMBU_LOG_CHANNEL', False)
  42. @python_2_unicode_compatible
  43. class Connection(object):
  44. """A connection to the broker.
  45. Example:
  46. >>> Connection('amqp://guest:guest@localhost:5672//')
  47. >>> Connection('amqp://foo;amqp://bar',
  48. ... failover_strategy='round-robin')
  49. >>> Connection('redis://', transport_options={
  50. ... 'visibility_timeout': 3000,
  51. ... })
  52. >>> import ssl
  53. >>> Connection('amqp://', login_method='EXTERNAL', ssl={
  54. ... 'ca_certs': '/etc/pki/tls/certs/something.crt',
  55. ... 'keyfile': '/etc/something/system.key',
  56. ... 'certfile': '/etc/something/system.cert',
  57. ... 'cert_reqs': ssl.CERT_REQUIRED,
  58. ... })
  59. Note:
  60. SSL currently only works with the py-amqp, and qpid
  61. transports. For other transports you can use stunnel.
  62. Arguments:
  63. URL (str, Sequence): Broker URL, or a list of URLs.
  64. Keyword Arguments:
  65. ssl (bool): Use SSL to connect to the server. Default is ``False``.
  66. May not be supported by the specified transport.
  67. transport (Transport): Default transport if not specified in the URL.
  68. connect_timeout (float): Timeout in seconds for connecting to the
  69. server. May not be supported by the specified transport.
  70. transport_options (Dict): A dict of additional connection arguments to
  71. pass to alternate kombu channel implementations. Consult the
  72. transport documentation for available options.
  73. heartbeat (float): Heartbeat interval in int/float seconds.
  74. Note that if heartbeats are enabled then the
  75. :meth:`heartbeat_check` method must be called regularly,
  76. around once per second.
  77. Note:
  78. The connection is established lazily when needed. If you need the
  79. connection to be established, then force it by calling
  80. :meth:`connect`::
  81. >>> conn = Connection('amqp://')
  82. >>> conn.connect()
  83. and always remember to close the connection::
  84. >>> conn.release()
  85. These options have been replaced by the URL argument, but are still
  86. supported for backwards compatibility:
  87. :keyword hostname: Host name/address.
  88. NOTE: You cannot specify both the URL argument and use the hostname
  89. keyword argument at the same time.
  90. :keyword userid: Default user name if not provided in the URL.
  91. :keyword password: Default password if not provided in the URL.
  92. :keyword virtual_host: Default virtual host if not provided in the URL.
  93. :keyword port: Default port if not provided in the URL.
  94. """
  95. port = None
  96. virtual_host = '/'
  97. connect_timeout = 5
  98. _closed = None
  99. _connection = None
  100. _default_channel = None
  101. _transport = None
  102. _logger = False
  103. uri_prefix = None
  104. #: The cache of declared entities is per connection,
  105. #: in case the server loses data.
  106. declared_entities = None
  107. #: Iterator returning the next broker URL to try in the event
  108. #: of connection failure (initialized by :attr:`failover_strategy`).
  109. cycle = None
  110. #: Additional transport specific options,
  111. #: passed on to the transport instance.
  112. transport_options = None
  113. #: Strategy used to select new hosts when reconnecting after connection
  114. #: failure. One of "round-robin", "shuffle" or any custom iterator
  115. #: constantly yielding new URLs to try.
  116. failover_strategy = 'round-robin'
  117. #: Heartbeat value, currently only supported by the py-amqp transport.
  118. heartbeat = None
  119. resolve_aliases = resolve_aliases
  120. failover_strategies = failover_strategies
  121. hostname = userid = password = ssl = login_method = None
  122. def __init__(self, hostname='localhost', userid=None,
  123. password=None, virtual_host=None, port=None, insist=False,
  124. ssl=False, transport=None, connect_timeout=5,
  125. transport_options=None, login_method=None, uri_prefix=None,
  126. heartbeat=0, failover_strategy='round-robin',
  127. alternates=None, **kwargs):
  128. alt = [] if alternates is None else alternates
  129. # have to spell the args out, just to get nice docstrings :(
  130. params = self._initial_params = {
  131. 'hostname': hostname, 'userid': userid,
  132. 'password': password, 'virtual_host': virtual_host,
  133. 'port': port, 'insist': insist, 'ssl': ssl,
  134. 'transport': transport, 'connect_timeout': connect_timeout,
  135. 'login_method': login_method, 'heartbeat': heartbeat
  136. }
  137. if hostname and not isinstance(hostname, string_t):
  138. alt.extend(hostname)
  139. hostname = alt[0]
  140. if hostname and '://' in hostname:
  141. if ';' in hostname:
  142. alt.extend(hostname.split(';'))
  143. hostname = alt[0]
  144. if '+' in hostname[:hostname.index('://')]:
  145. # e.g. sqla+mysql://root:masterkey@localhost/
  146. params['transport'], params['hostname'] = \
  147. hostname.split('+', 1)
  148. self.uri_prefix = params['transport']
  149. else:
  150. transport = transport or urlparse(hostname).scheme
  151. if not get_transport_cls(transport).can_parse_url:
  152. # we must parse the URL
  153. url_params = parse_url(hostname)
  154. params.update(
  155. dictfilter(url_params),
  156. hostname=url_params['hostname'],
  157. )
  158. params['transport'] = transport
  159. self._init_params(**params)
  160. # fallback hosts
  161. self.alt = alt
  162. # keep text representation for .info
  163. # only temporary solution as this won't work when
  164. # passing a custom object (Issue celery/celery#3320).
  165. self._failover_strategy = failover_strategy or 'round-robin'
  166. self.failover_strategy = self.failover_strategies.get(
  167. self._failover_strategy) or self._failover_strategy
  168. if self.alt:
  169. self.cycle = self.failover_strategy(self.alt)
  170. next(self.cycle) # skip first entry
  171. if transport_options is None:
  172. transport_options = {}
  173. self.transport_options = transport_options
  174. if _log_connection: # pragma: no cover
  175. self._logger = True
  176. if uri_prefix:
  177. self.uri_prefix = uri_prefix
  178. self.declared_entities = set()
  179. def switch(self, url):
  180. """Switch connection parameters to use a new URL.
  181. Note:
  182. Does not reconnect!
  183. """
  184. self.close()
  185. self.declared_entities.clear()
  186. self._closed = False
  187. self._init_params(**dict(self._initial_params, **parse_url(url)))
  188. def maybe_switch_next(self):
  189. """Switch to next URL given by the current failover strategy."""
  190. if self.cycle:
  191. self.switch(next(self.cycle))
  192. def _init_params(self, hostname, userid, password, virtual_host, port,
  193. insist, ssl, transport, connect_timeout,
  194. login_method, heartbeat):
  195. transport = transport or 'amqp'
  196. if transport == 'amqp' and supports_librabbitmq():
  197. transport = 'librabbitmq'
  198. if transport == 'rediss' and ssl_available and not ssl:
  199. logger.warning(
  200. 'Secure redis scheme specified (rediss) with no ssl '
  201. 'options, defaulting to insecure SSL behaviour.'
  202. )
  203. ssl = {'ssl_cert_reqs': CERT_NONE}
  204. self.hostname = hostname
  205. self.userid = userid
  206. self.password = password
  207. self.login_method = login_method
  208. self.virtual_host = virtual_host or self.virtual_host
  209. self.port = port or self.port
  210. self.insist = insist
  211. self.connect_timeout = connect_timeout
  212. self.ssl = ssl
  213. self.transport_cls = transport
  214. self.heartbeat = heartbeat and float(heartbeat)
  215. def register_with_event_loop(self, loop):
  216. self.transport.register_with_event_loop(self.connection, loop)
  217. def _debug(self, msg, *args, **kwargs):
  218. if self._logger: # pragma: no cover
  219. fmt = '[Kombu connection:{id:#x}] {msg}'
  220. logger.debug(fmt.format(id=id(self), msg=text_t(msg)),
  221. *args, **kwargs)
  222. def connect(self):
  223. """Establish connection to server immediately."""
  224. self._closed = False
  225. return self.connection
  226. def channel(self):
  227. """Create and return a new channel."""
  228. self._debug('create channel')
  229. chan = self.transport.create_channel(self.connection)
  230. if _log_channel: # pragma: no cover
  231. from .utils.debug import Logwrapped
  232. return Logwrapped(chan, 'kombu.channel',
  233. '[Kombu channel:{0.channel_id}] ')
  234. return chan
  235. def heartbeat_check(self, rate=2):
  236. """Check heartbeats.
  237. Allow the transport to perform any periodic tasks
  238. required to make heartbeats work. This should be called
  239. approximately every second.
  240. If the current transport does not support heartbeats then
  241. this is a noop operation.
  242. Arguments:
  243. rate (int): Rate is how often the tick is called
  244. compared to the actual heartbeat value. E.g. if
  245. the heartbeat is set to 3 seconds, and the tick
  246. is called every 3 / 2 seconds, then the rate is 2.
  247. This value is currently unused by any transports.
  248. """
  249. return self.transport.heartbeat_check(self.connection, rate=rate)
  250. def drain_events(self, **kwargs):
  251. """Wait for a single event from the server.
  252. Arguments:
  253. timeout (float): Timeout in seconds before we give up.
  254. Raises:
  255. socket.timeout: if the timeout is exceeded.
  256. """
  257. return self.transport.drain_events(self.connection, **kwargs)
  258. def maybe_close_channel(self, channel):
  259. """Close given channel, but ignore connection and channel errors."""
  260. try:
  261. channel.close()
  262. except (self.connection_errors + self.channel_errors):
  263. pass
  264. def _do_close_self(self):
  265. # Close only connection and channel(s), but not transport.
  266. self.declared_entities.clear()
  267. if self._default_channel:
  268. self.maybe_close_channel(self._default_channel)
  269. if self._connection:
  270. try:
  271. self.transport.close_connection(self._connection)
  272. except self.connection_errors + (AttributeError, socket.error):
  273. pass
  274. self._connection = None
  275. def _close(self):
  276. """Really close connection, even if part of a connection pool."""
  277. self._do_close_self()
  278. self._do_close_transport()
  279. self._debug('closed')
  280. self._closed = True
  281. def _do_close_transport(self):
  282. if self._transport:
  283. self._transport.client = None
  284. self._transport = None
  285. def collect(self, socket_timeout=None):
  286. # amqp requires communication to close, we don't need that just
  287. # to clear out references, Transport._collect can also be implemented
  288. # by other transports that want fast after fork
  289. try:
  290. gc_transport = self._transport._collect
  291. except AttributeError:
  292. _timeo = socket.getdefaulttimeout()
  293. socket.setdefaulttimeout(socket_timeout)
  294. try:
  295. self._do_close_self()
  296. except socket.timeout:
  297. pass
  298. finally:
  299. socket.setdefaulttimeout(_timeo)
  300. else:
  301. gc_transport(self._connection)
  302. self._do_close_transport()
  303. self.declared_entities.clear()
  304. self._connection = None
  305. def release(self):
  306. """Close the connection (if open)."""
  307. self._close()
  308. close = release
  309. def ensure_connection(self, errback=None, max_retries=None,
  310. interval_start=2, interval_step=2, interval_max=30,
  311. callback=None, reraise_as_library_errors=True,
  312. timeout=None):
  313. """Ensure we have a connection to the server.
  314. If not retry establishing the connection with the settings
  315. specified.
  316. Arguments:
  317. errback (Callable): Optional callback called each time the
  318. connection can't be established. Arguments provided are
  319. the exception raised and the interval that will be
  320. slept ``(exc, interval)``.
  321. max_retries (int): Maximum number of times to retry.
  322. If this limit is exceeded the connection error
  323. will be re-raised.
  324. interval_start (float): The number of seconds we start
  325. sleeping for.
  326. interval_step (float): How many seconds added to the interval
  327. for each retry.
  328. interval_max (float): Maximum number of seconds to sleep between
  329. each retry.
  330. callback (Callable): Optional callback that is called for every
  331. internal iteration (1 s).
  332. timeout (int): Maximum amount of time in seconds to spend
  333. waiting for connection
  334. """
  335. def on_error(exc, intervals, retries, interval=0):
  336. round = self.completes_cycle(retries)
  337. if round:
  338. interval = next(intervals)
  339. if errback:
  340. errback(exc, interval)
  341. self.maybe_switch_next() # select next host
  342. return interval if round else 0
  343. ctx = self._reraise_as_library_errors
  344. if not reraise_as_library_errors:
  345. ctx = self._dummy_context
  346. with ctx():
  347. retry_over_time(self.connect, self.recoverable_connection_errors,
  348. (), {}, on_error, max_retries,
  349. interval_start, interval_step, interval_max,
  350. callback, timeout=timeout)
  351. return self
  352. @contextmanager
  353. def _reraise_as_library_errors(
  354. self,
  355. ConnectionError=exceptions.OperationalError,
  356. ChannelError=exceptions.OperationalError):
  357. try:
  358. yield
  359. except (ConnectionError, ChannelError):
  360. raise
  361. except self.recoverable_connection_errors as exc:
  362. reraise(ConnectionError, ConnectionError(text_t(exc)),
  363. sys.exc_info()[2])
  364. except self.recoverable_channel_errors as exc:
  365. reraise(ChannelError, ChannelError(text_t(exc)),
  366. sys.exc_info()[2])
  367. @contextmanager
  368. def _dummy_context(self):
  369. yield
  370. def completes_cycle(self, retries):
  371. """Return true if the cycle is complete after number of `retries`."""
  372. return not (retries + 1) % len(self.alt) if self.alt else True
  373. def revive(self, new_channel):
  374. """Revive connection after connection re-established."""
  375. if self._default_channel and new_channel is not self._default_channel:
  376. self.maybe_close_channel(self._default_channel)
  377. self._default_channel = None
  378. def ensure(self, obj, fun, errback=None, max_retries=None,
  379. interval_start=1, interval_step=1, interval_max=1,
  380. on_revive=None):
  381. """Ensure operation completes.
  382. Regardless of any channel/connection errors occurring.
  383. Retries by establishing the connection, and reapplying
  384. the function.
  385. Arguments:
  386. obj: The object to ensure an action on.
  387. fun (Callable): Method to apply.
  388. errback (Callable): Optional callback called each time the
  389. connection can't be established. Arguments provided are
  390. the exception raised and the interval that will
  391. be slept ``(exc, interval)``.
  392. max_retries (int): Maximum number of times to retry.
  393. If this limit is exceeded the connection error
  394. will be re-raised.
  395. interval_start (float): The number of seconds we start
  396. sleeping for.
  397. interval_step (float): How many seconds added to the interval
  398. for each retry.
  399. interval_max (float): Maximum number of seconds to sleep between
  400. each retry.
  401. on_revive (Callable): Optional callback called whenever
  402. revival completes successfully
  403. Examples:
  404. >>> from kombu import Connection, Producer
  405. >>> conn = Connection('amqp://')
  406. >>> producer = Producer(conn)
  407. >>> def errback(exc, interval):
  408. ... logger.error('Error: %r', exc, exc_info=1)
  409. ... logger.info('Retry in %s seconds.', interval)
  410. >>> publish = conn.ensure(producer, producer.publish,
  411. ... errback=errback, max_retries=3)
  412. >>> publish({'hello': 'world'}, routing_key='dest')
  413. """
  414. def _ensured(*args, **kwargs):
  415. got_connection = 0
  416. conn_errors = self.recoverable_connection_errors
  417. chan_errors = self.recoverable_channel_errors
  418. has_modern_errors = hasattr(
  419. self.transport, 'recoverable_connection_errors',
  420. )
  421. with self._reraise_as_library_errors():
  422. for retries in count(0): # for infinity
  423. try:
  424. return fun(*args, **kwargs)
  425. except conn_errors as exc:
  426. if got_connection and not has_modern_errors:
  427. # transport can not distinguish between
  428. # recoverable/irrecoverable errors, so we propagate
  429. # the error if it persists after a new connection
  430. # was successfully established.
  431. raise
  432. if max_retries is not None and retries > max_retries:
  433. raise
  434. self._debug('ensure connection error: %r',
  435. exc, exc_info=1)
  436. self.collect()
  437. errback and errback(exc, 0)
  438. remaining_retries = None
  439. if max_retries is not None:
  440. remaining_retries = max(max_retries - retries, 1)
  441. self.ensure_connection(
  442. errback,
  443. remaining_retries,
  444. interval_start, interval_step, interval_max,
  445. reraise_as_library_errors=False,
  446. )
  447. channel = self.default_channel
  448. obj.revive(channel)
  449. if on_revive:
  450. on_revive(channel)
  451. got_connection += 1
  452. except chan_errors as exc:
  453. if max_retries is not None and retries > max_retries:
  454. raise
  455. self._debug('ensure channel error: %r',
  456. exc, exc_info=1)
  457. errback and errback(exc, 0)
  458. _ensured.__name__ = bytes_if_py2('{0}(ensured)'.format(fun.__name__))
  459. _ensured.__doc__ = fun.__doc__
  460. _ensured.__module__ = fun.__module__
  461. return _ensured
  462. def autoretry(self, fun, channel=None, **ensure_options):
  463. """Decorator for functions supporting a ``channel`` keyword argument.
  464. The resulting callable will retry calling the function if
  465. it raises connection or channel related errors.
  466. The return value will be a tuple of ``(retval, last_created_channel)``.
  467. If a ``channel`` is not provided, then one will be automatically
  468. acquired (remember to close it afterwards).
  469. See Also:
  470. :meth:`ensure` for the full list of supported keyword arguments.
  471. Example:
  472. >>> channel = connection.channel()
  473. >>> try:
  474. ... ret, channel = connection.autoretry(
  475. ... publish_messages, channel)
  476. ... finally:
  477. ... channel.close()
  478. """
  479. channels = [channel]
  480. class Revival(object):
  481. __name__ = getattr(fun, '__name__', None)
  482. __module__ = getattr(fun, '__module__', None)
  483. __doc__ = getattr(fun, '__doc__', None)
  484. def __init__(self, connection):
  485. self.connection = connection
  486. def revive(self, channel):
  487. channels[0] = channel
  488. def __call__(self, *args, **kwargs):
  489. if channels[0] is None:
  490. self.revive(self.connection.default_channel)
  491. return fun(*args, channel=channels[0], **kwargs), channels[0]
  492. revive = Revival(self)
  493. return self.ensure(revive, revive, **ensure_options)
  494. def create_transport(self):
  495. return self.get_transport_cls()(client=self)
  496. def get_transport_cls(self):
  497. """Get the currently used transport class."""
  498. transport_cls = self.transport_cls
  499. if not transport_cls or isinstance(transport_cls, string_t):
  500. transport_cls = get_transport_cls(transport_cls)
  501. return transport_cls
  502. def clone(self, **kwargs):
  503. """Create a copy of the connection with same settings."""
  504. return self.__class__(**dict(self._info(resolve=False), **kwargs))
  505. def get_heartbeat_interval(self):
  506. return self.transport.get_heartbeat_interval(self.connection)
  507. def _info(self, resolve=True):
  508. transport_cls = self.transport_cls
  509. if resolve:
  510. transport_cls = self.resolve_aliases.get(
  511. transport_cls, transport_cls)
  512. D = self.transport.default_connection_params
  513. hostname = self.hostname or D.get('hostname')
  514. if self.uri_prefix:
  515. hostname = '%s+%s' % (self.uri_prefix, hostname)
  516. info = (
  517. ('hostname', hostname),
  518. ('userid', self.userid or D.get('userid')),
  519. ('password', self.password or D.get('password')),
  520. ('virtual_host', self.virtual_host or D.get('virtual_host')),
  521. ('port', self.port or D.get('port')),
  522. ('insist', self.insist),
  523. ('ssl', self.ssl),
  524. ('transport', transport_cls),
  525. ('connect_timeout', self.connect_timeout),
  526. ('transport_options', self.transport_options),
  527. ('login_method', self.login_method or D.get('login_method')),
  528. ('uri_prefix', self.uri_prefix),
  529. ('heartbeat', self.heartbeat),
  530. ('failover_strategy', self._failover_strategy),
  531. ('alternates', self.alt),
  532. )
  533. return info
  534. def info(self):
  535. """Get connection info."""
  536. return OrderedDict(self._info())
  537. def __eqhash__(self):
  538. return HashedSeq(self.transport_cls, self.hostname, self.userid,
  539. self.password, self.virtual_host, self.port,
  540. repr(self.transport_options))
  541. def as_uri(self, include_password=False, mask='**',
  542. getfields=itemgetter('port', 'userid', 'password',
  543. 'virtual_host', 'transport')):
  544. """Convert connection parameters to URL form."""
  545. hostname = self.hostname or 'localhost'
  546. if self.transport.can_parse_url:
  547. if self.uri_prefix:
  548. return '%s+%s' % (self.uri_prefix, hostname)
  549. return self.hostname
  550. if self.uri_prefix:
  551. return '%s+%s' % (self.uri_prefix, hostname)
  552. fields = self.info()
  553. port, userid, password, vhost, transport = getfields(fields)
  554. return as_url(
  555. transport, hostname, port, userid, password, quote(vhost),
  556. sanitize=not include_password, mask=mask,
  557. )
  558. def Pool(self, limit=None, **kwargs):
  559. """Pool of connections.
  560. See Also:
  561. :class:`ConnectionPool`.
  562. Arguments:
  563. limit (int): Maximum number of active connections.
  564. Default is no limit.
  565. Example:
  566. >>> connection = Connection('amqp://')
  567. >>> pool = connection.Pool(2)
  568. >>> c1 = pool.acquire()
  569. >>> c2 = pool.acquire()
  570. >>> c3 = pool.acquire()
  571. Traceback (most recent call last):
  572. File "<stdin>", line 1, in <module>
  573. File "kombu/connection.py", line 354, in acquire
  574. raise ConnectionLimitExceeded(self.limit)
  575. kombu.exceptions.ConnectionLimitExceeded: 2
  576. >>> c1.release()
  577. >>> c3 = pool.acquire()
  578. """
  579. return ConnectionPool(self, limit, **kwargs)
  580. def ChannelPool(self, limit=None, **kwargs):
  581. """Pool of channels.
  582. See Also:
  583. :class:`ChannelPool`.
  584. Arguments:
  585. limit (int): Maximum number of active channels.
  586. Default is no limit.
  587. Example:
  588. >>> connection = Connection('amqp://')
  589. >>> pool = connection.ChannelPool(2)
  590. >>> c1 = pool.acquire()
  591. >>> c2 = pool.acquire()
  592. >>> c3 = pool.acquire()
  593. Traceback (most recent call last):
  594. File "<stdin>", line 1, in <module>
  595. File "kombu/connection.py", line 354, in acquire
  596. raise ChannelLimitExceeded(self.limit)
  597. kombu.connection.ChannelLimitExceeded: 2
  598. >>> c1.release()
  599. >>> c3 = pool.acquire()
  600. """
  601. return ChannelPool(self, limit, **kwargs)
  602. def Producer(self, channel=None, *args, **kwargs):
  603. """Create new :class:`kombu.Producer` instance."""
  604. from .messaging import Producer
  605. return Producer(channel or self, *args, **kwargs)
  606. def Consumer(self, queues=None, channel=None, *args, **kwargs):
  607. """Create new :class:`kombu.Consumer` instance."""
  608. from .messaging import Consumer
  609. return Consumer(channel or self, queues, *args, **kwargs)
  610. def SimpleQueue(self, name, no_ack=None, queue_opts=None,
  611. queue_args=None,
  612. exchange_opts=None, channel=None, **kwargs):
  613. """Simple persistent queue API.
  614. Create new :class:`~kombu.simple.SimpleQueue`, using a channel
  615. from this connection.
  616. If ``name`` is a string, a queue and exchange will be automatically
  617. created using that name as the name of the queue and exchange,
  618. also it will be used as the default routing key.
  619. Arguments:
  620. name (str, kombu.Queue): Name of the queue/or a queue.
  621. no_ack (bool): Disable acknowledgments. Default is false.
  622. queue_opts (Dict): Additional keyword arguments passed to the
  623. constructor of the automatically created :class:`~kombu.Queue`.
  624. queue_args (Dict): Additional keyword arguments passed to the
  625. constructor of the automatically created :class:`~kombu.Queue`
  626. for setting implementation extensions (e.g., in RabbitMQ).
  627. exchange_opts (Dict): Additional keyword arguments passed to the
  628. constructor of the automatically created
  629. :class:`~kombu.Exchange`.
  630. channel (ChannelT): Custom channel to use. If not specified the
  631. connection default channel is used.
  632. """
  633. from .simple import SimpleQueue
  634. return SimpleQueue(channel or self, name, no_ack, queue_opts,
  635. queue_args,
  636. exchange_opts, **kwargs)
  637. def SimpleBuffer(self, name, no_ack=None, queue_opts=None,
  638. exchange_opts=None, channel=None, **kwargs):
  639. """Simple ephemeral queue API.
  640. Create new :class:`~kombu.simple.SimpleQueue` using a channel
  641. from this connection.
  642. See Also:
  643. Same as :meth:`SimpleQueue`, but configured with buffering
  644. semantics. The resulting queue and exchange will not be durable,
  645. also auto delete is enabled. Messages will be transient (not
  646. persistent), and acknowledgments are disabled (``no_ack``).
  647. """
  648. from .simple import SimpleBuffer
  649. return SimpleBuffer(channel or self, name, no_ack, queue_opts,
  650. exchange_opts, **kwargs)
  651. def _establish_connection(self):
  652. self._debug('establishing connection...')
  653. conn = self.transport.establish_connection()
  654. self._debug('connection established: %r', self)
  655. return conn
  656. def supports_exchange_type(self, exchange_type):
  657. return exchange_type in self.transport.implements.exchange_type
  658. def __repr__(self):
  659. return '<Connection: {0} at {1:#x}>'.format(self.as_uri(), id(self))
  660. def __copy__(self):
  661. return self.clone()
  662. def __reduce__(self):
  663. return self.__class__, tuple(self.info().values()), None
  664. def __enter__(self):
  665. return self
  666. def __exit__(self, *args):
  667. self.release()
  668. @property
  669. def qos_semantics_matches_spec(self):
  670. return self.transport.qos_semantics_matches_spec(self.connection)
  671. @property
  672. def connected(self):
  673. """Return true if the connection has been established."""
  674. return (not self._closed and
  675. self._connection is not None and
  676. self.transport.verify_connection(self._connection))
  677. @property
  678. def connection(self):
  679. """The underlying connection object.
  680. Warning:
  681. This instance is transport specific, so do not
  682. depend on the interface of this object.
  683. """
  684. if not self._closed:
  685. if not self.connected:
  686. self.declared_entities.clear()
  687. self._default_channel = None
  688. self._connection = self._establish_connection()
  689. self._closed = False
  690. return self._connection
  691. @property
  692. def default_channel(self):
  693. """Default channel.
  694. Created upon access and closed when the connection is closed.
  695. Note:
  696. Can be used for automatic channel handling when you only need one
  697. channel, and also it is the channel implicitly used if
  698. a connection is passed instead of a channel, to functions that
  699. require a channel.
  700. """
  701. conn_opts = {}
  702. transport_opts = self.transport_options
  703. if transport_opts:
  704. if 'max_retries' in transport_opts:
  705. conn_opts['max_retries'] = transport_opts['max_retries']
  706. if 'interval_start' in transport_opts:
  707. conn_opts['interval_start'] = transport_opts['interval_start']
  708. if 'interval_step' in transport_opts:
  709. conn_opts['interval_step'] = transport_opts['interval_step']
  710. if 'interval_max' in transport_opts:
  711. conn_opts['interval_max'] = transport_opts['interval_max']
  712. # make sure we're still connected, and if not refresh.
  713. self.ensure_connection(**conn_opts)
  714. if self._default_channel is None:
  715. self._default_channel = self.channel()
  716. return self._default_channel
  717. @property
  718. def host(self):
  719. """The host as a host name/port pair separated by colon."""
  720. return ':'.join([self.hostname, str(self.port)])
  721. @property
  722. def transport(self):
  723. if self._transport is None:
  724. self._transport = self.create_transport()
  725. return self._transport
  726. @cached_property
  727. def manager(self):
  728. """AMQP Management API.
  729. Experimental manager that can be used to manage/monitor the broker
  730. instance.
  731. Not available for all transports.
  732. """
  733. return self.transport.manager
  734. def get_manager(self, *args, **kwargs):
  735. return self.transport.get_manager(*args, **kwargs)
  736. @cached_property
  737. def recoverable_connection_errors(self):
  738. """Recoverable connection errors.
  739. List of connection related exceptions that can be recovered from,
  740. but where the connection must be closed and re-established first.
  741. """
  742. try:
  743. return self.transport.recoverable_connection_errors
  744. except AttributeError:
  745. # There were no such classification before,
  746. # and all errors were assumed to be recoverable,
  747. # so this is a fallback for transports that do
  748. # not support the new recoverable/irrecoverable classes.
  749. return self.connection_errors + self.channel_errors
  750. @cached_property
  751. def recoverable_channel_errors(self):
  752. """Recoverable channel errors.
  753. List of channel related exceptions that can be automatically
  754. recovered from without re-establishing the connection.
  755. """
  756. try:
  757. return self.transport.recoverable_channel_errors
  758. except AttributeError:
  759. return ()
  760. @cached_property
  761. def connection_errors(self):
  762. """List of exceptions that may be raised by the connection."""
  763. return self.transport.connection_errors
  764. @cached_property
  765. def channel_errors(self):
  766. """List of exceptions that may be raised by the channel."""
  767. return self.transport.channel_errors
  768. @property
  769. def supports_heartbeats(self):
  770. return self.transport.implements.heartbeats
  771. @property
  772. def is_evented(self):
  773. return self.transport.implements.asynchronous
  774. BrokerConnection = Connection # noqa: E305
  775. class ConnectionPool(Resource):
  776. """Pool of connections."""
  777. LimitExceeded = exceptions.ConnectionLimitExceeded
  778. close_after_fork = True
  779. def __init__(self, connection, limit=None, **kwargs):
  780. self.connection = connection
  781. super(ConnectionPool, self).__init__(limit=limit)
  782. def new(self):
  783. return self.connection.clone()
  784. def release_resource(self, resource):
  785. try:
  786. resource._debug('released')
  787. except AttributeError:
  788. pass
  789. def close_resource(self, resource):
  790. resource._close()
  791. def collect_resource(self, resource, socket_timeout=0.1):
  792. if not isinstance(resource, lazy):
  793. return resource.collect(socket_timeout)
  794. @contextmanager
  795. def acquire_channel(self, block=False):
  796. with self.acquire(block=block) as connection:
  797. yield connection, connection.default_channel
  798. def setup(self):
  799. if self.limit:
  800. q = self._resource.queue
  801. while len(q) < self.limit:
  802. self._resource.put_nowait(lazy(self.new))
  803. def prepare(self, resource):
  804. if callable(resource):
  805. resource = resource()
  806. resource._debug('acquired')
  807. return resource
  808. class ChannelPool(Resource):
  809. """Pool of channels."""
  810. LimitExceeded = exceptions.ChannelLimitExceeded
  811. def __init__(self, connection, limit=None, **kwargs):
  812. self.connection = connection
  813. super(ChannelPool, self).__init__(limit=limit)
  814. def new(self):
  815. return lazy(self.connection.channel)
  816. def setup(self):
  817. channel = self.new()
  818. if self.limit:
  819. q = self._resource.queue
  820. while len(q) < self.limit:
  821. self._resource.put_nowait(lazy(channel))
  822. def prepare(self, channel):
  823. if callable(channel):
  824. channel = channel()
  825. return channel
  826. def maybe_channel(channel):
  827. """Get channel from object.
  828. Return the default channel if argument is a connection instance,
  829. otherwise just return the channel given.
  830. """
  831. if is_connection(channel):
  832. return channel.default_channel
  833. return channel
  834. def is_connection(obj):
  835. return isinstance(obj, Connection)