Visitar URL original
Merge pull request #800 from tseaver/744-pubsub-message_class · googleapis/google-cloud-python@9c7f1fd · GitHub
Skip to content

Commit 9c7f1fd

Browse files
tseavertseaver
committed
Merge pull request #800 from tseaver/744-pubsub-message_class
#744: Add 'pubsub.message.Message' class
2 parents f3c3543 + 6a5f480 commit 9c7f1fd

9 files changed

Lines changed: 328 additions & 71 deletions

File tree

‎docs/pubsub-surface.rst‎

Lines changed: 18 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -199,11 +199,7 @@ Delete a subscription:
199199
Pull messages from a subscription
200200
---------------------------------
201201

202-
Fetch pending messages for a pull subscription
203-
204-
.. note::
205-
206-
The messages will have been ACKed already.
202+
Fetch pending messages for a pull subscription:
207203

208204
.. doctest::
209205

@@ -215,14 +211,23 @@ Fetch pending messages for a pull subscription
215211
... topic.publish('this is the first message_payload')
216212
... topic.publish('this is the second message_payload',
217213
... attr1='value1', attr2='value2')
218-
>>> messages = subscription.pull() # API request
214+
>>> received = subscription.pull() # API request
215+
>>> messages = [recv[1] for recv in received]
219216
>>> [message.id for message in messages]
220217
[<message_id1>, <message_id2>]
221218
>>> [message.data for message in messages]
222219
['this is the first message_payload', 'this is the second message_payload']
223-
>>> [message.attrs for message in messages]
220+
>>> [message.attributes for message in messages]
224221
[{}, {'attr1': 'value1', 'attr2': 'value2'}]
225222

223+
Note that received messages must be acknowledged, or else the back-end
224+
will re-send them later:
225+
226+
.. doctest::
227+
228+
>>> ack_ids = [recv[0] for recv in received]
229+
>>> subscription.acknowledge(ack_ids)
230+
226231
Fetch a limited number of pending messages for a pull subscription:
227232

228233
.. doctest::
@@ -235,8 +240,9 @@ Fetch a limited number of pending messages for a pull subscription:
235240
... topic.publish('this is the first message_payload')
236241
... topic.publish('this is the second message_payload',
237242
... attr1='value1', attr2='value2')
238-
>>> [message.id for message in subscription.pull(max_messages=1)]
239-
[<message_id1>]
243+
>>> received = subscription.pull(max_messages=1) # API request
244+
>>> messages = [recv[1] for recv in received]
245+
>>> [message.id for message in messages]
240246

241247
Fetch messages for a pull subscription without blocking (none pending):
242248

@@ -246,6 +252,7 @@ Fetch messages for a pull subscription without blocking (none pending):
246252
>>> from gcloud.pubsub.subscription import Subscription
247253
>>> topic = Topic('topic_name')
248254
>>> subscription = Subscription('subscription_name', topic)
249-
>>> [message.id for message in subscription.pull(return_immediately=True)]
255+
>>> received = subscription.pull(max_messages=1) # API request
256+
>>> messages = [recv[1] for recv in received]
257+
>>> [message.id for message in messages]
250258
[]
251-

‎gcloud/pubsub/api.py‎

Lines changed: 4 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -66,7 +66,7 @@ def list_topics(page_size=None, page_token=None,
6666

6767
path = '/projects/%s/topics' % project
6868
resp = connection.api_request(method='GET', path=path, query_params=params)
69-
topics = [_topic_from_resource(resource, connection)
69+
topics = [Topic.from_api_repr(resource, connection)
7070
for resource in resp['topics']]
7171
return topics, resp.get('nextPageToken')
7272

@@ -128,47 +128,8 @@ def list_subscriptions(page_size=None, page_token=None, topic_name=None,
128128

129129
resp = connection.api_request(method='GET', path=path, query_params=params)
130130
topics = {}
131-
subscriptions = [_subscription_from_resource(resource, topics, connection)
131+
subscriptions = [Subscription.from_api_repr(resource,
132+
connection=connection,
133+
topics=topics)
132134
for resource in resp['subscriptions']]
133135
return subscriptions, resp.get('nextPageToken')
134-
135-
136-
def _topic_from_resource(resource, connection):
137-
"""Construct a topic given its full path-like name.
138-
139-
:type resource: dict
140-
:param resource: topic resource representation returned from the API
141-
142-
:type connection: :class:`gcloud.pubsub.connection.Connection`
143-
:param connection: connection to use for the topic.
144-
145-
:rtype: :class:`gcloud.pubsub.topic.Topic`
146-
"""
147-
_, project, _, name = resource['name'].split('/')
148-
return Topic(name, project, connection)
149-
150-
151-
def _subscription_from_resource(resource, topics, connection):
152-
"""Construct a topic given its full path-like name.
153-
154-
:type resource: string
155-
:param resource: subscription resource representation returned from the API
156-
157-
:type topics: dict, full_name -> :class:`gcloud.pubsub.topic.Topic`
158-
:param topics: the topics to which subscriptions have been bound
159-
160-
:type connection: :class:`gcloud.pubsub.connection.Connection`
161-
:param connection: connection to use for the topic.
162-
163-
:rtype: :class:`gcloud.pubsub.subscription.Subscription`
164-
"""
165-
t_name = resource['topic']
166-
topic = topics.get(t_name)
167-
if topic is None:
168-
topic = topics[t_name] = _topic_from_resource({'name': t_name},
169-
connection)
170-
_, _, _, name = resource['name'].split('/')
171-
ack_deadline = resource.get('ackDeadlineSeconds')
172-
push_config = resource.get('pushConfig', {})
173-
push_endpoint = push_config.get('pushEndpoint')
174-
return Subscription(name, topic, ack_deadline, push_endpoint)

‎gcloud/pubsub/message.py‎

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,56 @@
1+
# Copyright 2015 Google Inc. All rights reserved.
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
"""Define API Topics."""
16+
17+
import base64
18+
19+
20+
class Message(object):
21+
"""Messages can be published to a topic and received by subscribers.
22+
23+
See:
24+
https://cloud.google.com/pubsub/reference/rest/google/pubsub/v1beta2/PubsubMessage
25+
26+
:type name: bytes
27+
:param name: the payload of the message
28+
29+
:type message_id: string
30+
:param message_id: An ID assigned to the message by the API.
31+
32+
:type attrs: dict or None
33+
:param attrs: Extra metadata associated by the publisher with the message.
34+
"""
35+
def __init__(self, data, message_id, attributes=None):
36+
self.data = data
37+
self.message_id = message_id
38+
self._attrs = attributes
39+
40+
@property
41+
def attrs(self):
42+
"""Lazily-constructed attribute dictionary"""
43+
if self._attrs is None:
44+
self._attrs = {}
45+
return self._attrs
46+
47+
@classmethod
48+
def from_api_repr(cls, api_repr):
49+
"""Factory: construct message from API representation.
50+
51+
:type api_repr: dict or None
52+
:param api_repr: The API representation of the message
53+
"""
54+
data = base64.b64decode(api_repr['data'])
55+
return cls(data=data, message_id=api_repr['messageId'],
56+
attributes=api_repr.get('attributes'))

‎gcloud/pubsub/subscription.py‎

Lines changed: 39 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@
1515
"""Define API Subscriptions."""
1616

1717
from gcloud.exceptions import NotFound
18+
from gcloud.pubsub.message import Message
19+
from gcloud.pubsub.topic import Topic
1820

1921

2022
class Subscription(object):
@@ -43,6 +45,37 @@ def __init__(self, name, topic, ack_deadline=None, push_endpoint=None):
4345
self.ack_deadline = ack_deadline
4446
self.push_endpoint = push_endpoint
4547

48+
@classmethod
49+
def from_api_repr(cls, resource, connection=None, topics=None):
50+
"""Factory: construct a topic given its API representation
51+
52+
:type resource: dict
53+
:param resource: topic resource representation returned from the API
54+
55+
:type connection: :class:`gcloud.pubsub.connection.Connection` or None
56+
:param connection: the connection to use. If not passed,
57+
falls back to the default inferred from the
58+
environment.
59+
60+
:type topics: dict or None
61+
:param topics: A mapping of topic names -> topics. If not passed,
62+
the subscription will have a newly-created topic.
63+
64+
:rtype: :class:`gcloud.pubsub.subscription.Subscription`
65+
"""
66+
if topics is None:
67+
topics = {}
68+
t_name = resource['topic']
69+
topic = topics.get(t_name)
70+
if topic is None:
71+
topic = topics[t_name] = Topic.from_api_repr({'name': t_name},
72+
connection)
73+
_, _, _, name = resource['name'].split('/')
74+
ack_deadline = resource.get('ackDeadlineSeconds')
75+
push_config = resource.get('pushConfig', {})
76+
push_endpoint = push_config.get('pushEndpoint')
77+
return cls(name, topic, ack_deadline, push_endpoint)
78+
4679
@property
4780
def path(self):
4881
"""URL path for the subscription's APIs"""
@@ -130,18 +163,19 @@ def pull(self, return_immediately=False, max_messages=1):
130163
:type max_messages: int
131164
:param max_messages: the maximum number of messages to return.
132165
133-
:rtype: list of dict
134-
:returns: sequence of mappings, each containing keys ``ackId`` (the
135-
ID to be used in a subsequent call to :meth:`acknowledge`)
136-
and ``message``.
166+
:rtype: list of (ack_id, message) tuples
167+
:returns: sequence of tuples: ``ack_id`` is the ID to be used in a
168+
subsequent call to :meth:`acknowledge`, and ``message``
169+
is an instance of :class:`gcloud.pubsub.message.Message`.
137170
"""
138171
data = {'returnImmediately': return_immediately,
139172
'maxMessages': max_messages}
140173
conn = self.topic.connection
141174
response = conn.api_request(method='POST',
142175
path='%s:pull' % self.path,
143176
data=data)
144-
return response['receivedMessages']
177+
return [(info['ackId'], Message.from_api_repr(info['message']))
178+
for info in response['receivedMessages']]
145179

146180
def acknowledge(self, ack_ids):
147181
"""API call: acknowledge retrieved messages for the subscription.

‎gcloud/pubsub/test_message.py‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
# Copyright 2015 Google Inc. All rights reserved.
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
import unittest2
16+
17+
18+
class TestMessage(unittest2.TestCase):
19+
20+
def _getTargetClass(self):
21+
from gcloud.pubsub.message import Message
22+
return Message
23+
24+
def _makeOne(self, *args, **kw):
25+
return self._getTargetClass()(*args, **kw)
26+
27+
def test_ctor_no_attrs(self):
28+
DATA = b'DEADBEEF'
29+
MESSAGE_ID = b'12345'
30+
message = self._makeOne(data=DATA, message_id=MESSAGE_ID)
31+
self.assertEqual(message.data, DATA)
32+
self.assertEqual(message.message_id, MESSAGE_ID)
33+
self.assertEqual(message.attrs, {})
34+
35+
def test_ctor_w_attrs(self):
36+
DATA = b'DEADBEEF'
37+
MESSAGE_ID = b'12345'
38+
ATTRS = {'a': 'b'}
39+
message = self._makeOne(data=DATA, message_id=MESSAGE_ID,
40+
attributes=ATTRS)
41+
self.assertEqual(message.data, DATA)
42+
self.assertEqual(message.message_id, MESSAGE_ID)
43+
self.assertEqual(message.attrs, ATTRS)
44+
45+
def test_from_api_repr_no_attrs(self):
46+
from base64 import b64encode as b64
47+
DATA = b'DEADBEEF'
48+
B64_DATA = b64(DATA)
49+
MESSAGE_ID = '12345'
50+
api_repr = {'data': B64_DATA, 'messageId': MESSAGE_ID}
51+
message = self._getTargetClass().from_api_repr(api_repr)
52+
self.assertEqual(message.data, DATA)
53+
self.assertEqual(message.message_id, MESSAGE_ID)
54+
self.assertEqual(message.attrs, {})
55+
56+
def test_from_api_repr_w_attrs(self):
57+
from base64 import b64encode as b64
58+
DATA = b'DEADBEEF'
59+
B64_DATA = b64(DATA)
60+
MESSAGE_ID = '12345'
61+
ATTRS = {'a': 'b'}
62+
api_repr = {'data': B64_DATA,
63+
'messageId': MESSAGE_ID,
64+
'attributes': ATTRS}
65+
message = self._getTargetClass().from_api_repr(api_repr)
66+
self.assertEqual(message.data, DATA)
67+
self.assertEqual(message.message_id, MESSAGE_ID)
68+
self.assertEqual(message.attrs, ATTRS)

0 commit comments

Comments
 (0)