jcgruenhage/void-packages
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

python3-aioamqp: update to 0.15.0.

Browse source
This commit is contained in:
Andrew J. Hesford 2022-04-05 21:24:10 -04:00
parent 4560508470
commit 0f4887548d
3 changed files with 3 additions and 817 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

345
srcpkgs/python3-aioamqp/patches/01-python-3.10.patch
View file

@ -1,345 +0,0 @@
From a3ab4768a13540e8c8f566bca990a98f96cb3802 Mon Sep 17 00:00:00 2001
From: dzen <benoit.calvez@polyconseil.fr>
Date: Thu, 5 Mar 2020 14:40:50 +0100
Subject: [PATCH 1/2] Support newer Python
Combines upstream commits to suppress asyncio deprecations:
- f004c52
- 388d38f
- 0cdfe91
- b4f0120
- 23d84ca
---
.travis.yml | 6 ++---
Dockerfile | 2 +-
aioamqp/__init__.py | 4 ++--
aioamqp/channel.py | 4 ++--
aioamqp/protocol.py | 33 ++++++++++++++------------
aioamqp/tests/test_basic.py | 2 +-
aioamqp/tests/test_connection_close.py | 2 +-
aioamqp/tests/test_connection_lost.py | 2 +-
aioamqp/tests/test_protocol.py | 4 ++--
docs/changelog.rst | 4 ++++
docs/introduction.rst | 2 +-
setup.cfg | 2 +-
setup.py | 2 +-
tox.ini | 2 +-
14 files changed, 39 insertions(+), 32 deletions(-)
diff --git a/.travis.yml b/.travis.yml
index 1069e7c..9d6e4ef 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,10 +1,10 @@
language: python
-dist: bionic
+dist: focal
python:
-- 3.5
- 3.6
-- 3.7-dev
+- 3.7
- 3.8
+- 3.9
services:
- rabbitmq
env:
diff --git a/Dockerfile b/Dockerfile
index 7ec4545..1ea92f1 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,4 +1,4 @@
-FROM python:3.5
+FROM python:3.9
WORKDIR /usr/src/app
diff --git a/aioamqp/__init__.py b/aioamqp/__init__.py
index 3e67936..030e7db 100644
--- a/aioamqp/__init__.py
+++ b/aioamqp/__init__.py
@@ -10,7 +10,7 @@ from .version import __packagename__
async def connect(host='localhost', port=None, login='guest', password='guest',
- virtualhost='/', ssl=None, login_method='AMQPLAIN', insist=False,
+ virtualhost='/', ssl=None, login_method='PLAIN', insist=False,
protocol_factory=AmqpProtocol, *, loop=None, **kwargs):
"""Convenient method to connect to an AMQP broker
@@ -69,7 +69,7 @@ async def connect(host='localhost', port=None, login='guest', password='guest',
async def from_url(
- url, login_method='AMQPLAIN', insist=False, protocol_factory=AmqpProtocol, **kwargs):
+ url, login_method='PLAIN', insist=False, protocol_factory=AmqpProtocol, **kwargs):
""" Connect to the AMQP using a single url parameter and return the client.
For instance:
diff --git a/aioamqp/channel.py b/aioamqp/channel.py
index 7f0f402..b58bd5a 100644
--- a/aioamqp/channel.py
+++ b/aioamqp/channel.py
@@ -31,7 +31,7 @@ class Channel:
self.cancellation_callbacks = []
self.return_callback = return_callback
self.response_future = None
- self.close_event = asyncio.Event(loop=self._loop)
+ self.close_event = asyncio.Event()
self.cancelled_consumers = set()
self.last_consumer_tag = None
self.publisher_confirms = False
@@ -518,7 +518,7 @@ class Channel:
}
future = self._get_waiter('basic_consume' + ctag)
future.set_result(results)
- self._ctag_events[ctag] = asyncio.Event(loop=self._loop)
+ self._ctag_events[ctag] = asyncio.Event()
async def basic_deliver(self, frame):
consumer_tag = frame.consumer_tag
diff --git a/aioamqp/protocol.py b/aioamqp/protocol.py
index ddf1285..4938833 100644
--- a/aioamqp/protocol.py
+++ b/aioamqp/protocol.py
@@ -79,8 +79,8 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
self.connection_tunning['heartbeat'] = kwargs.get('heartbeat')
self.connecting = asyncio.Future(loop=self._loop)
- self.connection_closed = asyncio.Event(loop=self._loop)
- self.stop_now = asyncio.Future(loop=self._loop)
+ self.connection_closed = asyncio.Event()
+ self.stop_now = asyncio.Event()
self.state = CONNECTING
self.version_major = None
self.version_minor = None
@@ -91,14 +91,14 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
self.server_heartbeat = None
self._heartbeat_timer_recv = None
self._heartbeat_timer_send = None
- self._heartbeat_trigger_send = asyncio.Event(loop=self._loop)
+ self._heartbeat_trigger_send = asyncio.Event()
self._heartbeat_worker = None
self.channels = {}
self.server_frame_max = None
self.server_channel_max = None
self.channels_ids_ceil = 0
self.channels_ids_free = set()
- self._drain_lock = asyncio.Lock(loop=self._loop)
+ self._drain_lock = asyncio.Lock()
def connection_made(self, transport):
super().connection_made(transport)
@@ -171,16 +171,13 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
await self.wait_closed(timeout=timeout)
async def wait_closed(self, timeout=None):
- await asyncio.wait_for(self.connection_closed.wait(), timeout=timeout, loop=self._loop)
+ await asyncio.wait_for(self.connection_closed.wait(), timeout=timeout)
if self._heartbeat_worker is not None:
try:
- await asyncio.wait_for(self._heartbeat_worker, timeout=timeout, loop=self._loop)
+ await asyncio.wait_for(self._heartbeat_worker, timeout=timeout)
except asyncio.CancelledError:
pass
- async def close_ok(self, frame):
- self._stream_writer.close()
-
async def start_connection(self, host, port, login, password, virtualhost, ssl=False,
login_method='PLAIN', insist=False):
"""Initiate a connection at the protocol level
@@ -188,7 +185,7 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
"""
if login_method != 'PLAIN':
- logger.warning('only PLAIN login_method is supported, falling back to AMQPLAIN')
+ logger.warning('login_method %s is not supported, falling back to PLAIN', login_method)
self._stream_writer.write(amqp_constants.PROTOCOL_HEADER)
@@ -311,12 +308,12 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
channel.connection_closed(reply_code, reply_text, exception)
async def run(self):
- while not self.stop_now.done():
+ while not self.stop_now.is_set():
try:
await self.dispatch_frame()
except exceptions.AmqpClosedConnection as exc:
logger.info("Close connection")
- self.stop_now.set_result(None)
+ self.stop_now.set()
self._close_channels(exception=exc)
except Exception: # pylint: disable=broad-except
@@ -329,7 +326,7 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
the rest of the AmqpProtocol class. This is kept around for backwards
compatibility purposes only.
"""
- await self.stop_now
+ await self.stop_now.wait()
async def send_heartbeat(self):
"""Sends an heartbeat message.
@@ -403,6 +400,11 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
)
await self._write_frame(0, request)
+ async def close_ok(self, frame):
+ """In response to server close confirmation"""
+ self.stop_now.set()
+ self._stream_writer.close()
+
async def server_close(self, frame):
"""The server is closing the connection"""
self.state = CLOSING
@@ -414,6 +416,7 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
reply_text, reply_code, class_id, method_id)
self._close_channels(reply_code, reply_text)
await self._close_ok()
+ self.stop_now.set()
self._stream_writer.close()
async def _close_ok(self):
@@ -456,11 +459,11 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
await self.ensure_open()
try:
channel_id = self.channels_ids_free.pop()
- except KeyError:
+ except KeyError as ex:
assert self.server_channel_max is not None, 'connection channel-max tuning not performed'
# channel-max = 0 means no limit
if self.server_channel_max and self.channels_ids_ceil > self.server_channel_max:
- raise exceptions.NoChannelAvailable()
+ raise exceptions.NoChannelAvailable() from ex
self.channels_ids_ceil += 1
channel_id = self.channels_ids_ceil
channel = self.CHANNEL_FACTORY(self, channel_id, **kwargs)
diff --git a/aioamqp/tests/test_basic.py b/aioamqp/tests/test_basic.py
index d59ce19..00ec880 100644
--- a/aioamqp/tests/test_basic.py
+++ b/aioamqp/tests/test_basic.py
@@ -58,7 +58,7 @@ class BasicCancelTestCase(testcase.RabbitTestCaseMixin, asynctest.TestCase):
result = await self.channel.publish("payload", exchange_name, routing_key='')
- await asyncio.sleep(5, loop=self.loop)
+ await asyncio.sleep(5)
result = await self.channel.queue_declare(queue_name, passive=True)
self.assertEqual(result['message_count'], 1)
diff --git a/aioamqp/tests/test_connection_close.py b/aioamqp/tests/test_connection_close.py
index 9491548..c5d4188 100644
--- a/aioamqp/tests/test_connection_close.py
+++ b/aioamqp/tests/test_connection_close.py
@@ -22,7 +22,7 @@ class CloseTestCase(testcase.RabbitTestCaseMixin, asynctest.TestCase):
# TODO: remove with python <3.4.4 support
self.assertTrue(transport._closing)
# make sure those 2 tasks/futures are properly set as finished
- await amqp.stop_now
+ await amqp.stop_now.wait()
await amqp.worker
async def test_multiple_close(self):
diff --git a/aioamqp/tests/test_connection_lost.py b/aioamqp/tests/test_connection_lost.py
index 21c7819..d16d51b 100644
--- a/aioamqp/tests/test_connection_lost.py
+++ b/aioamqp/tests/test_connection_lost.py
@@ -24,7 +24,7 @@ class ConnectionLostTestCase(testcase.RabbitTestCaseMixin, asynctest.TestCase):
self.assertEqual(amqp.state, OPEN)
self.assertTrue(channel.is_open)
amqp._stream_reader._transport.close() # this should have the same effect as the tcp connection being lost
- await asyncio.wait_for(amqp.worker, 1, loop=self.loop)
+ await asyncio.wait_for(amqp.worker, 1)
self.assertEqual(amqp.state, CLOSED)
self.assertFalse(channel.is_open)
self.assertTrue(self.callback_called)
diff --git a/aioamqp/tests/test_protocol.py b/aioamqp/tests/test_protocol.py
index 766a5d9..80d4187 100644
--- a/aioamqp/tests/test_protocol.py
+++ b/aioamqp/tests/test_protocol.py
@@ -55,7 +55,7 @@ class ProtocolTestCase(testcase.RabbitTestCaseMixin, asynctest.TestCase):
connect.assert_called_once_with(
insist=False,
password='pass',
- login_method='AMQPLAIN',
+ login_method='PLAIN',
login='tom',
host='example.com',
protocol_factory=AmqpProtocol,
@@ -74,7 +74,7 @@ class ProtocolTestCase(testcase.RabbitTestCaseMixin, asynctest.TestCase):
connect.assert_called_once_with(
insist=False,
password='pass',
- login_method='AMQPLAIN',
+ login_method='PLAIN',
ssl=ssl_context,
login='tom',
host='example.com',
diff --git a/docs/changelog.rst b/docs/changelog.rst
index 0939ca4..8ac4f7b 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -4,6 +4,10 @@ Changelog
Next release
------------
+ * Add support for Python 3.9.
+ * Drop support for Python 3.5.
+ * Fix annoying auth method warning because of a wrong defined default argument (closes #214).
+
Aioamqp 0.14.0
--------------
diff --git a/docs/introduction.rst b/docs/introduction.rst
index ec86f61..d0e183a 100644
--- a/docs/introduction.rst
+++ b/docs/introduction.rst
@@ -7,7 +7,7 @@ Aioamqp library is a pure-Python implementation of the AMQP 0.9.1 protocol using
Prerequisites
-------------
-Aioamqp works only with python >= 3.5 using asyncio library.
+Aioamqp works only with python >= 3.6 using asyncio library.
Installation
------------
diff --git a/setup.cfg b/setup.cfg
index 0ab7d0b..9a9f5ab 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,2 +1,2 @@
[bdist_wheel]
-python-tag = py35.py36.py37.py38
+python-tag = py36.py37.py38.py39
diff --git a/setup.py b/setup.py
index a740243..ea0e79f 100644
--- a/setup.py
+++ b/setup.py
@@ -34,10 +34,10 @@ setuptools.setup(
"Operating System :: OS Independent",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
- "Programming Language :: Python :: 3.5",
"Programming Language :: Python :: 3.6",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
+ "Programming Language :: Python :: 3.9",
],
platforms='all',
license='BSD'
diff --git a/tox.ini b/tox.ini
index 127cc7b..a1f5fcd 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,5 +1,5 @@
[tox]
-envlist = py35, py36, py37, py38
+envlist = py36, py37, py38, py39
skipsdist = true
skip_missing_interpreters = true
--
2.34.1

469
srcpkgs/python3-aioamqp/patches/02-pamqp-3.1.0.patch
View file

@ -1,469 +0,0 @@
From bb59736dc0127f3bd5b632d0531058f7d98384f1 Mon Sep 17 00:00:00 2001
From: dzen <benoit.calvez@polyconseil.fr>
Date: Thu, 12 Mar 2020 08:32:29 +0100
Subject: [PATCH 2/2] Require pamqp 3.1.0
Combines upstream commits:
- 6de2e1a
- 95cb160
- 9f08434
---
aioamqp/channel.py | 114 +++++++++++++++++++-------------------
aioamqp/frame.py | 1 -
aioamqp/properties.py | 6 +-
aioamqp/protocol.py | 22 ++++----
aioamqp/tests/testcase.py | 2 +-
setup.py | 3 +-
6 files changed, 76 insertions(+), 72 deletions(-)
diff --git a/aioamqp/channel.py b/aioamqp/channel.py
index b58bd5a..b87392d 100644
--- a/aioamqp/channel.py
+++ b/aioamqp/channel.py
@@ -9,7 +9,7 @@ import io
from itertools import count
import warnings
-import pamqp.specification
+import pamqp.commands
from . import frame as amqp_frame
from . import exceptions
@@ -78,35 +78,35 @@ class Channel:
async def dispatch_frame(self, frame):
methods = {
- pamqp.specification.Channel.OpenOk.name: self.open_ok,
- pamqp.specification.Channel.FlowOk.name: self.flow_ok,
- pamqp.specification.Channel.CloseOk.name: self.close_ok,
- pamqp.specification.Channel.Close.name: self.server_channel_close,
-
- pamqp.specification.Exchange.DeclareOk.name: self.exchange_declare_ok,
- pamqp.specification.Exchange.BindOk.name: self.exchange_bind_ok,
- pamqp.specification.Exchange.UnbindOk.name: self.exchange_unbind_ok,
- pamqp.specification.Exchange.DeleteOk.name: self.exchange_delete_ok,
-
- pamqp.specification.Queue.DeclareOk.name: self.queue_declare_ok,
- pamqp.specification.Queue.DeleteOk.name: self.queue_delete_ok,
- pamqp.specification.Queue.BindOk.name: self.queue_bind_ok,
- pamqp.specification.Queue.UnbindOk.name: self.queue_unbind_ok,
- pamqp.specification.Queue.PurgeOk.name: self.queue_purge_ok,
-
- pamqp.specification.Basic.QosOk.name: self.basic_qos_ok,
- pamqp.specification.Basic.ConsumeOk.name: self.basic_consume_ok,
- pamqp.specification.Basic.CancelOk.name: self.basic_cancel_ok,
- pamqp.specification.Basic.GetOk.name: self.basic_get_ok,
- pamqp.specification.Basic.GetEmpty.name: self.basic_get_empty,
- pamqp.specification.Basic.Deliver.name: self.basic_deliver,
- pamqp.specification.Basic.Cancel.name: self.server_basic_cancel,
- pamqp.specification.Basic.Ack.name: self.basic_server_ack,
- pamqp.specification.Basic.Nack.name: self.basic_server_nack,
- pamqp.specification.Basic.RecoverOk.name: self.basic_recover_ok,
- pamqp.specification.Basic.Return.name: self.basic_return,
-
- pamqp.specification.Confirm.SelectOk.name: self.confirm_select_ok,
+ pamqp.commands.Channel.OpenOk.name: self.open_ok,
+ pamqp.commands.Channel.FlowOk.name: self.flow_ok,
+ pamqp.commands.Channel.CloseOk.name: self.close_ok,
+ pamqp.commands.Channel.Close.name: self.server_channel_close,
+
+ pamqp.commands.Exchange.DeclareOk.name: self.exchange_declare_ok,
+ pamqp.commands.Exchange.BindOk.name: self.exchange_bind_ok,
+ pamqp.commands.Exchange.UnbindOk.name: self.exchange_unbind_ok,
+ pamqp.commands.Exchange.DeleteOk.name: self.exchange_delete_ok,
+
+ pamqp.commands.Queue.DeclareOk.name: self.queue_declare_ok,
+ pamqp.commands.Queue.DeleteOk.name: self.queue_delete_ok,
+ pamqp.commands.Queue.BindOk.name: self.queue_bind_ok,
+ pamqp.commands.Queue.UnbindOk.name: self.queue_unbind_ok,
+ pamqp.commands.Queue.PurgeOk.name: self.queue_purge_ok,
+
+ pamqp.commands.Basic.QosOk.name: self.basic_qos_ok,
+ pamqp.commands.Basic.ConsumeOk.name: self.basic_consume_ok,
+ pamqp.commands.Basic.CancelOk.name: self.basic_cancel_ok,
+ pamqp.commands.Basic.GetOk.name: self.basic_get_ok,
+ pamqp.commands.Basic.GetEmpty.name: self.basic_get_empty,
+ pamqp.commands.Basic.Deliver.name: self.basic_deliver,
+ pamqp.commands.Basic.Cancel.name: self.server_basic_cancel,
+ pamqp.commands.Basic.Ack.name: self.basic_server_ack,
+ pamqp.commands.Basic.Nack.name: self.basic_server_nack,
+ pamqp.commands.Basic.RecoverOk.name: self.basic_recover_ok,
+ pamqp.commands.Basic.Return.name: self.basic_return,
+
+ pamqp.commands.Confirm.SelectOk.name: self.confirm_select_ok,
}
if frame.name not in methods:
@@ -144,7 +144,7 @@ class Channel:
async def open(self):
"""Open the channel on the server."""
- request = pamqp.specification.Channel.Open()
+ request = pamqp.commands.Channel.Open()
return (await self._write_frame_awaiting_response(
'open', self.channel_id, request, no_wait=False, check_open=False))
@@ -159,7 +159,7 @@ class Channel:
if not self.is_open:
raise exceptions.ChannelClosed("channel already closed or closing")
self.close_event.set()
- request = pamqp.specification.Channel.Close(reply_code, reply_text, class_id=0, method_id=0)
+ request = pamqp.commands.Channel.Close(reply_code, reply_text, class_id=0, method_id=0)
return (await self._write_frame_awaiting_response(
'close', self.channel_id, request, no_wait=False, check_open=False))
@@ -169,7 +169,7 @@ class Channel:
self.protocol.release_channel_id(self.channel_id)
async def _send_channel_close_ok(self):
- request = pamqp.specification.Channel.CloseOk()
+ request = pamqp.commands.Channel.CloseOk()
await self._write_frame(self.channel_id, request)
async def server_channel_close(self, frame):
@@ -183,7 +183,7 @@ class Channel:
self.connection_closed(results['reply_code'], results['reply_text'])
async def flow(self, active):
- request = pamqp.specification.Channel.Flow(active)
+ request = pamqp.commands.Channel.Flow(active)
return (await self._write_frame_awaiting_response(
'flow', self.channel_id, request, no_wait=False,
check_open=False))
@@ -201,7 +201,7 @@ class Channel:
async def exchange_declare(self, exchange_name, type_name, passive=False, durable=False,
auto_delete=False, no_wait=False, arguments=None):
- request = pamqp.specification.Exchange.Declare(
+ request = pamqp.commands.Exchange.Declare(
exchange=exchange_name,
exchange_type=type_name,
passive=passive,
@@ -222,7 +222,7 @@ class Channel:
return future
async def exchange_delete(self, exchange_name, if_unused=False, no_wait=False):
- request = pamqp.specification.Exchange.Delete(exchange=exchange_name, if_unused=if_unused, nowait=no_wait)
+ request = pamqp.commands.Exchange.Delete(exchange=exchange_name, if_unused=if_unused, nowait=no_wait)
return await self._write_frame_awaiting_response(
'exchange_delete', self.channel_id, request, no_wait)
@@ -235,7 +235,7 @@ class Channel:
no_wait=False, arguments=None):
if arguments is None:
arguments = {}
- request = pamqp.specification.Exchange.Bind(
+ request = pamqp.commands.Exchange.Bind(
destination=exchange_destination,
source=exchange_source,
routing_key=routing_key,
@@ -255,7 +255,7 @@ class Channel:
if arguments is None:
arguments = {}
- request = pamqp.specification.Exchange.Unbind(
+ request = pamqp.commands.Exchange.Unbind(
destination=exchange_destination,
source=exchange_source,
routing_key=routing_key,
@@ -297,7 +297,7 @@ class Channel:
if not queue_name:
queue_name = 'aioamqp.gen-' + str(uuid.uuid4())
- request = pamqp.specification.Queue.Declare(
+ request = pamqp.commands.Queue.Declare(
queue=queue_name,
passive=passive,
durable=durable,
@@ -327,7 +327,7 @@ class Channel:
if_empty: bool, the queue is deleted if it has no messages. Raise if not.
no_wait: bool, if set, the server will not respond to the method
"""
- request = pamqp.specification.Queue.Delete(
+ request = pamqp.commands.Queue.Delete(
queue=queue_name,
if_unused=if_unused,
if_empty=if_empty,
@@ -346,7 +346,7 @@ class Channel:
if arguments is None:
arguments = {}
- request = pamqp.specification.Queue.Bind(
+ request = pamqp.commands.Queue.Bind(
queue=queue_name,
exchange=exchange_name,
routing_key=routing_key,
@@ -367,7 +367,7 @@ class Channel:
if arguments is None:
arguments = {}
- request = pamqp.specification.Queue.Unbind(
+ request = pamqp.commands.Queue.Unbind(
queue=queue_name,
exchange=exchange_name,
routing_key=routing_key,
@@ -383,7 +383,7 @@ class Channel:
logger.debug("Queue unbound")
async def queue_purge(self, queue_name, no_wait=False):
- request = pamqp.specification.Queue.Purge(
+ request = pamqp.commands.Queue.Purge(
queue=queue_name, nowait=no_wait
)
return (await self._write_frame_awaiting_response(
@@ -406,7 +406,7 @@ class Channel:
if properties is None:
properties = {}
- method_request = pamqp.specification.Basic.Publish(
+ method_request = pamqp.commands.Basic.Publish(
exchange=exchange_name,
routing_key=routing_key,
mandatory=mandatory,
@@ -417,7 +417,7 @@ class Channel:
header_request = pamqp.header.ContentHeader(
body_size=len(payload),
- properties=pamqp.specification.Basic.Properties(**properties)
+ properties=pamqp.commands.Basic.Properties(**properties)
)
await self._write_frame(self.channel_id, header_request, drain=False)
@@ -446,7 +446,7 @@ class Channel:
settings should apply per-consumer channel; and global=true to mean
that the QoS settings should apply per-channel.
"""
- request = pamqp.specification.Basic.Qos(
+ request = pamqp.commands.Basic.Qos(
prefetch_size, prefetch_count, connection_global
)
return (await self._write_frame_awaiting_response(
@@ -490,7 +490,7 @@ class Channel:
if arguments is None:
arguments = {}
- request = pamqp.specification.Basic.Consume(
+ request = pamqp.commands.Basic.Consume(
queue=queue_name,
consumer_tag=consumer_tag,
no_local=no_local,
@@ -561,7 +561,7 @@ class Channel:
callback, error)
async def basic_cancel(self, consumer_tag, no_wait=False):
- request = pamqp.specification.Basic.Cancel(consumer_tag, no_wait)
+ request = pamqp.commands.Basic.Cancel(consumer_tag, no_wait)
return (await self._write_frame_awaiting_response(
'basic_cancel', self.channel_id, request, no_wait=no_wait)
)
@@ -575,7 +575,7 @@ class Channel:
logger.debug("Cancel ok")
async def basic_get(self, queue_name='', no_ack=False):
- request = pamqp.specification.Basic.Get(queue=queue_name, no_ack=no_ack)
+ request = pamqp.commands.Basic.Get(queue=queue_name, no_ack=no_ack)
return (await self._write_frame_awaiting_response(
'basic_get', self.channel_id, request, no_wait=False)
)
@@ -606,11 +606,11 @@ class Channel:
future.set_exception(exceptions.EmptyQueue)
async def basic_client_ack(self, delivery_tag, multiple=False):
- request = pamqp.specification.Basic.Ack(delivery_tag, multiple)
+ request = pamqp.commands.Basic.Ack(delivery_tag, multiple)
await self._write_frame(self.channel_id, request)
async def basic_client_nack(self, delivery_tag, multiple=False, requeue=True):
- request = pamqp.specification.Basic.Nack(delivery_tag, multiple, requeue)
+ request = pamqp.commands.Basic.Nack(delivery_tag, multiple, requeue)
await self._write_frame(self.channel_id, request)
async def basic_server_ack(self, frame):
@@ -620,15 +620,15 @@ class Channel:
fut.set_result(True)
async def basic_reject(self, delivery_tag, requeue=False):
- request = pamqp.specification.Basic.Reject(delivery_tag, requeue)
+ request = pamqp.commands.Basic.Reject(delivery_tag, requeue)
await self._write_frame(self.channel_id, request)
async def basic_recover_async(self, requeue=True):
- request = pamqp.specification.Basic.RecoverAsync(requeue)
+ request = pamqp.commands.Basic.RecoverAsync(requeue)
await self._write_frame(self.channel_id, request)
async def basic_recover(self, requeue=True):
- request = pamqp.specification.Basic.Recover(requeue)
+ request = pamqp.commands.Basic.Recover(requeue)
return (await self._write_frame_awaiting_response(
'basic_recover', self.channel_id, request, no_wait=False)
)
@@ -681,7 +681,7 @@ class Channel:
delivery_tag = next(self.delivery_tag_iter) # pylint: disable=stop-iteration-return
fut = self._set_waiter('basic_server_ack_{}'.format(delivery_tag))
- method_request = pamqp.specification.Basic.Publish(
+ method_request = pamqp.commands.Basic.Publish(
exchange=exchange_name,
routing_key=routing_key,
mandatory=mandatory,
@@ -689,7 +689,7 @@ class Channel:
)
await self._write_frame(self.channel_id, method_request, drain=False)
- properties = pamqp.specification.Basic.Properties(**properties)
+ properties = pamqp.commands.Basic.Properties(**properties)
header_request = pamqp.header.ContentHeader(
body_size=len(payload), properties=properties
)
@@ -710,7 +710,7 @@ class Channel:
async def confirm_select(self, *, no_wait=False):
if self.publisher_confirms:
raise ValueError('publisher confirms already enabled')
- request = pamqp.specification.Confirm.Select(nowait=no_wait)
+ request = pamqp.commands.Confirm.Select(nowait=no_wait)
return (await self._write_frame_awaiting_response(
'confirm_select', self.channel_id, request, no_wait)
diff --git a/aioamqp/frame.py b/aioamqp/frame.py
index d70cfd7..af27ab5 100644
--- a/aioamqp/frame.py
+++ b/aioamqp/frame.py
@@ -42,7 +42,6 @@ import asyncio
import socket
import pamqp.encode
-import pamqp.specification
import pamqp.frame
from . import exceptions
diff --git a/aioamqp/properties.py b/aioamqp/properties.py
index 56a3484..4232040 100644
--- a/aioamqp/properties.py
+++ b/aioamqp/properties.py
@@ -1,4 +1,6 @@
# pylint: disable=redefined-builtin
+import datetime
+
from .constants import MESSAGE_PROPERTIES
@@ -37,7 +39,9 @@ def from_pamqp(instance):
props.reply_to = instance.reply_to
props.expiration = instance.expiration
props.message_id = instance.message_id
- props.timestamp = instance.timestamp
+ if instance.timestamp is not None:
+ # pamqp uses naive datetimes representing UTC, let's use TZ-aware datetimes
+ props.timestamp = instance.timestamp.replace(tzinfo=datetime.timezone.utc)
props.message_type = instance.message_type
props.user_id = instance.user_id
props.app_id = instance.app_id
diff --git a/aioamqp/protocol.py b/aioamqp/protocol.py
index 4938833..6465400 100644
--- a/aioamqp/protocol.py
+++ b/aioamqp/protocol.py
@@ -5,9 +5,9 @@
import asyncio
import logging
+import pamqp.commands
import pamqp.frame
import pamqp.heartbeat
-import pamqp.specification
from . import channel as amqp_channel
from . import constants as amqp_constants
@@ -159,7 +159,7 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
"""Close connection (and all channels)"""
await self.ensure_open()
self.state = CLOSING
- request = pamqp.specification.Connection.Close(
+ request = pamqp.commands.Connection.Close(
reply_code=0,
reply_text='',
class_id=0,
@@ -251,11 +251,11 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
"""Dispatch the received frame to the corresponding handler"""
method_dispatch = {
- pamqp.specification.Connection.Close.name: self.server_close,
- pamqp.specification.Connection.CloseOk.name: self.close_ok,
- pamqp.specification.Connection.Tune.name: self.tune,
- pamqp.specification.Connection.Start.name: self.start,
- pamqp.specification.Connection.OpenOk.name: self.open_ok,
+ pamqp.commands.Connection.Close.name: self.server_close,
+ pamqp.commands.Connection.CloseOk.name: self.close_ok,
+ pamqp.commands.Connection.Tune.name: self.tune,
+ pamqp.commands.Connection.Start.name: self.start,
+ pamqp.commands.Connection.OpenOk.name: self.open_ok,
}
if frame_channel is None and frame is None:
frame_channel, frame = await self.get_frame()
@@ -392,7 +392,7 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
def credentials():
return '\0{LOGIN}\0{PASSWORD}'.format(**auth)
- request = pamqp.specification.Connection.StartOk(
+ request = pamqp.commands.Connection.StartOk(
client_properties=client_properties,
mechanism=mechanism,
locale=locale,
@@ -420,7 +420,7 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
self._stream_writer.close()
async def _close_ok(self):
- request = pamqp.specification.Connection.CloseOk()
+ request = pamqp.commands.Connection.CloseOk()
await self._write_frame(0, request)
async def tune(self, frame):
@@ -429,7 +429,7 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
self.server_heartbeat = frame.heartbeat
async def tune_ok(self, channel_max, frame_max, heartbeat):
- request = pamqp.specification.Connection.TuneOk(
+ request = pamqp.commands.Connection.TuneOk(
channel_max, frame_max, heartbeat
)
await self._write_frame(0, request)
@@ -439,7 +439,7 @@ class AmqpProtocol(asyncio.StreamReaderProtocol):
async def open(self, virtual_host, capabilities='', insist=False):
"""Open connection to virtual host."""
- request = pamqp.specification.Connection.Open(
+ request = pamqp.commands.Connection.Open(
virtual_host, capabilities, insist
)
await self._write_frame(0, request)
diff --git a/aioamqp/tests/testcase.py b/aioamqp/tests/testcase.py
index 120104b..d6d702b 100644
--- a/aioamqp/tests/testcase.py
+++ b/aioamqp/tests/testcase.py
@@ -147,7 +147,7 @@ class RabbitTestCaseMixin:
if amqp is None:
amqp = self.amqp
- server_version = tuple(int(x) for x in amqp.server_properties['version'].decode().split('.'))
+ server_version = tuple(int(x) for x in amqp.server_properties['version'].split('.'))
return server_version
async def check_exchange_exists(self, exchange_name):
diff --git a/setup.py b/setup.py
index ea0e79f..1d0e923 100644
--- a/setup.py
+++ b/setup.py
@@ -25,8 +25,9 @@ setuptools.setup(
'aioamqp',
],
install_requires=[
- 'pamqp>=2.2.0,<3',
+ 'pamqp>=3.1.0',
],
+ python_requires=">=3.6",
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
--
2.34.1

6
srcpkgs/python3-aioamqp/template
View file

@ -1,7 +1,7 @@
# Template file for 'python3-aioamqp'
pkgname=python3-aioamqp
version=0.14.0
revision=4
version=0.15.0
revision=1
wrksrc="aioamqp-aioamqp-${version}"
build_style=python3-module
hostmakedepends="python3-setuptools"
@ -11,7 +11,7 @@ maintainer="Andrew J. Hesford <ajh@sideband.org>"
license="BSD-3-Clause"
homepage="https://github.com/polyconseil/aioamqp"
distfiles="${homepage}/archive/${pkgname#*-}-${version}.tar.gz"
checksum=9fce69be1ed9bcc06b74683094f738556305215a1affd0fd7789c23a01c53ecd
checksum=4d30fe210c742496b5b78de0337f49524571240701eb6c2dc348b1b9d8f819a9
# Tests require unpackaged asynctest package
make_check=no