Pub/Sub: add publish retry sample (GoogleCloudPlatform#2273)

* Publish retry sample

* double to single quotes

* double to single quotes

* license year

Author: anguillanneuf

pubsub/cloud-client/iam.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Copyright 2016 Google Inc. All Rights Reserved.
+# Copyright 2019 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.

pubsub/cloud-client/iam_test.py

@@ -1,4 +1,4 @@
-# Copyright 2016 Google Inc. All Rights Reserved.
+# Copyright 2019 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.

pubsub/cloud-client/publisher.py

@@ -120,7 +120,8 @@ def publish_messages_with_custom_attributes(project_id, topic_name):
         data = data.encode('utf-8')
         # Add two attributes, origin and username, to the message
         future = publisher.publish(
-            topic_path, data, origin='python-sample', username='gcp')
+            topic_path, data, origin='python-sample', username='gcp'
+        )
         print(future.result())
 
     print('Published messages with custom attributes.')
@@ -147,7 +148,7 @@ def publish_messages_with_futures(project_id, topic_name):
         future = publisher.publish(topic_path, data=data)
         print(future.result())
 
-    print("Published messages with futures.")
+    print('Published messages with futures.')
     # [END pubsub_publisher_concurrency_control]
 
 
@@ -171,17 +172,17 @@ def callback(f):
             try:
                 print(f.result())
                 futures.pop(data)
-            except: # noqa
-                print("Please handle {} for {}.".format(f.exception(), data))
+            except:  # noqa
+                print('Please handle {} for {}.'.format(f.exception(), data))
+
         return callback
 
     for i in range(10):
         data = str(i)
         futures.update({data: None})
         # When you publish a message, the client returns a future.
         future = publisher.publish(
-            topic_path,
-            data=data.encode("utf-8"),  # data must be a bytestring.
+            topic_path, data=data.encode('utf-8')  # data must be a bytestring.
         )
         futures[data] = future
         # Publish failures shall be handled in the callback function.
@@ -191,7 +192,7 @@ def callback(f):
     while futures:
         time.sleep(5)
 
-    print("Published message with error handler.")
+    print('Published message with error handler.')
     # [END pubsub_publish_messages_error_handler]
 
 
@@ -207,7 +208,7 @@ def publish_messages_with_batch_settings(project_id, topic_name):
     # of data or one second has passed.
     batch_settings = pubsub_v1.types.BatchSettings(
         max_bytes=1024,  # One kilobyte
-        max_latency=1,  # One second
+        max_latency=1,   # One second
     )
     publisher = pubsub_v1.PublisherClient(batch_settings)
     topic_path = publisher.topic_path(project_id, topic_name)
@@ -223,7 +224,65 @@ def publish_messages_with_batch_settings(project_id, topic_name):
     # [END pubsub_publisher_batch_settings]
 
 
-if __name__ == '__main__':
+def publish_messages_with_retry_settings(project_id, topic_name):
+    """Publishes messages with custom retry settings."""
+    # [START pubsub_publisher_retry_settings]
+    from google.cloud import pubsub_v1
+
+    # TODO project_id = "Your Google Cloud Project ID"
+    # TODO topic_name = "Your Pub/Sub topic name"
+
+    # Configure the retry settings. Defaults will be overwritten.
+    retry_settings = {
+        'interfaces': {
+            'google.pubsub.v1.Publisher': {
+                'retry_codes': {
+                    'publish': [
+                        'ABORTED',
+                        'CANCELLED',
+                        'DEADLINE_EXCEEDED',
+                        'INTERNAL',
+                        'RESOURCE_EXHAUSTED',
+                        'UNAVAILABLE',
+                        'UNKNOWN',
+                    ]
+                },
+                'retry_params': {
+                    'messaging': {
+                        'initial_retry_delay_millis': 150,  # default: 100
+                        'retry_delay_multiplier': 1.5,  # default: 1.3
+                        'max_retry_delay_millis': 65000,  # default: 60000
+                        'initial_rpc_timeout_millis': 25000,  # default: 25000
+                        'rpc_timeout_multiplier': 1.0,  # default: 1.0
+                        'max_rpc_timeout_millis': 35000,  # default: 30000
+                        'total_timeout_millis': 650000,  # default: 600000
+                    }
+                },
+                'methods': {
+                    'Publish': {
+                        'retry_codes_name': 'publish',
+                        'retry_params_name': 'messaging',
+                    }
+                },
+            }
+        }
+    }
+
+    publisher = pubsub_v1.PublisherClient(client_config=retry_settings)
+    topic_path = publisher.topic_path(project_id, topic_name)
+
+    for n in range(1, 10):
+        data = u'Message number {}'.format(n)
+        # Data must be a bytestring
+        data = data.encode('utf-8')
+        future = publisher.publish(topic_path, data=data)
+        print(future.result())
+
+    print('Published messages with retry settings.')
+    # [END pubsub_publisher_retry_settings]
+
+
+if __name__ == "__main__":
     parser = argparse.ArgumentParser(
         description=__doc__,
         formatter_class=argparse.RawDescriptionHelpFormatter
@@ -233,36 +292,47 @@ def publish_messages_with_batch_settings(project_id, topic_name):
     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 = 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 = subparsers.add_parser('delete',
+                                          help=delete_topic.__doc__)
     delete_parser.add_argument('topic_name')
 
-    publish_parser = subparsers.add_parser(
-        'publish', help=publish_messages.__doc__)
+    publish_parser = subparsers.add_parser('publish',
+                                           help=publish_messages.__doc__)
     publish_parser.add_argument('topic_name')
 
     publish_with_custom_attributes_parser = subparsers.add_parser(
         'publish-with-custom-attributes',
-        help=publish_messages_with_custom_attributes.__doc__)
+        help=publish_messages_with_custom_attributes.__doc__,
+    )
     publish_with_custom_attributes_parser.add_argument('topic_name')
 
     publish_with_futures_parser = subparsers.add_parser(
-        'publish-with-futures',
-        help=publish_messages_with_futures.__doc__)
+        'publish-with-futures', help=publish_messages_with_futures.__doc__
+    )
     publish_with_futures_parser.add_argument('topic_name')
 
     publish_with_error_handler_parser = subparsers.add_parser(
         'publish-with-error-handler',
-        help=publish_messages_with_error_handler.__doc__)
+        help=publish_messages_with_error_handler.__doc__
+    )
     publish_with_error_handler_parser.add_argument('topic_name')
 
     publish_with_batch_settings_parser = subparsers.add_parser(
         'publish-with-batch-settings',
-        help=publish_messages_with_batch_settings.__doc__)
+        help=publish_messages_with_batch_settings.__doc__
+    )
     publish_with_batch_settings_parser.add_argument('topic_name')
 
+    publish_with_retry_settings_parser = subparsers.add_parser(
+        'publish-with-retry-settings',
+        help=publish_messages_with_retry_settings.__doc__
+    )
+    publish_with_retry_settings_parser.add_argument('topic_name')
+
     args = parser.parse_args()
 
     if args.command == 'list':
@@ -274,11 +344,13 @@ def publish_messages_with_batch_settings(project_id, topic_name):
     elif args.command == 'publish':
         publish_messages(args.project_id, args.topic_name)
     elif args.command == 'publish-with-custom-attributes':
-        publish_messages_with_custom_attributes(
-            args.project_id, args.topic_name)
+        publish_messages_with_custom_attributes(args.project_id,
+                                                args.topic_name)
     elif args.command == 'publish-with-futures':
         publish_messages_with_futures(args.project_id, args.topic_name)
     elif args.command == 'publish-with-error-handler':
         publish_messages_with_error_handler(args.project_id, args.topic_name)
     elif args.command == 'publish-with-batch-settings':
         publish_messages_with_batch_settings(args.project_id, args.topic_name)
+    elif args.command == 'publish-with-retry-settings':
+        publish_messages_with_retry_settings(args.project_id, args.topic_name)

pubsub/cloud-client/publisher_test.py

@@ -1,4 +1,4 @@
-# Copyright 2016 Google Inc. All Rights Reserved.
+# Copyright 2019 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.
@@ -110,6 +110,13 @@ def test_publish_with_batch_settings(topic, capsys):
     assert 'Published' in out
 
 
+def test_publish_with_retry_settings(topic, capsys):
+    publisher.publish_messages_with_retry_settings(PROJECT, TOPIC)
+
+    out, _ = capsys.readouterr()
+    assert 'Published' in out
+
+
 def test_publish_with_error_handler(topic, capsys):
     publisher.publish_messages_with_error_handler(PROJECT, TOPIC)
 

pubsub/cloud-client/quickstart.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Copyright 2016 Google Inc. All Rights Reserved.
+# Copyright 2019 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.

pubsub/cloud-client/quickstart_test.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Copyright 2018 Google Inc. All Rights Reserved.
+# Copyright 2019 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.

pubsub/cloud-client/subscriber.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python
 
-# Copyright 2016 Google Inc. All Rights Reserved.
+# Copyright 2019 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.

pubsub/cloud-client/subscriber_test.py

@@ -1,4 +1,4 @@
-# Copyright 2016 Google Inc. All Rights Reserved.
+# Copyright 2019 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.