-
Notifications
You must be signed in to change notification settings - Fork 12
/
async_publisher_impl.py
33 lines (25 loc) · 1.06 KB
/
async_publisher_impl.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from typing import Mapping
from google.pubsub_v1 import PubsubMessage
from google.cloud.pubsublite.cloudpubsub.message_transforms import (
from_cps_publish_message,
)
from google.cloud.pubsublite.cloudpubsub.publisher import AsyncPublisher
from google.cloud.pubsublite.internal.wire.publisher import Publisher
class AsyncPublisherImpl(AsyncPublisher):
_publisher: Publisher
def __init__(self, publisher: Publisher):
super().__init__()
self._publisher = publisher
async def publish(
self, data: bytes, ordering_key: str = "", **attrs: Mapping[str, str]
) -> str:
cps_message = PubsubMessage(
data=data, ordering_key=ordering_key, attributes=attrs
)
psl_message = from_cps_publish_message(cps_message)
return (await self._publisher.publish(psl_message)).encode()
async def __aenter__(self):
await self._publisher.__aenter__()
return self
async def __aexit__(self, exc_type, exc_value, traceback):
await self._publisher.__aexit__(exc_type, exc_value, traceback)