Merge remote-tracking branch 'upstream/master'

storage/cloud-client/README.rst

@@ -225,7 +225,7 @@ To run this sample:
 
     $ python notification_polling.py
 
-    usage: notification_polling.py [-h] [--project PROJECT] subscription
+    usage: notification_polling.py [-h] subscription
 
     This application demonstrates how to poll for GCS notifications from a Cloud
     Pub/Sub subscription, parse the incoming message, and acknowledge the
@@ -245,12 +245,10 @@ To run this sample:
     changes scroll by in the app.
 
     positional arguments:
-      subscription       The ID of the Pub/Sub subscription
+      subscription  The ID of the Pub/Sub subscription
 
     optional arguments:
-      -h, --help         show this help message and exit
-      --project PROJECT  The project of the subscription, if not in your default
-                         project
+      -h, --help    show this help message and exit
 
 
 

storage/cloud-client/notification_polling.py

@@ -46,14 +46,14 @@
 
 import argparse
 import json
-import sys
+import time
 
-from google.cloud import pubsub
+from google.cloud import pubsub_v1
 
 
 def summarize(message):
     # [START parse_message]
-    data = message.data
+    data = message.data.decode('utf-8')
     attributes = message.attributes
 
     event_type = attributes['eventType']
@@ -87,24 +87,24 @@ def summarize(message):
     # [END parse_message]
 
 
-def poll_notifications(subscription_id):
+def poll_notifications(project, subscription_name):
     """Polls a Cloud Pub/Sub subscription for new GCS events for display."""
     # [BEGIN poll_notifications]
-    client = pubsub.Client()
-    subscription = pubsub.subscription.Subscription(
-        subscription_id, client=client)
+    subscriber = pubsub_v1.SubscriberClient()
+    subscription_path = subscriber.subscription_path(
+        project, subscription_name)
 
-    if not subscription.exists():
-        sys.stderr.write('Cannot find subscription {0}\n'.format(sys.argv[1]))
-        return
+    def callback(message):
+        print('Received message:\n{1}'.format(summarize(message)))
+        message.ack()
 
-    print('Polling for messages. Press ctrl+c to exit.')
+    subscriber.subscribe(subscription_path, callback=callback)
+
+    # The subscriber is non-blocking, so we must keep the main thread from
+    # exiting to allow it to process messages in the background.
+    print('Listening for messages on {}'.format(subscription_path))
     while True:
-        pulled = subscription.pull(max_messages=100)
-        for ack_id, message in pulled:
-            print('Received message {0}:\n{1}'.format(
-                message.message_id, summarize(message)))
-            subscription.acknowledge([ack_id])
+        time.sleep(60)
     # [END poll_notifications]
 
 

storage/cloud-client/notification_polling_test.py

@@ -13,7 +13,8 @@
 # limitations under the License.
 
 
-from google.cloud.pubsub.message import Message
+from google.cloud.pubsub_v1.subscriber.message import Message
+import mock
 
 from notification_polling import summarize
 
@@ -31,12 +32,15 @@ def test_parse_json_message():
         'notificationConfig': ('projects/_/buckets/mybucket/'
                                'notificationConfigs/5'),
         'payloadFormat': 'JSON_API_V1'}
-    data = ('{'
-            '  "size": 12345,'
-            '  "contentType": "text/html",'
-            '  "metageneration": 1'
-            '}')
-    message = Message(data, MESSAGE_ID, attributes=attributes)
+    data = (b'{'
+            b'  "size": 12345,'
+            b'  "contentType": "text/html",'
+            b'  "metageneration": 1'
+            b'}')
+    message = Message(
+        mock.Mock(data=data, attributes=attributes),
+        MESSAGE_ID,
+        mock.Mock())
     assert summarize(message) == (
         '\tEvent type: OBJECT_FINALIZE\n'
         '\tBucket ID: mybucket\n'

storage/cloud-client/requirements.txt

@@ -1,2 +1,2 @@
-google-cloud-storage==0.22.0
-google-cloud-pubsub==0.22.0
+google-cloud-storage==1.4.0
+google-cloud-pubsub==0.28.3