diff --git a/asyncio_dgram/aio.py b/asyncio_dgram/aio.py
index 8117282..7487a98 100644
--- a/asyncio_dgram/aio.py
+++ b/asyncio_dgram/aio.py
@@ -22,17 +22,20 @@ class DatagramStream:
     raised.
     """
 
-    def __init__(self, transport, recvq, excq):
+    def __init__(self, transport, recvq, excq, drained):
         """
         @param transport    - asyncio transport
         @param recvq        - asyncio queue that gets populated by the
                               DatagramProtocol with received datagrams.
         @param excq         - asyncio queue that gets populated with any errors
                               detected by the DatagramProtocol.
+        @param drained      - asyncio event that is unset when writing is
+                              paused and set otherwise.
         """
         self._transport = transport
         self._recvq = recvq
         self._excq = excq
+        self._drained = drained
 
     def __del__(self):
         self._transport.close()
@@ -87,6 +90,7 @@ class DatagramStream:
         """
         _ = self.exception
         self._transport.sendto(data, addr)
+        await self._drained.wait()
 
     async def recv(self):
         """
@@ -132,13 +136,18 @@ class Protocol(asyncio.DatagramProtocol):
     based asyncio into higher level coroutines.
     """
 
-    def __init__(self, recvq, excq):
+    def __init__(self, recvq, excq, drained):
         """
         @param recvq    - asyncio.Queue for new datagrams
         @param excq     - asyncio.Queue for exceptions
+        @param drained  - asyncio.Event set when the write buffer is below the
+                          high watermark.
         """
         self._recvq = recvq
         self._excq = excq
+        self._drained = drained
+
+        self._drained.set()
 
         # Transports are connected at the time a connection is made.
         self._transport = None
@@ -167,12 +176,19 @@ class Protocol(asyncio.DatagramProtocol):
     def error_received(self, exc):
         self._excq.put_nowait(exc)
 
+    def pause_writing(self):
+        self._drained.clear()
+        super().pause_writing()
+
+    def resume_writing(self):
+        self._drained.set()
+        super().resume_writing()
+
 
 async def bind(addr):
     """
     Bind a socket to a local address for datagrams.  The socket will be either
-    AF_INET or AF_INET6 depending upon the type of address specified.  The
-    socket will be reusable (SO_REUSEADDR) once it enters TIME_WAIT.
+    AF_INET or AF_INET6 depending upon the type of address specified.
 
     @param addr - For AF_INET or AF_INET6, a tuple with the the host and port to
                   to bind; port may be set to 0 to get any free port.
@@ -181,12 +197,13 @@ async def bind(addr):
     loop = asyncio.get_event_loop()
     recvq = asyncio.Queue()
     excq = asyncio.Queue()
+    drained = asyncio.Event()
 
     transport, protocol = await loop.create_datagram_endpoint(
-        lambda: Protocol(recvq, excq), local_addr=addr, reuse_address=True
+        lambda: Protocol(recvq, excq, drained), local_addr=addr, reuse_address=False
     )
 
-    return DatagramServer(transport, recvq, excq)
+    return DatagramServer(transport, recvq, excq, drained)
 
 
 async def connect(addr):
@@ -201,12 +218,13 @@ async def connect(addr):
     loop = asyncio.get_event_loop()
     recvq = asyncio.Queue()
     excq = asyncio.Queue()
+    drained = asyncio.Event()
 
     transport, protocol = await loop.create_datagram_endpoint(
-        lambda: Protocol(recvq, excq), remote_addr=addr
+        lambda: Protocol(recvq, excq, drained), remote_addr=addr
     )
 
-    return DatagramClient(transport, recvq, excq)
+    return DatagramClient(transport, recvq, excq, drained)
 
 
 async def from_socket(sock):
@@ -224,6 +242,7 @@ async def from_socket(sock):
     loop = asyncio.get_event_loop()
     recvq = asyncio.Queue()
     excq = asyncio.Queue()
+    drained = asyncio.Event()
 
     if sock.family not in (socket.AF_INET, socket.AF_INET6):
         raise TypeError(
@@ -234,12 +253,12 @@ async def from_socket(sock):
         raise TypeError("socket must be %s" % (socket.SOCK_DGRAM,))
 
     transport, protocol = await loop.create_datagram_endpoint(
-        lambda: Protocol(recvq, excq), sock=sock
+        lambda: Protocol(recvq, excq, drained), sock=sock
     )
 
     if transport.get_extra_info("peername") is not None:
         # Workaround transport ignoring the peer address of the socket.
         transport._address = transport.get_extra_info("peername")
-        return DatagramClient(transport, recvq, excq)
+        return DatagramClient(transport, recvq, excq, drained)
     else:
-        return DatagramServer(transport, recvq, excq)
+        return DatagramServer(transport, recvq, excq, drained)

DatagramStream.send should respect flow control
currently send is a coroutine that `awaits` nothing. it should be a `send/drain` pair of methods instead
diff --git a/test/test_aio.py b/test/test_aio.py
index 7d52a99..7b71ffc 100644
--- a/test/test_aio.py
+++ b/test/test_aio.py
@@ -1,12 +1,23 @@
 import asyncio
 import contextlib
+import os
 import socket
+import unittest.mock
 
 import pytest
 
 import asyncio_dgram
 
 
+@pytest.fixture
+def mock_socket():
+    s = unittest.mock.create_autospec(socket.socket)
+    s.family = socket.AF_INET
+    s.type = socket.SOCK_DGRAM
+
+    return s
+
+
 @contextlib.contextmanager
 def loop_exception_handler():
     """
@@ -260,3 +271,81 @@ async def test_unconnected_sender(addr):
 
     with pytest.raises(asyncio.TimeoutError):
         await asyncio.wait_for(connected.recv(), 0.05)
+
+
+@pytest.mark.asyncio
+async def test_protocol_pause_resume(monkeypatch, mock_socket, tmp_path):
+    # This is a little involved, but necessary to make sure that the Protocol
+    # is correctly noticing when writing as been paused and resumed.  In
+    # summary:
+    #
+    # - Mock the Protocol with one that sets the write buffer limits to 0 and
+    # records when pause and recume writing are called.
+    #
+    # - Use a mock socket so that we can inject a BlockingIOError on send.
+    # Ideally we'd mock method itself, but it's read-only the entire object
+    # needs to be mocked.  Due to this, we need to use a temporary file that we
+    # can write to in order to kick the event loop to consider it ready for
+    # writing.
+
+    class TestableProtocol(asyncio_dgram.aio.Protocol):
+        pause_writing_called = 0
+        resume_writing_called = 0
+        instance = None
+
+        def __init__(self, *args, **kwds):
+            TestableProtocol.instance = self
+            super().__init__(*args, **kwds)
+
+        def connection_made(self, transport):
+            transport.set_write_buffer_limits(low=0, high=0)
+            super().connection_made(transport)
+
+        def pause_writing(self):
+            self.pause_writing_called += 1
+            super().pause_writing()
+
+        def resume_writing(self):
+            self.resume_writing_called += 1
+            super().resume_writing()
+
+    async def passthrough():
+        """
+        Used to mock the wait method on the asyncio.Event tracking if the write
+        buffer is past the high water mark or not.  Given we're testing how
+        that case is handled, we know it's safe locally to mock it.
+        """
+        pass
+
+    with monkeypatch.context() as ctx:
+        ctx.setattr(asyncio_dgram.aio, "Protocol", TestableProtocol)
+
+        client = await asyncio_dgram.from_socket(mock_socket)
+        mock_socket.send.side_effect = BlockingIOError
+        mock_socket.fileno.return_value = os.open(
+            tmp_path / "socket", os.O_RDONLY | os.O_CREAT
+        )
+
+        with monkeypatch.context() as ctx2:
+            ctx2.setattr(client._drained, "wait", passthrough)
+            await client.send(b"foo")
+
+        assert TestableProtocol.instance.pause_writing_called == 1
+        assert TestableProtocol.instance.resume_writing_called == 0
+        assert not TestableProtocol.instance._drained.is_set()
+
+        mock_socket.send.side_effect = None
+        fd = os.open(tmp_path / "socket", os.O_WRONLY)
+        os.write(fd, b"\n")
+        os.close(fd)
+
+        with monkeypatch.context() as ctx2:
+            ctx2.setattr(client._drained, "wait", passthrough)
+            await client.send(b"foo")
+        await asyncio.sleep(0.1)
+
+        assert TestableProtocol.instance.pause_writing_called == 1
+        assert TestableProtocol.instance.resume_writing_called == 1
+        assert TestableProtocol.instance._drained.is_set()
+
+        os.close(mock_socket.fileno.return_value)

pip install -e .[test]
parse_log_pytest_v2
requirements.txt
3.7
requirements-test.txt
pytest --log-level=DEBUG -W default -v -s
-e git+https://github.com/jsbronder/asyncio-dgram.git@40cabedc88195bf85a3912752c02fed09385dc94#egg=asyncio_dgram
black==23.3.0
certifi @ file:///croot/certifi_1671487769961/work/certifi
click==8.1.8
exceptiongroup==1.2.2
flake8==5.0.4
importlib-metadata==4.2.0
iniconfig==2.0.0
mccabe==0.7.0
mypy-extensions==1.0.0
packaging==24.0
pathspec==0.11.2
platformdirs==4.0.0
pluggy==1.2.0
pycodestyle==2.9.1
pyflakes==2.5.0
pytest==7.4.4
pytest-asyncio==0.21.2
tomli==2.0.1
typed-ast==1.5.5
typing_extensions==4.7.1
zipp==3.15.0

name: asyncio-dgram
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2022.12.7=py37h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=22.3.1=py37h06a4308_0
  - python=3.7.16=h7a1cb2a_0
  - readline=8.2=h5eee18b_0
  - setuptools=65.6.3=py37h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.38.4=py37h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - black==23.3.0
      - click==8.1.8
      - exceptiongroup==1.2.2
      - flake8==5.0.4
      - importlib-metadata==4.2.0
      - iniconfig==2.0.0
      - mccabe==0.7.0
      - mypy-extensions==1.0.0
      - packaging==24.0
      - pathspec==0.11.2
      - platformdirs==4.0.0
      - pluggy==1.2.0
      - pycodestyle==2.9.1
      - pyflakes==2.5.0
      - pytest==7.4.4
      - pytest-asyncio==0.21.2
      - tomli==2.0.1
      - typed-ast==1.5.5
      - typing-extensions==4.7.1
      - zipp==3.15.0
prefix: /opt/conda/envs/asyncio-dgram

test/test_aio.py::test_bind_sync[INET] test/test_aio.py::test_bind_sync[INET6] test/test_aio.py::test_echo[INET] test/test_aio.py::test_echo[INET6] test/test_aio.py::test_echo_bind[INET] test/test_aio.py::test_echo_bind[INET6] test/test_aio.py::test_unconnected_sender[INET] test/test_aio.py::test_unconnected_sender[INET6] test/test_aio.py::test_protocol_pause_resume
test/test_aio.py::test_connect_sync[INET] test/test_aio.py::test_connect_sync[INET6] test/test_aio.py::test_from_socket_streamtype[INET] test/test_aio.py::test_from_socket_streamtype[INET6] test/test_aio.py::test_from_socket_bad_socket test/test_aio.py::test_no_server[INET] test/test_aio.py::test_no_server[INET6]
asyncio_dgram/aio.py
asyncio_dgram/aio.py