Repository: gurumitts/pylutron-caseta

Issue:
Timeout error not trapped?

After receiving this timeout error, smartbridge.is_connected still says connected?

```
2020-04-03 19:56:46,242 [Controller] [INFO ] is_connected=True
2020-04-03 19:56:46,242 [Controller] [DEBUG] set_value: {'CommuniqueType': 'CreateRequest', 'Header': {'Url': '/zone/5/commandprocessor'}, 'Body': {'Command': {'CommandType': 'GoToLevel', 'Parameter': [{'Type': 'Level', 'Value': 0}]}}}
2020-04-03 19:56:46,243 [Controller] [DEBUG] sending b'{"CommuniqueType": "CreateRequest", "Header": {"Url": "/zone/5/commandprocessor"}, "Body": {"Command": {"CommandType": "GoToLevel", "Parameter": [{"Type": "Level", "Value": 0}]}}}'
2020-04-03 19:56:46,244 [Controller] [ERROR] Fatal write error on socket transport
protocol: <asyncio.sslproto.SSLProtocol object at 0x75863d90>
transport: <_SelectorSocketTransport fd=11 read=polling write=<idle, bufsize=0>>
Traceback (most recent call last):
  File "/usr/lib/python3.7/asyncio/selector_events.py", line 857, in write
    n = self._sock.send(data)
TimeoutError: [Errno 110] Connection timed out
```

Note that the "sending" line is from
https://github.com/gurumitts/pylutron-caseta/blob/dev/pylutron_caseta/leap.py#L88

How should this be handled? Is there any way for the caller to know that the command failed?



Discussion:

Patch:
diff --git a/pylintrc b/pylintrc
index 5a366ce..de598db 100644
--- a/pylintrc
+++ b/pylintrc
@@ -5,3 +5,7 @@ disable=
   too-many-arguments,
   too-many-instance-attributes,
   too-many-public-methods,
+
+good-names=
+  i,j,k,ex,Run,_, # defaults
+  T
diff --git a/pylutron_caseta/__init__.py b/pylutron_caseta/__init__.py
index 3ef86ea..704ff72 100644
--- a/pylutron_caseta/__init__.py
+++ b/pylutron_caseta/__init__.py
@@ -1,5 +1,9 @@
 """An API to communicate with the Lutron Caseta Smart Bridge."""
 
+from typing import Optional
+
+from .messages import Response, ResponseStatus
+
 _LEAP_DEVICE_TYPES = {
     "light": ["WallDimmer", "PlugInDimmer"],
     "switch": ["WallSwitch"],
@@ -34,3 +38,21 @@ FAN_HIGH = "High"
 OCCUPANCY_GROUP_OCCUPIED = "Occupied"
 OCCUPANCY_GROUP_UNOCCUPIED = "Unoccupied"
 OCCUPANCY_GROUP_UNKNOWN = "Unknown"
+
+
+class BridgeDisconnectedError(Exception):
+    """Raised when the connection is lost while waiting for a response."""
+
+
+class BridgeResponseError(Exception):
+    """Raised when the bridge sends an error response."""
+
+    def __init__(self, response: Response):
+        """Create a BridgeResponseError."""
+        super().__init__(str(response.Header.StatusCode))
+        self.response = response
+
+    @property
+    def code(self) -> Optional[ResponseStatus]:
+        """Get the status code returned by the server."""
+        return self.response.Header.StatusCode
diff --git a/pylutron_caseta/leap.py b/pylutron_caseta/leap.py
index 6f166b6..8dfc2db 100644
--- a/pylutron_caseta/leap.py
+++ b/pylutron_caseta/leap.py
@@ -4,102 +4,138 @@ import asyncio
 import json
 import logging
 import re
+import uuid
+from typing import Callable, Dict, List, Optional
+
+from . import BridgeDisconnectedError
+from .messages import Response
 
 _LOG = logging.getLogger(__name__)
 _DEFAULT_LIMIT = 2 ** 16
 
 
-async def open_connection(host=None, port=None, *, limit=_DEFAULT_LIMIT, **kwds):
-    """Open a stream and wrap it with LEAP."""
-    connection = await asyncio.open_connection(host, port, limit=limit, **kwds)
-    return LeapReader(connection[0]), LeapWriter(connection[1])
-
-
-class LeapReader:
-    """A wrapper for reading the LEAP protocol."""
+class LeapProtocol:
+    """A wrapper for making LEAP calls."""
 
-    def __init__(self, reader):
-        """Initialize the reader."""
+    def __init__(
+        self,
+        reader: asyncio.StreamReader,
+        writer: asyncio.StreamWriter,
+    ):
+        """Wrap a reader and writer with a LEAP request and response protocol."""
         self._reader = reader
+        self._writer = writer
+        self._in_flight_requests: Dict[str, "asyncio.Future[Response]"] = {}
+        self._unsolicited_subs: List[Callable[[Response], None]] = []
 
-    def exception(self):
-        """Get the exception."""
-        return self._reader.exception()
+    async def request(
+        self, communique_type: str, url: str, body: Optional[dict] = None
+    ) -> Response:
+        """Make a request to the bridge and return the response."""
+        tag = str(uuid.uuid4())
+        future: asyncio.Future = asyncio.get_running_loop().create_future()
 
-    async def read(self):
-        """
-        Read a single object.
+        cmd = {
+            "CommuniqueType": communique_type,
+            "Header": {"ClientTag": tag, "Url": url},
+        }
 
-        If EOF is received, return `None`.
+        if body is not None:
+            cmd["Body"] = body
 
-        If invaid data is received, raise ValueError.
-        """
-        received = await self._reader.readline()
+        self._in_flight_requests[tag] = future
 
-        if received == b"":
-            return None
-        _LOG.debug("received %s", received)
+        # remove cancelled tasks
+        def clean_up(future):
+            if future.cancelled():
+                self._in_flight_requests.pop(tag, None)
 
-        try:
-            return json.loads(received.decode("UTF-8"))
-        except ValueError as err:
-            _LOG.error("Invalid LEAP response: %s", received)
-            self._reader.set_exception(err)
-            raise err
+        future.add_done_callback(clean_up)
 
-    async def wait_for(self, communique_type):
+        try:
+            text = json.dumps(cmd).encode("UTF-8")
+            _LOG.debug("sending %s", text)
+            self._writer.write(text + b"\r\n")
+
+            return await future
+        finally:
+            self._in_flight_requests.pop(tag, None)
+
+    async def run(self):
+        """Event monitoring loop."""
+        while not self._reader.at_eof():
+            received = await self._reader.readline()
+
+            if received == b"":
+                break
+
+            resp_json = json.loads(received.decode("UTF-8"))
+
+            if isinstance(resp_json, dict):
+                tag = resp_json.get("Header", {}).pop("ClientTag", None)
+                if tag is not None:
+                    in_flight = self._in_flight_requests.pop(tag, None)
+                    if in_flight is not None and not in_flight.done():
+                        _LOG.debug("received: %s", resp_json)
+                        in_flight.set_result(Response.from_json(resp_json))
+                    else:
+                        _LOG.error(
+                            "Was not expecting message with tag %s: %s", tag, resp_json
+                        )
+                else:
+                    _LOG.debug("Received message with no tag: %s", resp_json)
+                    obj = Response.from_json(resp_json)
+                    for handler in self._unsolicited_subs:
+                        try:
+                            handler(obj)
+                        except Exception:  # pylint: disable=broad-except
+                            _LOG.exception(
+                                "Got exception from unsolicited message handler"
+                            )
+
+    def subscribe_unsolicited(self, callback: Callable[[Response], None]):
         """
-        Read for a specific communique type.
+        Subscribe to notifications of unsolicited events.
 
-        Discards all messages not matching the communique type until the
-        specified communique type is received
+        The provided callback will be executed when the bridge sends an untagged
+        response message.
         """
-        while True:
-            received = await self.read()
-            if received.get("CommuniqueType", None) == communique_type:
-                return received
-            _LOG.info("Ignoring message %s", received)
-
-    def at_eof(self):
-        """Return `True` if the underlying stream is at EOF."""
-        return self._reader.at_eof()
-
-
-class LeapWriter:
-    """A wrapper for writing the LEAP protocol."""
+        if not callable(callback):
+            raise TypeError("callback must be callable")
+        self._unsolicited_subs.append(callback)
 
-    def __init__(self, writer):
-        """Initialize the writer."""
-        self._writer = writer
+    def unsubscribe_unsolicited(self, callback: Callable[[Response], None]):
+        """Unsubscribe from notifications of unsolicited events."""
+        self._unsolicited_subs.remove(callback)
 
-    def abort(self):
-        """Abort the underlying stream."""
-        self._writer.transport.abort()
+    def close(self):
+        """Disconnect."""
+        self._writer.close()
 
-    async def drain(self):
-        """Let the underlying stream drain its buffer."""
-        await self._writer.drain()
+        for request in self._in_flight_requests.values():
+            request.set_exception(BridgeDisconnectedError())
+        self._in_flight_requests.clear()
 
-    def write(self, obj):
-        """Write a single object."""
-        text = json.dumps(obj).encode("UTF-8")
-        _LOG.debug("sending %s", text)
-        self._writer.write(text + b"\r\n")
 
-    def write_eof(self):
-        """Write EOF to the underlying stream."""
-        self._writer.write_eof()
+async def open_connection(
+    host: str, port: int, *, limit: int = _DEFAULT_LIMIT, **kwds
+) -> LeapProtocol:
+    """Open a stream and wrap it with LEAP."""
+    reader, writer = await asyncio.open_connection(host, port, limit=limit, **kwds)
+    return LeapProtocol(reader, writer)
 
 
 _HREFRE = re.compile(r"/(?:\D+)/(\d+)(?:\/\D+)?")
 
 
-def id_from_href(href):
+def id_from_href(href: str) -> str:
     """Get an id from any kind of href.
 
     Raises ValueError if id cannot be determined from the format
     """
-    try:
-        return _HREFRE.match(href).group(1)
-    except IndexError as ex:
-        raise ValueError("Cannot find ID from href {}".format(href)) from ex
+    match = _HREFRE.match(href)
+
+    if match is None:
+        raise ValueError(f"Cannot find ID from href {href!r}")
+
+    return match.group(1)
diff --git a/pylutron_caseta/messages.py b/pylutron_caseta/messages.py
new file mode 100644
index 0000000..b7b4e6c
--- /dev/null
+++ b/pylutron_caseta/messages.py
@@ -0,0 +1,88 @@
+"""Models for messages exchanged with the bridge."""
+
+from typing import NamedTuple, Optional
+
+
+class ResponseStatus:
+    """A response status split into its code and message parts."""
+
+    def __init__(self, code: Optional[int], message: str):
+        """Create a new ResponseStatus."""
+        self.code = code
+        self.message = message
+
+    @classmethod
+    def from_str(cls, data: str) -> "ResponseStatus":
+        """Convert a str to a ResponseStatus."""
+        space = data.find(" ")
+        if space == -1:
+            code = None
+        else:
+            try:
+                code = int(data[:space])
+                data = data[space + 1 :]
+            except ValueError:
+                code = None
+
+        return ResponseStatus(code, data)
+
+    def is_successful(self) -> bool:
+        """Check if the status code is in the range [200, 300)."""
+        return self.code is not None and self.code >= 200 and self.code < 300
+
+    def __repr__(self):
+        """Get a string representation of the ResponseStatus."""
+        return f"ResponseStatus({self.code!r}, {self.message!r})"
+
+    def __str__(self):
+        """Format the response status as a string in the LEAP header format."""
+        return f"{self.code} {self.message}"
+
+    def __eq__(self, other):
+        """Check if this ResponseStatus is equal to another ResponseStatus."""
+        return (
+            isinstance(other, ResponseStatus)
+            and self.code == other.code
+            and self.message == other.message
+        )
+
+
+class ResponseHeader(NamedTuple):
+    """A LEAP response header."""
+
+    # pylint: disable=invalid-name
+
+    StatusCode: Optional[ResponseStatus] = None
+    Url: Optional[str] = None
+    MessageBodyType: Optional[str] = None
+
+    @classmethod
+    def from_json(cls, data: dict) -> "ResponseHeader":
+        """Convert a JSON dictionary to a ResponseHeader."""
+        status = data.get("StatusCode", None)
+        StatusCode = ResponseStatus.from_str(status) if status is not None else None
+
+        return ResponseHeader(
+            StatusCode=StatusCode,
+            Url=data.get("Url", None),
+            MessageBodyType=data.get("MessageBodyType", None),
+        )
+
+
+class Response(NamedTuple):
+    """A LEAP response."""
+
+    # pylint: disable=invalid-name
+
+    Header: ResponseHeader
+    CommuniqueType: Optional[str] = None
+    Body: dict = {}
+
+    @classmethod
+    def from_json(cls, data: dict) -> "Response":
+        """Convert a JSON dictionary to a Response."""
+        return Response(
+            Header=ResponseHeader.from_json(data.get("Header", {})),
+            CommuniqueType=data.get("CommuniqueType", None),
+            Body=data.get("Body", None),
+        )
diff --git a/pylutron_caseta/smartbridge.py b/pylutron_caseta/smartbridge.py
index 52dcee3..0762eab 100644
--- a/pylutron_caseta/smartbridge.py
+++ b/pylutron_caseta/smartbridge.py
@@ -4,6 +4,7 @@ import asyncio
 import logging
 import socket
 import ssl
+from typing import Callable, Dict, List, Optional
 
 try:
     from asyncio import get_running_loop as get_loop
@@ -11,15 +12,22 @@ except ImportError:
     # For Python 3.6 and earlier, we have to use get_event_loop instead
     from asyncio import get_event_loop as get_loop
 
-from . import _LEAP_DEVICE_TYPES, FAN_OFF, OCCUPANCY_GROUP_UNKNOWN
-from .leap import open_connection, id_from_href
+from . import (
+    _LEAP_DEVICE_TYPES,
+    FAN_OFF,
+    OCCUPANCY_GROUP_UNKNOWN,
+    BridgeDisconnectedError,
+    BridgeResponseError,
+)
+from .leap import open_connection, id_from_href, LeapProtocol
+from .messages import Response
 
 _LOG = logging.getLogger(__name__)
 
 LEAP_PORT = 8081
 PING_INTERVAL = 60.0
-PING_DELAY = 5.0
 CONNECT_TIMEOUT = 5.0
+REQUEST_TIMEOUT = 5.0
 RECONNECT_DELAY = 2.0
 
 
@@ -30,28 +38,65 @@ class Smartbridge:
     It uses an SSL interface known as the LEAP server.
     """
 
-    def __init__(self, connect):
+    def __init__(self, connect: Callable[[], LeapProtocol]):
         """Initialize the Smart Bridge."""
-        self.devices = {}
-        self.scenes = {}
-        self.occupancy_groups = {}
-        self.areas = {}
-        self.logged_in = False
+        self.devices: Dict[str, dict] = {}
+        self.scenes: Dict[str, dict] = {}
+        self.occupancy_groups: Dict[str, dict] = {}
+        self.areas: Dict[str, dict] = {}
         self._connect = connect
-        self._subscribers = {}
-        self._occupancy_subscribers = {}
-        self._login_lock = asyncio.Lock()
-        self._reader = None
-        self._writer = None
-        self._monitor_task = None
-        self._ping_task = None
-        self._got_ping = asyncio.Event()
+        self._subscribers: Dict[str, Callable[[], None]] = {}
+        self._occupancy_subscribers: Dict[str, Callable[[], None]] = {}
+        self._login_task: Optional[asyncio.Task] = None
+        # Use future so we can wait before the login starts and
+        # don't need to wait for "login" on reconnect.
+        self._login_completed: asyncio.Future = (
+            asyncio.get_running_loop().create_future()
+        )
+        self._leap: Optional[LeapProtocol] = None
+        self._monitor_task: Optional[asyncio.Task] = None
+        self._ping_task: Optional[asyncio.Task] = None
+
+    @property
+    def logged_in(self):
+        """Check if the bridge is connected and ready."""
+        return (
+            # are we connected?
+            self._monitor_task is not None
+            and not self._monitor_task.done()
+            # are we ready?
+            and self._login_completed.done()
+            and not self._login_completed.cancelled()
+            and self._login_completed.exception() is None
+        )
 
     async def connect(self):
         """Connect to the bridge."""
-        await self._login()
+        # reset any existing connection state
+        if self._login_task is not None:
+            self._login_task.cancel()
+            self._login_task = None
+
+        if self._monitor_task is not None:
+            self._monitor_task.cancel()
+            self._monitor_task = None
+
+        if self._ping_task is not None:
+            self._ping_task.cancel()
+            self._ping_task = None
+
+        if self._leap is not None:
+            self._leap.close()
+            self._leap = None
+
+        if not self._login_completed.done():
+            self._login_completed.cancel()
+            self._login_completed = asyncio.get_running_loop().create_future()
+
         self._monitor_task = get_loop().create_task(self._monitor())
 
+        await self._login_completed
+
     @classmethod
     def create_tls(cls, hostname, keyfile, certfile, ca_certs, port=LEAP_PORT):
         """Initialize the Smart Bridge using TLS over IPv4."""
@@ -72,7 +117,7 @@ class Smartbridge:
 
         return cls(_connect)
 
-    def add_subscriber(self, device_id, callback_):
+    def add_subscriber(self, device_id: str, callback_: Callable[[], None]):
         """
         Add a listener to be notified of state changes.
 
@@ -81,7 +126,9 @@ class Smartbridge:
         """
         self._subscribers[device_id] = callback_
 
-    def add_occupancy_subscriber(self, occupancy_group_id, callback_):
+    def add_occupancy_subscriber(
+        self, occupancy_group_id: str, callback_: Callable[[], None]
+    ):
         """
         Add a listener to be notified of occupancy state changes.
 
@@ -90,42 +137,34 @@ class Smartbridge:
         """
         self._occupancy_subscribers[occupancy_group_id] = callback_
 
-    def get_devices(self):
+    def get_devices(self) -> Dict[str, dict]:
         """Will return all known devices connected to the Smart Bridge."""
         return self.devices
 
-    def get_devices_by_domain(self, domain):
+    def get_devices_by_domain(self, domain: str) -> List[dict]:
         """
         Return a list of devices for the given domain.
 
         :param domain: one of 'light', 'switch', 'cover', 'fan' or 'sensor'
         :returns list of zero or more of the devices
         """
-        devs = []
+        types = _LEAP_DEVICE_TYPES.get(domain, None)
 
         # return immediately if not a supported domain
-        if domain not in _LEAP_DEVICE_TYPES:
-            return devs
+        if types is None:
+            return []
 
-        # loop over all devices and check their type
-        for device_id in self.devices:
-            if self.devices[device_id]["type"] in _LEAP_DEVICE_TYPES[domain]:
-                devs.append(self.devices[device_id])
-        return devs
+        return self.get_devices_by_types(types)
 
-    def get_devices_by_type(self, type_):
+    def get_devices_by_type(self, type_: str) -> List[dict]:
         """
         Will return all devices of a given device type.
 
         :param type_: LEAP device type, e.g. WallSwitch
         """
-        devs = []
-        for device_id in self.devices:
-            if self.devices[device_id]["type"] == type_:
-                devs.append(self.devices[device_id])
-        return devs
+        return [device for device in self.devices.values() if device["type"] == type_]
 
-    def get_device_by_zone_id(self, zone_id):
+    def get_device_by_zone_id(self, zone_id: str) -> dict:
         """
         Return the first device associated with a given zone.
 
@@ -137,21 +176,17 @@ class Smartbridge:
         for device in self.devices.values():
             if zone_id == device.get("zone"):
                 return device
-        raise KeyError("No device associated with zone {}".format(zone_id))
+        raise KeyError(f"No device associated with zone {zone_id}")
 
-    def get_devices_by_types(self, types):
+    def get_devices_by_types(self, types: List[str]) -> List[dict]:
         """
-        Will return all devices of for a list of given device types.
+        Will return all devices for a list of given device types.
 
         :param types: list of LEAP device types such as WallSwitch, WallDimmer
         """
-        devs = []
-        for device_id in self.devices:
-            if self.devices[device_id]["type"] in types:
-                devs.append(self.devices[device_id])
-        return devs
+        return [device for device in self.devices.values() if device["type"] in types]
 
-    def get_device_by_id(self, device_id):
+    def get_device_by_id(self, device_id: str) -> dict:
         """
         Will return a device with the given ID.
 
@@ -159,11 +194,11 @@ class Smartbridge:
         """
         return self.devices[device_id]
 
-    def get_scenes(self):
+    def get_scenes(self) -> Dict[str, dict]:
         """Will return all known scenes from the Smart Bridge."""
         return self.scenes
 
-    def get_scene_by_id(self, scene_id):
+    def get_scene_by_id(self, scene_id: str) -> dict:
         """
         Will return a scene with the given scene ID.
 
@@ -171,11 +206,11 @@ class Smartbridge:
         """
         return self.scenes[scene_id]
 
-    def is_connected(self):
+    def is_connected(self) -> bool:
         """Will return True if currently connected to the Smart Bridge."""
         return self.logged_in
 
-    def is_on(self, device_id):
+    def is_on(self, device_id: str) -> bool:
         """
         Will return True is the device with the given ID is 'on'.
 
@@ -187,7 +222,24 @@ class Smartbridge:
             or (self.devices[device_id]["fan_speed"] or FAN_OFF) != FAN_OFF
         )
 
-    def set_value(self, device_id, value):
+    async def _request(
+        self, communique_type: str, url: str, body: Optional[dict] = None
+    ) -> Response:
+        if self._leap is None:
+            raise BridgeDisconnectedError()
+
+        response = await asyncio.wait_for(
+            self._leap.request(communique_type, url, body),
+            timeout=REQUEST_TIMEOUT,
+        )
+
+        status = response.Header.StatusCode
+        if status is None or not status.is_successful():
+            raise BridgeResponseError(response)
+
+        return response
+
+    async def set_value(self, device_id: str, value: int):
         """
         Will set the value for a device with the given ID.
 
@@ -195,52 +247,52 @@ class Smartbridge:
         :param value: integer value from 0 to 100 to set
         """
         zone_id = self._get_zone_id(device_id)
-        if zone_id:
-            cmd = {
-                "CommuniqueType": "CreateRequest",
-                "Header": {"Url": "/zone/%s/commandprocessor" % zone_id},
-                "Body": {
-                    "Command": {
-                        "CommandType": "GoToLevel",
-                        "Parameter": [{"Type": "Level", "Value": value}],
-                    }
-                },
-            }
-            self._writer.write(cmd)
+        if not zone_id:
+            return
 
-    def _send_zone_create_request(self, device_id, command):
+        await self._request(
+            "CreateRequest",
+            f"/zone/{zone_id}/commandprocessor",
+            {
+                "Command": {
+                    "CommandType": "GoToLevel",
+                    "Parameter": [{"Type": "Level", "Value": value}],
+                }
+            },
+        )
+
+    async def _send_zone_create_request(self, device_id: str, command: str):
         zone_id = self._get_zone_id(device_id)
         if not zone_id:
             return
-        self._writer.write(
-            {
-                "CommuniqueType": "CreateRequest",
-                "Header": {"Url": "/zone/%s/commandprocessor" % zone_id},
-                "Body": {"Command": {"CommandType": command}},
-            }
+
+        await self._request(
+            "CreateRequest",
+            f"/zone/{zone_id}/commandprocessor",
+            {"Command": {"CommandType": command}},
         )
 
-    def stop_cover(self, device_id):
+    async def stop_cover(self, device_id: str):
         """Will stop a cover."""
-        self._send_zone_create_request(device_id, "Stop")
+        await self._send_zone_create_request(device_id, "Stop")
 
-    def raise_cover(self, device_id):
+    async def raise_cover(self, device_id: str):
         """Will raise a cover."""
-        self._send_zone_create_request(device_id, "Raise")
+        await self._send_zone_create_request(device_id, "Raise")
         # If set_value is called, we get an optimistic callback right
         # away with the value, if we use Raise we have to set it
         # as one won't come unless Stop is called or something goes wrong.
         self.devices[device_id]["current_state"] = 100
 
-    def lower_cover(self, device_id):
+    async def lower_cover(self, device_id: str):
         """Will lower a cover."""
-        self._send_zone_create_request(device_id, "Lower")
+        await self._send_zone_create_request(device_id, "Lower")
         # If set_value is called, we get an optimistic callback right
         # away with the value, if we use Lower we have to set it
         # as one won't come unless Stop is called or something goes wrong.
         self.devices[device_id]["current_state"] = 0
 
-    def set_fan(self, device_id, value):
+    async def set_fan(self, device_id: str, value: str):
         """
         Will set the value for a fan device with the given device ID.
 
@@ -250,49 +302,47 @@ class Smartbridge:
         """
         zone_id = self._get_zone_id(device_id)
         if zone_id:
-            cmd = {
-                "CommuniqueType": "CreateRequest",
-                "Header": {"Url": "/zone/%s/commandprocessor" % zone_id},
-                "Body": {
+            await self._request(
+                "CreateRequest",
+                f"/zone/{zone_id}/commandprocessor",
+                {
                     "Command": {
                         "CommandType": "GoToFanSpeed",
                         "FanSpeedParameters": {"FanSpeed": value},
                     }
                 },
-            }
-            self._writer.write(cmd)
+            )
 
-    def turn_on(self, device_id):
+    async def turn_on(self, device_id: str):
         """
         Will turn 'on' the device with the given ID.
 
         :param device_id: device id to turn on
         """
-        return self.set_value(device_id, 100)
+        await self.set_value(device_id, 100)
 
-    def turn_off(self, device_id):
+    async def turn_off(self, device_id: str):
         """
         Will turn 'off' the device with the given ID.
 
         :param device_id: device id to turn off
         """
-        return self.set_value(device_id, 0)
+        await self.set_value(device_id, 0)
 
-    def activate_scene(self, scene_id):
+    async def activate_scene(self, scene_id: str):
         """
         Will activate the scene with the given ID.
 
         :param scene_id: scene id, e.g. 23
         """
         if scene_id in self.scenes:
-            cmd = {
-                "CommuniqueType": "CreateRequest",
-                "Header": {"Url": "/virtualbutton/%s/commandprocessor" % scene_id},
-                "Body": {"Command": {"CommandType": "PressAndRelease"}},
-            }
-            self._writer.write(cmd)
+            await self._request(
+                "CreateRequest",
+                f"/virtualbutton/{scene_id}/commandprocessor",
+                {"Command": {"CommandType": "PressAndRelease"}},
+            )
 
-    def _get_zone_id(self, device_id):
+    def _get_zone_id(self, device_id: str) -> Optional[str]:
         """
         Return the zone id for an given device.
 
@@ -300,47 +350,67 @@ class Smartbridge:
         """
         return self.devices[device_id].get("zone")
 
-    def _send_command(self, cmd):
-        """Send a command to the bridge."""
-        self._writer.write(cmd)
-
     async def _monitor(self):
         """Event monitoring loop."""
         try:
             while True:
-                try:
-                    await asyncio.wait_for(self._login(), timeout=CONNECT_TIMEOUT)
-                    received = await self._reader.read()
-                    _LOG.debug("received LEAP: %s", received)
-                    if received is not None:
-                        self._handle_response(received)
-                # ignore OSError too.
-                # sometimes you get OSError instead of ConnectionError.
-                except (ValueError, ConnectionError, OSError, asyncio.TimeoutError):
-                    _LOG.warning("reconnecting", exc_info=1)
-                    await asyncio.sleep(RECONNECT_DELAY)
+                await self._monitor_once()
         except asyncio.CancelledError:
             pass
-        except Exception:
+        except Exception as ex:
             _LOG.critical("monitor loop has exited", exc_info=1)
+            if not self._login_completed.done():
+                self._login_completed.set_exception(ex)
             raise
+        finally:
+            self._login_completed.cancel()
 
-    def _handle_response(self, resp_json):
-        """
-        Handle an event from the ssl interface.
-
-        If a zone level was changed either by external means such as a Pico
-        remote or by a command sent from us, the new level will appear on the
-        reader and the response is handled by this function.
-
-        :param resp_json: full JSON response from the LEAP connection
-        """
-        comm_type = resp_json["CommuniqueType"]
-        if comm_type == "ReadResponse":
-            self._handle_read_response(resp_json)
+    async def _monitor_once(self):
+        """Monitor for events until an error occurs."""
+        try:
+            _LOG.debug("Connecting to Smart Bridge via SSL")
+            self._leap = await self._connect()
+            self._leap.subscribe_unsolicited(self._handle_unsolicited)
+            _LOG.debug("Successfully connected to Smart Bridge.")
 
-    def _handle_one_zone_status(self, resp_json):
-        body = resp_json["Body"]
+            if self._login_task is not None:
+                self._login_task.cancel()
+
+            if self._ping_task is not None:
+                self._ping_task.cancel()
+
+            self._login_task = asyncio.get_running_loop().create_task(self._login())
+            self._ping_task = asyncio.get_running_loop().create_task(self._ping())
+
+            await self._leap.run()
+            _LOG.warning("LEAP session ended. Reconnecting...")
+            await asyncio.sleep(RECONNECT_DELAY)
+        # ignore OSError too.
+        # sometimes you get OSError instead of ConnectionError.
+        except (
+            ValueError,
+            ConnectionError,
+            OSError,
+            asyncio.TimeoutError,
+            BridgeDisconnectedError,
+        ):
+            _LOG.warning("Reconnecting...", exc_info=1)
+            await asyncio.sleep(RECONNECT_DELAY)
+        finally:
+            if self._login_task is not None:
+                self._login_task.cancel()
+                self._login_task = None
+
+            if self._ping_task is not None:
+                self._ping_task.cancel()
+                self._ping_task = None
+
+            if self._leap is not None:
+                self._leap.close()
+                self._leap = None
+
+    def _handle_one_zone_status(self, response: Response):
+        body = response.Body
         status = body["ZoneStatus"]
         zone = id_from_href(status["Zone"]["href"])
         level = status.get("Level", -1)
@@ -352,19 +422,16 @@ class Smartbridge:
         if device["device_id"] in self._subscribers:
             self._subscribers[device["device_id"]]()
 
-    def _handle_one_ping_response(self, _):
-        self._got_ping.set()
-
-    def _handle_occupancy_group_status(self, resp_json):
-        _LOG.debug("Handling occupancy group status: %s", resp_json)
-        statuses = resp_json.get("Body", {}).get("OccupancyGroupStatuses", {})
+    def _handle_occupancy_group_status(self, response: Response):
+        _LOG.debug("Handling occupancy group status: %s", response)
+        statuses = response.Body.get("OccupancyGroupStatuses", {})
         for status in statuses:
             occgroup_id = id_from_href(status["OccupancyGroup"]["href"])
             ostat = status["OccupancyStatus"]
             if occgroup_id not in self.occupancy_groups:
                 if ostat != OCCUPANCY_GROUP_UNKNOWN:
                     _LOG.warning(
-                        "Occupancy group %s has a status but no " "sensors", occgroup_id
+                        "Occupancy group %s has a status but no sensors", occgroup_id
                     )
                 continue
             if ostat == OCCUPANCY_GROUP_UNKNOWN:
@@ -378,85 +445,68 @@ class Smartbridge:
 
     _read_response_handler_callbacks = dict(
         OneZoneStatus=_handle_one_zone_status,
-        OnePingResponse=_handle_one_ping_response,
         MultipleOccupancyGroupStatus=_handle_occupancy_group_status,
     )
 
-    def _handle_read_response(self, resp_json):
-        body_type = resp_json.get("Header", {}).get("MessageBodyType")
-        if body_type in self._read_response_handler_callbacks:
-            self._read_response_handler_callbacks[body_type](self, resp_json)
+    def _handle_read_response(self, response: Response):
+        if response.Header.MessageBodyType is None:
+            return
+        callback = self._read_response_handler_callbacks.get(
+            response.Header.MessageBodyType, None
+        )
+        if callback is not None:
+            callback(self, response)
+
+    def _handle_unsolicited(self, response: Response):
+        if response.CommuniqueType == "ReadResponse":
+            self._handle_read_response(response)
 
     async def _login(self):
         """Connect and login to the Smart Bridge LEAP server using SSL."""
-        async with self._login_lock:
-            if self._reader is not None and self._writer is not None:
-                if (
-                    self.logged_in
-                    and self._reader.exception() is None
-                    and not self._reader.at_eof()
-                ):
-                    return
-                self._writer.abort()
-                self._reader = self._writer = None
-
-            self.logged_in = False
-            _LOG.debug("Connecting to Smart Bridge via SSL")
-            self._reader, self._writer = await self._connect()
-            _LOG.debug("Successfully connected to Smart Bridge.")
+        await self._load_devices()
+        await self._load_scenes()
+        await self._load_areas()
+        await self._load_occupancy_groups()
+        await self._subscribe_to_occupancy_groups()
 
-            await self._load_devices()
-            await self._load_scenes()
-            await self._load_areas()
-            await self._load_occupancy_groups()
-            await self._subscribe_to_occupancy_groups()
+        try:
             for device in self.devices.values():
                 if device.get("zone") is not None:
                     _LOG.debug("Requesting zone information from %s", device)
-                    cmd = {
-                        "CommuniqueType": "ReadRequest",
-                        "Header": {"Url": "/zone/%s/status" % device["zone"]},
-                    }
-                    self._writer.write(cmd)
-            self._ping_task = get_loop().create_task(self._ping())
-            self.logged_in = True
+                    response = await self._request(
+                        "ReadRequest", f"/zone/{device['zone']}/status"
+                    )
+                    self._handle_read_response(response)
+
+            if not self._login_completed.done():
+                self._login_completed.set_result(None)
+        except asyncio.CancelledError:
+            pass
+        except Exception as ex:
+            self._login_completed.set_exception(ex)
+            raise
 
     async def _ping(self):
         """Periodically ping the LEAP server to keep the connection open."""
-        writer = self._writer
         try:
-            try:
-                while True:
-                    await asyncio.sleep(PING_INTERVAL)
-                    self._got_ping.clear()
-                    writer.write(
-                        {
-                            "CommuniqueType": "ReadRequest",
-                            "Header": {"Url": "/server/1/status/ping"},
-                        }
-                    )
-                    await asyncio.wait_for(self._got_ping.wait(), PING_DELAY)
-            except asyncio.TimeoutError:
-                _LOG.warning("ping was not answered. closing connection.")
-                writer.abort()
-            except ConnectionError:
-                _LOG.warning("ping failed. closing connection.")
-                writer.abort()
+            while True:
+                await asyncio.sleep(PING_INTERVAL)
+                await self._request("ReadRequest", "/server/1/status/ping")
+        except asyncio.TimeoutError:
+            _LOG.warning("ping was not answered. closing connection.")
+            self._leap.close()
         except asyncio.CancelledError:
             pass
         except Exception:
-            _LOG.error("ping failed. closing connection.", exc_info=1)
-            writer.abort()
+            _LOG.warning("ping failed. closing connection.", exc_info=1)
+            self._leap.close()
             raise
 
     async def _load_devices(self):
         """Load the device list from the SSL LEAP server interface."""
         _LOG.debug("Loading devices")
-        self._writer.write(
-            {"CommuniqueType": "ReadRequest", "Header": {"Url": "/device"}}
-        )
-        device_json = await self._reader.wait_for("ReadResponse")
-        for device in device_json["Body"]["Devices"]:
+        device_json = await self._request("ReadRequest", "/device")
+        for device in device_json.Body["Devices"]:
             _LOG.debug(device)
             device_id = id_from_href(device["href"])
             device_zone = None
@@ -481,11 +531,8 @@ class Smartbridge:
         Scenes are known as virtual buttons in the SSL LEAP interface.
         """
         _LOG.debug("Loading scenes from the Smart Bridge")
-        self._writer.write(
-            {"CommuniqueType": "ReadRequest", "Header": {"Url": "/virtualbutton"}}
-        )
-        scene_json = await self._reader.wait_for("ReadResponse")
-        for scene in scene_json["Body"]["VirtualButtons"]:
+        scene_json = await self._request("ReadRequest", "/virtualbutton")
+        for scene in scene_json.Body["VirtualButtons"]:
             _LOG.debug(scene)
             # If 'Name' is not a key in scene, then it is likely a scene pico
             # vbutton. For now, simply ignore these scenes.
@@ -497,9 +544,8 @@ class Smartbridge:
     async def _load_areas(self):
         """Load the areas from the Smart Bridge."""
         _LOG.debug("Loading areas from the Smart Bridge")
-        self._writer.write(dict(CommuniqueType="ReadRequest", Header=dict(Url="/area")))
-        area_json = await self._reader.wait_for("ReadResponse")
-        for area in area_json["Body"]["Areas"]:
+        area_json = await self._request("ReadRequest", "/area")
+        for area in area_json.Body["Areas"]:
             area_id = id_from_href(area["href"])
             # We currently only need the name, so just load that
             self.areas[area_id] = dict(name=area["Name"])
@@ -507,11 +553,8 @@ class Smartbridge:
     async def _load_occupancy_groups(self):
         """Load the occupancy groups from the Smart Bridge."""
         _LOG.debug("Loading occupancy groups from the Smart Bridge")
-        self._writer.write(
-            dict(CommuniqueType="ReadRequest", Header=dict(Url="/occupancygroup"))
-        )
-        occgroup_json = await self._reader.wait_for("ReadResponse")
-        occgroups = occgroup_json.get("Body", {}).get("OccupancyGroups", {})
+        occgroup_json = await self._request("ReadRequest", "/occupancygroup")
+        occgroups = occgroup_json.Body.get("OccupancyGroups", {})
         for occgroup in occgroups:
             self._process_occupancy_group(occgroup)
 
@@ -553,17 +596,11 @@ class Smartbridge:
     async def _subscribe_to_occupancy_groups(self):
         """Subscribe to occupancy group status updates."""
         _LOG.debug("Subscribing to occupancy group status updates")
-        self._writer.write(
-            dict(
-                CommuniqueType="SubscribeRequest",
-                Header=dict(Url="/occupancygroup/status"),
-            )
-        )
-        response = await self._reader.wait_for("SubscribeResponse")
-        if response["Header"]["StatusCode"].startswith("20"):
+        try:
+            response = await self._request("SubscribeRequest", "/occupancygroup/status")
             _LOG.debug("Subscribed to occupancygroup status")
-        else:
-            _LOG.error("Failed occupancy subscription: %s", response)
+        except BridgeResponseError as ex:
+            _LOG.error("Failed occupancy subscription: %s", ex.response)
             return
         self._handle_occupancy_group_status(response)
 
@@ -574,5 +611,3 @@ class Smartbridge:
             self._monitor_task.cancel()
         if self._ping_task is not None and not self._ping_task.cancelled():
             self._ping_task.cancel()
-        await self._writer.drain()
-        self._writer.abort()


Test patch:
diff --git a/tests/test_leap.py b/tests/test_leap.py
index accb8c0..5e2182f 100644
--- a/tests/test_leap.py
+++ b/tests/test_leap.py
@@ -1,12 +1,22 @@
-"""Tests to validate ssl interactions."""
+"""Tests to validate low-level network interactions."""
 import asyncio
-from collections import namedtuple
+import json
+from typing import AsyncGenerator, Iterable, NamedTuple, Tuple
 
 import pytest
 
-import pylutron_caseta.leap
+from pylutron_caseta import BridgeDisconnectedError
+from pylutron_caseta.leap import LeapProtocol
+from pylutron_caseta.messages import Response, ResponseHeader, ResponseStatus
 
-Pipe = namedtuple("Pipe", ("leap_reader", "leap_writer", "test_reader", "test_writer"))
+
+class Pipe(NamedTuple):
+    """A LeapProtocol that communicates to a stream reader/writer pair."""
+
+    leap: LeapProtocol
+    leap_loop: asyncio.Task
+    test_reader: asyncio.StreamReader
+    test_writer: asyncio.StreamWriter
 
 
 class _PipeTransport(asyncio.Transport):
@@ -21,7 +31,7 @@ class _PipeTransport(asyncio.Transport):
         self._closing = True
         self.other.get_protocol().connection_lost(None)
 
-    def is_closing(self):
+    def is_closing(self) -> bool:
         return self._closing
 
     def pause_reading(self):
@@ -33,43 +43,45 @@ class _PipeTransport(asyncio.Transport):
     def abort(self):
         self.close()
 
-    def can_write_eof(self):
+    def can_write_eof(self) -> bool:
         return False
 
-    def get_write_buffer_size(self):
+    def get_write_buffer_size(self) -> int:
         return 0
 
-    def get_write_buffer_limits(self):  # pylint: disable=no-self-use
+    def get_write_buffer_limits(self) -> Tuple[int, int]:  # pylint: disable=no-self-use
         """Return (0, 0)."""
         return (0, 0)
 
     def set_write_buffer_limits(self, high=None, low=None):
         raise NotImplementedError()
 
-    def write(self, data):
+    def write(self, data: bytes):
         self.other.get_protocol().data_received(data)
 
-    def writelines(self, list_of_data):
+    def writelines(self, list_of_data: Iterable[bytes]):
         for line in list_of_data:
             self.write(line)
 
     def write_eof(self):
         raise NotImplementedError()
 
-    def set_protocol(self, protocol):
+    def set_protocol(self, protocol: asyncio.BaseProtocol):
         self._protocol = protocol
 
-    def get_protocol(self):
+    def get_protocol(self) -> asyncio.BaseProtocol:
         return self._protocol
 
 
 @pytest.fixture(name="pipe")
-def fixture_pipe(event_loop):
+async def fixture_pipe(
+    event_loop: asyncio.AbstractEventLoop,
+) -> AsyncGenerator[Pipe, None]:
     """Create linked readers and writers for tests."""
-    test_reader = asyncio.StreamReader(loop=event_loop)
-    impl_reader = asyncio.StreamReader(loop=event_loop)
-    test_protocol = asyncio.StreamReaderProtocol(test_reader, loop=event_loop)
-    impl_protocol = asyncio.StreamReaderProtocol(impl_reader, loop=event_loop)
+    test_reader = asyncio.StreamReader()
+    impl_reader = asyncio.StreamReader()
+    test_protocol = asyncio.StreamReaderProtocol(test_reader)
+    impl_protocol = asyncio.StreamReaderProtocol(impl_reader)
     test_pipe = _PipeTransport()
     impl_pipe = _PipeTransport()
     test_pipe.other = impl_pipe
@@ -79,54 +91,124 @@ def fixture_pipe(event_loop):
     test_protocol.connection_made(test_pipe)
     impl_protocol.connection_made(impl_pipe)
     test_writer = asyncio.StreamWriter(
-        test_pipe, test_protocol, test_reader, loop=event_loop
+        test_pipe, test_protocol, test_reader, event_loop
     )
     impl_writer = asyncio.StreamWriter(
-        impl_pipe, impl_protocol, impl_reader, loop=event_loop
+        impl_pipe, impl_protocol, impl_reader, event_loop
     )
-    leap_reader = pylutron_caseta.leap.LeapReader(impl_reader)
-    leap_writer = pylutron_caseta.leap.LeapWriter(impl_writer)
-    return Pipe(leap_reader, leap_writer, test_reader, test_writer)
+
+    leap = LeapProtocol(impl_reader, impl_writer)
+    leap_task = asyncio.create_task(leap.run())
+
+    yield Pipe(leap, leap_task, test_reader, test_writer)
+
+    leap_task.cancel()
 
 
 @pytest.mark.asyncio
-async def test_read(pipe):
-    """Test basic object reading."""
-    pipe.test_writer.write(b'{"test": true}\r\n')
-    result = await pipe.leap_reader.read()
-    assert result == {"test": True}
+async def test_call(pipe: Pipe):
+    """Test basic call and response."""
+    task = asyncio.create_task(pipe.leap.request("ReadRequest", "/test"))
+
+    received = json.loads((await pipe.test_reader.readline()).decode("utf-8"))
+
+    # message should contain ClientTag
+    tag = received.get("Header", {}).pop("ClientTag", None)
+    assert tag
+    assert isinstance(tag, str)
+
+    assert received == {"CommuniqueType": "ReadRequest", "Header": {"Url": "/test"}}
+
+    response_obj = {
+        "CommuniqueType": "ReadResponse",
+        "Header": {"ClientTag": tag, "StatusCode": "200 OK", "Url": "/test"},
+        "Body": {"ok": True},
+    }
+    response_bytes = f"{json.dumps(response_obj)}\r\n".encode("utf-8")
+    pipe.test_writer.write(response_bytes)
+
+    result = await task
+
+    assert result == Response(
+        Header=ResponseHeader(StatusCode=ResponseStatus(200, "OK"), Url="/test"),
+        CommuniqueType="ReadResponse",
+        Body={"ok": True},
+    )
 
 
 @pytest.mark.asyncio
 async def test_read_eof(pipe):
     """Test reading when EOF is encountered."""
     pipe.test_writer.close()
-    result = await pipe.leap_reader.read()
-    assert result is None
+
+    await pipe.leap_loop
 
 
 @pytest.mark.asyncio
 async def test_read_invalid(pipe):
     """Test reading when invalid data is received."""
-    pipe.test_writer.write(b"?")
-    pipe.test_writer.close()
-    with pytest.raises(ValueError):
-        await pipe.leap_reader.read()
+    pipe.test_writer.write(b"?\r\n")
+
+    with pytest.raises(json.JSONDecodeError):
+        await pipe.leap_loop
 
 
 @pytest.mark.asyncio
-async def test_write(pipe):
-    """Test basic object writing."""
-    pipe.leap_writer.write({"test": True})
-    result = await pipe.test_reader.readline()
-    assert result == b'{"test": true}\r\n'
+async def test_busy_close(pipe):
+    """Test closing the session while there are in-flight requests."""
+    task = asyncio.create_task(pipe.leap.request("ReadRequest", "/test"))
+
+    await pipe.test_reader.readline()
+    pipe.test_writer.close()
+    await pipe.leap_loop
+    pipe.leap.close()
+
+    with pytest.raises(BridgeDisconnectedError):
+        await task
 
 
 @pytest.mark.asyncio
-async def test_wait_for(pipe):
-    """Test the wait_for method."""
-    pipe.test_writer.write(b'{"test": true}\r\n')
-    pipe.test_writer.write(b'{"CommuniqueType": "TheAnswerIs42"}\r\n')
-    pipe.test_writer.write(b'{"CommuniqueType": "ReadRequest", ' b'"foo": "bar"}\r\n')
-    result = await pipe.leap_reader.wait_for("ReadRequest")
-    assert result == {"CommuniqueType": "ReadRequest", "foo": "bar"}
+async def test_unsolicited(pipe):
+    """Test subscribing and unsubscribing unsolicited message handlers."""
+    handler1_message = None
+    handler2_message = None
+    handler2_called = asyncio.Event()
+
+    def handler1(response):
+        nonlocal handler1_message
+        handler1_message = response
+
+    def handler2(response):
+        nonlocal handler2_message
+        handler2_message = response
+        handler2_called.set()
+
+    pipe.leap.subscribe_unsolicited(handler1)
+    pipe.leap.subscribe_unsolicited(handler2)
+
+    response_dict = {
+        "CommuniqueType": "ReadResponse",
+        "Header": {"StatusCode": "200 OK", "Url": "/test"},
+        "Body": {"Index": 0},
+    }
+    response_bytes = f"{json.dumps(response_dict)}\r\n".encode("utf-8")
+    pipe.test_writer.write(response_bytes)
+    response = Response.from_json(response_dict)
+
+    await asyncio.wait_for(handler2_called.wait(), 1.0)
+    handler2_called.clear()
+
+    assert handler1_message == response, "handler1 did not receive correct message"
+    assert handler2_message == response, "handler2 did not receive correct message"
+
+    pipe.leap.unsubscribe_unsolicited(handler1)
+
+    response_dict["Body"]["Index"] = 1
+    response_bytes = f"{json.dumps(response_dict)}\r\n".encode("utf-8")
+    pipe.test_writer.write(response_bytes)
+    response = Response.from_json(response_dict)
+
+    await asyncio.wait_for(handler2_called.wait(), 1.0)
+
+    assert handler1_message != response, "handler1 was not unsubscribed"
+    assert handler2_message == response, "handler2 did not receive correct message"
diff --git a/tests/test_smartbridge.py b/tests/test_smartbridge.py
index 0a39f3d..aa4bacc 100644
--- a/tests/test_smartbridge.py
+++ b/tests/test_smartbridge.py
@@ -3,30 +3,103 @@ import asyncio
 import json
 import logging
 import os
-import pytest
+from typing import (
+    AsyncGenerator,
+    Awaitable,
+    Callable,
+    List,
+    NamedTuple,
+    Optional,
+    TypeVar,
+    TYPE_CHECKING,
+)
 
-try:
-    from asyncio import get_running_loop as get_loop
-except ImportError:
-    # get_running_loop is better, but it was introduced in Python 3.7
-    from asyncio import get_event_loop as get_loop
+import pytest
 
+from pylutron_caseta.messages import Response, ResponseHeader, ResponseStatus
 import pylutron_caseta.smartbridge as smartbridge
 from pylutron_caseta import (
     FAN_MEDIUM,
     OCCUPANCY_GROUP_OCCUPIED,
     OCCUPANCY_GROUP_UNOCCUPIED,
+    BridgeDisconnectedError,
 )
 
+if TYPE_CHECKING:
+    from typing import Tuple
+
 logging.getLogger().setLevel(logging.DEBUG)
-logging.getLogger().addHandler(logging.StreamHandler())
+_LOG = logging.getLogger(__name__)
 
 
-def response_from_json_file(filename):
+def response_from_json_file(filename: str) -> Response:
     """Fetch a response from a saved JSON file."""
     responsedir = os.path.join(os.path.split(__file__)[0], "responses")
     with open(os.path.join(responsedir, filename), "r") as ifh:
-        return json.load(ifh)
+        return Response.from_json(json.load(ifh))
+
+
+class Request(NamedTuple):
+    """An in-flight LEAP request."""
+
+    communique_type: str
+    url: str
+    body: Optional[dict] = None
+
+
+class _FakeLeap:
+    def __init__(self):
+        self.requests: "asyncio.Queue[Tuple[Request, asyncio.Future[Response]]]" = (
+            asyncio.Queue()
+        )
+        self.running = None
+        self._unsolicited: List[Callable[[Response], None]] = []
+
+    async def request(
+        self, communique_type: str, url: str, body: Optional[dict] = None
+    ) -> dict:
+        """Make a request to the bridge and return the response."""
+        future: asyncio.Future = asyncio.get_running_loop().create_future()
+        obj = Request(communique_type=communique_type, url=url, body=body)
+
+        await self.requests.put((obj, future))
+
+        return await future
+
+    async def run(self):
+        """Event monitoring loop."""
+        self.running = asyncio.get_running_loop().create_future()
+        await self.running
+
+    def subscribe_unsolicited(self, callback: Callable[[Response], None]):
+        """Subscribe to unsolicited responses."""
+        if not callable(callback):
+            raise TypeError("callback must be callable")
+        self._unsolicited.append(callback)
+
+    def unsubscribe_unsolicited(self, callback: Callable[[Response], None]):
+        """Unsubscribe from unsolicited responses."""
+        self._unsolicited.remove(callback)
+
+    def send_unsolicited(self, response: Response):
+        """Send an unsolicited response message to SmartBridge."""
+        for handler in self._unsolicited:
+            handler(response)
+
+    def close(self):
+        """Disconnect."""
+        if self.running is not None and not self.running.done():
+            self.running.set_result(None)
+            self.running = None
+
+        while not self.requests.empty():
+            (_, response) = self.requests.get_nowait()
+            if not response.done():
+                response.set_exception(BridgeDisconnectedError())
+            self.requests.task_done()
+
+
+T = TypeVar("T")
 
 
 class Bridge:
@@ -34,222 +107,140 @@ class Bridge:
 
     def __init__(self):
         """Create a new Bridge in a disconnected state."""
-        self._connections = None
-        self.reader = self.writer = None
+        self.connections = asyncio.Queue()
+        self.leap: _FakeLeap = None
 
         async def fake_connect():
-            """Use by SmartBridge to connect to the test."""
-            closed = asyncio.Event()
-            reader = _FakeLeapReader(closed, get_loop())
-            writer = _FakeLeapWriter(closed, get_loop())
-            await self.connections.put((reader, writer))
-            return (reader, writer)
+            """Open a fake LEAP connection for the test."""
+            leap = _FakeLeap()
+            await self.connections.put(leap)
+            return leap
 
         self.target = smartbridge.Smartbridge(fake_connect)
 
-    @property
-    def connections(self):
-        """Defer creating the connection queue until we are in a loop."""
-        if self._connections is None:
-            self._connections = asyncio.Queue()
-        return self._connections
-
     async def initialize(self):
         """Perform the initial connection with SmartBridge."""
-        connect_task = get_loop().create_task(self.target.connect())
-        reader, writer = await self.connections.get()
+        connect_task = asyncio.get_running_loop().create_task(self.target.connect())
+        fake_leap = await self.connections.get()
 
-        async def wait(coro):
+        async def wait(coro: Awaitable[T]) -> T:
             # abort if SmartBridge reports it has finished connecting early
-            task = get_loop().create_task(coro)
-            done, _ = await asyncio.wait(
+            task = asyncio.get_running_loop().create_task(coro)
+            race = await asyncio.wait(
                 (connect_task, task), timeout=10, return_when=asyncio.FIRST_COMPLETED
             )
+            done, _ = race
             assert len(done) > 0, "operation timed out"
             if len(done) == 1 and connect_task in done:
                 raise connect_task.exception()
             result = await task
             return result
 
-        await self._accept_connection(reader, writer, wait)
+        await self._accept_connection(fake_leap, wait)
         await connect_task
 
-        self.reader = reader
-        self.writer = writer
+        self.leap = fake_leap
         self.connections.task_done()
 
-    async def _accept_connection(self, reader, writer, wait):
+    async def _accept_connection(self, leap, wait):
         """Accept a connection from SmartBridge (implementation)."""
         # First message should be read request on /device
-        value = await wait(writer.queue.get())
-        assert value == {"CommuniqueType": "ReadRequest", "Header": {"Url": "/device"}}
-        writer.queue.task_done()
-        await reader.write(response_from_json_file("devices.json"))
+        request, response = await wait(leap.requests.get())
+        assert request == Request(communique_type="ReadRequest", url="/device")
+        response.set_result(response_from_json_file("devices.json"))
+        leap.requests.task_done()
+
         # Second message should be read request on /virtualbutton
-        value = await wait(writer.queue.get())
-        assert value == {
-            "CommuniqueType": "ReadRequest",
-            "Header": {"Url": "/virtualbutton"},
-        }
-        writer.queue.task_done()
-        await reader.write(response_from_json_file("scenes.json"))
+        request, response = await wait(leap.requests.get())
+        assert request == Request(communique_type="ReadRequest", url="/virtualbutton")
+        response.set_result(response_from_json_file("scenes.json"))
+        leap.requests.task_done()
+
         # Third message should be read request on /areas
-        value = await wait(writer.queue.get())
-        assert value == {"CommuniqueType": "ReadRequest", "Header": {"Url": "/area"}}
-        writer.queue.task_done()
-        await reader.write(response_from_json_file("areas.json"))
+        request, response = await wait(leap.requests.get())
+        assert request == Request(communique_type="ReadRequest", url="/area")
+        response.set_result(response_from_json_file("areas.json"))
+        leap.requests.task_done()
+
         # Fourth message should be read request on /occupancygroup
-        value = await wait(writer.queue.get())
-        assert value == {
-            "CommuniqueType": "ReadRequest",
-            "Header": {"Url": "/occupancygroup"},
-        }
-        writer.queue.task_done()
-        await reader.write(response_from_json_file("occupancygroups.json"))
+        request, response = await wait(leap.requests.get())
+        assert request == Request(communique_type="ReadRequest", url="/occupancygroup")
+        response.set_result(response_from_json_file("occupancygroups.json"))
+        leap.requests.task_done()
+
         # Fifth message should be subscribe request on /occupancygroup/status
-        value = await wait(writer.queue.get())
-        assert value == {
-            "CommuniqueType": "SubscribeRequest",
-            "Header": {"Url": "/occupancygroup/status"},
-        }
-        writer.queue.task_done()
-        await reader.write(response_from_json_file("occupancygroupsubscribe.json"))
+        request, response = await wait(leap.requests.get())
+        assert request == Request(
+            communique_type="SubscribeRequest", url="/occupancygroup/status"
+        )
+        response.set_result(response_from_json_file("occupancygroupsubscribe.json"))
+        leap.requests.task_done()
+
         # Finally, we should check the zone status on each zone
         requested_zones = []
         for _ in range(0, 3):
-            value = await wait(writer.queue.get())
-            logging.info("Read %s", value)
-            assert value["CommuniqueType"] == "ReadRequest"
-            requested_zones.append(value["Header"]["Url"])
-            writer.queue.task_done()
+            request, response = await wait(leap.requests.get())
+            logging.info("Read %s", request)
+            assert request.communique_type == "ReadRequest"
+            requested_zones.append(request.url)
+            response.set_result(
+                Response(
+                    CommuniqueType="ReadResponse",
+                    Header=ResponseHeader(
+                        MessageBodyType="OneZoneStatus",
+                        StatusCode=ResponseStatus(200, "OK"),
+                        Url=request.url,
+                    ),
+                    Body={
+                        "ZoneStatus": {
+                            "href": request.url,
+                            "Level": 0,
+                            "Zone": {"href": request.url.replace("/status", "")},
+                            "StatusAccuracy": "Good",
+                        }
+                    },
+                )
+            )
+            leap.requests.task_done()
         requested_zones.sort()
         assert requested_zones == ["/zone/1/status", "/zone/2/status", "/zone/6/status"]
 
-    async def disconnect(self, exception=None):
+    def disconnect(self, exception=None):
         """Disconnect SmartBridge."""
-        await self.reader.end(exception)
+        if exception is None:
+            self.leap.running.set_result(None)
+        else:
+            self.leap.running.set_exception(exception)
 
     async def accept_connection(self):
         """Wait for SmartBridge to reconnect."""
-        reader, writer = await self.connections.get()
+        leap = await self.connections.get()
 
         async def wait(coro):
             # nothing special
             result = await coro
             return result
 
-        await self._accept_connection(reader, writer, wait)
+        await self._accept_connection(leap, wait)
 
-        self.reader = reader
-        self.writer = writer
+        self.leap = leap
         self.connections.task_done()
 
 
-class _FakeLeapWriter:
-    """A "Writer" which just puts messages onto a queue."""
-
-    def __init__(self, closed, loop):
-        self.queue = asyncio.Queue()
-        self.closed = closed
-        self._loop = loop
-
-    def write(self, obj):
-        """Send an object to the bridge."""
-        self.queue.put_nowait(obj)
-
-    async def drain(self):
-        """Wait for all objects to be received by the bridge."""
-        task = self._loop.create_task(self.queue.join())
-        await asyncio.wait(
-            (self.closed.wait(), task), return_when=asyncio.FIRST_COMPLETED
-        )
-
-    def abort(self):
-        """Close the connection."""
-        self.closed.set()
-
-
-class _FakeLeapReader:
-    """A "Reader" which just pulls messages from a queue."""
-
-    def __init__(self, closed, loop):
-        self._loop = loop
-        self.closed = closed
-        self.queue = asyncio.Queue()
-        self.exception_value = None
-        self.eof = False
-
-    def exception(self):
-        """Get the exception."""
-        return self.exception_value
-
-    async def read(self):
-        """Read an object from the bridge."""
-        task = self._loop.create_task(self.queue.get())
-        done, _ = await asyncio.wait(
-            (self.closed.wait(), task), return_when=asyncio.FIRST_COMPLETED
-        )
-        if task not in done:
-            return None
-
-        action = await task
-        self._loop.call_soon(self.queue.task_done)
-        try:
-            value = action()
-        except Exception as exception:
-            self.exception_value = exception
-            self.eof = True
-            raise
-        else:
-            if value is None:
-                self.eof = True
-            return value
-
-    async def wait_for(self, communique_type):
-        """Wait for a type of message."""
-        while True:
-            response = await self.read()
-            if response.get("CommuniqueType", None) == communique_type:
-                return response
-
-    async def write(self, item):
-        """Write an object to the queue."""
-
-        def action():
-            return item
-
-        await self.queue.put(action)
-
-    def at_eof(self):
-        """Check if the connection is closed."""
-        return self.closed.is_set() or self.eof
-
-    async def end(self, exception=None):
-        """Close the connection."""
-        if exception is None:
-            await self.write(None)
-        else:
-
-            def action():
-                raise exception
-
-            await self.queue.put(action)
-
-
-@pytest.yield_fixture(name="bridge")
-def fixture_bridge(event_loop):
+@pytest.fixture(name="bridge")
+async def fixture_bridge() -> AsyncGenerator[Bridge, None]:
     """Create a bridge attached to a fake reader and writer."""
     harness = Bridge()
 
-    event_loop.run_until_complete(harness.initialize())
+    await harness.initialize()
 
     yield harness
 
-    event_loop.run_until_complete(harness.target.close())
+    await harness.target.close()
 
 
 @pytest.mark.asyncio
-async def test_notifications(bridge):
+async def test_notifications(bridge: Bridge):
     """Test notifications are sent to subscribers."""
     notified = False
 
@@ -258,23 +249,23 @@ async def test_notifications(bridge):
         notified = True
 
     bridge.target.add_subscriber("2", callback)
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "OneZoneStatus",
-                "StatusCode": "200 OK",
-                "Url": "/zone/1/status",
-            },
-            "Body": {"ZoneStatus": {"Level": 100, "Zone": {"href": "/zone/1"}}},
-        }
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/zone/1/status",
+            ),
+            Body={"ZoneStatus": {"Level": 100, "Zone": {"href": "/zone/1"}}},
+        )
     )
-    await asyncio.wait_for(bridge.reader.queue.join(), 10)
+    await asyncio.wait_for(bridge.leap.requests.join(), 10)
     assert notified
 
 
 @pytest.mark.asyncio
-async def test_device_list(bridge):
+async def test_device_list(bridge: Bridge):
     """Test methods getting devices."""
     devices = bridge.target.get_devices()
     expected_devices = {
@@ -295,7 +286,7 @@ async def test_device_list(bridge):
             "zone": "1",
             "model": "PD-6WCL-XX",
             "serial": 2345,
-            "current_state": -1,
+            "current_state": 0,
             "fan_speed": None,
         },
         "3": {
@@ -305,7 +296,7 @@ async def test_device_list(bridge):
             "zone": "2",
             "model": "PD-FSQN-XX",
             "serial": 3456,
-            "current_state": -1,
+            "current_state": 0,
             "fan_speed": None,
         },
         "4": {
@@ -344,7 +335,7 @@ async def test_device_list(bridge):
             "type": "QsWirelessShade",
             "model": "QSYC-J-RCVR",
             "serial": 1234,
-            "current_state": -1,
+            "current_state": 0,
             "fan_speed": None,
             "zone": "6",
         },
@@ -352,29 +343,28 @@ async def test_device_list(bridge):
 
     assert devices == expected_devices
 
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "OneZoneStatus",
-                "StatusCode": "200 OK",
-                "Url": "/zone/1/status",
-            },
-            "Body": {"ZoneStatus": {"Level": 100, "Zone": {"href": "/zone/1"}}},
-        }
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/zone/1/status",
+            ),
+            Body={"ZoneStatus": {"Level": 100, "Zone": {"href": "/zone/1"}}},
+        )
     )
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "OneZoneStatus",
-                "StatusCode": "200 OK",
-                "Url": "/zone/2/status",
-            },
-            "Body": {"ZoneStatus": {"FanSpeed": "Medium", "Zone": {"href": "/zone/2"}}},
-        }
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/zone/2/status",
+            ),
+            Body={"ZoneStatus": {"FanSpeed": "Medium", "Zone": {"href": "/zone/2"}}},
+        )
     )
-    await asyncio.wait_for(bridge.reader.queue.join(), 10)
     devices = bridge.target.get_devices()
     assert devices["2"]["current_state"] == 100
     assert devices["2"]["fan_speed"] is None
@@ -404,7 +394,7 @@ async def test_device_list(bridge):
     assert devices[0]["device_id"] == "3"
 
 
-def test_scene_list(bridge):
+def test_scene_list(bridge: Bridge):
     """Test methods getting scenes."""
     scenes = bridge.target.get_scenes()
     assert scenes == {"1": {"scene_id": "1", "name": "scene 1"}}
@@ -412,16 +402,20 @@ def test_scene_list(bridge):
     assert scene == {"scene_id": "1", "name": "scene 1"}
 
 
-def test_is_connected(bridge):
+@pytest.mark.asyncio
+async def test_is_connected(bridge: Bridge):
     """Test the is_connected method returns connection state."""
     assert bridge.target.is_connected() is True
 
-    other = smartbridge.Smartbridge(None)
+    def connect():
+        raise NotImplementedError()
+
+    other = smartbridge.Smartbridge(connect)
     assert other.is_connected() is False
 
 
 @pytest.mark.asyncio
-async def test_area_list(bridge):
+async def test_area_list(bridge: Bridge):
     """Test the list of areas loaded by the bridge."""
     expected_areas = {
         "1": {"name": "root"},
@@ -434,7 +428,7 @@ async def test_area_list(bridge):
 
 
 @pytest.mark.asyncio
-async def test_occupancy_group_list(bridge):
+async def test_occupancy_group_list(bridge: Bridge):
     """Test the list of occupancy groups loaded by the bridge."""
     # Occupancy group 1 has no sensors, so it shouldn't appear here
     expected_groups = {
@@ -454,17 +448,17 @@ async def test_occupancy_group_list(bridge):
 
 
 @pytest.mark.asyncio
-async def test_occupancy_group_status_change(bridge):
+async def test_occupancy_group_status_change(bridge: Bridge):
     """Test that the status is updated when occupancy changes."""
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "MultipleOccupancyGroupStatus",
-                "StatusCode": "200 OK",
-                "Url": "/occupancygroup/status",
-            },
-            "Body": {
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="MultipleOccupancyGroupStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/occupancygroup/status",
+            ),
+            Body={
                 "OccupancyGroupStatuses": [
                     {
                         "href": "/occupancygroup/2/status",
@@ -473,15 +467,14 @@ async def test_occupancy_group_status_change(bridge):
                     }
                 ]
             },
-        }
+        )
     )
-    await asyncio.wait_for(bridge.reader.queue.join(), 10)
     new_status = bridge.target.occupancy_groups["2"]["status"]
     assert new_status == OCCUPANCY_GROUP_UNOCCUPIED
 
 
 @pytest.mark.asyncio
-async def test_occupancy_group_status_change_notification(bridge):
+async def test_occupancy_group_status_change_notification(bridge: Bridge):
     """Test that occupancy status changes send notifications."""
     notified = False
 
@@ -490,15 +483,15 @@ async def test_occupancy_group_status_change_notification(bridge):
         notified = True
 
     bridge.target.add_occupancy_subscriber("2", notify)
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "MultipleOccupancyGroupStatus",
-                "StatusCode": "200 OK",
-                "Url": "/occupancygroup/status",
-            },
-            "Body": {
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="MultipleOccupancyGroupStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/occupancygroup/status",
+            ),
+            Body={
                 "OccupancyGroupStatuses": [
                     {
                         "href": "/occupancygroup/2/status",
@@ -507,242 +500,358 @@ async def test_occupancy_group_status_change_notification(bridge):
                     }
                 ]
             },
-        }
+        )
     )
-    await asyncio.wait_for(bridge.reader.queue.join(), 10)
     assert notified
 
 
 @pytest.mark.asyncio
-async def test_is_on(bridge):
+async def test_is_on(bridge: Bridge):
     """Test the is_on method returns device state."""
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "OneZoneStatus",
-                "StatusCode": "200 OK",
-                "Url": "/zone/1/status",
-            },
-            "Body": {"ZoneStatus": {"Level": 50, "Zone": {"href": "/zone/1"}}},
-        }
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/zone/1/status",
+            ),
+            Body={"ZoneStatus": {"Level": 50, "Zone": {"href": "/zone/1"}}},
+        )
     )
-    await asyncio.wait_for(bridge.reader.queue.join(), 10)
     assert bridge.target.is_on("2") is True
 
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "OneZoneStatus",
-                "StatusCode": "200 OK",
-                "Url": "/zone/1/status",
-            },
-            "Body": {"ZoneStatus": {"Level": 0, "Zone": {"href": "/zone/1"}}},
-        }
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/zone/1/status",
+            ),
+            Body={"ZoneStatus": {"Level": 0, "Zone": {"href": "/zone/1"}}},
+        )
     )
-    await asyncio.wait_for(bridge.reader.queue.join(), 10)
     assert bridge.target.is_on("2") is False
 
 
 @pytest.mark.asyncio
-async def test_is_on_fan(bridge):
+async def test_is_on_fan(bridge: Bridge):
     """Test the is_on method returns device state for fans."""
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "OneZoneStatus",
-                "StatusCode": "200 OK",
-                "Url": "/zone/1/status",
-            },
-            "Body": {"ZoneStatus": {"FanSpeed": "Medium", "Zone": {"href": "/zone/1"}}},
-        }
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/zone/1/status",
+            ),
+            Body={"ZoneStatus": {"FanSpeed": "Medium", "Zone": {"href": "/zone/1"}}},
+        )
     )
-    await asyncio.wait_for(bridge.reader.queue.join(), 10)
     assert bridge.target.is_on("2") is True
 
-    await bridge.reader.write(
-        {
-            "CommuniqueType": "ReadResponse",
-            "Header": {
-                "MessageBodyType": "OneZoneStatus",
-                "StatusCode": "200 OK",
-                "Url": "/zone/1/status",
-            },
-            "Body": {"ZoneStatus": {"FanSpeed": "Off", "Zone": {"href": "/zone/1"}}},
-        }
+    bridge.leap.send_unsolicited(
+        Response(
+            CommuniqueType="ReadResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(200, "OK"),
+                Url="/zone/1/status",
+            ),
+            Body={"ZoneStatus": {"FanSpeed": "Off", "Zone": {"href": "/zone/1"}}},
+        )
     )
-    await asyncio.wait_for(bridge.reader.queue.join(), 10)
     assert bridge.target.is_on("2") is False
 
 
 @pytest.mark.asyncio
-async def test_set_value(bridge):
+async def test_set_value(bridge: Bridge, event_loop):
     """Test that setting values produces the right commands."""
-    bridge.target.set_value("2", 50)
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
-    assert command == {
-        "CommuniqueType": "CreateRequest",
-        "Header": {"Url": "/zone/1/commandprocessor"},
-        "Body": {
+    task = event_loop.create_task(bridge.target.set_value("2", 50))
+    command, response = await bridge.leap.requests.get()
+    assert command == Request(
+        communique_type="CreateRequest",
+        url="/zone/1/commandprocessor",
+        body={
             "Command": {
                 "CommandType": "GoToLevel",
                 "Parameter": [{"Type": "Level", "Value": 50}],
             }
         },
-    }
-
-    bridge.target.turn_on("2")
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
-    assert command == {
-        "CommuniqueType": "CreateRequest",
-        "Header": {"Url": "/zone/1/commandprocessor"},
-        "Body": {
+    )
+    response.set_result(
+        Response(
+            CommuniqueType="CreateResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(201, "Created"),
+                Url="/zone/1/commandprocessor",
+            ),
+            Body={
+                "ZoneStatus": {
+                    "href": "/zone/1/status",
+                    "Level": 50,
+                    "Zone": {"href": "/zone/1"},
+                }
+            },
+        )
+    )
+    bridge.leap.requests.task_done()
+    await task
+
+    task = event_loop.create_task(bridge.target.turn_on("2"))
+    command, response = await bridge.leap.requests.get()
+    assert command == Request(
+        communique_type="CreateRequest",
+        url="/zone/1/commandprocessor",
+        body={
             "Command": {
                 "CommandType": "GoToLevel",
                 "Parameter": [{"Type": "Level", "Value": 100}],
             }
         },
-    }
-
-    bridge.target.turn_off("2")
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
-    assert command == {
-        "CommuniqueType": "CreateRequest",
-        "Header": {"Url": "/zone/1/commandprocessor"},
-        "Body": {
+    )
+    response.set_result(
+        Response(
+            CommuniqueType="CreateResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(201, "Created"),
+                Url="/zone/1/commandprocessor",
+            ),
+            Body={
+                "ZoneStatus": {
+                    "href": "/zone/1/status",
+                    "Level": 100,
+                    "Zone": {"href": "/zone/1"},
+                }
+            },
+        ),
+    )
+    bridge.leap.requests.task_done()
+    await task
+
+    task = event_loop.create_task(bridge.target.turn_off("2"))
+    command, response = await bridge.leap.requests.get()
+    assert command == Request(
+        communique_type="CreateRequest",
+        url="/zone/1/commandprocessor",
+        body={
             "Command": {
                 "CommandType": "GoToLevel",
                 "Parameter": [{"Type": "Level", "Value": 0}],
             }
         },
-    }
+    )
+    response.set_result(
+        Response(
+            CommuniqueType="CreateResponse",
+            Header=ResponseHeader(
+                MessageBodyType="OneZoneStatus",
+                StatusCode=ResponseStatus(201, "Created"),
+                Url="/zone/1/commandprocessor",
+            ),
+            Body={
+                "ZoneStatus": {
+                    "href": "/zone/1/status",
+                    "Level": 0,
+                    "Zone": {"href": "/zone/1"},
+                }
+            },
+        ),
+    )
+    bridge.leap.requests.task_done()
+    await task
 
 
 @pytest.mark.asyncio
-async def test_set_fan(bridge):
+async def test_set_fan(bridge: Bridge, event_loop):
     """Test that setting fan speed produces the right commands."""
-    bridge.target.set_fan("2", FAN_MEDIUM)
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
-    assert command == {
-        "CommuniqueType": "CreateRequest",
-        "Header": {"Url": "/zone/1/commandprocessor"},
-        "Body": {
+    task = event_loop.create_task(bridge.target.set_fan("2", FAN_MEDIUM))
+    command, _ = await bridge.leap.requests.get()
+    assert command == Request(
+        communique_type="CreateRequest",
+        url="/zone/1/commandprocessor",
+        body={
             "Command": {
                 "CommandType": "GoToFanSpeed",
                 "FanSpeedParameters": {"FanSpeed": "Medium"},
             }
         },
-    }
+    )
+    bridge.leap.requests.task_done()
+    task.cancel()
 
 
 @pytest.mark.asyncio
-async def test_lower_cover(bridge):
+async def test_lower_cover(bridge: Bridge, event_loop):
     """Test that lowering a cover produces the right commands."""
     devices = bridge.target.get_devices()
-    bridge.target.lower_cover("7")
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
-    assert command == {
-        "CommuniqueType": "CreateRequest",
-        "Header": {"Url": "/zone/6/commandprocessor"},
-        "Body": {"Command": {"CommandType": "Lower"}},
-    }
+    task = event_loop.create_task(bridge.target.lower_cover("7"))
+    command, response = await bridge.leap.requests.get()
+    assert command == Request(
+        communique_type="CreateRequest",
+        url="/zone/6/commandprocessor",
+        body={"Command": {"CommandType": "Lower"}},
+    )
+    # the real response probably contains more data
+    response.set_result(
+        Response(
+            CommuniqueType="CreateResponse",
+            Header=ResponseHeader(
+                StatusCode=ResponseStatus(201, "Created"),
+                Url="/zone/6/commandprocessor",
+            ),
+        ),
+    )
+    bridge.leap.requests.task_done()
+    await task
     assert devices["7"]["current_state"] == 0
 
 
 @pytest.mark.asyncio
-async def test_raise_cover(bridge):
+async def test_raise_cover(bridge: Bridge, event_loop):
     """Test that raising a cover produces the right commands."""
     devices = bridge.target.get_devices()
-    bridge.target.raise_cover("7")
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
-    assert command == {
-        "CommuniqueType": "CreateRequest",
-        "Header": {"Url": "/zone/6/commandprocessor"},
-        "Body": {"Command": {"CommandType": "Raise"}},
-    }
+    task = event_loop.create_task(bridge.target.raise_cover("7"))
+    command, response = await bridge.leap.requests.get()
+    assert command == Request(
+        communique_type="CreateRequest",
+        url="/zone/6/commandprocessor",
+        body={"Command": {"CommandType": "Raise"}},
+    )
+    # the real response probably contains more data
+    response.set_result(
+        Response(
+            CommuniqueType="CreateResponse",
+            Header=ResponseHeader(
+                StatusCode=ResponseStatus(201, "Created"),
+                Url="/zone/6/commandprocessor",
+            ),
+        ),
+    )
+    bridge.leap.requests.task_done()
+    await task
     assert devices["7"]["current_state"] == 100
 
 
 @pytest.mark.asyncio
-async def test_stop_cover(bridge):
+async def test_stop_cover(bridge: Bridge, event_loop):
     """Test that stopping a cover produces the right commands."""
-    bridge.target.stop_cover("7")
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
-    assert command == {
-        "CommuniqueType": "CreateRequest",
-        "Header": {"Url": "/zone/6/commandprocessor"},
-        "Body": {"Command": {"CommandType": "Stop"}},
-    }
+    task = event_loop.create_task(bridge.target.stop_cover("7"))
+    command, _ = await bridge.leap.requests.get()
+    assert command == Request(
+        communique_type="CreateRequest",
+        url="/zone/6/commandprocessor",
+        body={"Command": {"CommandType": "Stop"}},
+    )
+    bridge.leap.requests.task_done()
+    task.cancel()
 
 
 @pytest.mark.asyncio
-async def test_activate_scene(bridge):
+async def test_activate_scene(bridge: Bridge, event_loop):
     """Test that activating scenes produces the right commands."""
-    bridge.target.activate_scene("1")
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
-    assert command == {
-        "CommuniqueType": "CreateRequest",
-        "Header": {"Url": "/virtualbutton/1/commandprocessor"},
-        "Body": {"Command": {"CommandType": "PressAndRelease"}},
-    }
+    task = event_loop.create_task(bridge.target.activate_scene("1"))
+    command, _ = await bridge.leap.requests.get()
+    assert command == Request(
+        communique_type="CreateRequest",
+        url="/virtualbutton/1/commandprocessor",
+        body={"Command": {"CommandType": "PressAndRelease"}},
+    )
+    bridge.leap.requests.task_done()
+    task.cancel()
 
 
 @pytest.mark.asyncio
-async def test_reconnect_eof(bridge):
+async def test_reconnect_eof(bridge: Bridge, event_loop):
     """Test that SmartBridge can reconnect on disconnect."""
-    await bridge.disconnect()
+    time = 0.0
+    event_loop.time = lambda: time
+
+    bridge.disconnect()
+
+    await asyncio.sleep(0.0)
+    time += smartbridge.RECONNECT_DELAY
+
     await bridge.accept_connection()
-    bridge.target.set_value("2", 50)
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
+
+    task = event_loop.create_task(bridge.target.set_value("2", 50))
+    command, _ = await bridge.leap.requests.get()
     assert command is not None
+    bridge.leap.requests.task_done()
+    task.cancel()
 
 
 @pytest.mark.asyncio
-async def test_reconnect_error(bridge):
+async def test_connect_error(event_loop):
+    """Test that SmartBridge can retry failed connections."""
+    time = 0.0
+    event_loop.time = lambda: time
+
+    tried = asyncio.Event()
+
+    async def fake_connect():
+        """Simulate connection error for the test."""
+        tried.set()
+        raise OSError()
+
+    target = smartbridge.Smartbridge(fake_connect)
+    connect_task = event_loop.create_task(target.connect())
+
+    await tried.wait()
+    tried.clear()
+    time += smartbridge.RECONNECT_DELAY
+
+    await tried.wait()
+    connect_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_reconnect_error(bridge: Bridge, event_loop):
     """Test that SmartBridge can reconnect on error."""
-    await bridge.disconnect()
+    time = 0.0
+    event_loop.time = lambda: time
+
+    bridge.disconnect()
+
+    await asyncio.sleep(0.0)
+    time += smartbridge.RECONNECT_DELAY
+
     await bridge.accept_connection()
-    bridge.target.set_value("2", 50)
-    command = await asyncio.wait_for(bridge.writer.queue.get(), 10)
-    bridge.writer.queue.task_done()
+
+    task = event_loop.create_task(bridge.target.set_value("2", 50))
+    command, _ = await bridge.leap.requests.get()
     assert command is not None
+    bridge.leap.requests.task_done()
+    task.cancel()
 
 
 @pytest.mark.asyncio
-async def test_reconnect_timeout():
+async def test_reconnect_timeout(event_loop):
     """Test that SmartBridge can reconnect if the remote does not respond."""
     bridge = Bridge()
 
     time = 0.0
-
-    get_loop().time = lambda: time
+    event_loop.time = lambda: time
 
     await bridge.initialize()
 
     time = smartbridge.PING_INTERVAL
-    ping = await bridge.writer.queue.get()
-    assert ping == {
-        "CommuniqueType": "ReadRequest",
-        "Header": {"Url": "/server/1/status/ping"},
-    }
+    ping, _ = await bridge.leap.requests.get()
+    assert ping == Request(communique_type="ReadRequest", url="/server/1/status/ping")
+    bridge.leap.requests.task_done()
 
-    time += smartbridge.PING_DELAY
+    time += smartbridge.REQUEST_TIMEOUT
+    await bridge.leap.running
+    time += smartbridge.RECONNECT_DELAY
     await bridge.accept_connection()
 
-    bridge.target.set_value("2", 50)
-    command = await bridge.writer.queue.get()
-    bridge.writer.queue.task_done()
+    task = event_loop.create_task(bridge.target.set_value("2", 50))
+    command, _ = await bridge.leap.requests.get()
     assert command is not None
+    bridge.leap.requests.task_done()
+    task.cancel()
 
     await bridge.target.close()