ref(channel): reduce adaptation

Author: emptybutton

src/tgdb/application/horizon/commit_transaction.py

@@ -55,8 +55,8 @@ async def __call__(
             self.channel.wait(commit.xid),
             self.commit_buffer.add(commit),
         )
-        if notification.error is not None:
-            raise notification.error from notification.error
+        if notification is not None:
+            raise notification from notification
 
     async def _effect(
         self, operator: Operator

src/tgdb/application/horizon/output_commits.py

@@ -10,7 +10,7 @@
     NoTransactionError,
     TransactionNotCommittingError,
 )
-from tgdb.entities.horizon.transaction import XID, Commit, PreparedCommit
+from tgdb.entities.horizon.transaction import Commit, PreparedCommit
 
 
 @dataclass(frozen=True)
@@ -22,26 +22,23 @@ class OutputCommits:
     clock: Clock
 
     async def __call__(self) -> None:
-        async for prepared_commits in self.commit_buffer:
-            await self.output_commits.push(prepared_commits)
+        async for commits in self.commit_buffer:
+            await self.output_commits.push(commits)
             await self.output_commits.sync()
 
-            ok_commit_xids = list[XID]()
-            error_commit_map = dict[
-                XID, NoTransactionError | TransactionNotCommittingError
-            ]()
-
             async with self.shared_horizon as horizon:
-                for prepared_commit in prepared_commits:
+                for commit in commits:
+                    if isinstance(commit, Commit):
+                        await self.channel.publish(commit.xid, None)
+                        continue
+
                     time = await self.clock
 
                     try:
-                        horizon.complete_commit(time, prepared_commit.xid)
+                        horizon.complete_commit(time, commit.xid)
                     except (
                         NoTransactionError, TransactionNotCommittingError
                     ) as error:
-                        error_commit_map[prepared_commit.xid] = error
+                        await self.channel.publish(commit.xid, error)
                     else:
-                        ok_commit_xids.append(prepared_commit.xid)
-
-                await self.channel.publish(ok_commit_xids, error_commit_map)
+                        await self.channel.publish(commit.xid, None)

src/tgdb/application/horizon/ports/channel.py

@@ -1,6 +1,4 @@
 from abc import ABC, abstractmethod
-from collections.abc import Mapping, Sequence
-from dataclasses import dataclass
 
 from tgdb.entities.horizon.horizon import (
     NoTransactionError,
@@ -9,19 +7,15 @@
 from tgdb.entities.horizon.transaction import XID
 
 
-@dataclass(frozen=True)
-class Notification:
-    error: NoTransactionError | TransactionNotCommittingError | None
+type Notification = NoTransactionError | TransactionNotCommittingError | None
 
 
 class Channel(ABC):
     @abstractmethod
     async def publish(
         self,
-        ok_commit_xids: Sequence[XID],
-        error_commit_map: Mapping[
-            XID, NoTransactionError | TransactionNotCommittingError
-        ],
+        xid: XID,
+        notification: Notification,
         /,
     ) -> None: ...
 

src/tgdb/infrastructure/adapters/channel.py

@@ -1,5 +1,4 @@
 from asyncio import wait_for
-from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 
 from tgdb.application.horizon.ports.channel import Channel, Notification
@@ -20,22 +19,11 @@ class AsyncMapChannel(Channel):
 
     async def publish(
         self,
-        ok_commit_xids: Sequence[XID],
-        error_commit_map: Mapping[
-            XID, NoTransactionError | TransactionNotCommittingError
-        ],
+        xid: XID,
+        notification: Notification,
     ) -> None:
-        for ok_commit_xid in ok_commit_xids:
-            self._async_map[ok_commit_xid] = None
-            del self._async_map[ok_commit_xid]
-
-        for error_commit_xid, error in error_commit_map.items():
-            self._async_map[error_commit_xid] = error
-            del self._async_map[error_commit_xid]
+        self._async_map[xid] = notification
+        del self._async_map[xid]
 
     async def wait(self, xid: XID) -> Notification:
-        notification_error = await wait_for(
-            self._async_map[xid], self._timeout_seconds
-        )
-
-        return Notification(notification_error)
+        return await wait_for(self._async_map[xid], self._timeout_seconds)