[Python-modules-commits] [python-pika] 01/09: Import python-pika_0.11.0.orig.tar.gz
Christopher Stuart Hoskin
mans0954 at moszumanska.debian.org
Wed Sep 27 03:31:13 UTC 2017
This is an automated email from the git hooks/post-receive script.
mans0954 pushed a commit to branch master
in repository python-pika.
commit 60eafcb3a71aa6a017e272bd7d1c185a7529b628
Author: Christopher Hoskin <mans0954 at debian.org>
Date: Wed Sep 27 03:50:53 2017 +0100
Import python-pika_0.11.0.orig.tar.gz
---
.checkignore | 8 +-
.codeclimate.yml | 14 +-
.gitignore | 2 +
.travis.yml | 42 +-
CHANGELOG.rst | 79 +-
CONTRIBUTING.md | 45 +
LICENSE | 2 +-
README.rst | 37 +-
appveyor.yml | 86 +
docs/examples.rst | 2 +
docs/examples/asynchronous_publisher_example.rst | 119 +-
..._publisher_example.rst => asyncio_consumer.rst} | 357 ++---
docs/examples/blocking_basic_get.rst | 4 +-
docs/examples/blocking_consume.rst | 28 +-
docs/examples/blocking_consumer_generator.rst | 8 +-
docs/examples/blocking_delivery_confirmations.rst | 4 +-
docs/examples/blocking_publish_mandatory.rst | 6 +-
docs/examples/direct_reply_to.rst | 81 +
docs/examples/heartbeat_and_blocked_timeouts.rst | 37 +
docs/examples/tls_mutual_authentication.rst | 61 +
docs/examples/tls_server_uathentication.rst | 60 +
docs/examples/twisted_example.rst | 2 +-
docs/examples/using_urlparameters.rst | 15 +-
docs/intro.rst | 10 +-
docs/modules/adapters/asyncio.rst | 9 +
docs/version_history.rst | 86 +-
.../{tmp.py => asynchronous_publisher_example.py} | 272 ++--
examples/direct_reply_to.py | 80 +
examples/heatbeat_and_blocked_timeouts.py | 48 +
examples/send.py | 6 +-
examples/twisted_service.py | 2 +-
nose.cfg | 3 -
pika/__init__.py | 2 +-
pika/adapters/__init__.py | 17 +-
pika/adapters/asyncio_connection.py | 221 +++
pika/adapters/base_connection.py | 175 ++-
pika/adapters/blocking_connection.py | 321 ++--
pika/adapters/libev_connection.py | 43 +-
pika/adapters/select_connection.py | 865 +++++++---
pika/adapters/tornado_connection.py | 9 +-
pika/adapters/twisted_connection.py | 21 +-
pika/channel.py | 568 ++++---
pika/compat.py | 35 +-
pika/connection.py | 1649 +++++++++++++-------
pika/credentials.py | 16 +
pika/exceptions.py | 26 +-
pika/heartbeat.py | 13 +-
pika/spec.py | 21 +-
pylintrc | 391 +++++
setup.cfg | 8 +
setup.py | 5 +-
test-requirements.txt | 2 +-
tests/acceptance/async_adapter_tests.py | 190 ++-
tests/acceptance/async_test_base.py | 159 +-
tests/acceptance/blocking_adapter_test.py | 328 +++-
tests/acceptance/enforce_one_basicget_test.py | 32 +
tests/acceptance/forward_server.py | 238 +--
tests/acceptance/test_utils.py | 73 +
tests/unit/base_connection_tests.py | 48 +
tests/unit/blocking_channel_tests.py | 24 +-
tests/unit/blocking_connection_tests.py | 164 +-
tests/unit/channel_tests.py | 658 +++++---
tests/unit/connection_parameters_tests.py | 642 ++++++++
tests/unit/connection_tests.py | 851 +++++++---
tests/unit/connection_timeout_tests.py | 149 +-
...r_tests.py => content_frame_assembler_tests.py} | 10 +-
tests/unit/credentials_tests.py | 110 ++
tests/unit/heartbeat_tests.py | 28 +-
tests/unit/parameter_tests.py | 69 -
tests/unit/select_connection_ioloop_tests.py | 314 +++-
tests/unit/tornado_tests.py | 5 +-
tests/unit/twisted_tests.py | 36 +
utils/codegen.py | 29 +-
73 files changed, 7620 insertions(+), 2560 deletions(-)
diff --git a/.checkignore b/.checkignore
index c28842f..e5a2365 100644
--- a/.checkignore
+++ b/.checkignore
@@ -1,5 +1,5 @@
-tests
-utils
-examples
-docs
+**/docs
+**/examples
+**/test
+**/utils
setup.py
diff --git a/.codeclimate.yml b/.codeclimate.yml
index 9f4bbca..ba9e7cc 100644
--- a/.codeclimate.yml
+++ b/.codeclimate.yml
@@ -1,8 +1,8 @@
languages:
- Python: true
-exclude_paths:
- - docs/*
- - tests/*
- - utils/*
- - pika/examples/*
- - pika/spec.py
+ - python
+exclude_paths:
+ - docs/*
+ - tests/*
+ - utils/*
+ - pika/examples/*
+ - pika/spec.py
diff --git a/.gitignore b/.gitignore
index 37efc2a..96467da 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,6 +4,7 @@
.coverage
.tox
.DS_Store
+.python-version
pika.iml
codegen
pika.egg-info
@@ -14,3 +15,4 @@ build
dist
docs/_build
*.conf.in
+venvs/
diff --git a/.travis.yml b/.travis.yml
index 8f8cfd7..2cb242c 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,23 +1,41 @@
language: python
+
python:
- - 2.6
- - 2.7
- - 3.3
- - 3.4
+ - 2.6
+ - 2.7
+ - 3.3
+ - 3.4
+ - 3.5
+ - 3.6
+
before_install:
- - sudo add-apt-repository "deb http://us.archive.ubuntu.com/ubuntu/ trusty main restricted universe multiverse"
- - sudo add-apt-repository "deb http://us.archive.ubuntu.com/ubuntu/ trusty-updates main restricted universe multiverse"
- - sudo apt-get update -qq
- - sudo apt-get install libev-dev/trusty
+ - sudo add-apt-repository "deb http://us.archive.ubuntu.com/ubuntu/ trusty main restricted universe multiverse"
+ - sudo add-apt-repository "deb http://us.archive.ubuntu.com/ubuntu/ trusty-updates main restricted universe multiverse"
+ - sudo apt-get update -qq
+ - sudo apt-get install libev-dev/trusty
+
install:
- - if [[ $TRAVIS_PYTHON_VERSION == '2.6' ]]; then pip install unittest2 ordereddict; fi
- - if [[ $TRAVIS_PYTHON_VERSION != '2.6' ]]; then pip install pyev; fi
- - pip install -r test-requirements.txt
+ - which -a python
+ - python --version
+ - which pip
+ - pip --version
+ - if [[ $TRAVIS_PYTHON_VERSION == '2.6' ]]; then pip install unittest2 ordereddict; fi
+ - if [[ $TRAVIS_PYTHON_VERSION != '2.6' ]]; then pip install pyev; fi
+ - pip install -r test-requirements.txt
+ - pip freeze
+
services:
- rabbitmq
-script: nosetests -c nose.cfg --with-coverage --cover-package=pika --cover-branches
+
+before_script:
+ - sudo rabbitmqctl status
+
+script:
+ - nosetests
+
after_success:
- codecov
+
deploy:
distributions: sdist bdist_wheel
provider: pypi
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index c20c512..26e54e0 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -1,6 +1,62 @@
+0.11.0 2017-07-29
+-----------------
+
+`GitHub milestone <https://github.com/pika/pika/milestone/9>`_
+
+ - Simplify Travis CI configuration for OS X.
+ - Add `asyncio` connection adapter for Python 3.4 and newer.
+ - Connection failures that occur after the socket is opened and before the
+ AMQP connection is ready to go are now reported by calling the connection
+ error callback. Previously these were not consistently reported.
+ - In BaseConnection.close, call _handle_ioloop_stop only if the connection is
+ already closed to allow the asynchronous close operation to complete
+ gracefully.
+ - Pass error information from failed socket connection to user callbacks
+ on_open_error_callback and on_close_callback with result_code=-1.
+ - ValueError is raised when a completion callback is passed to an asynchronous
+ (nowait) Channel operation. It's an application error to pass a non-None
+ completion callback with an asynchronous request, because this callback can
+ never be serviced in the asynchronous scenario.
+ - `Channel.basic_reject` fixed to allow `delivery_tag` to be of type `long`
+ as well as `int`. (by quantum5)
+ - Implemented support for blocked connection timeouts in
+ `pika.connection.Connection`. This feature is available to all pika adapters.
+ See `pika.connection.ConnectionParameters` docstring to learn more about
+ `blocked_connection_timeout` configuration.
+ - Deprecated the `heartbeat_interval` arg in `pika.ConnectionParameters` in
+ favor of the `heartbeat` arg for consistency with the other connection
+ parameters classes `pika.connection.Parameters` and `pika.URLParameters`.
+ - When the `port` arg is not set explicitly in `ConnectionParameters`
+ constructor, but the `ssl` arg is set explicitly, then set the port value to
+ to the default AMQP SSL port if SSL is enabled, otherwise to the default
+ AMQP plaintext port.
+ - `URLParameters` will raise ValueError if a non-empty URL scheme other than
+ {amqp | amqps | http | https} is specified.
+ - `InvalidMinimumFrameSize` and `InvalidMaximumFrameSize` exceptions are
+ deprecated. pika.connection.Parameters.frame_max property setter now raises
+ the standard `ValueError` exception when the value is out of bounds.
+ - Removed deprecated parameter `type` in `Channel.exchange_declare` and
+ `BlockingChannel.exchange_declare` in favor of the `exchange_type` arg that
+ doesn't overshadow the builtin `type` keyword.
+ - Channel.close() on OPENING channel transitions it to CLOSING instead of
+ raising ChannelClosed.
+ - Channel.close() on CLOSING channel raises `ChannelAlreadyClosing`; used to
+ raise `ChannelClosed`.
+ - Connection.channel() raises `ConnectionClosed` if connection is not in OPEN
+ state.
+ - When performing graceful close on a channel and `Channel.Close` from broker
+ arrives while waiting for CloseOk, don't release the channel number until
+ CloseOk arrives to avoid race condition that may lead to a new channel
+ receiving the CloseOk that was destined for the closing channel.
+ - The `backpressure_detection` option of `ConnectionParameters` and
+ `URLParameters` property is DEPRECATED in favor of `Connection.Blocked` and
+ `Connection.Unblocked`. See `Connection.add_on_connection_blocked_callback`.
+
0.10.0 2015-09-02
-----------------
+`0.10.0 <https://github.com/pika/pika/compare/0.9.14...0.10.0>`_
+
- a9bf96d - LibevConnection: Fixed dict chgd size during iteration (Michael Laing)
- 388c55d - SelectConnection: Fixed KeyError exceptions in IOLoop timeout executions (Shinji Suzuki)
- 4780de3 - BlockingConnection: Add support to make BlockingConnection a Context Manager (@reddec)
@@ -41,7 +97,7 @@ parameters channel, method, properties, and body instead of a tuple of those
values for congruence with other similar callbacks.
`BlockingConnection`: This adapter underwent a makeover under the hood and
-gained significant performance improvements as well as ehnanced timer
+gained significant performance improvements as well as enhanced timer
resolution. It is now implemented as a client of the `SelectConnection` adapter.
Below is an overview of the `BlockingConnection` and `BlockingChannel` API
@@ -226,6 +282,8 @@ changes:
0.9.14 - 2014-07-11
-------------------
+`0.9.14 <https://github.com/pika/pika/compare/0.9.13...0.9.14>`_
+
- 57fe43e - fix test to generate a correct range of random ints (ml)
- 0d68dee - fix async watcher for libev_connection (ml)
- 01710ad - Use default username and password if not specified in URLParameters (Sean Dwyer)
@@ -365,6 +423,9 @@ changes:
0.9.13 - 2013-05-15
-------------------
+
+`0.9.13 <https://github.com/pika/pika/compare/0.9.12...0.9.13>`_
+
**Major Changes**
- IPv6 Support with thanks to Alessandro Tagliapietra for initial prototype
@@ -404,6 +465,8 @@ changes:
0.9.12 - 2013-03-18
-------------------
+`0.9.12 <https://github.com/pika/pika/compare/0.9.11...0.9.12>`_
+
**Bugfixes**
- New timeout id hashing was not unique
@@ -411,6 +474,8 @@ changes:
0.9.11 - 2013-03-17
-------------------
+`0.9.11 <https://github.com/pika/pika/compare/0.9.10...0.9.11>`_
+
**Bugfixes**
- Address inconsistent channel close callback documentation and add the signature
@@ -422,6 +487,8 @@ changes:
0.9.10 - 2013-03-16
-------------------
+`0.9.10 <https://github.com/pika/pika/compare/0.9.9...0.9.10>`_
+
**Bugfixes**
- Fix timeout in twisted adapter (Submitted by cellscape)
@@ -456,6 +523,8 @@ changes:
0.9.9 - 2013-01-29
------------------
+`0.9.9 <https://github.com/pika/pika/compare/0.9.8...0.9.9>`_
+
**Bugfixes**
- Only remove the tornado_connection.TornadoConnection file descriptor from the IOLoop if it's still open (Issue #221)
@@ -479,6 +548,8 @@ changes:
0.9.8 - 2012-11-18
------------------
+`0.9.8 <https://github.com/pika/pika/compare/0.9.7...0.9.8>`_
+
**Bugfixes**
- Channel.queue_declare/BlockingChannel.queue_declare not setting up callbacks property for empty queue name (Issue #218)
@@ -489,6 +560,8 @@ changes:
0.9.7 - 2012-11-11
------------------
+`0.9.7 <https://github.com/pika/pika/compare/0.9.6...0.9.7>`_
+
**New features**
- generator based consumer in BlockingChannel (See :doc:`examples/blocking_consumer_generator` for example)
@@ -512,6 +585,8 @@ changes:
0.9.6 - 2012-10-29
------------------
+`0.9.6 <https://github.com/pika/pika/compare/0.9.5...0.9.6>`_
+
**New features**
- URLParameters
@@ -537,6 +612,8 @@ changes:
0.9.5 - 2011-03-29
------------------
+`0.9.5 <https://github.com/pika/pika/compare/0.9.4...0.9.5>`_
+
**Changelog**
- Scope changes with adapter IOLoops and CallbackManager allowing for cleaner, multi-threaded operation
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
new file mode 100644
index 0000000..326678e
--- /dev/null
+++ b/CONTRIBUTING.md
@@ -0,0 +1,45 @@
+# Contributing
+
+## Test Coverage
+
+To contribute to Pika, please make sure that any new features or changes
+to existing functionality **include test coverage**.
+
+*Pull requests that add or change code without coverage have a much lower chance
+of being accepted.*
+
+
+## Prerequisites
+
+Pika test suite has a couple of requirements:
+
+ * Dependencies from `test-dependencies.txt` are installed
+ * A RabbitMQ node with all defaults is running on `localhost:5672`
+
+
+## Installing Dependencies
+
+To install the dependencies needed to run Pika tests, use
+
+ pip install -r test-requirements.txt
+
+which on Python 3 might look like this
+
+ pip3 install -r test-requirements.txt
+
+
+## Running Tests
+
+To run all test suites, use
+
+ nosetests
+
+Note that some tests are OS-specific (e.g. epoll on Linux
+or kqueue on MacOS and BSD). Those will be skipped
+automatically.
+
+
+## Code Formatting
+
+Please format your code using [yapf](http://pypi.python.org/pypi/yapf)
+with ``google`` style prior to issuing your pull request.
diff --git a/LICENSE b/LICENSE
index 6d671a4..0aed110 100644
--- a/LICENSE
+++ b/LICENSE
@@ -1,4 +1,4 @@
-Copyright (c) 2009-2015, Tony Garnock-Jones, Gavin M. Roy, Pivotal and others.
+Copyright (c) 2009-2017, Tony Garnock-Jones, Gavin M. Roy, Pivotal and others.
All rights reserved.
Redistribution and use in source and binary forms, with or without modification,
diff --git a/README.rst b/README.rst
index 94c35eb..9abdf75 100644
--- a/README.rst
+++ b/README.rst
@@ -1,12 +1,13 @@
-Pika, an AMQP 0-9-1 client library for Python
-=============================================
+Pika
+====
+Pika is a RabbitMQ (AMQP-0-9-1) client library for Python.
-|Version| |Downloads| |Status| |Coverage| |License|
+|Version| |Status| |Coverage| |License| |Docs|
Introduction
-------------
-Pika is a pure-Python implementation of the AMQP 0-9-1 protocol that tries
-to stay fairly independent of the underlying network support library.
+Pika is a pure-Python implementation of the AMQP 0-9-1 protocol including RabbitMQ's
+extensions.
- Python 2.6+ and 3.3+ are supported.
@@ -22,8 +23,7 @@ to stay fairly independent of the underlying network support library.
Documentation
-------------
-
-Pika's documentation can be found at `https://pika.readthedocs.org <https://pika.readthedocs.org>`_
+Pika's documentation can be found at `https://pika.readthedocs.io <https://pika.readthedocs.io>`_
Example
-------
@@ -50,7 +50,7 @@ And an example of writing a blocking consumer:
for method_frame, properties, body in channel.consume('test'):
# Display the message parts and ack the message
- print method_frame, properties, body
+ print(method_frame, properties, body)
channel.basic_ack(method_frame.delivery_tag)
# Escape out of the loop after 10 messages
@@ -59,12 +59,13 @@ And an example of writing a blocking consumer:
# Cancel the consumer and return any pending messages
requeued_messages = channel.cancel()
- print 'Requeued %i messages' % requeued_messages
+ print('Requeued %i messages' % requeued_messages)
connection.close()
Pika provides the following adapters
------------------------------------
+- AsyncioConnection - adapter for the Python3 AsyncIO event loop
- BlockingConnection - enables blocking, synchronous operation on top of library for simple uses
- LibevConnection - adapter for use with the libev event loop http://libev.schmorp.de
- SelectConnection - fast asynchronous adapter
@@ -74,9 +75,12 @@ Pika provides the following adapters
Contributing
------------
To contribute to pika, please make sure that any new features or changes
-to existing functionality include test coverage. Additionally, please format
-your code using `yapf <http://pypi.python.org/pypi/yapf>`_ with ``google`` style
-prior to issuing your pull request.
+to existing functionality **include test coverage**.
+
+*Pull requests that add or change code without coverage will most likely be rejected.*
+
+Additionally, please format your code using `yapf <http://pypi.python.org/pypi/yapf>`_
+with ``google`` style prior to issuing your pull request.
.. |Version| image:: https://img.shields.io/pypi/v/pika.svg?
:target: http://badge.fury.io/py/pika
@@ -87,8 +91,9 @@ prior to issuing your pull request.
.. |Coverage| image:: https://img.shields.io/codecov/c/github/pika/pika.svg?
:target: https://codecov.io/github/pika/pika?branch=master
-.. |Downloads| image:: https://img.shields.io/pypi/dm/pika.svg?
- :target: https://pypi.python.org/pypi/pika
-
.. |License| image:: https://img.shields.io/pypi/l/pika.svg?
- :target: https://pika.readthedocs.org
+ :target: https://pika.readthedocs.io
+
+.. |Docs| image:: https://readthedocs.org/projects/pika/badge/?version=stable
+ :target: https://pika.readthedocs.io
+ :alt: Documentation Status
diff --git a/appveyor.yml b/appveyor.yml
new file mode 100644
index 0000000..a73286c
--- /dev/null
+++ b/appveyor.yml
@@ -0,0 +1,86 @@
+# Windows build and test of Pika
+
+environment:
+ erlang_download_url: "http://erlang.org/download/otp_win64_18.3.exe"
+ erlang_exe_path: "C:\\Users\\appveyor\\erlang.exe"
+ erlang_home_dir: "C:\\Users\\appveyor\\erlang"
+
+ rabbitmq_installer_download_url: "https://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/rabbitmq-server-3.6.1.exe"
+ rabbitmq_installer_path: "C:\\Users\\appveyor\\rabbitmq-server-3.6.1.exe"
+
+ matrix:
+ - PYTHON_ARCH: "32"
+ PYTHONHOME: "C:\\Python27"
+
+
+cache:
+ # RabbitMQ is a pretty big package, so caching it in hopes of expediting the
+ # runtime
+ - "%erlang_exe_path%"
+ - "%rabbitmq_installer_path%"
+
+
+install:
+ - SET PYTHONPATH=%PYTHONHOME%
+ - SET PATH=%PYTHONHOME%\Scripts;%PYTHONHOME%;%PATH%
+
+ # For diagnostics
+ - ECHO %PYTHONPATH%
+ - ECHO %PATH%
+ - python --version
+
+ - ECHO Upgrading pip...
+ - python -m pip install --upgrade pip setuptools
+ - pip --version
+
+ - ECHO Installing wheel...
+ - pip install wheel
+
+
+build_script:
+
+ - ECHO Building distributions...
+ - python setup.py sdist bdist bdist_wheel
+ - DIR /s *.whl
+
+
+artifacts:
+ - path: 'dist\*.whl'
+ name: pika wheel
+
+
+before_test:
+ # Install test requirements
+
+ - ECHO Installing pika...
+ - python setup.py install
+
+ - ECHO Installing pika test requirements...
+ - pip install -r test-requirements.txt
+
+ # List conents of C:\ to help debug caching of rabbitmq artifacts
+ - DIR C:\
+
+ - ps: $webclient=New-Object System.Net.WebClient
+
+ - ECHO Downloading Erlang...
+ - ps: if (-Not (Test-Path "$env:erlang_exe_path")) { $webclient.DownloadFile("$env:erlang_download_url", "$env:erlang_exe_path") } else { Write-Host "Found" $env:erlang_exe_path "in cache." }
+
+ - ECHO Starting Erlang...
+ - start /B /WAIT %erlang_exe_path% /S /D=%erlang_home_dir%
+ - set ERLANG_HOME=%erlang_home_dir%
+
+ - ECHO Downloading RabbitMQ...
+ - ps: if (-Not (Test-Path "$env:rabbitmq_installer_path")) { $webclient.DownloadFile("$env:rabbitmq_installer_download_url", "$env:rabbitmq_installer_path") } else { Write-Host "Found" $env:rabbitmq_installer_path "in cache." }
+
+ - ECHO Installing and starting RabbitMQ with default config...
+ - start /B /WAIT %rabbitmq_installer_path% /S
+ - ps: (Get-Service -Name RabbitMQ).Status
+
+
+test_script:
+ - nosetests
+
+
+# Not deploying Windows builds yet TODO
+deploy: false
diff --git a/docs/examples.rst b/docs/examples.rst
index e8fd62b..5df6740 100644
--- a/docs/examples.rst
+++ b/docs/examples.rst
@@ -19,3 +19,5 @@ Pika has various methods of use, between the synchronous BlockingConnection adap
examples/asynchronous_publisher_example
examples/twisted_example
examples/tornado_consumer
+ examples/tls_mutual_authentication
+ examples/tls_server_authentication
diff --git a/docs/examples/asynchronous_publisher_example.rst b/docs/examples/asynchronous_publisher_example.rst
index 6cba07c..887148d 100644
--- a/docs/examples/asynchronous_publisher_example.rst
+++ b/docs/examples/asynchronous_publisher_example.rst
@@ -11,7 +11,7 @@ publisher.py::
import json
LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
- '-35s %(lineno) -5d: %(message)s')
+ '-35s %(lineno) -5d: %(message)s')
LOGGER = logging.getLogger(__name__)
@@ -43,13 +43,14 @@ publisher.py::
"""
self._connection = None
self._channel = None
- self._deliveries = []
- self._acked = 0
- self._nacked = 0
- self._message_number = 0
+
+ self._deliveries = None
+ self._acked = None
+ self._nacked = None
+ self._message_number = None
+
self._stopping = False
self._url = amqp_url
- self._closing = False
def connect(self):
"""This method connects to RabbitMQ, returning the connection handle.
@@ -63,7 +64,8 @@ publisher.py::
"""
LOGGER.info('Connecting to %s', self._url)
return pika.SelectConnection(pika.URLParameters(self._url),
- self.on_connection_open,
+ on_open_callback=self.on_connection_open,
+ on_close_callback=self.on_connection_closed,
stop_ioloop_on_close=False)
def on_connection_open(self, unused_connection):
@@ -75,17 +77,8 @@ publisher.py::
"""
LOGGER.info('Connection opened')
- self.add_on_connection_close_callback()
self.open_channel()
- def add_on_connection_close_callback(self):
- """This method adds an on close callback that will be invoked by pika
- when RabbitMQ closes the connection to the publisher unexpectedly.
-
- """
- LOGGER.info('Adding connection close callback')
- self._connection.add_on_close_callback(self.on_connection_closed)
-
def on_connection_closed(self, connection, reply_code, reply_text):
"""This method is invoked by pika when the connection to RabbitMQ is
closed unexpectedly. Since it is unexpected, we will reconnect to
@@ -97,31 +90,12 @@ publisher.py::
"""
self._channel = None
- if self._closing:
+ if self._stopping:
self._connection.ioloop.stop()
else:
LOGGER.warning('Connection closed, reopening in 5 seconds: (%s) %s',
reply_code, reply_text)
- self._connection.add_timeout(5, self.reconnect)
-
- def reconnect(self):
- """Will be invoked by the IOLoop timer if the connection is
- closed. See the on_connection_closed method.
-
- """
- self._deliveries = []
- self._acked = 0
- self._nacked = 0
- self._message_number = 0
-
- # This is the old connection IOLoop instance, stop its ioloop
- self._connection.ioloop.stop()
-
- # Create a new connection
- self._connection = self.connect()
-
- # There is now a new connection, needs a new ioloop to run
- self._connection.ioloop.start()
+ self._connection.add_timeout(5, self._connection.ioloop.stop)
def open_channel(self):
"""This method will open a new channel with RabbitMQ by issuing the
@@ -162,13 +136,14 @@ publisher.py::
different parameters. In this case, we'll close the connection
to shutdown the object.
- :param pika.channel.Channel: The closed channel
+ :param pika.channel.Channel channel: The closed channel
:param int reply_code: The numeric reason the channel was closed
:param str reply_text: The text reason the channel was closed
"""
LOGGER.warning('Channel was closed: (%s) %s', reply_code, reply_text)
- if not self._closing:
+ self._channel = None
+ if not self._stopping:
self._connection.close()
def setup_exchange(self, exchange_name):
@@ -282,8 +257,6 @@ publisher.py::
message to be delivered in PUBLISH_INTERVAL seconds.
"""
- if self._stopping:
- return
LOGGER.info('Scheduling next message for %0.1f seconds',
self.PUBLISH_INTERVAL)
self._connection.add_timeout(self.PUBLISH_INTERVAL,
@@ -302,16 +275,17 @@ publisher.py::
class.
"""
- if self._stopping:
+ if self._channel is None or not self._channel.is_open:
return
- message = {u'مفتاح': u' قيمة',
- u'键': u'值',
- u'キー': u'値'}
+ hdrs = {u'مفتاح': u' قيمة',
+ u'键': u'值',
+ u'キー': u'値'}
properties = pika.BasicProperties(app_id='example-publisher',
content_type='application/json',
- headers=message)
+ headers=hdrs)
+ message = u'مفتاح قيمة 键 值 キー 値'
self._channel.basic_publish(self.EXCHANGE, self.ROUTING_KEY,
json.dumps(message, ensure_ascii=False),
properties)
@@ -320,21 +294,28 @@ publisher.py::
LOGGER.info('Published message # %i', self._message_number)
self.schedule_next_message()
- def close_channel(self):
- """Invoke this command to close the channel with RabbitMQ by sending
- the Channel.Close RPC command.
-
- """
- LOGGER.info('Closing the channel')
- if self._channel:
- self._channel.close()
-
def run(self):
"""Run the example code by connecting and then starting the IOLoop.
"""
- self._connection = self.connect()
- self._connection.ioloop.start()
+ while not self._stopping:
+ self._connection = None
+ self._deliveries = []
+ self._acked = 0
+ self._nacked = 0
+ self._message_number = 0
+
+ try:
+ self._connection = self.connect()
+ self._connection.ioloop.start()
+ except KeyboardInterrupt:
+ self.stop()
+ if (self._connection is not None and
+ not self._connection.is_closed):
+ # Finish closing
+ self._connection.ioloop.start()
+
+ LOGGER.info('Stopped')
def stop(self):
"""Stop the example by closing the channel and connection. We
@@ -349,14 +330,21 @@ publisher.py::
self._stopping = True
self.close_channel()
self.close_connection()
- self._connection.ioloop.start()
- LOGGER.info('Stopped')
+
+ def close_channel(self):
+ """Invoke this command to close the channel with RabbitMQ by sending
+ the Channel.Close RPC command.
+
+ """
+ if self._channel is not None:
+ LOGGER.info('Closing the channel')
+ self._channel.close()
def close_connection(self):
"""This method closes the connection to RabbitMQ."""
- LOGGER.info('Closing connection')
- self._closing = True
- self._connection.close()
+ if self._connection is not None:
+ LOGGER.info('Closing connection')
+ self._connection.close()
def main():
@@ -364,11 +352,8 @@ publisher.py::
# Connect to localhost:5672 as guest with the password guest and virtual host "/" (%2F)
example = ExamplePublisher('amqp://guest:guest@localhost:5672/%2F?connection_attempts=3&heartbeat_interval=3600')
- try:
- example.run()
- except KeyboardInterrupt:
- example.stop()
+ example.run()
+
if __name__ == '__main__':
main()
-
diff --git a/docs/examples/asynchronous_publisher_example.rst b/docs/examples/asyncio_consumer.rst
similarity index 50%
copy from docs/examples/asynchronous_publisher_example.rst
copy to docs/examples/asyncio_consumer.rst
index 6cba07c..cb4914e 100644
--- a/docs/examples/asynchronous_publisher_example.rst
+++ b/docs/examples/asyncio_consumer.rst
@@ -1,22 +1,26 @@
-Asynchronous publisher example
-==============================
-The following example implements a publisher that will respond to RPC commands sent from RabbitMQ and uses delivery confirmations. It will reconnect if RabbitMQ closes the connection and will shutdown if RabbitMQ closes the channel. While it may look intimidating, each method is very short and represents a individual actions that a publisher can do.
-
-publisher.py::
-
- # -*- coding: utf-8 -*-
-
- import logging
+Asyncio Consumer
+================
+The following example implements a consumer using the
+:class:`Asyncio adapter <pika.adapters.asyncio_connection.AsyncioConnection>` for the
+`Asyncio library <https://docs.python.org/3/library/asyncio.html>`_ that will respond to RPC commands sent
+ from RabbitMQ. For example, it will reconnect if RabbitMQ closes the connection and will shutdown if
+ RabbitMQ cancels the consumer or closes the channel. While it may look intimidating, each method is
+ very short and represents a individual actions that a consumer can do.
+
+consumer.py::
+
+ from pika import adapters
import pika
- import json
+ import logging
LOG_FORMAT = ('%(levelname) -10s %(asctime)s %(name) -30s %(funcName) '
'-35s %(lineno) -5d: %(message)s')
+
LOGGER = logging.getLogger(__name__)
- class ExamplePublisher(object):
- """This is an example publisher that will handle unexpected interactions
+ class ExampleConsumer(object):
+ """This is an example consumer that will handle unexpected interactions
with RabbitMQ such as channel and connection closures.
If RabbitMQ closes the connection, it will reopen it. You should
@@ -24,59 +28,44 @@ publisher.py::
be closed, which usually are tied to permission related issues or
socket timeouts.
- It uses delivery confirmations and illustrates one way to keep track of
- messages that have been sent and if they've been confirmed by RabbitMQ.
+ If the channel is closed, it will indicate a problem with one of the
+ commands that were issued and that should surface in the output as well.
"""
EXCHANGE = 'message'
EXCHANGE_TYPE = 'topic'
- PUBLISH_INTERVAL = 1
QUEUE = 'text'
ROUTING_KEY = 'example.text'
def __init__(self, amqp_url):
- """Setup the example publisher object, passing in the URL we will use
- to connect to RabbitMQ.
+ """Create a new instance of the consumer class, passing in the AMQP
+ URL used to connect to RabbitMQ.
- :param str amqp_url: The URL for connecting to RabbitMQ
+ :param str amqp_url: The AMQP url to connect with
"""
self._connection = None
self._channel = None
- self._deliveries = []
- self._acked = 0
- self._nacked = 0
- self._message_number = 0
- self._stopping = False
- self._url = amqp_url
self._closing = False
+ self._consumer_tag = None
+ self._url = amqp_url
def connect(self):
"""This method connects to RabbitMQ, returning the connection handle.
When the connection is established, the on_connection_open method
- will be invoked by pika. If you want the reconnection to work, make
- sure you set stop_ioloop_on_close to False, which is not the default
- behavior of this adapter.
+ will be invoked by pika.
:rtype: pika.SelectConnection
"""
LOGGER.info('Connecting to %s', self._url)
- return pika.SelectConnection(pika.URLParameters(self._url),
- self.on_connection_open,
- stop_ioloop_on_close=False)
-
- def on_connection_open(self, unused_connection):
- """This method is called by pika once the connection to RabbitMQ has
- been established. It passes the handle to the connection object in
- case we need it, but in this case, we'll just mark it unused.
-
- :type unused_connection: pika.SelectConnection
+ return adapters.AsyncioConnection(pika.URLParameters(self._url),
+ self.on_connection_open)
- """
- LOGGER.info('Connection opened')
- self.add_on_connection_close_callback()
- self.open_channel()
+ def close_connection(self):
+ """This method closes the connection to RabbitMQ."""
+ LOGGER.info('Closing connection')
+ self._connection.close()
def add_on_connection_close_callback(self):
"""This method adds an on close callback that will be invoked by pika
@@ -104,48 +93,27 @@ publisher.py::
reply_code, reply_text)
self._connection.add_timeout(5, self.reconnect)
- def reconnect(self):
- """Will be invoked by the IOLoop timer if the connection is
- closed. See the on_connection_closed method.
-
- """
- self._deliveries = []
- self._acked = 0
- self._nacked = 0
- self._message_number = 0
-
- # This is the old connection IOLoop instance, stop its ioloop
- self._connection.ioloop.stop()
-
- # Create a new connection
- self._connection = self.connect()
-
- # There is now a new connection, needs a new ioloop to run
- self._connection.ioloop.start()
+ def on_connection_open(self, unused_connection):
+ """This method is called by pika once the connection to RabbitMQ has
+ been established. It passes the handle to the connection object in
+ case we need it, but in this case, we'll just mark it unused.
- def open_channel(self):
- """This method will open a new channel with RabbitMQ by issuing the
- Channel.Open RPC command. When RabbitMQ confirms the channel is open
- by sending the Channel.OpenOK RPC reply, the on_channel_open method
- will be invoked.
+ :type unused_connection: pika.SelectConnection
"""
- LOGGER.info('Creating a new channel')
- self._connection.channel(on_open_callback=self.on_channel_open)
-
- def on_channel_open(self, channel):
- """This method is invoked by pika when the channel has been opened.
- The channel object is passed in so we can make use of it.
-
- Since the channel is now open, we'll declare the exchange to use.
+ LOGGER.info('Connection opened')
+ self.add_on_connection_close_callback()
+ self.open_channel()
- :param pika.channel.Channel channel: The channel object
... 14518 lines suppressed ...
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/python-modules/packages/python-pika.git
More information about the Python-modules-commits
mailing list