Added sample for synchronous pull

pubsub/cloud-client/subscriber.py

@@ -207,6 +207,31 @@ def callback(message):
     # [END pubsub_subscriber_flow_settings]
 
 
+def receive_messages_synchronously(project, subscription_name):
+    """Pulling messages synchronously."""
+    # [START pubsub_subscriber_sync_pull]
+    subscriber = pubsub_v1.SubscriberClient()
+    subscription_path = subscriber.subscription_path(
+        project, subscription_name)
+
+    # Builds a pull request with a specific number of messages to return.
+    # `return_immediately` is set to False so that the system waits (for a
+    # bounded amount of time) until at lease one message is available.
+    response = subscriber.pull(
+        subscription_path,
+        max_messages=3,
+        return_immediately=False)
+
+    ack_ids = []
+    for received_message in response.received_messages:
+        print("Received: {}".format(received_message.message.data))
+        ack_ids.append(received_message.ack_id)
+
+    # Acknowledges the received messages so they will not be sent again.
+    subscriber.acknowledge(subscription_path, ack_ids)
+    # [END pubsub_subscriber_sync_pull]
+
+
 def listen_for_errors(project, subscription_name):
     """Receives messages and catches errors from a pull subscription."""
     # [START pubsub_subscriber_error_listener]
@@ -281,6 +306,11 @@ def callback(message):
         help=receive_messages_with_flow_control.__doc__)
     receive_with_flow_control_parser.add_argument('subscription_name')
 
+    receive_messages_synchronously_parser = subparsers.add_parser(
+        'receive-synchronously',
+        help=receive_messages_synchronously.__doc__)
+    receive_messages_synchronously_parser.add_argument('subscription_name')
+
     listen_for_errors_parser = subparsers.add_parser(
         'listen_for_errors', help=listen_for_errors.__doc__)
     listen_for_errors_parser.add_argument('subscription_name')
@@ -314,5 +344,8 @@ def callback(message):
     elif args.command == 'receive-flow-control':
         receive_messages_with_flow_control(
             args.project, args.subscription_name)
+    elif args.command == 'receive-synchronously':
+        receive_messages_synchronously(
+            args.project, args.subscription_name)
     elif args.command == 'listen_for_errors':
         listen_for_errors(args.project, args.subscription_name)

pubsub/cloud-client/subscriber_test.py

@@ -188,3 +188,18 @@ def test_receive_with_flow_control(
     assert 'Listening' in out
     assert subscription in out
     assert 'Message 1' in out
+
+
+def test_receive_synchronously(
+        publisher_client, topic, subscription, capsys):
+    _publish_messages(publisher_client, topic)
+
+    with _make_sleep_patch():
+        with pytest.raises(RuntimeError, match='sigil'):
+            subscriber.receive_messages_with_flow_control(
+                PROJECT, SUBSCRIPTION)
+
+    out, _ = capsys.readouterr()
+    assert 'Message 1' in out
+    assert 'Message 2' in out
+    assert 'Message 3' in out