sethmaxwl
diff --git a/Collapse file
‎google/cloud/pubsub_v1/gapic/subscriber_client.py‎
Copy file name to clipboardExpand all lines: google/cloud/pubsub_v1/gapic/subscriber_client.py
+5Lines changed: 5 additions & 0 deletions b/Collapse file
‎google/cloud/pubsub_v1/gapic/subscriber_client.py‎
Copy file name to clipboardExpand all lines: google/cloud/pubsub_v1/gapic/subscriber_client.py
+5Lines changed: 5 additions & 0 deletions
diff --git a/Collapse file
‎setup.py‎
Copy file name to clipboardExpand all lines: setup.py
+3-3Lines changed: 3 additions & 3 deletions b/Collapse file
‎setup.py‎
Copy file name to clipboardExpand all lines: setup.py
+3-3Lines changed: 3 additions & 3 deletions
diff --git a/Collapse file
‎synth.py‎
Copy file name to clipboardExpand all lines: synth.py
+14Lines changed: 14 additions & 0 deletions b/Collapse file
‎synth.py‎
Copy file name to clipboardExpand all lines: synth.py
+14Lines changed: 14 additions & 0 deletions
diff --git a/Collapse file
‎tests/unit/pubsub_v1/subscriber/test_subscriber_client.py‎
Copy file name to clipboardExpand all lines: tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
+12Lines changed: 12 additions & 0 deletions b/Collapse file
‎tests/unit/pubsub_v1/subscriber/test_subscriber_client.py‎
Copy file name to clipboardExpand all lines: tests/unit/pubsub_v1/subscriber/test_subscriber_client.py
+12Lines changed: 12 additions & 0 deletions
@@ -1175,6 +1175,11 @@ def streaming_pull(
                 client_info=self._client_info,
             )
 
+            # Wrappers in api-core should not automatically pre-fetch the first
+            # stream result, as this breaks the stream when re-opening it.
+            # https://github.com/googleapis/python-pubsub/issues/93#issuecomment-630762257
+            self.transport.streaming_pull._prefetch_first_result_ = False
+
         return self._inner_api_calls["streaming_pull"](
             requests, retry=retry, timeout=timeout, metadata=metadata
         )
 
@@ -29,10 +29,10 @@
 # 'Development Status :: 5 - Production/Stable'
 release_status = "Development Status :: 5 - Production/Stable"
 dependencies = [
-    # google-api-core[grpc] 1.17.0 causes problems, thus restricting its
-    # version until the issue gets fixed.
+    # google-api-core[grpc] 1.17.0 up to 1.19.1 causes problems with stream
+    # recovery, thus those versions should not be used.
     # https://github.com/googleapis/python-pubsub/issues/74
-    "google-api-core[grpc] >= 1.14.0, < 1.17.0",
+    "google-api-core[grpc] >= 1.14.0, != 1.17.*, != 1.18.*, != 1.19.*",
     "grpc-google-iam-v1 >= 0.12.3, < 0.13dev",
     'enum34; python_version < "3.4"',
 ]
 
@@ -185,6 +185,20 @@ def _merge_dict(d1, d2):
     "from google.iam.v1 import iam_policy_pb2_grpc as iam_policy_pb2",
 )
 
+# Monkey patch the streaming_pull() GAPIC method to disable pre-fetching stream
+# results.
+s.replace(
+    "google/cloud/pubsub_v1/gapic/subscriber_client.py",
+    r"return self\._inner_api_calls\['streaming_pull'\]\(.*",
+    """
+            # Wrappers in api-core should not automatically pre-fetch the first
+            # stream result, as this breaks the stream when re-opening it.
+            # https://github.com/googleapis/python-pubsub/issues/93#issuecomment-630762257
+            self.transport.streaming_pull._prefetch_first_result_ = False
+
+        \g<0>"""
+)
+
 # Add missing blank line before Attributes: in generated docstrings
 # https://github.com/googleapis/protoc-docs-plugin/pull/31
 s.replace(
 
@@ -154,3 +154,15 @@ def test_closes_channel_as_context_manager():
         pass
 
     mock_transport.channel.close.assert_called()
+
+
+def test_streaming_pull_gapic_monkeypatch():
+    transport = mock.NonCallableMock(spec=["streaming_pull"])
+    transport.streaming_pull = mock.Mock(spec=[])
+    client = subscriber.Client(transport=transport)
+
+    client.streaming_pull(requests=iter([]))
+
+    assert client.api.transport is transport
+    assert hasattr(transport.streaming_pull, "_prefetch_first_result_")
+    assert not transport.streaming_pull._prefetch_first_result_
Original file line number	Diff line number	Diff line change
`@@ -1175,6 +1175,11 @@ def streaming_pull(`
`1175`	`1175`	`client_info=self._client_info,`
`1176`	`1176`	`)`
`1177`	`1177`
	`1178`	`+ # Wrappers in api-core should not automatically pre-fetch the first`
	`1179`	`+ # stream result, as this breaks the stream when re-opening it.`
	`1180`	`+ # https://github.com/googleapis/python-pubsub/issues/93#issuecomment-630762257`
	`1181`	`+ self.transport.streaming_pull._prefetch_first_result_ = False`
	`1182`	`+`
`1178`	`1183`	`return self._inner_api_calls["streaming_pull"](`
`1179`	`1184`	`requests, retry=retry, timeout=timeout, metadata=metadata`
`1180`	`1185`	`)`