fix function signature for streaming rpc (#279)

* fix function signature for streaming rpc

generated code should take request iterator for client streaming, and
return response iterator for server streaming

Author: arithmetic1728

packages/gapic-generator/.gitignore

@@ -59,3 +59,6 @@ pylintrc.test
 
 # Mypy
 .mypy_cache
+
+# pyenv
+.python-version

packages/gapic-generator/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2

@@ -2,7 +2,7 @@
 
 {% block content %}
 from collections import OrderedDict
-from typing import Dict, Sequence, Tuple, Type, Union
+from typing import Dict, Iterable, Iterator, Sequence, Tuple, Type, Union
 import pkg_resources
 
 import google.api_core.client_options as ClientOptions # type: ignore
@@ -11,7 +11,7 @@ from google.api_core import gapic_v1                   # type: ignore
 from google.api_core import retry as retries           # type: ignore
 from google.auth import credentials                    # type: ignore
 from google.oauth2 import service_account              # type: ignore
-  
+
 {% filter sort_lines -%}
 {% for method in service.methods.values() -%}
 {% for ref_type in method.ref_types_legacy -%}
@@ -126,18 +126,28 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
     {% for method in service.methods.values() -%}
     def {{ method.name|snake_case }}(self,
+            {%- if not method.client_streaming %}
             request: {{ method.input.ident }} = None,
             *,
             {% for field in method.flattened_fields.values() -%}
             {{ field.name }}: {{ field.ident }} = None,
             {% endfor -%}
+            {%- else %}
+            requests: Iterator[{{ method.input.ident }}] = None,
+            *,
+            {% endif -%}
             retry: retries.Retry = gapic_v1.method.DEFAULT,
             timeout: float = None,
             metadata: Sequence[Tuple[str, str]] = (),
+            {%- if not method.server_streaming %}
             ) -> {{ method.client_output.ident }}:
+            {%- else %}
+            ) -> Iterable[{{ method.client_output.ident }}]:
+            {%- endif %}
         r"""{{ method.meta.doc|rst(width=72, indent=8) }}
 
         Args:
+            {%- if not method.client_streaming %}
             request (:class:`{{ method.input.ident.sphinx }}`):
                 The request object.{{ ' ' -}}
                 {{ method.input.meta.doc|wrap(width=72, offset=36, indent=16) }}
@@ -148,6 +158,11 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
                 on the ``request`` instance; if ``request`` is provided, this
                 should not be set.
             {% endfor -%}
+            {%- else %}
+            requests (Iterator[`{{ method.input.ident.sphinx }}`]):
+                The request object iterator.{{ ' ' -}}
+                {{ method.input.meta.doc|wrap(width=72, offset=36, indent=16) }}
+            {%- endif %}
             retry (google.api_core.retry.Retry): Designation of what errors, if any,
                 should be retried.
             timeout (float): The timeout for this request.
@@ -156,10 +171,15 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         {%- if not method.void %}
 
         Returns:
+            {%- if not method.server_streaming %}
             {{ method.client_output.ident.sphinx }}:
+            {%- else %}
+            Iterable[{{ method.client_output.ident.sphinx }}]:
+            {%- endif %}
                 {{ method.client_output.meta.doc|rst(width=72, indent=16) }}
         {%- endif %}
         """
+        {%- if not method.client_streaming %}
         # Create or coerce a protobuf request object.
         {% if method.flattened_fields -%}
         # Sanity check: If we got a request object, we should *not* have
@@ -176,6 +196,7 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
         if {{ field.name }} is not None:
             request.{{ key }} = {{ field.name }}
         {%- endfor %}
+        {%- endif %}
 
         # Wrap the RPC method; this adds retry and timeout information,
         # and friendly error handling.
@@ -213,7 +234,11 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
         # Send the request.
         {% if not method.void %}response = {% endif %}rpc(
+            {%- if not method.client_streaming %}
             request,
+            {%- else %}
+            requests,
+            {%- endif %}
             retry=retry,
             timeout=timeout,
             metadata=metadata,

packages/gapic-generator/gapic/templates/tests/unit/%name_%version/%sub/test_%service.py.j2

@@ -79,6 +79,9 @@ def test_{{ method.name|snake_case }}(transport: str = 'grpc'):
     # Everything is optional in proto3 as far as the runtime is concerned,
     # and we are mocking out the actual API, so just send an empty request.
     request = {{ method.input.ident }}()
+    {% if method.client_streaming %}
+    requests = [request]
+    {% endif %}
 
     # Mock the actual call within the gRPC stub, and fake the request.
     with mock.patch.object(
@@ -98,12 +101,20 @@ def test_{{ method.name|snake_case }}(transport: str = 'grpc'):
             {%- endfor %}
         )
         {% endif -%}
+        {% if method.client_streaming %}
+        response = client.{{ method.name|snake_case }}(iter(requests))
+        {% else %}
         response = client.{{ method.name|snake_case }}(request)
-        
+        {% endif %}
+
         # Establish that the underlying gRPC stub method was called.
         assert len(call.mock_calls) == 1
         _, args, _ = call.mock_calls[0]
+        {% if method.client_streaming %}
+        assert next(args[0]) == request
+        {% else %}
         assert args[0] == request
+        {% endif %}
 
     # Establish that the response is the type that we expect.
     {% if method.void -%}

packages/gapic-generator/tests/system/test_grpc_streams.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from google import showcase
+
 
 def test_unary_stream(echo):
     content = 'The hail in Wales falls mainly on the snails.'
@@ -24,3 +26,51 @@ def test_unary_stream(echo):
     for ground_truth, response in zip(content.split(' '), responses):
         assert response.content == ground_truth
     assert ground_truth == 'snails.'
+
+    # TODO. Check responses.trailing_metadata() content once gapic-showcase
+    # server returns non-empty trailing metadata.
+    assert len(responses.trailing_metadata()) == 0
+
+
+def test_stream_unary(echo):
+    requests = []
+    requests.append(showcase.EchoRequest(content="hello"))
+    requests.append(showcase.EchoRequest(content="world!"))
+    response = echo.collect(iter(requests))
+    assert response.content == 'hello world!'
+
+
+def test_stream_unary_passing_dict(echo):
+    requests = [{'content': 'hello'}, {'content': 'world!'}]
+    response = echo.collect(iter(requests))
+    assert response.content == 'hello world!'
+
+
+def test_stream_stream(echo):
+    requests = []
+    requests.append(showcase.EchoRequest(content="hello"))
+    requests.append(showcase.EchoRequest(content="world!"))
+    responses = echo.chat(iter(requests))
+
+    contents = []
+    for response in responses:
+        contents.append(response.content)
+    assert contents == ['hello', 'world!']
+
+    # TODO. Check responses.trailing_metadata() content once gapic-showcase
+    # server returns non-empty trailing metadata.
+    assert len(responses.trailing_metadata()) == 0
+
+
+def test_stream_stream_passing_dict(echo):
+    requests = [{'content': 'hello'}, {'content': 'world!'}]
+    responses = echo.chat(iter(requests))
+
+    contents = []
+    for response in responses:
+        contents.append(response.content)
+    assert contents == ['hello', 'world!']
+
+    # TODO. Check responses.trailing_metadata() content once gapic-showcase
+    # server returns non-empty trailing metadata.
+    assert len(responses.trailing_metadata()) == 0