diff --git a/openstack/message/v1/_proxy.py b/openstack/message/v1/_proxy.py index 81979e0a..ea6c205e 100644 --- a/openstack/message/v1/_proxy.py +++ b/openstack/message/v1/_proxy.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +from openstack.message.v1 import claim from openstack.message.v1 import message from openstack.message.v1 import queue from openstack import proxy @@ -44,29 +45,36 @@ class Proxy(proxy.BaseProxy): """ return self._delete(queue.Queue, value, ignore_missing=ignore_missing) - def create_messages(self, client, value, messages): + def create_messages(self, values): """Create new messages - :param uuid client: A UUID for each client instance. The UUID must - be submitted in its canonical form (for - example, 3381af92-2b9e-11e3-b191-71861300734c). - The client generates this UUID once. - The client UUID persists between restarts of the - client so the client should reuse that same - UUID. All message-related operations - require the use of the client UUID in the headers - to ensure that messages are not echoed back - to the client that posted them, unless the - client explicitly requests this. - :param value: The value can be either the name of a queue or a - :class:`~openstack.message.v1.queue.Queue` instance. - :param list messages: The list of + :param list values: The list of :class:`~openstack.message.v1.message.Message`s to create. :returns: The results of message creation - :rtype: list ids: A list of ids that correspond to the messages - created, in order. + :rtype: list messages: The list of + :class:`~openstack.message.v1.message.Message`s created. """ - queue_name = queue.Queue.get_id(value) - return message.Message.create_from_messages(self.session, client, - queue_name, messages) + return message.Message.create_messages(self.session, values) + + def claim_messages(self, value): + """Claims a set of messages. + + :param value: The value must be a + :class:`~openstack.message.v1.claim.Claim` instance. + + :returns: The results of a claim + :rtype: list messages: The list of + :class:`~openstack.message.v1.message.Message`s claimed. + """ + return claim.Claim.claim_messages(self.session, value) + + def delete_message(self, value): + """Delete a message + + :param value: The value must be a + :class:`~openstack.message.v1.message.Message` instance. + + :returns: ``None`` + """ + message.Message.delete_by_id(self.session, value) diff --git a/openstack/message/v1/claim.py b/openstack/message/v1/claim.py new file mode 100644 index 00000000..9dd56e90 --- /dev/null +++ b/openstack/message/v1/claim.py @@ -0,0 +1,71 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import json + +from openstack.message import message_service +from openstack.message.v1 import message +from openstack import resource + + +class Claim(resource.Resource): + resources_key = 'claims' + base_path = "/queues/%(queue_name)s/claims" + service = message_service.MessageService() + + # capabilities + allow_create = True + allow_list = False + allow_retrieve = False + allow_delete = False + + #: A UUID for each client instance. The UUID must be submitted in its + #: canonical form (for example, 3381af92-2b9e-11e3-b191-71861300734c). + #: The client generates this UUID once. The client UUID persists between + #: restarts of the client so the client should reuse that same UUID. + #: All message-related operations require the use of the client UUID in + #: the headers to ensure that messages are not echoed back to the client + #: that posted them, unless the client explicitly requests this. + client = None + + #: The queue this Claim belongs to. + queue = None + + #: Specifies the number of Messages to return. + limit = None + + #: Specifies how long the server waits before releasing the claim, + #: in seconds. + ttl = resource.prop("ttl") + + #: Specifies the message grace period, in seconds. + grace = resource.prop("grace") + + @classmethod + def claim_messages(cls, session, claim): + """Create a remote resource from this instance.""" + url = cls._get_url({'queue_name': claim.queue}) + headers = {'Client-ID': claim.client} + params = {'limit': claim.limit} if claim.limit else None + + resp = session.post(url, service=cls.service, headers=headers, + data=json.dumps(claim, cls=ClaimEncoder), + params=params) + + for message_attrs in resp.body: + yield message.Message.new( + client=claim.client, queue=claim.queue, **message_attrs) + + +class ClaimEncoder(json.JSONEncoder): + def default(self, claim): + return {'ttl': claim.ttl, 'grace': claim.grace} diff --git a/openstack/message/v1/message.py b/openstack/message/v1/message.py index e9cfad31..e9f89b1d 100644 --- a/openstack/message/v1/message.py +++ b/openstack/message/v1/message.py @@ -10,13 +10,14 @@ # License for the specific language governing permissions and limitations # under the License. +import copy import json +from six.moves.urllib import parse + from openstack.message import message_service from openstack import resource -from six.moves.urllib import parse - class Message(resource.Resource): resources_key = 'messages' @@ -29,6 +30,21 @@ class Message(resource.Resource): allow_retrieve = False allow_delete = False + #: A UUID for each client instance. The UUID must be submitted in its + #: canonical form (for example, 3381af92-2b9e-11e3-b191-71861300734c). + #: The client generates this UUID once. The client UUID persists between + #: restarts of the client so the client should reuse that same UUID. + #: All message-related operations require the use of the client UUID in + #: the headers to ensure that messages are not echoed back to the client + #: that posted them, unless the client explicitly requests this. + client = None + + #: The queue this Message belongs to. + queue = None + + #: A relative href that references this Message. + href = resource.prop("href") + #: An arbitrary JSON document that constitutes the body of the message #: being sent. body = resource.prop("body") @@ -40,28 +56,49 @@ class Message(resource.Resource): #: Specifies how long the message has been in the queue, in seconds. age = resource.prop("age") - @staticmethod - def get_message_id(href): - """Get the ID of a message, which is the last component in an href.""" - path = parse.urlparse(href).path - return path[path.rfind('/')+1:] - @classmethod - def create_from_messages(cls, session, client_id=None, queue_name=None, - messages=None): - """Create a remote resource from this instance.""" - url = cls._get_url({'queue_name': queue_name}) - headers = {'Client-ID': client_id} + def create_messages(cls, session, messages): + if len(messages) == 0: + raise ValueError('messages cannot be empty') + + for i, message in enumerate(messages, -1): + if message.queue != messages[i].queue: + raise ValueError('All queues in messages must be equal') + if message.client != messages[i].client: + raise ValueError('All clients in messages must be equal') + + url = cls._get_url({'queue_name': messages[0].queue}) + headers = {'Client-ID': messages[0].client} resp = session.post(url, service=cls.service, headers=headers, data=json.dumps(messages, cls=MessageEncoder)) + messages_deepcopy = copy.deepcopy(messages) hrefs = resp.body['resources'] - ids = [cls.get_message_id(href) for href in hrefs] - return ids + for i, href in enumerate(hrefs): + messages_deepcopy[i].href = href + + return messages_deepcopy + + @classmethod + def _strip_version(cls, href): + path = parse.urlparse(href).path + + if path.startswith('/v'): + return href[href.find('/', 1):] + else: + return href + + @classmethod + def delete_by_id(cls, session, message, path_args=None): + url = cls._strip_version(message.href) + headers = {'Client-ID': message.client} + + session.delete(url, service=cls.service, + headers=headers, accept=None) class MessageEncoder(json.JSONEncoder): - def default(self, obj): - return obj._attrs + def default(self, message): + return {'body': message.body, 'ttl': message.ttl} diff --git a/openstack/tests/unit/message/v1/test_claim.py b/openstack/tests/unit/message/v1/test_claim.py new file mode 100644 index 00000000..8011f74c --- /dev/null +++ b/openstack/tests/unit/message/v1/test_claim.py @@ -0,0 +1,62 @@ +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import json +import mock +import testtools + +from openstack.message.v1 import claim + +CLIENT = '3381af92-2b9e-11e3-b191-71861300734c' +QUEUE = 'test_queue' +LIMIT = 2 +FAKE = { + 'ttl': 300, + 'grace': 60 +} + + +class TestClaim(testtools.TestCase): + + def test_basic(self): + sot = claim.Claim() + self.assertEqual('claims', sot.resources_key) + self.assertEqual('/queues/%(queue_name)s/claims', sot.base_path) + self.assertEqual('messaging', sot.service.service_type) + self.assertTrue(sot.allow_create) + self.assertFalse(sot.allow_retrieve) + self.assertFalse(sot.allow_update) + self.assertFalse(sot.allow_delete) + self.assertFalse(sot.allow_list) + + def test_make_it(self): + sot = claim.Claim.new(client=CLIENT, queue=QUEUE, limit=LIMIT, **FAKE) + self.assertEqual(CLIENT, sot.client) + self.assertEqual(QUEUE, sot.queue) + self.assertEqual(LIMIT, sot.limit) + self.assertEqual(FAKE['ttl'], sot.ttl) + self.assertEqual(FAKE['grace'], sot.grace) + + def test_create(self): + sess = mock.Mock() + sess.post = mock.Mock() + sess.post.return_value = mock.MagicMock() + sot = claim.Claim() + + list(sot.claim_messages( + sess, claim.Claim.new(client=CLIENT, queue=QUEUE, **FAKE))) + + url = '/queues/%s/claims' % QUEUE + sess.post.assert_called_with( + url, service=sot.service, + headers={'Client-ID': CLIENT}, params=None, + data=json.dumps(FAKE, cls=claim.ClaimEncoder)) diff --git a/openstack/tests/unit/message/v1/test_message.py b/openstack/tests/unit/message/v1/test_message.py index 7bc19e91..dce358e7 100644 --- a/openstack/tests/unit/message/v1/test_message.py +++ b/openstack/tests/unit/message/v1/test_message.py @@ -16,12 +16,17 @@ import testtools from openstack.message.v1 import message -CLIENT_ID = '3381af92-2b9e-11e3-b191-71861300734c' -QUEUE_NAME = 'test_queue' +CLIENT = '3381af92-2b9e-11e3-b191-71861300734c' +QUEUE = 'test_queue' FAKE = { 'ttl': 300, 'body': {'key': 'value'} } +FAKE_HREF = { + 'href': '/v1/queues/test_queue/messages/1234', + 'ttl': 300, + 'body': {'key': 'value'} +} class TestMessage(testtools.TestCase): @@ -48,11 +53,25 @@ class TestMessage(testtools.TestCase): sess.post.return_value = mock.MagicMock() sot = message.Message() - sot.create_from_messages( - sess, CLIENT_ID, QUEUE_NAME, [message.Message.new(**FAKE)]) + sot.create_messages( + sess, [message.Message.new(client=CLIENT, queue=QUEUE, **FAKE)]) - url = '/queues/%s/messages' % QUEUE_NAME + url = '/queues/%s/messages' % QUEUE sess.post.assert_called_with( url, service=sot.service, - headers={'Client-ID': CLIENT_ID}, + headers={'Client-ID': CLIENT}, data=json.dumps([FAKE], cls=message.MessageEncoder)) + + def test_delete(self): + sess = mock.Mock() + sess.delete = mock.Mock() + sess.delete.return_value = mock.Mock() + sot = message.Message() + + sot.delete_by_id( + sess, message.Message.new(client=CLIENT, queue=QUEUE, **FAKE_HREF)) + + url = '/queues/%s/messages/1234' % QUEUE + sess.delete.assert_called_with( + url, service=sot.service, accept=None, + headers={'Client-ID': CLIENT}) diff --git a/openstack/tests/unit/message/v1/test_proxy.py b/openstack/tests/unit/message/v1/test_proxy.py index f079d3cc..54e8df20 100644 --- a/openstack/tests/unit/message/v1/test_proxy.py +++ b/openstack/tests/unit/message/v1/test_proxy.py @@ -11,6 +11,8 @@ # under the License. from openstack.message.v1 import _proxy +from openstack.message.v1 import claim +from openstack.message.v1 import message from openstack.message.v1 import queue from openstack.tests.unit import test_proxy_base @@ -41,7 +43,21 @@ class TestMessageProxy(test_proxy_base.TestProxyBase): def test_messages_create(self): self.verify_create2( - 'openstack.message.v1.message.Message.create_from_messages', + 'openstack.message.v1.message.Message.create_messages', self.proxy.create_messages, - method_args=[CLIENT_ID, QUEUE_NAME, []], - expected_args=[self.session, CLIENT_ID, QUEUE_NAME, []]) + method_args=[[]], + expected_args=[self.session, []]) + + def test_messages_claim(self): + self.verify_create2( + 'openstack.message.v1.claim.Claim.claim_messages', + self.proxy.claim_messages, + method_args=[claim.Claim], + expected_args=[self.session, claim.Claim]) + + def test_message_delete(self): + self.verify_delete( + 'openstack.message.v1.message.Message.delete_by_id', + self.proxy.delete_message, + method_args=[message.Message], + expected_args=[message.Message])