googleapis / googleapis/google-cloud-python

Async Streaming Pull Usage

Open
#15,644 2 comments 1 reaction 1 assignee Claimed by @abbrowne126 View on GitHub
api: pubsub
Dominant language
Python
Stars
5.4k
Forks
1.8k
Avg merge
3d 4h
Merged PRs (30d)
122

Description

Has someone used this in peoduction scenarios? I am stuck and unable to pull messages using the async streaming client and any help would be benefecial.

Is wrapping the standard streaming pull in an asyncio executor gonna give me the same behaviour as below client?

https://github.com/googleapis/python-pubsub/blob/ff229a5fdd4deaff0ac97c74f313d04b62720ff7/google/pubsub_v1/services/subscriber/async_client.py#L1368

This is my current usage:
```python
from google import pubsub_v1
subscriber = pubsub_v1.SubscriberAsyncClient() #skipping the class code but this is the Subscriber client
async def fetch_and_decode_msgs(self):
try:
print(f"Listening for messages on {self.subscription_path}...")
request = pubsub_v1.StreamingPullRequest(
subscription=self.subscription_path,
)

# Not sure if there is any other way to do this
async def request_generator():
yield request
print(f"stream: {request}"). # Code gets blocked here

stream = await self.subscriber.streaming_pull(requests=request_generator())
print(f"stream: {stream}")
# Handle the response
async for response in stream:
for received_message in response.received_messages:
print("Received message: ", received_message.message.data.decode('utf-8'))
except Exception as e:
raise e
```

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.