From 458f9fdeaa926df2c25d7d2ad2b5eb7fb5c286f1 Mon Sep 17 00:00:00 2001
From: phi <phi.friday@gmail.com>
Date: Mon, 29 Jul 2024 21:44:12 +0900
Subject: [PATCH] fix: hash

---
 src/async_wrapper/pipe.py | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/src/async_wrapper/pipe.py b/src/async_wrapper/pipe.py
index 6cbae1a..64b62ae 100644
--- a/src/async_wrapper/pipe.py
+++ b/src/async_wrapper/pipe.py
@@ -64,6 +64,9 @@ async def next(self, value: InputT) -> OutputT:
     async def dispose(self) -> Any:
         """Disposes the resource and releases any associated resources."""
 
+    @override
+    def __hash__(self) -> int: ...
+
 
 @runtime_checkable
 class DisposableWithCallback(Disposable[InputT, OutputT], Protocol[InputT, OutputT]):
@@ -151,6 +154,10 @@ def prepare_callback(self, subscribable: Subscribable[InputT, OutputT]) -> Any:
         with self._thread_lock:
             self._journals.append(subscribable)
 
+    @override
+    def __hash__(self) -> int:
+        return hash((id(self), id(self._func)))
+
 
 class Pipe(Subscribable[InputT, OutputT], Generic[InputT, OutputT]):
     """
@@ -251,6 +258,10 @@ def subscribe(
     def unsubscribe(self, disposable: Disposable[Any, Any]) -> None:
         self._listeners.pop(disposable, None)
 
+    @override
+    def __hash__(self) -> int:
+        return hash((id(self), id(self._listener)))
+
 
 def create_disposable(
     func: Callable[[InputT], Awaitable[OutputT]],