diff --git a/hbmqtt/adapters.py b/hbmqtt/adapters.py
index 3c64583..496ad70 100644
--- a/hbmqtt/adapters.py
+++ b/hbmqtt/adapters.py
@@ -176,13 +176,13 @@ class StreamWriterAdapter(WriterAdapter):
     @asyncio.coroutine
     def close(self):
         if not self.is_closed:
+            self.is_closed = True # we first mark this closed so yields below don't cause races with waiting writes
             yield from self._writer.drain()
             if self._writer.can_write_eof():
                 self._writer.write_eof()
             self._writer.close()
             try: yield from self._writer.wait_closed() # py37+
             except AttributeError: pass
-            self.is_closed = True
 
 
 class BufferReader(ReaderAdapter):
diff --git a/hbmqtt/codecs.py b/hbmqtt/codecs.py
index ecebe3b..a6af603 100644
--- a/hbmqtt/codecs.py
+++ b/hbmqtt/codecs.py
@@ -51,7 +51,7 @@ def read_or_raise(reader, n=-1):
     """
     try:
         data = yield from reader.read(n)
-    except (asyncio.IncompleteReadError, ConnectionResetError):
+    except (asyncio.IncompleteReadError, ConnectionResetError, BrokenPipeError):
         data = None
     if not data:
         raise NoDataException("No more data")