sync

Author: rekby

examples/topic/reader_async_example.py

@@ -15,61 +15,52 @@ async def connect():
 
 
 async def create_reader_and_close_with_context_manager(db: ydb.aio.Driver):
-    with ydb.TopicClientAsyncIO(db).reader(
+    async with ydb.TopicClientAsyncIO(db).reader(
         "/database/topic/path", consumer="consumer"
     ) as reader:
-        async for message in reader.messages():
-            pass
+        ...
 
 
 async def print_message_content(reader: ydb.TopicReaderAsyncIO):
-    async for message in reader.messages():
+    while True:
+        message = await reader.receive_message()
         print("text", message.data.read().decode("utf-8"))
         # await and async_commit need only for sync commit mode - for wait ack from servr
         await reader.commit(message)
 
 
-async def process_messages_batch_explicit_commit(reader: ydb.TopicReaderAsyncIO):
+async def process_messages_batch_with_commit(reader: ydb.TopicReaderAsyncIO):
     # Explicit commit example
-    async for batch in reader.batches(max_messages=100, timeout=2):
-        async with asyncio.TaskGroup() as tg:
-            for message in batch.messages:
-                tg.create_task(_process(message))
-
-        # wait complete of process all messages from batch be taskgroup context manager
-        # and commit complete batch
+    while True:
+        batch = await reader.receive_batch()
+        ...
         await reader.commit(batch)
 
 
-async def process_messages_batch_context_manager_commit(reader: ydb.TopicReaderAsyncIO):
-    # Commit with context manager
-    async for batch in reader.batches():
-        async with reader.commit_on_exit(batch), asyncio.TaskGroup() as tg:
-            for message in batch.messages:
-                tg.create_task(_process(message))
-
-
 async def get_message_with_timeout(reader: ydb.TopicReaderAsyncIO):
     try:
         message = await asyncio.wait_for(reader.receive_message(), timeout=1)
-    except TimeoutError:
+    except asyncio.TimeoutError:
         print("Have no new messages in a second")
         return
 
     print("mess", message.data)
 
 
 async def get_all_messages_with_small_wait(reader: ydb.TopicReaderAsyncIO):
-    async for message in reader.messages(timeout=1):
-        await _process(message)
-    print("Have no new messages in a second")
+    while True:
+        try:
+            message = await reader.receive_message()
+            await _process(message)
+        except asyncio.TimeoutError:
+            print("Have no new messages in a second")
 
 
 async def get_a_message_from_external_loop(reader: ydb.TopicReaderAsyncIO):
     for i in range(10):
         try:
             message = await asyncio.wait_for(reader.receive_message(), timeout=1)
-        except TimeoutError:
+        except asyncio.TimeoutError:
             return
         await _process(message)
 
@@ -78,7 +69,7 @@ async def get_one_batch_from_external_loop_async(reader: ydb.TopicReaderAsyncIO)
     for i in range(10):
         try:
             batch = await asyncio.wait_for(reader.receive_batch(), timeout=2)
-        except TimeoutError:
+        except asyncio.TimeoutError:
             return
 
         for message in batch.messages:
@@ -92,27 +83,20 @@ async def auto_deserialize_message(db: ydb.aio.Driver):
     async with ydb.TopicClientAsyncIO(db).reader(
         "/database/topic/path", consumer="asd", deserializer=json.loads
     ) as reader:
-        async for message in reader.messages():
+        while True:
+            message = await reader.receive_message()
             print(
                 message.data.Name
             )  # message.data replaces by json.loads(message.data) of raw message
             reader.commit(message)
 
 
-async def commit_batch_with_context(reader: ydb.TopicReaderAsyncIO):
-    async for batch in reader.batches():
-        async with reader.commit_on_exit(batch):
-            for message in batch.messages:
-                if not batch.is_alive:
-                    break
-                await _process(message)
-
-
 async def handle_partition_stop(reader: ydb.TopicReaderAsyncIO):
-    async for message in reader.messages():
-        time.sleep(1)  # some work
+    while True:
+        message = await reader.receive_message()
+        time.sleep(123)  # some work
         if message.is_alive:
-            time.sleep(123)  # some other work
+            time.sleep(1)  # some other work
             await reader.commit(message)
 
 
@@ -126,7 +110,8 @@ def process_batch(batch):
             _process(message)
         reader.commit(batch)
 
-    async for batch in reader.batches():
+    while True:
+        batch = await reader.receive_batch()
         process_batch(batch)
 
 
@@ -137,18 +122,12 @@ async def connect_and_read_few_topics(db: ydb.aio.Driver):
             ydb.TopicSelector("/database/second-topic", partitions=3),
         ]
     ) as reader:
-        async for message in reader.messages():
+        while True:
+            message = await reader.receive_message()
             await _process(message)
             await reader.commit(message)
 
 
-async def handle_partition_graceful_stop_batch(reader: ydb.TopicReaderAsyncIO):
-    # no special handle, but batch will contain less than prefer count messages
-    async for batch in reader.batches():
-        await _process(batch)
-        reader.commit(batch)
-
-
 async def advanced_commit_notify(db: ydb.aio.Driver):
     def on_commit(event: ydb.TopicReaderEvents.OnCommit) -> None:
         print(event.topic)
@@ -157,7 +136,8 @@ def on_commit(event: ydb.TopicReaderEvents.OnCommit) -> None:
     async with ydb.TopicClientAsyncIO(db).reader(
         "/local", consumer="consumer", commit_batch_time=4, on_commit=on_commit
     ) as reader:
-        async for message in reader.messages():
+        while True:
+            message = await reader.receive_message()
             await _process(message)
             await reader.commit(message)