Merge pull request #477 from GoogleCloudPlatform/pubsub-samples

Author: dpebot

.coveragerc

@@ -1,19 +1,21 @@
 [run]
 include =
     appengine/*
-    bigtable/*
     bigquery/*
+    bigtable/*
     blog/*
-    logging/*
     compute/*
-    dns/*
     datastore/*
+    dataproc/*
+    dns/*
     error_reporting/*
     language/*
-    managed_vms/*
+    logging/*
     monitoring/*
+    pubsub/*
     speech/*
     storage/*
+    vision/*
 [report]
 exclude_lines =
     pragma: NO COVER

nox.py

@@ -280,7 +280,10 @@ def session_reqcheck(session):
     else:
         command = 'check-requirements'
 
-    for reqfile in list_files('.', 'requirements*.txt'):
+    reqfiles = list(list_files('.', 'requirements*.txt'))
+    reqfiles.append('requirements-dev.in')
+
+    for reqfile in reqfiles:
         session.run('gcprepotools', command, reqfile)
 
 

pubsub/cloud-client/README.md

@@ -0,0 +1,17 @@
+# Google Cloud Pub/Sub Samples
+
+<!-- auto-doc-link -->
+<!-- end-auto-doc-link -->
+
+## Prerequisites
+
+All samples require a [Google Cloud Project](https://console.cloud.google.com).
+
+Use the [Cloud SDK](https://cloud.google.com/sdk) to provide authentication:
+
+    gcloud beta auth application-default login
+
+Run the samples:
+
+    python publisher.py -h
+    python subscriber.py -h

pubsub/cloud-client/publisher.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python
+
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# 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.
+
+"""This application demonstrates how to perform basic operations on topics
+with the Cloud Pub/Sub API.
+
+For more information, see the README.md under /pubsub and the documentation
+at https://cloud.google.com/pubsub/docs.
+"""
+
+import argparse
+
+from gcloud import pubsub
+
+
+def list_topics():
+    """Lists all Pub/Sub topics in the current project."""
+    pubsub_client = pubsub.Client()
+
+    topics = []
+    next_page_token = None
+    while True:
+        page, next_page_token = pubsub_client.list_topics()
+        topics.extend(page)
+        if not next_page_token:
+            break
+
+    for topic in topics:
+        print(topic.name)
+
+
+def create_topic(topic_name):
+    """Create a new Pub/Sub topic."""
+    pubsub_client = pubsub.Client()
+    topic = pubsub_client.topic(topic_name)
+
+    topic.create()
+
+    print('Topic {} created.'.format(topic.name))
+
+
+def delete_topic(topic_name):
+    """Deletes an existing Pub/Sub topic."""
+    pubsub_client = pubsub.Client()
+    topic = pubsub_client.topic(topic_name)
+
+    topic.delete()
+
+    print('Topic {} deleted.'.format(topic.name))
+
+
+def publish_message(topic_name, data):
+    """Publishes a message to a Pub/Sub topic with the given data."""
+    pubsub_client = pubsub.Client()
+    topic = pubsub_client.topic(topic_name)
+
+    # Data must be a bytestring
+    data = data.encode('utf-8')
+
+    message_id = topic.publish(data)
+
+    print('Message {} published.'.format(message_id))
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(
+        description=__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+
+    subparsers = parser.add_subparsers(dest='command')
+    subparsers.add_parser('list', help=list_topics.__doc__)
+
+    create_parser = subparsers.add_parser('create', help=create_topic.__doc__)
+    create_parser.add_argument('topic_name')
+
+    delete_parser = subparsers.add_parser('delete', help=delete_topic.__doc__)
+    delete_parser.add_argument('topic_name')
+
+    publish_parser = subparsers.add_parser(
+        'publish', help=publish_message.__doc__)
+    publish_parser.add_argument('topic_name')
+    publish_parser.add_argument('data')
+
+    args = parser.parse_args()
+
+    if args.command == 'list':
+        list_topics()
+    elif args.command == 'create':
+        create_topic(args.topic_name)
+    elif args.command == 'delete':
+        delete_topic(args.topic_name)
+    elif args.command == 'publish':
+        publish_message(args.topic_name, args.data)

pubsub/cloud-client/publisher_test.py

@@ -0,0 +1,67 @@
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# 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.
+
+from gcloud import pubsub
+from gcp.testing import eventually_consistent
+import pytest
+
+import publisher
+
+TEST_TOPIC = 'publisher-test-topic'
+
+
+@pytest.fixture
+def test_topic():
+    client = pubsub.Client()
+    topic = client.topic(TEST_TOPIC)
+    yield topic
+    if topic.exists():
+        topic.delete()
+
+
+def test_list(test_topic, capsys):
+    test_topic.create()
+
+    @eventually_consistent.call
+    def _():
+        publisher.list_topics()
+        out, _ = capsys.readouterr()
+        assert test_topic.name in out
+
+
+def test_create(test_topic):
+    publisher.create_topic(test_topic.name)
+
+    @eventually_consistent.call
+    def _():
+        assert test_topic.exists()
+
+
+def test_delete(test_topic):
+    test_topic.create()
+
+    publisher.delete_topic(test_topic.name)
+
+    @eventually_consistent.call
+    def _():
+        assert not test_topic.exists()
+
+
+def test_publish(test_topic, capsys):
+    test_topic.create()
+
+    publisher.publish_message(test_topic.name, 'hello')
+
+    out, _ = capsys.readouterr()
+    assert 'published' in out

pubsub/cloud-client/requirements.txt

@@ -0,0 +1 @@
+gcloud==0.18.1

pubsub/cloud-client/subscriber.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python
+
+# Copyright 2016 Google Inc. All Rights Reserved.
+#
+# 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.
+
+"""This application demonstrates how to perform basic operations on
+subscriptions with the Cloud Pub/Sub API.
+
+For more information, see the README.md under /pubsub and the documentation
+at https://cloud.google.com/pubsub/docs.
+"""
+
+import argparse
+
+from gcloud import pubsub
+
+
+def list_subscriptions(topic_name):
+    """Lists all subscriptions for a given topic."""
+    pubsub_client = pubsub.Client()
+    topic = pubsub_client.topic(topic_name)
+
+    subscriptions = []
+    next_page_token = None
+    while True:
+        page, next_page_token = topic.list_subscriptions()
+        subscriptions.extend(page)
+        if not next_page_token:
+            break
+
+    for subscription in subscriptions:
+        print(subscription.name)
+
+
+def create_subscription(topic_name, subscription_name):
+    """Create a new pull subscription on the given topic."""
+    pubsub_client = pubsub.Client()
+    topic = pubsub_client.topic(topic_name)
+
+    subscription = topic.subscription(subscription_name)
+    subscription.create()
+
+    print('Subscription {} created on topic {}.'.format(
+        subscription.name, topic.name))
+
+
+def delete_subscription(topic_name, subscription_name):
+    """Deletes an existing Pub/Sub topic."""
+    pubsub_client = pubsub.Client()
+    topic = pubsub_client.topic(topic_name)
+    subscription = topic.subscription(subscription_name)
+
+    subscription.delete()
+
+    print('Subscription {} deleted on topic {}.'.format(
+        subscription.name, topic.name))
+
+
+def receive_message(topic_name, subscription_name):
+    """Receives a message from a pull subscription."""
+    pubsub_client = pubsub.Client()
+    topic = pubsub_client.topic(topic_name)
+    subscription = topic.subscription(subscription_name)
+
+    # Change return_immediately=False to block until messages are
+    # received.
+    results = subscription.pull(return_immediately=True)
+
+    print('Received {} messages.'.format(len(results)))
+
+    for ack_id, message in results:
+        print('* {}: {}, {}'.format(
+            message.message_id, message.data, message.attributes))
+
+    # Acknowledge received messages. If you do not acknowledge, Pub/Sub will
+    # redeliver the message.
+    if results:
+        subscription.acknowledge([ack_id for ack_id, message in results])
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(
+        description=__doc__,
+        formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+
+    subparsers = parser.add_subparsers(dest='command')
+    list_parser = subparsers.add_parser(
+        'list', help=list_subscriptions.__doc__)
+    list_parser.add_argument('topic_name')
+
+    create_parser = subparsers.add_parser(
+        'create', help=create_subscription.__doc__)
+    create_parser.add_argument('topic_name')
+    create_parser.add_argument('subscription_name')
+
+    delete_parser = subparsers.add_parser(
+        'delete', help=delete_subscription.__doc__)
+    delete_parser.add_argument('topic_name')
+    delete_parser.add_argument('subscription_name')
+
+    receive_parser = subparsers.add_parser(
+        'receive', help=receive_message.__doc__)
+    receive_parser.add_argument('topic_name')
+    receive_parser.add_argument('subscription_name')
+
+    args = parser.parse_args()
+
+    if args.command == 'list':
+        list_subscriptions(args.topic_name)
+    elif args.command == 'create':
+        create_subscription(args.topic_name, args.subscription_name)
+    elif args.command == 'delete':
+        delete_subscription(args.topic_name, args.subscription_name)
+    elif args.command == 'receive':
+        receive_message(args.topic_name, args.subscription_name)