channel.py 72 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089
  1. """AMQP Channels."""
  2. # Copyright (C) 2007-2008 Barry Pederson <bp@barryp.org>
  3. from __future__ import absolute_import, unicode_literals
  4. import logging
  5. import socket
  6. from collections import defaultdict
  7. from warnings import warn
  8. from vine import ensure_promise
  9. from . import spec
  10. from .abstract_channel import AbstractChannel
  11. from .exceptions import (ChannelError, ConsumerCancelled,
  12. RecoverableChannelError, RecoverableConnectionError,
  13. error_for_code, MessageNacked)
  14. from .five import Queue
  15. from .protocol import queue_declare_ok_t
  16. __all__ = ['Channel']
  17. AMQP_LOGGER = logging.getLogger('amqp')
  18. EXCHANGE_AUTODELETE_DEPRECATED = """\
  19. The auto_delete flag for exchanges has been deprecated and will be removed
  20. from py-amqp v1.5.0.\
  21. """
  22. REJECTED_MESSAGE_WITHOUT_CALLBACK = """\
  23. Rejecting message with delivery tag %r for reason of having no callbacks.
  24. consumer_tag=%r exchange=%r routing_key=%r.\
  25. """
  26. class VDeprecationWarning(DeprecationWarning):
  27. pass
  28. class Channel(AbstractChannel):
  29. """AMQP Channel.
  30. The channel class provides methods for a client to establish a
  31. virtual connection - a channel - to a server and for both peers to
  32. operate the virtual connection thereafter.
  33. GRAMMAR::
  34. channel = open-channel *use-channel close-channel
  35. open-channel = C:OPEN S:OPEN-OK
  36. use-channel = C:FLOW S:FLOW-OK
  37. / S:FLOW C:FLOW-OK
  38. / functional-class
  39. close-channel = C:CLOSE S:CLOSE-OK
  40. / S:CLOSE C:CLOSE-OK
  41. Create a channel bound to a connection and using the specified
  42. numeric channel_id, and open on the server.
  43. The 'auto_decode' parameter (defaults to True), indicates
  44. whether the library should attempt to decode the body
  45. of Messages to a Unicode string if there's a 'content_encoding'
  46. property for the message. If there's no 'content_encoding'
  47. property, or the decode raises an Exception, the message body
  48. is left as plain bytes.
  49. """
  50. _METHODS = {
  51. spec.method(spec.Channel.Close, 'BsBB'),
  52. spec.method(spec.Channel.CloseOk),
  53. spec.method(spec.Channel.Flow, 'b'),
  54. spec.method(spec.Channel.FlowOk, 'b'),
  55. spec.method(spec.Channel.OpenOk),
  56. spec.method(spec.Exchange.DeclareOk),
  57. spec.method(spec.Exchange.DeleteOk),
  58. spec.method(spec.Exchange.BindOk),
  59. spec.method(spec.Exchange.UnbindOk),
  60. spec.method(spec.Queue.BindOk),
  61. spec.method(spec.Queue.UnbindOk),
  62. spec.method(spec.Queue.DeclareOk, 'sll'),
  63. spec.method(spec.Queue.DeleteOk, 'l'),
  64. spec.method(spec.Queue.PurgeOk, 'l'),
  65. spec.method(spec.Basic.Cancel, 's'),
  66. spec.method(spec.Basic.CancelOk, 's'),
  67. spec.method(spec.Basic.ConsumeOk, 's'),
  68. spec.method(spec.Basic.Deliver, 'sLbss', content=True),
  69. spec.method(spec.Basic.GetEmpty, 's'),
  70. spec.method(spec.Basic.GetOk, 'Lbssl', content=True),
  71. spec.method(spec.Basic.QosOk),
  72. spec.method(spec.Basic.RecoverOk),
  73. spec.method(spec.Basic.Return, 'Bsss', content=True),
  74. spec.method(spec.Tx.CommitOk),
  75. spec.method(spec.Tx.RollbackOk),
  76. spec.method(spec.Tx.SelectOk),
  77. spec.method(spec.Confirm.SelectOk),
  78. spec.method(spec.Basic.Ack, 'Lb'),
  79. spec.method(spec.Basic.Nack, 'Lb'),
  80. }
  81. _METHODS = {m.method_sig: m for m in _METHODS}
  82. def __init__(self, connection,
  83. channel_id=None, auto_decode=True, on_open=None):
  84. if channel_id:
  85. connection._claim_channel_id(channel_id)
  86. else:
  87. channel_id = connection._get_free_channel_id()
  88. AMQP_LOGGER.debug('using channel_id: %s', channel_id)
  89. super(Channel, self).__init__(connection, channel_id)
  90. self.is_open = False
  91. self.active = True # Flow control
  92. self.returned_messages = Queue()
  93. self.callbacks = {}
  94. self.cancel_callbacks = {}
  95. self.auto_decode = auto_decode
  96. self.events = defaultdict(set)
  97. self.no_ack_consumers = set()
  98. self.on_open = ensure_promise(on_open)
  99. # set first time basic_publish_confirm is called
  100. # and publisher confirms are enabled for this channel.
  101. self._confirm_selected = False
  102. if self.connection.confirm_publish:
  103. self.basic_publish = self.basic_publish_confirm
  104. def then(self, on_success, on_error=None):
  105. return self.on_open.then(on_success, on_error)
  106. def _setup_listeners(self):
  107. self._callbacks.update({
  108. spec.Channel.Close: self._on_close,
  109. spec.Channel.CloseOk: self._on_close_ok,
  110. spec.Channel.Flow: self._on_flow,
  111. spec.Channel.OpenOk: self._on_open_ok,
  112. spec.Basic.Cancel: self._on_basic_cancel,
  113. spec.Basic.CancelOk: self._on_basic_cancel_ok,
  114. spec.Basic.Deliver: self._on_basic_deliver,
  115. spec.Basic.Return: self._on_basic_return,
  116. spec.Basic.Ack: self._on_basic_ack,
  117. spec.Basic.Nack: self._on_basic_nack,
  118. })
  119. def collect(self):
  120. """Tear down this object.
  121. Best called after we've agreed to close with the server.
  122. """
  123. AMQP_LOGGER.debug('Closed channel #%s', self.channel_id)
  124. self.is_open = False
  125. channel_id, self.channel_id = self.channel_id, None
  126. connection, self.connection = self.connection, None
  127. if connection:
  128. connection.channels.pop(channel_id, None)
  129. connection._avail_channel_ids.append(channel_id)
  130. self.callbacks.clear()
  131. self.cancel_callbacks.clear()
  132. self.events.clear()
  133. self.no_ack_consumers.clear()
  134. def _do_revive(self):
  135. self.is_open = False
  136. self.open()
  137. def close(self, reply_code=0, reply_text='', method_sig=(0, 0),
  138. argsig='BsBB'):
  139. """Request a channel close.
  140. This method indicates that the sender wants to close the
  141. channel. This may be due to internal conditions (e.g. a forced
  142. shut-down) or due to an error handling a specific method, i.e.
  143. an exception. When a close is due to an exception, the sender
  144. provides the class and method id of the method which caused
  145. the exception.
  146. RULE:
  147. After sending this method any received method except
  148. Channel.Close-OK MUST be discarded.
  149. RULE:
  150. The peer sending this method MAY use a counter or timeout
  151. to detect failure of the other peer to respond correctly
  152. with Channel.Close-OK..
  153. PARAMETERS:
  154. reply_code: short
  155. The reply code. The AMQ reply codes are defined in AMQ
  156. RFC 011.
  157. reply_text: shortstr
  158. The localised reply text. This text can be logged as an
  159. aid to resolving issues.
  160. class_id: short
  161. failing method class
  162. When the close is provoked by a method exception, this
  163. is the class of the method.
  164. method_id: short
  165. failing method ID
  166. When the close is provoked by a method exception, this
  167. is the ID of the method.
  168. """
  169. try:
  170. is_closed = (
  171. not self.is_open or
  172. self.connection is None or
  173. self.connection.channels is None
  174. )
  175. if is_closed:
  176. return
  177. self.is_closing = True
  178. return self.send_method(
  179. spec.Channel.Close, argsig,
  180. (reply_code, reply_text, method_sig[0], method_sig[1]),
  181. wait=spec.Channel.CloseOk,
  182. )
  183. finally:
  184. self.is_closing = False
  185. self.connection = None
  186. def _on_close(self, reply_code, reply_text, class_id, method_id):
  187. """Request a channel close.
  188. This method indicates that the sender wants to close the
  189. channel. This may be due to internal conditions (e.g. a forced
  190. shut-down) or due to an error handling a specific method, i.e.
  191. an exception. When a close is due to an exception, the sender
  192. provides the class and method id of the method which caused
  193. the exception.
  194. RULE:
  195. After sending this method any received method except
  196. Channel.Close-OK MUST be discarded.
  197. RULE:
  198. The peer sending this method MAY use a counter or timeout
  199. to detect failure of the other peer to respond correctly
  200. with Channel.Close-OK..
  201. PARAMETERS:
  202. reply_code: short
  203. The reply code. The AMQ reply codes are defined in AMQ
  204. RFC 011.
  205. reply_text: shortstr
  206. The localised reply text. This text can be logged as an
  207. aid to resolving issues.
  208. class_id: short
  209. failing method class
  210. When the close is provoked by a method exception, this
  211. is the class of the method.
  212. method_id: short
  213. failing method ID
  214. When the close is provoked by a method exception, this
  215. is the ID of the method.
  216. """
  217. self.send_method(spec.Channel.CloseOk)
  218. if not self.connection.is_closing:
  219. self._do_revive()
  220. raise error_for_code(
  221. reply_code, reply_text, (class_id, method_id), ChannelError,
  222. )
  223. def _on_close_ok(self):
  224. """Confirm a channel close.
  225. This method confirms a Channel.Close method and tells the
  226. recipient that it is safe to release resources for the channel
  227. and close the socket.
  228. RULE:
  229. A peer that detects a socket closure without having
  230. received a Channel.Close-Ok handshake method SHOULD log
  231. the error.
  232. """
  233. self.collect()
  234. def flow(self, active):
  235. """Enable/disable flow from peer.
  236. This method asks the peer to pause or restart the flow of
  237. content data. This is a simple flow-control mechanism that a
  238. peer can use to avoid oveflowing its queues or otherwise
  239. finding itself receiving more messages than it can process.
  240. Note that this method is not intended for window control. The
  241. peer that receives a request to stop sending content should
  242. finish sending the current content, if any, and then wait
  243. until it receives a Flow restart method.
  244. RULE:
  245. When a new channel is opened, it is active. Some
  246. applications assume that channels are inactive until
  247. started. To emulate this behaviour a client MAY open the
  248. channel, then pause it.
  249. RULE:
  250. When sending content data in multiple frames, a peer
  251. SHOULD monitor the channel for incoming methods and
  252. respond to a Channel.Flow as rapidly as possible.
  253. RULE:
  254. A peer MAY use the Channel.Flow method to throttle
  255. incoming content data for internal reasons, for example,
  256. when exchangeing data over a slower connection.
  257. RULE:
  258. The peer that requests a Channel.Flow method MAY
  259. disconnect and/or ban a peer that does not respect the
  260. request.
  261. PARAMETERS:
  262. active: boolean
  263. start/stop content frames
  264. If True, the peer starts sending content frames. If
  265. False, the peer stops sending content frames.
  266. """
  267. return self.send_method(
  268. spec.Channel.Flow, 'b', (active,), wait=spec.Channel.FlowOk,
  269. )
  270. def _on_flow(self, active):
  271. """Enable/disable flow from peer.
  272. This method asks the peer to pause or restart the flow of
  273. content data. This is a simple flow-control mechanism that a
  274. peer can use to avoid oveflowing its queues or otherwise
  275. finding itself receiving more messages than it can process.
  276. Note that this method is not intended for window control. The
  277. peer that receives a request to stop sending content should
  278. finish sending the current content, if any, and then wait
  279. until it receives a Flow restart method.
  280. RULE:
  281. When a new channel is opened, it is active. Some
  282. applications assume that channels are inactive until
  283. started. To emulate this behaviour a client MAY open the
  284. channel, then pause it.
  285. RULE:
  286. When sending content data in multiple frames, a peer
  287. SHOULD monitor the channel for incoming methods and
  288. respond to a Channel.Flow as rapidly as possible.
  289. RULE:
  290. A peer MAY use the Channel.Flow method to throttle
  291. incoming content data for internal reasons, for example,
  292. when exchangeing data over a slower connection.
  293. RULE:
  294. The peer that requests a Channel.Flow method MAY
  295. disconnect and/or ban a peer that does not respect the
  296. request.
  297. PARAMETERS:
  298. active: boolean
  299. start/stop content frames
  300. If True, the peer starts sending content frames. If
  301. False, the peer stops sending content frames.
  302. """
  303. self.active = active
  304. self._x_flow_ok(self.active)
  305. def _x_flow_ok(self, active):
  306. """Confirm a flow method.
  307. Confirms to the peer that a flow command was received and
  308. processed.
  309. PARAMETERS:
  310. active: boolean
  311. current flow setting
  312. Confirms the setting of the processed flow method:
  313. True means the peer will start sending or continue
  314. to send content frames; False means it will not.
  315. """
  316. return self.send_method(spec.Channel.FlowOk, 'b', (active,))
  317. def open(self):
  318. """Open a channel for use.
  319. This method opens a virtual connection (a channel).
  320. RULE:
  321. This method MUST NOT be called when the channel is already
  322. open.
  323. PARAMETERS:
  324. out_of_band: shortstr (DEPRECATED)
  325. out-of-band settings
  326. Configures out-of-band transfers on this channel. The
  327. syntax and meaning of this field will be formally
  328. defined at a later date.
  329. """
  330. if self.is_open:
  331. return
  332. return self.send_method(
  333. spec.Channel.Open, 's', ('',), wait=spec.Channel.OpenOk,
  334. )
  335. def _on_open_ok(self):
  336. """Signal that the channel is ready.
  337. This method signals to the client that the channel is ready
  338. for use.
  339. """
  340. self.is_open = True
  341. self.on_open(self)
  342. AMQP_LOGGER.debug('Channel open')
  343. #############
  344. #
  345. # Exchange
  346. #
  347. #
  348. # work with exchanges
  349. #
  350. # Exchanges match and distribute messages across queues.
  351. # Exchanges can be configured in the server or created at runtime.
  352. #
  353. # GRAMMAR::
  354. #
  355. # exchange = C:DECLARE S:DECLARE-OK
  356. # / C:DELETE S:DELETE-OK
  357. #
  358. # RULE:
  359. #
  360. # The server MUST implement the direct and fanout exchange
  361. # types, and predeclare the corresponding exchanges named
  362. # amq.direct and amq.fanout in each virtual host. The server
  363. # MUST also predeclare a direct exchange to act as the default
  364. # exchange for content Publish methods and for default queue
  365. # bindings.
  366. #
  367. # RULE:
  368. #
  369. # The server SHOULD implement the topic exchange type, and
  370. # predeclare the corresponding exchange named amq.topic in
  371. # each virtual host.
  372. #
  373. # RULE:
  374. #
  375. # The server MAY implement the system exchange type, and
  376. # predeclare the corresponding exchanges named amq.system in
  377. # each virtual host. If the client attempts to bind a queue to
  378. # the system exchange, the server MUST raise a connection
  379. # exception with reply code 507 (not allowed).
  380. #
  381. def exchange_declare(self, exchange, type, passive=False, durable=False,
  382. auto_delete=True, nowait=False, arguments=None,
  383. argsig='BssbbbbbF'):
  384. """Declare exchange, create if needed.
  385. This method creates an exchange if it does not already exist,
  386. and if the exchange exists, verifies that it is of the correct
  387. and expected class.
  388. RULE:
  389. The server SHOULD support a minimum of 16 exchanges per
  390. virtual host and ideally, impose no limit except as
  391. defined by available resources.
  392. PARAMETERS:
  393. exchange: shortstr
  394. RULE:
  395. Exchange names starting with "amq." are reserved
  396. for predeclared and standardised exchanges. If
  397. the client attempts to create an exchange starting
  398. with "amq.", the server MUST raise a channel
  399. exception with reply code 403 (access refused).
  400. type: shortstr
  401. exchange type
  402. Each exchange belongs to one of a set of exchange
  403. types implemented by the server. The exchange types
  404. define the functionality of the exchange - i.e. how
  405. messages are routed through it. It is not valid or
  406. meaningful to attempt to change the type of an
  407. existing exchange.
  408. RULE:
  409. If the exchange already exists with a different
  410. type, the server MUST raise a connection exception
  411. with a reply code 507 (not allowed).
  412. RULE:
  413. If the server does not support the requested
  414. exchange type it MUST raise a connection exception
  415. with a reply code 503 (command invalid).
  416. passive: boolean
  417. do not create exchange
  418. If set, the server will not create the exchange. The
  419. client can use this to check whether an exchange
  420. exists without modifying the server state.
  421. RULE:
  422. If set, and the exchange does not already exist,
  423. the server MUST raise a channel exception with
  424. reply code 404 (not found).
  425. durable: boolean
  426. request a durable exchange
  427. If set when creating a new exchange, the exchange will
  428. be marked as durable. Durable exchanges remain active
  429. when a server restarts. Non-durable exchanges
  430. (transient exchanges) are purged if/when a server
  431. restarts.
  432. RULE:
  433. The server MUST support both durable and transient
  434. exchanges.
  435. RULE:
  436. The server MUST ignore the durable field if the
  437. exchange already exists.
  438. auto_delete: boolean
  439. auto-delete when unused
  440. If set, the exchange is deleted when all queues have
  441. finished using it.
  442. RULE:
  443. The server SHOULD allow for a reasonable delay
  444. between the point when it determines that an
  445. exchange is not being used (or no longer used),
  446. and the point when it deletes the exchange. At
  447. the least it must allow a client to create an
  448. exchange and then bind a queue to it, with a small
  449. but non-zero delay between these two actions.
  450. RULE:
  451. The server MUST ignore the auto-delete field if
  452. the exchange already exists.
  453. nowait: boolean
  454. do not send a reply method
  455. If set, the server will not respond to the method. The
  456. client should not wait for a reply method. If the
  457. server could not complete the method it will raise a
  458. channel or connection exception.
  459. arguments: table
  460. arguments for declaration
  461. A set of arguments for the declaration. The syntax and
  462. semantics of these arguments depends on the server
  463. implementation. This field is ignored if passive is
  464. True.
  465. """
  466. if auto_delete:
  467. warn(VDeprecationWarning(EXCHANGE_AUTODELETE_DEPRECATED))
  468. self.send_method(
  469. spec.Exchange.Declare, argsig,
  470. (0, exchange, type, passive, durable, auto_delete,
  471. False, nowait, arguments),
  472. wait=None if nowait else spec.Exchange.DeclareOk,
  473. )
  474. def exchange_delete(self, exchange, if_unused=False, nowait=False,
  475. argsig='Bsbb'):
  476. """Delete an exchange.
  477. This method deletes an exchange. When an exchange is deleted
  478. all queue bindings on the exchange are cancelled.
  479. PARAMETERS:
  480. exchange: shortstr
  481. RULE:
  482. The exchange MUST exist. Attempting to delete a
  483. non-existing exchange causes a channel exception.
  484. if_unused: boolean
  485. delete only if unused
  486. If set, the server will only delete the exchange if it
  487. has no queue bindings. If the exchange has queue
  488. bindings the server does not delete it but raises a
  489. channel exception instead.
  490. RULE:
  491. If set, the server SHOULD delete the exchange but
  492. only if it has no queue bindings.
  493. RULE:
  494. If set, the server SHOULD raise a channel
  495. exception if the exchange is in use.
  496. nowait: boolean
  497. do not send a reply method
  498. If set, the server will not respond to the method. The
  499. client should not wait for a reply method. If the
  500. server could not complete the method it will raise a
  501. channel or connection exception.
  502. """
  503. return self.send_method(
  504. spec.Exchange.Delete, argsig, (0, exchange, if_unused, nowait),
  505. wait=None if nowait else spec.Exchange.DeleteOk,
  506. )
  507. def exchange_bind(self, destination, source='', routing_key='',
  508. nowait=False, arguments=None, argsig='BsssbF'):
  509. """Bind an exchange to an exchange.
  510. RULE:
  511. A server MUST allow and ignore duplicate bindings - that
  512. is, two or more bind methods for a specific exchanges,
  513. with identical arguments - without treating these as an
  514. error.
  515. RULE:
  516. A server MUST allow cycles of exchange bindings to be
  517. created including allowing an exchange to be bound to
  518. itself.
  519. RULE:
  520. A server MUST not deliver the same message more than once
  521. to a destination exchange, even if the topology of
  522. exchanges and bindings results in multiple (even infinite)
  523. routes to that exchange.
  524. PARAMETERS:
  525. reserved-1: short
  526. destination: shortstr
  527. Specifies the name of the destination exchange to
  528. bind.
  529. RULE:
  530. A client MUST NOT be allowed to bind a non-
  531. existent destination exchange.
  532. RULE:
  533. The server MUST accept a blank exchange name to
  534. mean the default exchange.
  535. source: shortstr
  536. Specifies the name of the source exchange to bind.
  537. RULE:
  538. A client MUST NOT be allowed to bind a non-
  539. existent source exchange.
  540. RULE:
  541. The server MUST accept a blank exchange name to
  542. mean the default exchange.
  543. routing-key: shortstr
  544. Specifies the routing key for the binding. The routing
  545. key is used for routing messages depending on the
  546. exchange configuration. Not all exchanges use a
  547. routing key - refer to the specific exchange
  548. documentation.
  549. no-wait: bit
  550. arguments: table
  551. A set of arguments for the binding. The syntax and
  552. semantics of these arguments depends on the exchange
  553. class.
  554. """
  555. return self.send_method(
  556. spec.Exchange.Bind, argsig,
  557. (0, destination, source, routing_key, nowait, arguments),
  558. wait=None if nowait else spec.Exchange.BindOk,
  559. )
  560. def exchange_unbind(self, destination, source='', routing_key='',
  561. nowait=False, arguments=None, argsig='BsssbF'):
  562. """Unbind an exchange from an exchange.
  563. RULE:
  564. If a unbind fails, the server MUST raise a connection
  565. exception.
  566. PARAMETERS:
  567. reserved-1: short
  568. destination: shortstr
  569. Specifies the name of the destination exchange to
  570. unbind.
  571. RULE:
  572. The client MUST NOT attempt to unbind an exchange
  573. that does not exist from an exchange.
  574. RULE:
  575. The server MUST accept a blank exchange name to
  576. mean the default exchange.
  577. source: shortstr
  578. Specifies the name of the source exchange to unbind.
  579. RULE:
  580. The client MUST NOT attempt to unbind an exchange
  581. from an exchange that does not exist.
  582. RULE:
  583. The server MUST accept a blank exchange name to
  584. mean the default exchange.
  585. routing-key: shortstr
  586. Specifies the routing key of the binding to unbind.
  587. no-wait: bit
  588. arguments: table
  589. Specifies the arguments of the binding to unbind.
  590. """
  591. return self.send_method(
  592. spec.Exchange.Unbind, argsig,
  593. (0, destination, source, routing_key, nowait, arguments),
  594. wait=None if nowait else spec.Exchange.UnbindOk,
  595. )
  596. #############
  597. #
  598. # Queue
  599. #
  600. #
  601. # work with queues
  602. #
  603. # Queues store and forward messages. Queues can be configured in
  604. # the server or created at runtime. Queues must be attached to at
  605. # least one exchange in order to receive messages from publishers.
  606. #
  607. # GRAMMAR::
  608. #
  609. # queue = C:DECLARE S:DECLARE-OK
  610. # / C:BIND S:BIND-OK
  611. # / C:PURGE S:PURGE-OK
  612. # / C:DELETE S:DELETE-OK
  613. #
  614. # RULE:
  615. #
  616. # A server MUST allow any content class to be sent to any
  617. # queue, in any mix, and queue and delivery these content
  618. # classes independently. Note that all methods that fetch
  619. # content off queues are specific to a given content class.
  620. #
  621. def queue_bind(self, queue, exchange='', routing_key='',
  622. nowait=False, arguments=None, argsig='BsssbF'):
  623. """Bind queue to an exchange.
  624. This method binds a queue to an exchange. Until a queue is
  625. bound it will not receive any messages. In a classic
  626. messaging model, store-and-forward queues are bound to a dest
  627. exchange and subscription queues are bound to a dest_wild
  628. exchange.
  629. RULE:
  630. A server MUST allow ignore duplicate bindings - that is,
  631. two or more bind methods for a specific queue, with
  632. identical arguments - without treating these as an error.
  633. RULE:
  634. If a bind fails, the server MUST raise a connection
  635. exception.
  636. RULE:
  637. The server MUST NOT allow a durable queue to bind to a
  638. transient exchange. If the client attempts this the server
  639. MUST raise a channel exception.
  640. RULE:
  641. Bindings for durable queues are automatically durable and
  642. the server SHOULD restore such bindings after a server
  643. restart.
  644. RULE:
  645. The server SHOULD support at least 4 bindings per queue,
  646. and ideally, impose no limit except as defined by
  647. available resources.
  648. PARAMETERS:
  649. queue: shortstr
  650. Specifies the name of the queue to bind. If the queue
  651. name is empty, refers to the current queue for the
  652. channel, which is the last declared queue.
  653. RULE:
  654. If the client did not previously declare a queue,
  655. and the queue name in this method is empty, the
  656. server MUST raise a connection exception with
  657. reply code 530 (not allowed).
  658. RULE:
  659. If the queue does not exist the server MUST raise
  660. a channel exception with reply code 404 (not
  661. found).
  662. exchange: shortstr
  663. The name of the exchange to bind to.
  664. RULE:
  665. If the exchange does not exist the server MUST
  666. raise a channel exception with reply code 404 (not
  667. found).
  668. routing_key: shortstr
  669. message routing key
  670. Specifies the routing key for the binding. The
  671. routing key is used for routing messages depending on
  672. the exchange configuration. Not all exchanges use a
  673. routing key - refer to the specific exchange
  674. documentation. If the routing key is empty and the
  675. queue name is empty, the routing key will be the
  676. current queue for the channel, which is the last
  677. declared queue.
  678. nowait: boolean
  679. do not send a reply method
  680. If set, the server will not respond to the method. The
  681. client should not wait for a reply method. If the
  682. server could not complete the method it will raise a
  683. channel or connection exception.
  684. arguments: table
  685. arguments for binding
  686. A set of arguments for the binding. The syntax and
  687. semantics of these arguments depends on the exchange
  688. class.
  689. """
  690. return self.send_method(
  691. spec.Queue.Bind, argsig,
  692. (0, queue, exchange, routing_key, nowait, arguments),
  693. wait=None if nowait else spec.Queue.BindOk,
  694. )
  695. def queue_unbind(self, queue, exchange, routing_key='',
  696. nowait=False, arguments=None, argsig='BsssF'):
  697. """Unbind a queue from an exchange.
  698. This method unbinds a queue from an exchange.
  699. RULE:
  700. If a unbind fails, the server MUST raise a connection exception.
  701. PARAMETERS:
  702. queue: shortstr
  703. Specifies the name of the queue to unbind.
  704. RULE:
  705. The client MUST either specify a queue name or have
  706. previously declared a queue on the same channel
  707. RULE:
  708. The client MUST NOT attempt to unbind a queue that
  709. does not exist.
  710. exchange: shortstr
  711. The name of the exchange to unbind from.
  712. RULE:
  713. The client MUST NOT attempt to unbind a queue from an
  714. exchange that does not exist.
  715. RULE:
  716. The server MUST accept a blank exchange name to mean
  717. the default exchange.
  718. routing_key: shortstr
  719. routing key of binding
  720. Specifies the routing key of the binding to unbind.
  721. arguments: table
  722. arguments of binding
  723. Specifies the arguments of the binding to unbind.
  724. """
  725. return self.send_method(
  726. spec.Queue.Unbind, argsig,
  727. (0, queue, exchange, routing_key, arguments),
  728. wait=None if nowait else spec.Queue.UnbindOk,
  729. )
  730. def queue_declare(self, queue='', passive=False, durable=False,
  731. exclusive=False, auto_delete=True, nowait=False,
  732. arguments=None, argsig='BsbbbbbF'):
  733. """Declare queue, create if needed.
  734. This method creates or checks a queue. When creating a new
  735. queue the client can specify various properties that control
  736. the durability of the queue and its contents, and the level of
  737. sharing for the queue.
  738. RULE:
  739. The server MUST create a default binding for a newly-
  740. created queue to the default exchange, which is an
  741. exchange of type 'direct'.
  742. RULE:
  743. The server SHOULD support a minimum of 256 queues per
  744. virtual host and ideally, impose no limit except as
  745. defined by available resources.
  746. PARAMETERS:
  747. queue: shortstr
  748. RULE:
  749. The queue name MAY be empty, in which case the
  750. server MUST create a new queue with a unique
  751. generated name and return this to the client in
  752. the Declare-Ok method.
  753. RULE:
  754. Queue names starting with "amq." are reserved for
  755. predeclared and standardised server queues. If
  756. the queue name starts with "amq." and the passive
  757. option is False, the server MUST raise a connection
  758. exception with reply code 403 (access refused).
  759. passive: boolean
  760. do not create queue
  761. If set, the server will not create the queue. The
  762. client can use this to check whether a queue exists
  763. without modifying the server state.
  764. RULE:
  765. If set, and the queue does not already exist, the
  766. server MUST respond with a reply code 404 (not
  767. found) and raise a channel exception.
  768. durable: boolean
  769. request a durable queue
  770. If set when creating a new queue, the queue will be
  771. marked as durable. Durable queues remain active when
  772. a server restarts. Non-durable queues (transient
  773. queues) are purged if/when a server restarts. Note
  774. that durable queues do not necessarily hold persistent
  775. messages, although it does not make sense to send
  776. persistent messages to a transient queue.
  777. RULE:
  778. The server MUST recreate the durable queue after a
  779. restart.
  780. RULE:
  781. The server MUST support both durable and transient
  782. queues.
  783. RULE:
  784. The server MUST ignore the durable field if the
  785. queue already exists.
  786. exclusive: boolean
  787. request an exclusive queue
  788. Exclusive queues may only be consumed from by the
  789. current connection. Setting the 'exclusive' flag
  790. always implies 'auto-delete'.
  791. RULE:
  792. The server MUST support both exclusive (private)
  793. and non-exclusive (shared) queues.
  794. RULE:
  795. The server MUST raise a channel exception if
  796. 'exclusive' is specified and the queue already
  797. exists and is owned by a different connection.
  798. auto_delete: boolean
  799. auto-delete queue when unused
  800. If set, the queue is deleted when all consumers have
  801. finished using it. Last consumer can be cancelled
  802. either explicitly or because its channel is closed. If
  803. there was no consumer ever on the queue, it won't be
  804. deleted.
  805. RULE:
  806. The server SHOULD allow for a reasonable delay
  807. between the point when it determines that a queue
  808. is not being used (or no longer used), and the
  809. point when it deletes the queue. At the least it
  810. must allow a client to create a queue and then
  811. create a consumer to read from it, with a small
  812. but non-zero delay between these two actions. The
  813. server should equally allow for clients that may
  814. be disconnected prematurely, and wish to re-
  815. consume from the same queue without losing
  816. messages. We would recommend a configurable
  817. timeout, with a suitable default value being one
  818. minute.
  819. RULE:
  820. The server MUST ignore the auto-delete field if
  821. the queue already exists.
  822. nowait: boolean
  823. do not send a reply method
  824. If set, the server will not respond to the method. The
  825. client should not wait for a reply method. If the
  826. server could not complete the method it will raise a
  827. channel or connection exception.
  828. arguments: table
  829. arguments for declaration
  830. A set of arguments for the declaration. The syntax and
  831. semantics of these arguments depends on the server
  832. implementation. This field is ignored if passive is
  833. True.
  834. Returns a tuple containing 3 items:
  835. the name of the queue (essential for automatically-named queues),
  836. message count and
  837. consumer count
  838. """
  839. self.send_method(
  840. spec.Queue.Declare, argsig,
  841. (0, queue, passive, durable, exclusive, auto_delete,
  842. nowait, arguments),
  843. )
  844. if not nowait:
  845. return queue_declare_ok_t(*self.wait(
  846. spec.Queue.DeclareOk, returns_tuple=True,
  847. ))
  848. def queue_delete(self, queue='',
  849. if_unused=False, if_empty=False, nowait=False,
  850. argsig='Bsbbb'):
  851. """Delete a queue.
  852. This method deletes a queue. When a queue is deleted any
  853. pending messages are sent to a dead-letter queue if this is
  854. defined in the server configuration, and all consumers on the
  855. queue are cancelled.
  856. RULE:
  857. The server SHOULD use a dead-letter queue to hold messages
  858. that were pending on a deleted queue, and MAY provide
  859. facilities for a system administrator to move these
  860. messages back to an active queue.
  861. PARAMETERS:
  862. queue: shortstr
  863. Specifies the name of the queue to delete. If the
  864. queue name is empty, refers to the current queue for
  865. the channel, which is the last declared queue.
  866. RULE:
  867. If the client did not previously declare a queue,
  868. and the queue name in this method is empty, the
  869. server MUST raise a connection exception with
  870. reply code 530 (not allowed).
  871. RULE:
  872. The queue must exist. Attempting to delete a non-
  873. existing queue causes a channel exception.
  874. if_unused: boolean
  875. delete only if unused
  876. If set, the server will only delete the queue if it
  877. has no consumers. If the queue has consumers the
  878. server does does not delete it but raises a channel
  879. exception instead.
  880. RULE:
  881. The server MUST respect the if-unused flag when
  882. deleting a queue.
  883. if_empty: boolean
  884. delete only if empty
  885. If set, the server will only delete the queue if it
  886. has no messages. If the queue is not empty the server
  887. raises a channel exception.
  888. nowait: boolean
  889. do not send a reply method
  890. If set, the server will not respond to the method. The
  891. client should not wait for a reply method. If the
  892. server could not complete the method it will raise a
  893. channel or connection exception.
  894. If nowait is False, returns the number of deleted messages.
  895. """
  896. return self.send_method(
  897. spec.Queue.Delete, argsig,
  898. (0, queue, if_unused, if_empty, nowait),
  899. wait=None if nowait else spec.Queue.DeleteOk,
  900. )
  901. def queue_purge(self, queue='', nowait=False, argsig='Bsb'):
  902. """Purge a queue.
  903. This method removes all messages from a queue. It does not
  904. cancel consumers. Purged messages are deleted without any
  905. formal "undo" mechanism.
  906. RULE:
  907. A call to purge MUST result in an empty queue.
  908. RULE:
  909. On transacted channels the server MUST not purge messages
  910. that have already been sent to a client but not yet
  911. acknowledged.
  912. RULE:
  913. The server MAY implement a purge queue or log that allows
  914. system administrators to recover accidentally-purged
  915. messages. The server SHOULD NOT keep purged messages in
  916. the same storage spaces as the live messages since the
  917. volumes of purged messages may get very large.
  918. PARAMETERS:
  919. queue: shortstr
  920. Specifies the name of the queue to purge. If the
  921. queue name is empty, refers to the current queue for
  922. the channel, which is the last declared queue.
  923. RULE:
  924. If the client did not previously declare a queue,
  925. and the queue name in this method is empty, the
  926. server MUST raise a connection exception with
  927. reply code 530 (not allowed).
  928. RULE:
  929. The queue must exist. Attempting to purge a non-
  930. existing queue causes a channel exception.
  931. nowait: boolean
  932. do not send a reply method
  933. If set, the server will not respond to the method. The
  934. client should not wait for a reply method. If the
  935. server could not complete the method it will raise a
  936. channel or connection exception.
  937. If nowait is False, returns a number of purged messages.
  938. """
  939. return self.send_method(
  940. spec.Queue.Purge, argsig, (0, queue, nowait),
  941. wait=None if nowait else spec.Queue.PurgeOk,
  942. )
  943. #############
  944. #
  945. # Basic
  946. #
  947. #
  948. # work with basic content
  949. #
  950. # The Basic class provides methods that support an industry-
  951. # standard messaging model.
  952. #
  953. # GRAMMAR::
  954. #
  955. # basic = C:QOS S:QOS-OK
  956. # / C:CONSUME S:CONSUME-OK
  957. # / C:CANCEL S:CANCEL-OK
  958. # / C:PUBLISH content
  959. # / S:RETURN content
  960. # / S:DELIVER content
  961. # / C:GET ( S:GET-OK content / S:GET-EMPTY )
  962. # / C:ACK
  963. # / C:REJECT
  964. #
  965. # RULE:
  966. #
  967. # The server SHOULD respect the persistent property of basic
  968. # messages and SHOULD make a best-effort to hold persistent
  969. # basic messages on a reliable storage mechanism.
  970. #
  971. # RULE:
  972. #
  973. # The server MUST NOT discard a persistent basic message in
  974. # case of a queue overflow. The server MAY use the
  975. # Channel.Flow method to slow or stop a basic message
  976. # publisher when necessary.
  977. #
  978. # RULE:
  979. #
  980. # The server MAY overflow non-persistent basic messages to
  981. # persistent storage and MAY discard or dead-letter non-
  982. # persistent basic messages on a priority basis if the queue
  983. # size exceeds some configured limit.
  984. #
  985. # RULE:
  986. #
  987. # The server MUST implement at least 2 priority levels for
  988. # basic messages, where priorities 0-4 and 5-9 are treated as
  989. # two distinct levels. The server MAY implement up to 10
  990. # priority levels.
  991. #
  992. # RULE:
  993. #
  994. # The server MUST deliver messages of the same priority in
  995. # order irrespective of their individual persistence.
  996. #
  997. # RULE:
  998. #
  999. # The server MUST support both automatic and explicit
  1000. # acknowledgments on Basic content.
  1001. #
  1002. def basic_ack(self, delivery_tag, multiple=False, argsig='Lb'):
  1003. """Acknowledge one or more messages.
  1004. This method acknowledges one or more messages delivered via
  1005. the Deliver or Get-Ok methods. The client can ask to confirm
  1006. a single message or a set of messages up to and including a
  1007. specific message.
  1008. PARAMETERS:
  1009. delivery_tag: longlong
  1010. server-assigned delivery tag
  1011. The server-assigned and channel-specific delivery tag
  1012. RULE:
  1013. The delivery tag is valid only within the channel
  1014. from which the message was received. I.e. a client
  1015. MUST NOT receive a message on one channel and then
  1016. acknowledge it on another.
  1017. RULE:
  1018. The server MUST NOT use a zero value for delivery
  1019. tags. Zero is reserved for client use, meaning "all
  1020. messages so far received".
  1021. multiple: boolean
  1022. acknowledge multiple messages
  1023. If set to True, the delivery tag is treated as "up to
  1024. and including", so that the client can acknowledge
  1025. multiple messages with a single method. If set to
  1026. False, the delivery tag refers to a single message.
  1027. If the multiple field is True, and the delivery tag
  1028. is zero, tells the server to acknowledge all
  1029. outstanding mesages.
  1030. RULE:
  1031. The server MUST validate that a non-zero delivery-
  1032. tag refers to an delivered message, and raise a
  1033. channel exception if this is not the case.
  1034. """
  1035. return self.send_method(
  1036. spec.Basic.Ack, argsig, (delivery_tag, multiple),
  1037. )
  1038. def basic_cancel(self, consumer_tag, nowait=False, argsig='sb'):
  1039. """End a queue consumer.
  1040. This method cancels a consumer. This does not affect already
  1041. delivered messages, but it does mean the server will not send
  1042. any more messages for that consumer. The client may receive
  1043. an abitrary number of messages in between sending the cancel
  1044. method and receiving the cancel-ok reply.
  1045. RULE:
  1046. If the queue no longer exists when the client sends a
  1047. cancel command, or the consumer has been cancelled for
  1048. other reasons, this command has no effect.
  1049. PARAMETERS:
  1050. consumer_tag: shortstr
  1051. consumer tag
  1052. Identifier for the consumer, valid within the current
  1053. connection.
  1054. RULE:
  1055. The consumer tag is valid only within the channel
  1056. from which the consumer was created. I.e. a client
  1057. MUST NOT create a consumer in one channel and then
  1058. use it in another.
  1059. nowait: boolean
  1060. do not send a reply method
  1061. If set, the server will not respond to the method. The
  1062. client should not wait for a reply method. If the
  1063. server could not complete the method it will raise a
  1064. channel or connection exception.
  1065. """
  1066. if self.connection is not None:
  1067. self.no_ack_consumers.discard(consumer_tag)
  1068. return self.send_method(
  1069. spec.Basic.Cancel, argsig, (consumer_tag, nowait),
  1070. wait=None if nowait else spec.Basic.CancelOk,
  1071. )
  1072. def _on_basic_cancel(self, consumer_tag):
  1073. """Consumer cancelled by server.
  1074. Most likely the queue was deleted.
  1075. """
  1076. callback = self._remove_tag(consumer_tag)
  1077. if callback:
  1078. callback(consumer_tag)
  1079. else:
  1080. raise ConsumerCancelled(consumer_tag, spec.Basic.Cancel)
  1081. def _on_basic_cancel_ok(self, consumer_tag):
  1082. self._remove_tag(consumer_tag)
  1083. def _remove_tag(self, consumer_tag):
  1084. self.callbacks.pop(consumer_tag, None)
  1085. return self.cancel_callbacks.pop(consumer_tag, None)
  1086. def basic_consume(self, queue='', consumer_tag='', no_local=False,
  1087. no_ack=False, exclusive=False, nowait=False,
  1088. callback=None, arguments=None, on_cancel=None,
  1089. argsig='BssbbbbF'):
  1090. """Start a queue consumer.
  1091. This method asks the server to start a "consumer", which is a
  1092. transient request for messages from a specific queue.
  1093. Consumers last as long as the channel they were created on, or
  1094. until the client cancels them.
  1095. RULE:
  1096. The server SHOULD support at least 16 consumers per queue,
  1097. unless the queue was declared as private, and ideally,
  1098. impose no limit except as defined by available resources.
  1099. PARAMETERS:
  1100. queue: shortstr
  1101. Specifies the name of the queue to consume from. If
  1102. the queue name is null, refers to the current queue
  1103. for the channel, which is the last declared queue.
  1104. RULE:
  1105. If the client did not previously declare a queue,
  1106. and the queue name in this method is empty, the
  1107. server MUST raise a connection exception with
  1108. reply code 530 (not allowed).
  1109. consumer_tag: shortstr
  1110. Specifies the identifier for the consumer. The
  1111. consumer tag is local to a connection, so two clients
  1112. can use the same consumer tags. If this field is empty
  1113. the server will generate a unique tag.
  1114. RULE:
  1115. The tag MUST NOT refer to an existing consumer. If
  1116. the client attempts to create two consumers with
  1117. the same non-empty tag the server MUST raise a
  1118. connection exception with reply code 530 (not
  1119. allowed).
  1120. no_local: boolean
  1121. do not deliver own messages
  1122. If the no-local field is set the server will not send
  1123. messages to the client that published them.
  1124. no_ack: boolean
  1125. no acknowledgment needed
  1126. If this field is set the server does not expect
  1127. acknowledgments for messages. That is, when a message
  1128. is delivered to the client the server automatically and
  1129. silently acknowledges it on behalf of the client. This
  1130. functionality increases performance but at the cost of
  1131. reliability. Messages can get lost if a client dies
  1132. before it can deliver them to the application.
  1133. exclusive: boolean
  1134. request exclusive access
  1135. Request exclusive consumer access, meaning only this
  1136. consumer can access the queue.
  1137. RULE:
  1138. If the server cannot grant exclusive access to the
  1139. queue when asked, - because there are other
  1140. consumers active - it MUST raise a channel
  1141. exception with return code 403 (access refused).
  1142. nowait: boolean
  1143. do not send a reply method
  1144. If set, the server will not respond to the method. The
  1145. client should not wait for a reply method. If the
  1146. server could not complete the method it will raise a
  1147. channel or connection exception.
  1148. callback: Python callable
  1149. function/method called with each delivered message
  1150. For each message delivered by the broker, the
  1151. callable will be called with a Message object
  1152. as the single argument. If no callable is specified,
  1153. messages are quietly discarded, no_ack should probably
  1154. be set to True in that case.
  1155. """
  1156. p = self.send_method(
  1157. spec.Basic.Consume, argsig,
  1158. (
  1159. 0, queue, consumer_tag, no_local, no_ack, exclusive,
  1160. nowait, arguments
  1161. ),
  1162. wait=None if nowait else spec.Basic.ConsumeOk,
  1163. returns_tuple=True
  1164. )
  1165. if not nowait:
  1166. # send_method() returns (consumer_tag,) tuple.
  1167. # consumer_tag is returned by broker using following rules:
  1168. # * consumer_tag is not specified by client, random one
  1169. # is generated by Broker
  1170. # * consumer_tag is provided by client, the same one
  1171. # is returned by broker
  1172. consumer_tag = p[0]
  1173. elif nowait and not consumer_tag:
  1174. raise ValueError(
  1175. 'Consumer tag must be specified when nowait is True'
  1176. )
  1177. self.callbacks[consumer_tag] = callback
  1178. if on_cancel:
  1179. self.cancel_callbacks[consumer_tag] = on_cancel
  1180. if no_ack:
  1181. self.no_ack_consumers.add(consumer_tag)
  1182. if not nowait:
  1183. return consumer_tag
  1184. else:
  1185. return p
  1186. def _on_basic_deliver(self, consumer_tag, delivery_tag, redelivered,
  1187. exchange, routing_key, msg):
  1188. msg.channel = self
  1189. msg.delivery_info = {
  1190. 'consumer_tag': consumer_tag,
  1191. 'delivery_tag': delivery_tag,
  1192. 'redelivered': redelivered,
  1193. 'exchange': exchange,
  1194. 'routing_key': routing_key,
  1195. }
  1196. try:
  1197. fun = self.callbacks[consumer_tag]
  1198. except KeyError:
  1199. AMQP_LOGGER.warning(
  1200. REJECTED_MESSAGE_WITHOUT_CALLBACK,
  1201. delivery_tag, consumer_tag, exchange, routing_key,
  1202. )
  1203. self.basic_reject(delivery_tag, requeue=True)
  1204. else:
  1205. fun(msg)
  1206. def basic_get(self, queue='', no_ack=False, argsig='Bsb'):
  1207. """Direct access to a queue.
  1208. This method provides a direct access to the messages in a
  1209. queue using a synchronous dialogue that is designed for
  1210. specific types of application where synchronous functionality
  1211. is more important than performance.
  1212. PARAMETERS:
  1213. queue: shortstr
  1214. Specifies the name of the queue to consume from. If
  1215. the queue name is null, refers to the current queue
  1216. for the channel, which is the last declared queue.
  1217. RULE:
  1218. If the client did not previously declare a queue,
  1219. and the queue name in this method is empty, the
  1220. server MUST raise a connection exception with
  1221. reply code 530 (not allowed).
  1222. no_ack: boolean
  1223. no acknowledgment needed
  1224. If this field is set the server does not expect
  1225. acknowledgments for messages. That is, when a message
  1226. is delivered to the client the server automatically and
  1227. silently acknowledges it on behalf of the client. This
  1228. functionality increases performance but at the cost of
  1229. reliability. Messages can get lost if a client dies
  1230. before it can deliver them to the application.
  1231. Non-blocking, returns a amqp.basic_message.Message object,
  1232. or None if queue is empty.
  1233. """
  1234. ret = self.send_method(
  1235. spec.Basic.Get, argsig, (0, queue, no_ack),
  1236. wait=[spec.Basic.GetOk, spec.Basic.GetEmpty], returns_tuple=True,
  1237. )
  1238. if not ret or len(ret) < 2:
  1239. return self._on_get_empty(*ret)
  1240. return self._on_get_ok(*ret)
  1241. def _on_get_empty(self, cluster_id=None):
  1242. pass
  1243. def _on_get_ok(self, delivery_tag, redelivered, exchange, routing_key,
  1244. message_count, msg):
  1245. msg.channel = self
  1246. msg.delivery_info = {
  1247. 'delivery_tag': delivery_tag,
  1248. 'redelivered': redelivered,
  1249. 'exchange': exchange,
  1250. 'routing_key': routing_key,
  1251. 'message_count': message_count
  1252. }
  1253. return msg
  1254. def _basic_publish(self, msg, exchange='', routing_key='',
  1255. mandatory=False, immediate=False, timeout=None,
  1256. argsig='Bssbb'):
  1257. """Publish a message.
  1258. This method publishes a message to a specific exchange. The
  1259. message will be routed to queues as defined by the exchange
  1260. configuration and distributed to any active consumers when the
  1261. transaction, if any, is committed.
  1262. When channel is in confirm mode (when Connection parameter
  1263. confirm_publish is set to True), each message is confirmed. When
  1264. broker rejects published message (e.g. due internal broker
  1265. constrains), MessageNacked exception is raised.
  1266. PARAMETERS:
  1267. exchange: shortstr
  1268. Specifies the name of the exchange to publish to. The
  1269. exchange name can be empty, meaning the default
  1270. exchange. If the exchange name is specified, and that
  1271. exchange does not exist, the server will raise a
  1272. channel exception.
  1273. RULE:
  1274. The server MUST accept a blank exchange name to
  1275. mean the default exchange.
  1276. RULE:
  1277. The exchange MAY refuse basic content in which
  1278. case it MUST raise a channel exception with reply
  1279. code 540 (not implemented).
  1280. routing_key: shortstr
  1281. Message routing key
  1282. Specifies the routing key for the message. The
  1283. routing key is used for routing messages depending on
  1284. the exchange configuration.
  1285. mandatory: boolean
  1286. indicate mandatory routing
  1287. This flag tells the server how to react if the message
  1288. cannot be routed to a queue. If this flag is True, the
  1289. server will return an unroutable message with a Return
  1290. method. If this flag is False, the server silently
  1291. drops the message.
  1292. RULE:
  1293. The server SHOULD implement the mandatory flag.
  1294. immediate: boolean
  1295. request immediate delivery
  1296. This flag tells the server how to react if the message
  1297. cannot be routed to a queue consumer immediately. If
  1298. this flag is set, the server will return an
  1299. undeliverable message with a Return method. If this
  1300. flag is zero, the server will queue the message, but
  1301. with no guarantee that it will ever be consumed.
  1302. RULE:
  1303. The server SHOULD implement the immediate flag.
  1304. """
  1305. if not self.connection:
  1306. raise RecoverableConnectionError(
  1307. 'basic_publish: connection closed')
  1308. capabilities = self.connection.\
  1309. client_properties.get('capabilities', {})
  1310. if capabilities.get('connection.blocked', False):
  1311. try:
  1312. # Check if an event was sent, such as the out of memory message
  1313. self.connection.drain_events(timeout=0)
  1314. except socket.timeout:
  1315. pass
  1316. try:
  1317. with self.connection.transport.having_timeout(timeout):
  1318. return self.send_method(
  1319. spec.Basic.Publish, argsig,
  1320. (0, exchange, routing_key, mandatory, immediate), msg
  1321. )
  1322. except socket.timeout:
  1323. raise RecoverableChannelError('basic_publish: timed out')
  1324. basic_publish = _basic_publish
  1325. def basic_publish_confirm(self, *args, **kwargs):
  1326. def confirm_handler(method, *args):
  1327. # When RMQ nacks message we are raising MessageNacked exception
  1328. if method == spec.Basic.Nack:
  1329. raise MessageNacked()
  1330. if not self._confirm_selected:
  1331. self._confirm_selected = True
  1332. self.confirm_select()
  1333. ret = self._basic_publish(*args, **kwargs)
  1334. # Waiting for confirmation of message.
  1335. self.wait([spec.Basic.Ack, spec.Basic.Nack], callback=confirm_handler)
  1336. return ret
  1337. def basic_qos(self, prefetch_size, prefetch_count, a_global,
  1338. argsig='lBb'):
  1339. """Specify quality of service.
  1340. This method requests a specific quality of service. The QoS
  1341. can be specified for the current channel or for all channels
  1342. on the connection. The particular properties and semantics of
  1343. a qos method always depend on the content class semantics.
  1344. Though the qos method could in principle apply to both peers,
  1345. it is currently meaningful only for the server.
  1346. PARAMETERS:
  1347. prefetch_size: long
  1348. prefetch window in octets
  1349. The client can request that messages be sent in
  1350. advance so that when the client finishes processing a
  1351. message, the following message is already held
  1352. locally, rather than needing to be sent down the
  1353. channel. Prefetching gives a performance improvement.
  1354. This field specifies the prefetch window size in
  1355. octets. The server will send a message in advance if
  1356. it is equal to or smaller in size than the available
  1357. prefetch size (and also falls into other prefetch
  1358. limits). May be set to zero, meaning "no specific
  1359. limit", although other prefetch limits may still
  1360. apply. The prefetch-size is ignored if the no-ack
  1361. option is set.
  1362. RULE:
  1363. The server MUST ignore this setting when the
  1364. client is not processing any messages - i.e. the
  1365. prefetch size does not limit the transfer of
  1366. single messages to a client, only the sending in
  1367. advance of more messages while the client still
  1368. has one or more unacknowledged messages.
  1369. prefetch_count: short
  1370. prefetch window in messages
  1371. Specifies a prefetch window in terms of whole
  1372. messages. This field may be used in combination with
  1373. the prefetch-size field; a message will only be sent
  1374. in advance if both prefetch windows (and those at the
  1375. channel and connection level) allow it. The prefetch-
  1376. count is ignored if the no-ack option is set.
  1377. RULE:
  1378. The server MAY send less data in advance than
  1379. allowed by the client's specified prefetch windows
  1380. but it MUST NOT send more.
  1381. a_global: boolean
  1382. apply to entire connection
  1383. By default the QoS settings apply to the current
  1384. channel only. If this field is set, they are applied
  1385. to the entire connection.
  1386. """
  1387. return self.send_method(
  1388. spec.Basic.Qos, argsig, (prefetch_size, prefetch_count, a_global),
  1389. wait=spec.Basic.QosOk,
  1390. )
  1391. def basic_recover(self, requeue=False):
  1392. """Redeliver unacknowledged messages.
  1393. This method asks the broker to redeliver all unacknowledged
  1394. messages on a specified channel. Zero or more messages may be
  1395. redelivered. This method is only allowed on non-transacted
  1396. channels.
  1397. RULE:
  1398. The server MUST set the redelivered flag on all messages
  1399. that are resent.
  1400. RULE:
  1401. The server MUST raise a channel exception if this is
  1402. called on a transacted channel.
  1403. PARAMETERS:
  1404. requeue: boolean
  1405. requeue the message
  1406. If this field is False, the message will be redelivered
  1407. to the original recipient. If this field is True, the
  1408. server will attempt to requeue the message,
  1409. potentially then delivering it to an alternative
  1410. subscriber.
  1411. """
  1412. return self.send_method(spec.Basic.Recover, 'b', (requeue,))
  1413. def basic_recover_async(self, requeue=False):
  1414. return self.send_method(spec.Basic.RecoverAsync, 'b', (requeue,))
  1415. def basic_reject(self, delivery_tag, requeue, argsig='Lb'):
  1416. """Reject an incoming message.
  1417. This method allows a client to reject a message. It can be
  1418. used to interrupt and cancel large incoming messages, or
  1419. return untreatable messages to their original queue.
  1420. RULE:
  1421. The server SHOULD be capable of accepting and process the
  1422. Reject method while sending message content with a Deliver
  1423. or Get-Ok method. I.e. the server should read and process
  1424. incoming methods while sending output frames. To cancel a
  1425. partially-send content, the server sends a content body
  1426. frame of size 1 (i.e. with no data except the frame-end
  1427. octet).
  1428. RULE:
  1429. The server SHOULD interpret this method as meaning that
  1430. the client is unable to process the message at this time.
  1431. RULE:
  1432. A client MUST NOT use this method as a means of selecting
  1433. messages to process. A rejected message MAY be discarded
  1434. or dead-lettered, not necessarily passed to another
  1435. client.
  1436. PARAMETERS:
  1437. delivery_tag: longlong
  1438. server-assigned delivery tag
  1439. The server-assigned and channel-specific delivery tag
  1440. RULE:
  1441. The delivery tag is valid only within the channel
  1442. from which the message was received. I.e. a client
  1443. MUST NOT receive a message on one channel and then
  1444. acknowledge it on another.
  1445. RULE:
  1446. The server MUST NOT use a zero value for delivery
  1447. tags. Zero is reserved for client use, meaning "all
  1448. messages so far received".
  1449. requeue: boolean
  1450. requeue the message
  1451. If this field is False, the message will be discarded.
  1452. If this field is True, the server will attempt to
  1453. requeue the message.
  1454. RULE:
  1455. The server MUST NOT deliver the message to the
  1456. same client within the context of the current
  1457. channel. The recommended strategy is to attempt
  1458. to deliver the message to an alternative consumer,
  1459. and if that is not possible, to move the message
  1460. to a dead-letter queue. The server MAY use more
  1461. sophisticated tracking to hold the message on the
  1462. queue and redeliver it to the same client at a
  1463. later stage.
  1464. """
  1465. return self.send_method(
  1466. spec.Basic.Reject, argsig, (delivery_tag, requeue),
  1467. )
  1468. def _on_basic_return(self, reply_code, reply_text,
  1469. exchange, routing_key, message):
  1470. """Return a failed message.
  1471. This method returns an undeliverable message that was
  1472. published with the "immediate" flag set, or an unroutable
  1473. message published with the "mandatory" flag set. The reply
  1474. code and text provide information about the reason that the
  1475. message was undeliverable.
  1476. PARAMETERS:
  1477. reply_code: short
  1478. The reply code. The AMQ reply codes are defined in AMQ
  1479. RFC 011.
  1480. reply_text: shortstr
  1481. The localised reply text. This text can be logged as an
  1482. aid to resolving issues.
  1483. exchange: shortstr
  1484. Specifies the name of the exchange that the message
  1485. was originally published to.
  1486. routing_key: shortstr
  1487. Message routing key
  1488. Specifies the routing key name specified when the
  1489. message was published.
  1490. """
  1491. exc = error_for_code(
  1492. reply_code, reply_text, spec.Basic.Return, ChannelError,
  1493. )
  1494. handlers = self.events.get('basic_return')
  1495. if not handlers:
  1496. raise exc
  1497. for callback in handlers:
  1498. callback(exc, exchange, routing_key, message)
  1499. #############
  1500. #
  1501. # Tx
  1502. #
  1503. #
  1504. # work with standard transactions
  1505. #
  1506. # Standard transactions provide so-called "1.5 phase commit". We
  1507. # can ensure that work is never lost, but there is a chance of
  1508. # confirmations being lost, so that messages may be resent.
  1509. # Applications that use standard transactions must be able to
  1510. # detect and ignore duplicate messages.
  1511. #
  1512. # GRAMMAR::
  1513. #
  1514. # tx = C:SELECT S:SELECT-OK
  1515. # / C:COMMIT S:COMMIT-OK
  1516. # / C:ROLLBACK S:ROLLBACK-OK
  1517. #
  1518. # RULE:
  1519. #
  1520. # An client using standard transactions SHOULD be able to
  1521. # track all messages received within a reasonable period, and
  1522. # thus detect and reject duplicates of the same message. It
  1523. # SHOULD NOT pass these to the application layer.
  1524. #
  1525. #
  1526. def tx_commit(self):
  1527. """Commit the current transaction.
  1528. This method commits all messages published and acknowledged in
  1529. the current transaction. A new transaction starts immediately
  1530. after a commit.
  1531. """
  1532. return self.send_method(spec.Tx.Commit, wait=spec.Tx.CommitOk)
  1533. def tx_rollback(self):
  1534. """Abandon the current transaction.
  1535. This method abandons all messages published and acknowledged
  1536. in the current transaction. A new transaction starts
  1537. immediately after a rollback.
  1538. """
  1539. return self.send_method(spec.Tx.Rollback, wait=spec.Tx.RollbackOk)
  1540. def tx_select(self):
  1541. """Select standard transaction mode.
  1542. This method sets the channel to use standard transactions.
  1543. The client must use this method at least once on a channel
  1544. before using the Commit or Rollback methods.
  1545. """
  1546. return self.send_method(spec.Tx.Select, wait=spec.Tx.SelectOk)
  1547. def confirm_select(self, nowait=False):
  1548. """Enable publisher confirms for this channel.
  1549. Note: This is an RabbitMQ extension.
  1550. Can now be used if the channel is in transactional mode.
  1551. :param nowait:
  1552. If set, the server will not respond to the method.
  1553. The client should not wait for a reply method. If the
  1554. server could not complete the method it will raise a channel
  1555. or connection exception.
  1556. """
  1557. return self.send_method(
  1558. spec.Confirm.Select, 'b', (nowait,),
  1559. wait=None if nowait else spec.Confirm.SelectOk,
  1560. )
  1561. def _on_basic_ack(self, delivery_tag, multiple):
  1562. for callback in self.events['basic_ack']:
  1563. callback(delivery_tag, multiple)
  1564. def _on_basic_nack(self, delivery_tag, multiple):
  1565. for callback in self.events['basic_nack']:
  1566. callback(delivery_tag, multiple)