simplify stream.py + correct lock.py

Author: slampy97

ydb/aio/coordination/lock.py

@@ -70,10 +70,8 @@ async def _ensure_session(self):
     async def _wait_for_response(self, req_id: int, *, kind: str):
         try:
             while True:
-                resp = await asyncio.wait_for(
-                    self._stream._incoming_queue.get(),
-                    timeout=self._wait_timeout,
-                )
+                resp = await self._stream.receive(self._wait_timeout)
+
                 fs = FromServer.from_proto(resp)
 
                 if kind == "acquire":

ydb/aio/coordination/stream.py

@@ -92,10 +92,7 @@ async def close(self):
         self._background_tasks.clear()
 
         if self._stream:
-            try:
-                self._stream.close()
-            except Exception:
-                pass
+            self._stream.close()
             self._stream = None
 
         self.session_id = None