Skip to content

Commit 7475da9

Browse files
committed
Adding subscription factory on Pub / Sub client.
Fixes #3369.
1 parent 520637c commit 7475da9

File tree

3 files changed

+108
-11
lines changed

3 files changed

+108
-11
lines changed

docs/pubsub_snippets.py

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,28 @@ def do_something_with(sub): # pylint: disable=unused-argument
6565
# [END client_list_subscriptions]
6666

6767

68+
@snippet
69+
def client_topic(client, to_delete): # pylint: disable=unused-argument
70+
"""Topic factory."""
71+
TOPIC_NAME = 'topic_factory-%d' % (_millis(),)
72+
73+
# [START client_topic]
74+
topic = client.topic(TOPIC_NAME)
75+
# [END client_topic]
76+
77+
78+
@snippet
79+
def client_subscription(client, to_delete): # pylint: disable=unused-argument
80+
"""Subscription factory."""
81+
SUBSCRIPTION_NAME = 'subscription_factory-%d' % (_millis(),)
82+
83+
# [START client_subscription]
84+
subscription = client.topic(
85+
SUBSCRIPTION_NAME, ack_deadline=60,
86+
retain_acked_messages=True)
87+
# [END client_subscription]
88+
89+
6890
@snippet
6991
def topic_create(client, to_delete):
7092
"""Create a topic."""

pubsub/google/cloud/pubsub/client.py

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
from google.cloud.pubsub._http import _PublisherAPI as JSONPublisherAPI
2323
from google.cloud.pubsub._http import _SubscriberAPI as JSONSubscriberAPI
2424
from google.cloud.pubsub._http import _IAMPolicyAPI
25+
from google.cloud.pubsub.subscription import Subscription
2526
from google.cloud.pubsub.topic import Topic
2627

2728
try:
@@ -225,6 +226,7 @@ def topic(self, name, timestamp_messages=False):
225226
.. literalinclude:: pubsub_snippets.py
226227
:start-after: [START client_topic]
227228
:end-before: [END client_topic]
229+
:dedent: 4
228230
229231
:type name: str
230232
:param name: the name of the topic to be constructed.
@@ -236,3 +238,48 @@ def topic(self, name, timestamp_messages=False):
236238
:returns: Topic created with the current client.
237239
"""
238240
return Topic(name, client=self, timestamp_messages=timestamp_messages)
241+
242+
def subscription(self, name, ack_deadline=None, push_endpoint=None,
243+
retain_acked_messages=None,
244+
message_retention_duration=None):
245+
"""Creates a subscription bound to the current client.
246+
247+
Example:
248+
249+
.. literalinclude:: pubsub_snippets.py
250+
:start-after: [START client_subscription]
251+
:end-before: [END client_subscription]
252+
:dedent: 4
253+
254+
:type name: str
255+
:param name: the name of the subscription to be constructed.
256+
257+
:type ack_deadline: int
258+
:param ack_deadline: (Optional) The deadline (in seconds) by which
259+
messages pulledfrom the back-end must be
260+
acknowledged.
261+
262+
:type push_endpoint: str
263+
:param push_endpoint:
264+
(Optional) URL to which messages will be pushed by the back-end.
265+
If not set, the application must pull messages.
266+
267+
:type retain_acked_messages: bool
268+
:param retain_acked_messages:
269+
(Optional) Whether to retain acked messages. If set, acked messages
270+
are retained in the subscription's backlog for a duration indicated
271+
by ``message_retention_duration``.
272+
273+
:type message_retention_duration: :class:`datetime.timedelta`
274+
:param message_retention_duration:
275+
(Optional) Whether to retain acked messages. If set, acked messages
276+
are retained in the subscription's backlog for a duration indicated
277+
by ``message_retention_duration``. If unset, defaults to 7 days.
278+
279+
:rtype: :class:`~google.cloud.pubsub.subscription.Subscription`
280+
:returns: Subscription created with the current client.
281+
"""
282+
return Subscription(
283+
name, ack_deadline=ack_deadline, push_endpoint=push_endpoint,
284+
retain_acked_messages=retain_acked_messages,
285+
message_retention_duration=message_retention_duration, client=self)

pubsub/tests/unit/test_client.py

Lines changed: 39 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
# See the License for the specific language governing permissions and
1313
# limitations under the License.
1414

15+
import datetime
1516
import unittest
1617

1718
import mock
@@ -365,7 +366,17 @@ def test_list_subscriptions_w_missing_key(self):
365366
self.assertEqual(api._listed_subscriptions,
366367
(self.PROJECT, None, None))
367368

368-
def test_topic(self):
369+
def test_list_snapshots(self):
370+
creds = _make_credentials()
371+
client = self._make_one(project=self.PROJECT, credentials=creds)
372+
client._connection = object()
373+
api = _FauxSubscriberAPI()
374+
response = api._list_snapshots_response = object()
375+
client._subscriber_api = api
376+
self.assertEqual(client.list_snapshots(), response)
377+
self.assertEqual(api._listed_snapshots, (self.PROJECT, None, None))
378+
379+
def test_topic_factory(self):
369380
PROJECT = 'PROJECT'
370381
TOPIC_NAME = 'TOPIC_NAME'
371382
creds = _make_credentials()
@@ -379,17 +390,33 @@ def test_topic(self):
379390
'projects/%s/topics/%s' % (PROJECT, TOPIC_NAME))
380391
self.assertFalse(new_topic.timestamp_messages)
381392

382-
def test_list_snapshots(self):
393+
def test_subscription_factory(self):
394+
project = 'PROJECT'
383395
creds = _make_credentials()
384-
client = self._make_one(project=self.PROJECT, credentials=creds)
385-
client._connection = object()
386-
api = _FauxSubscriberAPI()
387-
response = api._list_snapshots_response = object()
388-
client._subscriber_api = api
389-
self.assertEqual(client.list_snapshots(), response)
390-
self.assertEqual(api._listed_snapshots, (self.PROJECT, None, None))
396+
client_obj = self._make_one(project=project, credentials=creds)
397+
398+
sub_name = 'hoot-n-holler'
399+
ack_deadline = 60,
400+
push_endpoint = 'https://api.example.com/push'
401+
message_retention_duration = datetime.timedelta(3600)
402+
new_subscription = client_obj.subscription(
403+
sub_name, ack_deadline=ack_deadline,
404+
push_endpoint=push_endpoint,
405+
retain_acked_messages=True,
406+
message_retention_duration=message_retention_duration)
407+
408+
self.assertEqual(new_subscription.name, sub_name)
409+
self.assertIsNone(new_subscription.topic)
410+
self.assertIs(new_subscription._client, client_obj)
411+
self.assertEqual(new_subscription._project, project)
412+
self.assertEqual(new_subscription.ack_deadline, ack_deadline)
413+
self.assertEqual(new_subscription.push_endpoint, push_endpoint)
414+
self.assertTrue(new_subscription.retain_acked_messages)
415+
self.assertEqual(
416+
new_subscription.message_retention_duration,
417+
message_retention_duration)
418+
391419

392-
393420
class _Iterator(object):
394421

395422
def __init__(self, items, token):
@@ -419,7 +446,8 @@ def list_subscriptions(self, project, page_size, page_token):
419446

420447
def list_snapshots(self, project, page_size, page_token):
421448
self._listed_snapshots = (project, page_size, page_token)
422-
return self._list_snapshots_response
449+
return self._list_snapshots_response
450+
423451

424452
class _Connection(object):
425453

0 commit comments

Comments
 (0)