create topic reader from topic client

Author: rekby

tests/topics/test_topic_reader.py

@@ -7,9 +7,6 @@
 @pytest.mark.asyncio
 class TestTopicWriterAsyncIO:
     async def test_read_message(self, driver, topic_path, topic_with_messages, topic_consumer):
-        reader = PublicAsyncIOReader(driver, TopicReaderSettings(
-            consumer=topic_consumer,
-            topic=topic_path,
-        ))
+        reader = driver.topic_client.topic_reader(topic_consumer, topic_path)
 
         assert await reader.receive_batch() is not None

ydb/_topic_reader/topic_reader.py

@@ -32,104 +32,6 @@ def __init__(self, path, *, partitions: Union[None, int, List[int]] = None):
         self.partitions = partitions
 
 
-class ReaderAsyncIO(object):
-    async def __aenter__(self):
-        raise NotImplementedError()
-
-    async def __aexit__(self, exc_type, exc_val, exc_tb):
-        raise NotImplementedError()
-
-    async def sessions_stat(self) -> List["SessionStat"]:
-        """
-        Receive stat from the server
-
-        use asyncio.wait_for for wait with timeout.
-        """
-        raise NotImplementedError()
-
-    def messages(
-        self, *, timeout: Union[float, None] = None
-    ) -> AsyncIterable["PublicMessage"]:
-        """
-        Block until receive new message
-
-        if no new messages in timeout seconds: stop iteration by raise StopAsyncIteration
-        """
-        raise NotImplementedError()
-
-    async def receive_message(self) -> Union["PublicMessage", None]:
-        """
-        Block until receive new message
-
-        use asyncio.wait_for for wait with timeout.
-        """
-        raise NotImplementedError()
-
-    def batches(
-        self,
-        *,
-        max_messages: Union[int, None] = None,
-        max_bytes: Union[int, None] = None,
-        timeout: Union[float, None] = None,
-    ) -> AsyncIterable["PublicBatch"]:
-        """
-        Block until receive new batch.
-        All messages in a batch from same partition.
-
-        if no new message in timeout seconds (default - infinite): stop iterations by raise StopIteration
-        """
-        raise NotImplementedError()
-
-    async def receive_batch(
-        self, *, max_messages: Union[int, None] = None, max_bytes: Union[int, None]
-    ) -> Union["PublicBatch", None]:
-        """
-        Get one messages batch from reader.
-        All messages in a batch from same partition.
-
-        use asyncio.wait_for for wait with timeout.
-        """
-        raise NotImplementedError()
-
-    async def commit_on_exit(self, mess: "ICommittable") -> AsyncContextManager:
-        """
-        commit the mess match/message if exit from context manager without exceptions
-
-        reader will close if exit from context manager with exception
-        """
-        raise NotImplementedError()
-
-    def commit(self, mess: "ICommittable"):
-        """
-        Write commit message to a buffer.
-
-        For the method no way check the commit result
-        (for example if lost connection - commits will not re-send and committed messages will receive again)
-        """
-        raise NotImplementedError()
-
-    async def commit_with_ack(
-        self, mess: "ICommittable"
-    ) -> Union["CommitResult", List["CommitResult"]]:
-        """
-        write commit message to a buffer and wait ack from the server.
-
-        use asyncio.wait_for for wait with timeout.
-        """
-        raise NotImplementedError()
-
-    async def flush(self):
-        """
-        force send all commit messages from internal buffers to server and wait acks for all of them.
-
-        use asyncio.wait_for for wait with timeout.
-        """
-        raise NotImplementedError()
-
-    async def close(self):
-        raise NotImplementedError()
-
-
 class Reader(object):
     def async_sessions_stat(self) -> concurrent.futures.Future:
         """

ydb/_topic_reader/topic_reader_asyncio.py

@@ -26,15 +26,22 @@ class TopicReaderError(YdbError):
 
 class TopicReaderStreamClosedError(TopicReaderError):
     def __init__(self):
-        super().__init__("Topic reader is closed")
+        super().__init__("Topic reader stream is closed")
+
+
+class TopicReaderClosedError(TopicReaderError):
+    def __init__(self):
+        super().__init__("Topic reader is closed already")
 
 
 class PublicAsyncIOReader:
     _loop: asyncio.AbstractEventLoop
+    _closed: bool
     _reconnector: ReaderReconnector
 
     def __init__(self, driver: Driver, settings: PublicReaderSettings):
         self._loop = asyncio.get_running_loop()
+        self._closed = False
         self._reconnector = ReaderReconnector(driver, settings)
 
     async def __aenter__(self):
@@ -43,6 +50,10 @@ async def __aenter__(self):
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         raise NotImplementedError()
 
+    def __del__(self):
+        if not self._closed:
+            self._loop.create_task(self.close(), name="close reader")
+
     async def sessions_stat(self) -> typing.List["SessionStat"]:
         """
         Receive stat from the server
@@ -132,7 +143,11 @@ async def flush(self):
         raise NotImplementedError()
 
     async def close(self):
-        raise NotImplementedError()
+        if self._closed:
+            raise TopicReaderClosedError()
+
+        self._closed = True
+        await self._reconnector.close()
 
 
 class ReaderReconnector:

ydb/topic.py

@@ -1,16 +1,19 @@
-from typing import List, Callable, Union, Mapping, Any
+from typing import List, Callable, Union, Mapping, Any, Optional
 
 from . import aio, Credentials
 from ._topic_reader.topic_reader import (
     PublicReaderSettings as TopicReaderSettings,
     Reader as TopicReader,
-    ReaderAsyncIO as TopicReaderAsyncIO,
     Selector as TopicSelector,
     Events as TopicReaderEvents,
     RetryPolicy as TopicReaderRetryPolicy,
     StubEvent as TopicReaderStubEvent,
 )
 
+from ._topic_reader.topic_reader_asyncio import (
+    PublicAsyncIOReader as TopicReaderAsyncIO
+)
+from ._topic_wrapper.common import TokenGetterFuncType
 
 from ._topic_writer.topic_writer import (  # noqa: F401
     Writer as TopicWriter,
@@ -32,26 +35,27 @@ def __init__(self, driver: aio.Driver, settings: "TopicClientSettings" = None):
 
     def topic_reader(
         self,
-        topic: Union[str, TopicSelector, List[Union[str, TopicSelector]]],
         consumer: str,
-        commit_batch_time: Union[float, None] = 0.1,
-        commit_batch_count: Union[int, None] = 1000,
+        topic: str,
         buffer_size_bytes: int = 50 * 1024 * 1024,
-        sync_commit: bool = False,  # reader.commit(...) will wait commit ack from server
-        on_commit: Callable[["TopicReaderEvents.OnCommit"], None] = None,
-        on_get_partition_start_offset: Callable[
-            ["TopicReaderEvents.OnPartitionGetStartOffsetRequest"],
-            "TopicReaderEvents.OnPartitionGetStartOffsetResponse",
-        ] = None,
-        on_init_partition: Callable[["TopicReaderStubEvent"], None] = None,
-        on_shutdown_partition: Callable[["TopicReaderStubEvent"], None] = None,
-        decoder: Union[Mapping[int, Callable[[bytes], bytes]], None] = None,
-        deserializer: Union[Callable[[bytes], Any], None] = None,
-        one_attempt_connection_timeout: Union[float, None] = 1,
-        connection_timeout: Union[float, None] = None,
-        retry_policy: Union["TopicReaderRetryPolicy", None] = None,
+        # on_commit: Callable[["Events.OnCommit"], None] = None
+        # on_get_partition_start_offset: Callable[
+        #     ["Events.OnPartitionGetStartOffsetRequest"],
+        #     "Events.OnPartitionGetStartOffsetResponse",
+        # ] = None
+        # on_partition_session_start: Callable[["StubEvent"], None] = None
+        # on_partition_session_stop: Callable[["StubEvent"], None] = None
+        # on_partition_session_close: Callable[["StubEvent"], None] = None  # todo?
+        # decoder: Union[Mapping[int, Callable[[bytes], bytes]], None] = None
+        # deserializer: Union[Callable[[bytes], Any], None] = None
+        # one_attempt_connection_timeout: Union[float, None] = 1
+        # connection_timeout: Union[float, None] = None
+        # retry_policy: Union["RetryPolicy", None] = None
     ) -> TopicReaderAsyncIO:
-        raise NotImplementedError()
+        args = locals()
+        del args["self"]
+        settings = TopicReaderSettings(**args)
+        return TopicReaderAsyncIO(self._driver, settings)
 
     def topic_writer(
         self,