Repository: postlund/pyatv
Pull request #2036 (2023-06-08 12:47:16)

Basic AirPlay 2 audio streaming support
This is initial support for streaming audio via AirPlay 2. It is in unfinished and barely tested state so far (no tests or documentation either). But I hope most things are in place and that I have not broken anything else...

<a href="https://gitpod.io/#https://github.com/postlund/pyatv/pull/2036"><img src="https://gitpod.io/button/open-in-gitpod.svg"/></a>



codecov[bot]: ## [Codecov](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl) Report
Patch coverage: **`77.15`**% and project coverage change: **`-0.63`** :warning:
> Comparison is base [(`d508f35`)](https://app.codecov.io/gh/postlund/pyatv/commit/d508f35d98a1af411877db5dd483e433e208a0ac?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl) 87.69% compared to head [(`a3a4b8b`)](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl) 87.06%.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #2036      +/-   ##
==========================================
- Coverage   87.69%   87.06%   -0.63%     
==========================================
  Files         155      159       +4     
  Lines        9977    10092     +115     
==========================================
+ Hits         8749     8787      +38     
- Misses       1228     1305      +77     
```


| [Impacted Files](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl) | Coverage Δ | |
|---|---|---|
| [pyatv/protocols/airplay/auth/hap\_transient.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvcHJvdG9jb2xzL2FpcnBsYXkvYXV0aC9oYXBfdHJhbnNpZW50LnB5) | `42.85% <0.00%> (ø)` | |
| [pyatv/protocols/airplay/mrp\_connection.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvcHJvdG9jb2xzL2FpcnBsYXkvbXJwX2Nvbm5lY3Rpb24ucHk=) | `46.66% <25.00%> (ø)` | |
| [pyatv/protocols/raop/protocols/airplayv2.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvcHJvdG9jb2xzL3Jhb3AvcHJvdG9jb2xzL2FpcnBsYXl2Mi5weQ==) | `33.87% <33.87%> (ø)` | |
| [pyatv/protocols/airplay/ap2\_session.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvcHJvdG9jb2xzL2FpcnBsYXkvYXAyX3Nlc3Npb24ucHk=) | `37.20% <37.03%> (ø)` | |
| [pyatv/protocols/airplay/\_\_init\_\_.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvcHJvdG9jb2xzL2FpcnBsYXkvX19pbml0X18ucHk=) | `71.85% <42.85%> (-4.12%)` | :arrow_down: |
| [pyatv/support/rtsp.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvc3VwcG9ydC9ydHNwLnB5) | `94.73% <75.00%> (-1.60%)` | :arrow_down: |
| [pyatv/protocols/airplay/auth/legacy.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvcHJvdG9jb2xzL2FpcnBsYXkvYXV0aC9sZWdhY3kucHk=) | `97.91% <80.00%> (ø)` | |
| [pyatv/support/http.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvc3VwcG9ydC9odHRwLnB5) | `93.24% <90.90%> (-0.86%)` | :arrow_down: |
| [pyatv/protocols/raop/stream\_client.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvcHJvdG9jb2xzL3Jhb3Avc3RyZWFtX2NsaWVudC5weQ==) | `86.54% <95.00%> (ø)` | |
| [pyatv/protocols/raop/\_\_init\_\_.py](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl#diff-cHlhdHYvcHJvdG9jb2xzL3Jhb3AvX19pbml0X18ucHk=) | `94.30% <95.23%> (+0.08%)` | :arrow_up: |
| ... and [11 more](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl) | |

... and [14 files with indirect coverage changes](https://app.codecov.io/gh/postlund/pyatv/pull/2036/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/postlund/pyatv/pull/2036?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Pierre+St%C3%A5hl).

postlund: As this seems to work good enough, I'm gonna merge this now and fix tests later. Would like to get this included in the next Home Assistant version.

Patch:
diff --git a/base_versions.txt b/base_versions.txt
index ea47dc0e..72ab8f30 100644
--- a/base_versions.txt
+++ b/base_versions.txt
@@ -1,6 +1,6 @@
 aiohttp==3.8.3,<5
 async-timeout==4.0.2
-bitarray==2.1.2
+bitarray==2.3.0
 cryptography==36.0.2
 chacha20poly1305-reuseable==0.0.3
 ifaddr==0.1.7
diff --git a/docs/documentation/atvremote.md b/docs/documentation/atvremote.md
index 0b106f99..9b025a77 100644
--- a/docs/documentation/atvremote.md
+++ b/docs/documentation/atvremote.md
@@ -158,6 +158,23 @@ still specify something. A simple call example looks like this:
 $ atvremote --manual --address 10.0.0.10 --port 49152 --protocol mrp --id test playing
 ```
 
+For testing purposes, it's possible to specify custom MDNS properties using
+`--service-properties`. This might be useful to tinker with certain flags that
+alter protocol behavior. The format looks like this:
+
+    Xvar1=value1Xvar2=value2
+
+Where `X` is any character not present in a variable name or value. A typical example
+might use `:` or `,` like this:
+
+    :name=test:flags=123
+
+An example call to `atvremote` might look like this:
+
+```shell
+$ atvremote --id "aa:bb:cc:dd:ee:ff" --address 10.0.10.253 --port 7000 --manual --protocol raop --service-properties :features=0x4A7FCA00,0xBC354BD0 --debug stream_file=never_gonna_give_you_up.mp3
+```
+
 # Pairing with a device
 
 In most cases you have to pair with a device and obtain *credentials* in order to communicate
diff --git a/docs/documentation/supported_features.md b/docs/documentation/supported_features.md
index d2de4dc2..8a988181 100644
--- a/docs/documentation/supported_features.md
+++ b/docs/documentation/supported_features.md
@@ -229,6 +229,7 @@ Audio streaming has been verified to work with these devices:
 
 * Apple TV 3 (v8.4.4)
 * Apple TV 4K gen 1 (v14.5)
+* [airplay2-receiver](https://github.com/openairplay/airplay2-receiver) ([224b9660c31ed8624f46fdae9fcb5f44e5ec6c12](https://github.com/openairplay/airplay2-receiver/tree/224b9660c31ed8624f46fdae9fcb5f44e5ec6c12))
 * HomePod Mini (v14.5)
 * AirPort Express (v7.8.1)
 * Yamaha RX-V773 (v1.98)
diff --git a/pyatv/auth/hap_srp.py b/pyatv/auth/hap_srp.py
index bae06e0d..0a7e613e 100644
--- a/pyatv/auth/hap_srp.py
+++ b/pyatv/auth/hap_srp.py
@@ -92,7 +92,7 @@ class SRPAuthHandler:
         )
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
-        decrypted_tlv = read_tlv(chacha.decrypt(encrypted, nounce="PV-Msg02".encode()))
+        decrypted_tlv = read_tlv(chacha.decrypt(encrypted, nonce="PV-Msg02".encode()))
 
         identifier = decrypted_tlv[TlvValue.Identifier]
         signature = decrypted_tlv[TlvValue.Signature]
@@ -121,7 +121,7 @@ class SRPAuthHandler:
             }
         )
 
-        return chacha.encrypt(tlv, nounce="PV-Msg03".encode())
+        return chacha.encrypt(tlv, nonce="PV-Msg03".encode())
 
     def verify2(
         self, salt: str, output_info: str, input_info: str
@@ -200,14 +200,14 @@ class SRPAuthHandler:
             tlv.update(additional_data)
 
         chacha = chacha20.Chacha20Cipher(self._session_key, self._session_key)
-        encrypted_data = chacha.encrypt(write_tlv(tlv), nounce="PS-Msg05".encode())
+        encrypted_data = chacha.encrypt(write_tlv(tlv), nonce="PS-Msg05".encode())
         log_binary(_LOGGER, "Data", Encrypted=encrypted_data)
         return encrypted_data
 
     def step4(self, encrypted_data):
         """Last pairing step."""
         chacha = chacha20.Chacha20Cipher(self._session_key, self._session_key)
-        decrypted_tlv_bytes = chacha.decrypt(encrypted_data, nounce="PS-Msg06".encode())
+        decrypted_tlv_bytes = chacha.decrypt(encrypted_data, nonce="PS-Msg06".encode())
 
         if not decrypted_tlv_bytes:
             raise exceptions.AuthenticationError("data decrypt failed")
diff --git a/pyatv/auth/hap_tlv8.py b/pyatv/auth/hap_tlv8.py
index 2417870b..b18d7f9c 100644
--- a/pyatv/auth/hap_tlv8.py
+++ b/pyatv/auth/hap_tlv8.py
@@ -30,6 +30,13 @@ class TlvValue(IntEnum):
 
     # Apple internal(?)
     Name = 0x11
+    Flags = 0x13
+
+
+class Flags(IntEnum):
+    """Flags used with TlvValue.Flags."""
+
+    TransientPairing = 0x10
 
 
 class ErrorCode(IntEnum):
diff --git a/pyatv/protocols/airplay/__init__.py b/pyatv/protocols/airplay/__init__.py
index e4cdf57e..c9df8ef1 100644
--- a/pyatv/protocols/airplay/__init__.py
+++ b/pyatv/protocols/airplay/__init__.py
@@ -7,7 +7,7 @@ from typing import Any, Dict, Generator, Mapping, Optional, Set
 
 from pyatv import exceptions
 from pyatv.auth.hap_pairing import AuthenticationType, HapCredentials, parse_credentials
-from pyatv.const import DeviceModel, FeatureName, PairingRequirement, Protocol
+from pyatv.const import DeviceModel, FeatureName, Protocol
 from pyatv.core import Core, MutableService, SetupData, mdns
 from pyatv.core.scan import (
     ScanHandlerDeviceInfoName,
@@ -26,6 +26,7 @@ from pyatv.interface import (
     Stream,
 )
 from pyatv.protocols import mrp
+from pyatv.protocols.airplay.ap2_session import AP2Session
 from pyatv.protocols.airplay.auth import extract_credentials, verify_connection
 from pyatv.protocols.airplay.mrp_connection import AirPlayMrpConnection
 from pyatv.protocols.airplay.pairing import (
@@ -33,14 +34,13 @@ from pyatv.protocols.airplay.pairing import (
     get_preferred_auth_type,
 )
 from pyatv.protocols.airplay.player import AirPlayPlayer
-from pyatv.protocols.airplay.remote_control import RemoteControl
 from pyatv.protocols.airplay.utils import (
     AirPlayFlags,
-    get_pairing_requirement,
-    is_password_required,
     is_remote_control_supported,
     parse_features,
+    update_service_details,
 )
+from pyatv.protocols.raop import setup as raop_setup
 from pyatv.support import net
 from pyatv.support.device_info import lookup_model
 from pyatv.support.http import (
@@ -56,10 +56,9 @@ _LOGGER = logging.getLogger(__name__)
 class AirPlayFeatures(Features):
     """Implementation of supported feature functionality."""
 
-    def __init__(self, service: BaseService) -> None:
+    def __init__(self, features: AirPlayFlags) -> None:
         """Initialize a new AirPlayFeatures instance."""
-        self.service = service
-        self._features = parse_features(self.service.properties.get("features", "0x0"))
+        self._features = features
 
     def get_feature(self, feature_name: FeatureName) -> FeatureInfo:
         """Return current state of a feature."""
@@ -174,14 +173,7 @@ async def service_info(
     services: Mapping[Protocol, BaseService],
 ) -> None:
     """Update service with additional information."""
-    service.requires_password = is_password_required(service)
-
-    if service.properties.get("acl", "0") == "1":
-        # Access control might say that pairing is not possible, e.g. only devices
-        # belonging to the same home (not supported by pyatv)
-        service.pairing = PairingRequirement.Disabled
-    else:
-        service.pairing = get_pairing_requirement(service)
+    update_service_details(service)
 
 
 def setup(  # pylint: disable=too-many-locals
@@ -191,8 +183,11 @@ def setup(  # pylint: disable=too-many-locals
     # TODO: Split up in connect/protocol and Stream implementation
     stream = AirPlayStream(core.config, core.service)
 
+    features = parse_features(core.service.properties.get("features", "0x0"))
+    credentials = extract_credentials(core.service)
+
     interfaces = {
-        Features: AirPlayFeatures(core.service),
+        Features: AirPlayFeatures(features),
         Stream: stream,
     }
 
@@ -215,7 +210,41 @@ def setup(  # pylint: disable=too-many-locals
         set([FeatureName.PlayUrl]),
     )
 
-    credentials = extract_credentials(core.service)
+    # AirPlay 2 does not mandate that a separate RAOP service exists for streaming
+    # audio, instead the same service user by AirPlay can be used if a particular flag
+    # is set (HasUnifiedAdvertiserInfo). If that flag is set and no RAOP service has
+    # been found, manually add a service pointing to the AirPlay service. This just
+    # simplifies the internal handling, but is not very efficient as no connections
+    # are re-used amongst the protocols.
+    if (
+        AirPlayFlags.HasUnifiedAdvertiserInfo in features
+        and core.config.get_service(Protocol.RAOP) is None
+    ):
+        _LOGGER.debug("RAOP supported but no service present, adding new service")
+        # Create a RAOP service to satisfy internal RAOP service
+        raop_service = MutableService(
+            None,
+            Protocol.RAOP,
+            core.service.port,
+            core.service.properties,
+            credentials=core.service.credentials,
+            password=core.service.password,
+        )
+        core.config.add_service(raop_service)
+
+        # Re-map service in Core to the newly created raop service
+        raop_core = Core(
+            core.loop,
+            core.config,
+            raop_service,
+            core.device_listener,
+            core.session_manager,
+            core.takeover,
+            core.state_dispatcher.create_copy(Protocol.RAOP),
+        )
+
+        for setup_data in raop_setup(raop_core):
+            yield setup_data
 
     # Set up remote control channel if it is supported
     if not is_remote_control_supported(core.service, credentials):
@@ -225,14 +254,13 @@ def setup(  # pylint: disable=too-many-locals
     else:
         _LOGGER.debug("Remote control channel is supported")
 
-        control = RemoteControl(core.device_listener)
-        control_port = core.service.port
+        session = AP2Session(str(core.config.address), core.service.port, credentials)
 
-        # A protocol requires its correaponding service to function, so add a
+        # A protocol requires its corresponding service to function, so add a
         # dummy one if we don't have one yet
         mrp_service = core.config.get_service(Protocol.MRP)
         if mrp_service is None:
-            mrp_service = MutableService(None, Protocol.MRP, control_port, {})
+            mrp_service = MutableService(None, Protocol.MRP, core.service.port, {})
             core.config.add_service(mrp_service)
 
         (
@@ -252,13 +280,15 @@ def setup(  # pylint: disable=too-many-locals
                 core.takeover,
                 core.state_dispatcher.create_copy(Protocol.MRP),
             ),
-            AirPlayMrpConnection(control, core.device_listener),
+            AirPlayMrpConnection(session, core.device_listener),
             requires_heatbeat=False,  # Already have heartbeat on control channel
         )
 
         async def _connect_rc() -> bool:
             try:
-                await control.start(str(core.config.address), control_port, credentials)
+                await session.connect()
+                await session.setup_remote_control()
+                session.start_keep_alive(core.device_listener)
             except exceptions.HttpError as ex:
                 if ex.status_code == 470:
                     _LOGGER.debug(
@@ -276,7 +306,7 @@ def setup(  # pylint: disable=too-many-locals
         def _close_rc() -> Set[asyncio.Task]:
             tasks = set()
             tasks.update(mrp_close())
-            tasks.update(control.stop())
+            tasks.update(session.stop())
             return tasks
 
         yield SetupData(
diff --git a/pyatv/protocols/airplay/remote_control.py b/pyatv/protocols/airplay/ap2_session.py
similarity index 70%
rename from pyatv/protocols/airplay/remote_control.py
rename to pyatv/protocols/airplay/ap2_session.py
index f6a12e89..5e64ff2f 100644
--- a/pyatv/protocols/airplay/remote_control.py
+++ b/pyatv/protocols/airplay/ap2_session.py
@@ -1,17 +1,23 @@
-"""Implementation of Remote Control (channel) in AirPlay 2."""
+"""Implementation of "high-level" support for an AirPlay 2 session.
+
+This code is pretty messy right now and needs some re-structuring. The intention is
+however to set up a connection to an AirPlay 2 receiver and ensure encryption and
+most low-level stuff is taken care of.
+"""
 import asyncio
 import logging
-import plistlib
 from random import randint
 from typing import Any, Dict, List, Optional, Set, cast
 from uuid import uuid4
 
+from pyatv import exceptions
 from pyatv.auth.hap_channel import setup_channel
 from pyatv.auth.hap_pairing import HapCredentials, PairVerifyProcedure
 from pyatv.core.protocol import heartbeater
+from pyatv.interface import DeviceListener
 from pyatv.protocols.airplay.auth import verify_connection
 from pyatv.protocols.airplay.channels import DataStreamChannel, EventChannel
-from pyatv.support.http import HttpConnection, http_connect
+from pyatv.support.http import HttpConnection, decode_bplist_from_body, http_connect
 from pyatv.support.rtsp import RtspSession
 from pyatv.support.state_producer import StateProducer
 
@@ -29,55 +35,64 @@ DATASTREAM_OUTPUT_INFO = "DataStream-Output-Encryption-Key"
 DATASTREAM_INPUT_INFO = "DataStream-Input-Encryption-Key"
 
 
-class RemoteControl:
-    """MRP remote control session over AirPlay."""
+class AP2Session:
+    """High-level session for AirPlay 2."""
 
-    def __init__(self, device_listener: StateProducer) -> None:
-        """Initialize a new RemoteControl instance."""
+    def __init__(
+        self, address: str, control_port: int, credentials: HapCredentials
+    ) -> None:
+        """Initialize a new AP2Session instance."""
+        self._address = address
+        self._control_port = control_port
+        self._credentials = credentials
         self.connection: Optional[HttpConnection] = None
         self.verifier: Optional[PairVerifyProcedure] = None
         self.rtsp: Optional[RtspSession] = None
-        self.device_listener = device_listener
         self.data_channel: Optional[DataStreamChannel] = None
         self._channels: List[asyncio.BaseTransport] = []
         self._feedback_task: Optional[asyncio.Task] = None
 
-    async def start(
-        self, address: str, control_port: int, credentials: HapCredentials
-    ) -> None:
-        """Open remote control connection."""
+    async def connect(self) -> None:
+        """Open connection to receiver."""
         _LOGGER.debug(
-            "Setting up remote control connection to %s:%d",
-            address,
-            control_port,
+            "Setting up remote connection to %s:%d",
+            self._address,
+            self._control_port,
         )
 
-        self.connection = await http_connect(address, control_port)
-        self.verifier = await verify_connection(credentials, self.connection)
+        self.connection = await http_connect(self._address, self._control_port)
+        self.verifier = await verify_connection(self._credentials, self.connection)
 
         self.rtsp = RtspSession(self.connection)
 
-        await self._setup_event_channel(self.connection.remote_ip, self.verifier)
+    async def setup_remote_control(self) -> None:
+        """Set up remote control session over the data channel."""
+        if self.connection is None or self.rtsp is None:
+            raise exceptions.InvalidStateError("not connected to remote")
 
+        await self._setup_event_channel(self.connection.remote_ip)
         await self.rtsp.record()
-        await self._setup_data_channel(self.connection.remote_ip, self.verifier)
+        await self._setup_data_channel(self.connection.remote_ip)
+
+    def start_keep_alive(self, device_listener: StateProducer[DeviceListener]) -> None:
+        """Start sending keep alive messages."""
 
-        # Lambdas as needed here as accessing a method in the device listener will
-        # cause the device listener to handle that as a connection error happened
-        # and tear everything down. This is by design.
         def _finish_func() -> None:
-            self.device_listener.listener.connection_closed()
+            device_listener.listener.connection_closed()
 
         def _failure_func(exc: Exception) -> None:
-            self.device_listener.listener.connection_lost(exc)
+            device_listener.listener.connection_lost(exc)
 
         async def _send_feedback(message: Optional[Any]) -> None:
             if self.rtsp:
                 await self.rtsp.feedback()
 
+        # Lambdas as needed here as accessing a method in the device listener will
+        # cause the device listener to handle that as a connection error happened
+        # and tear everything down. This is by design.
         self._feedback_task = asyncio.ensure_future(
             heartbeater(
-                name=f"AirPlay:{address}",
+                name=f"AirPlay:{self._address}",
                 sender_func=_send_feedback,
                 finish_func=_finish_func,
                 failure_func=_failure_func,
@@ -87,21 +102,13 @@ class RemoteControl:
 
     async def _setup(self, body: Dict[str, Any]) -> Dict[str, Any]:
         assert self.rtsp
+        resp = await self.rtsp.setup(body=body)
+        return decode_bplist_from_body(resp)
 
-        resp = await self.rtsp.setup(
-            headers={"Content-Type": "application/x-apple-binary-plist"},
-            body=plistlib.dumps(
-                body, fmt=plistlib.FMT_BINARY  # pylint: disable=no-member
-            ),
-        )
-        resp_body = (
-            resp.body if isinstance(resp.body, bytes) else resp.body.encode("utf-8")
-        )
-        return plistlib.loads(resp_body)
+    async def _setup_event_channel(self, address: str) -> None:
+        if self.verifier is None:
+            raise exceptions.InvalidStateError("not in connected state")
 
-    async def _setup_event_channel(
-        self, address: str, verifier: PairVerifyProcedure
-    ) -> None:
         resp = await self._setup(
             {
                 "isRemoteControlOnly": True,
@@ -125,7 +132,7 @@ class RemoteControl:
         # Note: Read/Write info reversed here as connection originates from receiver!
         transport, _ = await setup_channel(
             EventChannel,
-            verifier,
+            self.verifier,
             address,
             event_port,
             EVENTS_SALT,
@@ -134,9 +141,10 @@ class RemoteControl:
         )
         self._channels.append(transport)
 
-    async def _setup_data_channel(
-        self, address: str, verifier: PairVerifyProcedure
-    ) -> None:
+    async def _setup_data_channel(self, address: str) -> None:
+        if self.verifier is None:
+            raise exceptions.InvalidStateError("not in connected state")
+
         # A 64 bit random seed is included and used as part of the salt in encryption
         seed = randint(0, 2**64)
 
@@ -160,7 +168,7 @@ class RemoteControl:
 
         transport, protocol = await setup_channel(
             DataStreamChannel,
-            verifier,
+            self.verifier,
             address,
             data_port,
             DATASTREAM_SALT + str(seed),
diff --git a/pyatv/protocols/airplay/auth/__init__.py b/pyatv/protocols/airplay/auth/__init__.py
index be918ba4..79352954 100644
--- a/pyatv/protocols/airplay/auth/__init__.py
+++ b/pyatv/protocols/airplay/auth/__init__.py
@@ -121,7 +121,9 @@ def extract_credentials(service: BaseService) -> HapCredentials:
     if service.credentials is not None:
         return parse_credentials(service.credentials)
 
-    flags = parse_features(service.properties.get("features", "0x0"))
+    flags = parse_features(
+        service.properties.get("features", service.properties.get("ft", "0x0"))
+    )
     if (
         AirPlayFlags.SupportsSystemPairing in flags
         or AirPlayFlags.SupportsCoreUtilsPairingAndEncryption in flags
diff --git a/pyatv/protocols/airplay/auth/hap_transient.py b/pyatv/protocols/airplay/auth/hap_transient.py
index cb1f0488..c9a986f8 100644
--- a/pyatv/protocols/airplay/auth/hap_transient.py
+++ b/pyatv/protocols/airplay/auth/hap_transient.py
@@ -47,7 +47,13 @@ class AirPlayHapTransientPairVerifyProcedure(PairVerifyProcedure):
 
         await self.http.post("/pair-pin-start", headers=_AIRPLAY_HEADERS)
 
-        data = {hap_tlv8.TlvValue.Method: b"\x00", hap_tlv8.TlvValue.SeqNo: b"\x01"}
+        data = {
+            hap_tlv8.TlvValue.Method: b"\x00",
+            hap_tlv8.TlvValue.SeqNo: b"\x01",
+            hap_tlv8.TlvValue.Flags: int.to_bytes(
+                hap_tlv8.Flags.TransientPairing.value, 1, byteorder="big"
+            ),
+        }
         resp = await self.http.post(
             "/pair-setup", body=hap_tlv8.write_tlv(data), headers=_AIRPLAY_HEADERS
         )
diff --git a/pyatv/protocols/airplay/auth/legacy.py b/pyatv/protocols/airplay/auth/legacy.py
index febbd328..cf582be5 100755
--- a/pyatv/protocols/airplay/auth/legacy.py
+++ b/pyatv/protocols/airplay/auth/legacy.py
@@ -12,7 +12,7 @@ from pyatv.auth.hap_pairing import (
     PairVerifyProcedure,
 )
 from pyatv.protocols.airplay.srp import LegacySRPAuthHandler
-from pyatv.support.http import HttpConnection, HttpResponse
+from pyatv.support.http import HttpConnection, HttpResponse, decode_bplist_from_body
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -53,14 +53,12 @@ class AirPlayLegacyPairSetupProcedure(PairSetupProcedure):
         )
         self.srp.step1(client_id, pin_code)
         resp = await self._send_plist(method="pin", user=client_id)
-        resp = plistlib.loads(
-            resp.body if isinstance(resp.body, bytes) else resp.body.encode("utf-8")
-        )
-        if not isinstance(resp, dict):
-            raise exceptions.ProtocolError(f"exoected dict, got {type(resp)}")
+        body = decode_bplist_from_body(resp)
+        if not isinstance(body, dict):
+            raise exceptions.ProtocolError(f"exoected dict, got {type(body).__name__}")
 
         # Step 2
-        pub_key, key_proof = self.srp.step2(resp["pk"], resp["salt"])
+        pub_key, key_proof = self.srp.step2(body["pk"], body["salt"])
         await self._send_plist(
             pk=binascii.unhexlify(pub_key), proof=binascii.unhexlify(key_proof)
         )
diff --git a/pyatv/protocols/airplay/mrp_connection.py b/pyatv/protocols/airplay/mrp_connection.py
index d6aaf5ec..dc345639 100644
--- a/pyatv/protocols/airplay/mrp_connection.py
+++ b/pyatv/protocols/airplay/mrp_connection.py
@@ -3,8 +3,8 @@ import logging
 from typing import Optional
 
 from pyatv import exceptions
+from pyatv.protocols.airplay.ap2_session import AP2Session
 from pyatv.protocols.airplay.channels import DataStreamChannel, DataStreamListener
-from pyatv.protocols.airplay.remote_control import RemoteControl
 from pyatv.protocols.mrp import protobuf
 from pyatv.protocols.mrp.connection import AbstractMrpConnection
 from pyatv.support import log_protobuf
@@ -17,13 +17,11 @@ class AirPlayMrpConnection(AbstractMrpConnection, DataStreamListener):
     """Transparent connection/channel for transporting MRP messages."""
 
     def __init__(
-        self,
-        remote_control: RemoteControl,
-        device_listener: Optional[StateProducer] = None,
+        self, session: AP2Session, device_listener: Optional[StateProducer] = None
     ):
         """Initialize a new MrpConnection."""
         super().__init__()
-        self.remote_control = remote_control
+        self.session = session
         self.data_channel: Optional[DataStreamChannel] = None
         self.device_listener = device_listener
 
@@ -33,10 +31,10 @@ class AirPlayMrpConnection(AbstractMrpConnection, DataStreamListener):
 
     async def connect(self) -> None:
         """Connect to device."""
-        if self.remote_control.data_channel is None:
+        if self.session.data_channel is None:
             raise exceptions.InvalidStateError("remote control channel not connected")
 
-        self.data_channel = self.remote_control.data_channel
+        self.data_channel = self.session.data_channel
         self.data_channel.listener = self
 
     def enable_encryption(self, output_key: bytes, input_key: bytes) -> None:
diff --git a/pyatv/protocols/airplay/player.py b/pyatv/protocols/airplay/player.py
index ba2926cb..ee9f8d07 100644
--- a/pyatv/protocols/airplay/player.py
+++ b/pyatv/protocols/airplay/player.py
@@ -6,7 +6,7 @@ import plistlib
 from uuid import uuid4
 
 from pyatv import exceptions
-from pyatv.support.http import HttpConnection
+from pyatv.support.http import HttpConnection, decode_bplist_from_body
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -76,11 +76,7 @@ class AirPlayPlayer:
             _LOGGER.debug("Playback-info: %s", resp)
 
             if resp.body:
-                parsed = plistlib.loads(
-                    resp.body.encode("utf-8")
-                    if isinstance(resp.body, str)
-                    else resp.body
-                )
+                parsed = decode_bplist_from_body(resp)
             else:
                 parsed = {}
                 _LOGGER.debug("Got playback-info response without content")
diff --git a/pyatv/protocols/airplay/server_auth.py b/pyatv/protocols/airplay/server_auth.py
index 3cfc217f..e5839cee 100644
--- a/pyatv/protocols/airplay/server_auth.py
+++ b/pyatv/protocols/airplay/server_auth.py
@@ -285,7 +285,7 @@ class BaseAirPlayServerAuth(http.HttpSimpleRouter, ABC):
         )
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
-        encrypted = chacha.encrypt(tlv, nounce="PV-Msg02".encode())
+        encrypted = chacha.encrypt(tlv, nonce="PV-Msg02".encode())
 
         tlv = {
             TlvValue.SeqNo: b"\x02",
@@ -370,7 +370,7 @@ class BaseAirPlayServerAuth(http.HttpSimpleRouter, ABC):
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
         decrypted_tlv_bytes = chacha.decrypt(
-            pairing_data[TlvValue.EncryptedData], nounce="PS-Msg05".encode()
+            pairing_data[TlvValue.EncryptedData], nonce="PS-Msg05".encode()
         )
 
         _LOGGER.debug("MSG5 EncryptedData=%s", read_tlv(decrypted_tlv_bytes))
@@ -397,7 +397,7 @@ class BaseAirPlayServerAuth(http.HttpSimpleRouter, ABC):
         )
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
-        encrypted = chacha.encrypt(tlv, nounce="PS-Msg06".encode())
+        encrypted = chacha.encrypt(tlv, nonce="PS-Msg06".encode())
 
         self.has_paired()
         return write_tlv({TlvValue.SeqNo: b"\x06", TlvValue.EncryptedData: encrypted})
diff --git a/pyatv/protocols/airplay/utils.py b/pyatv/protocols/airplay/utils.py
index f019eb0c..bf10d0af 100644
--- a/pyatv/protocols/airplay/utils.py
+++ b/pyatv/protocols/airplay/utils.py
@@ -1,9 +1,10 @@
 """Manage announced AirPlay features."""
-from enum import IntFlag
+from enum import Enum, IntFlag, auto
 import logging
+import math
 import plistlib
 import re
-from typing import Any, Mapping, Union
+from typing import Any, Dict, Mapping, Union
 
 from pyatv.auth.hap_pairing import (
     TRANSIENT_CREDENTIALS,
@@ -11,7 +12,9 @@ from pyatv.auth.hap_pairing import (
     HapCredentials,
 )
 from pyatv.const import PairingRequirement
+from pyatv.core import MutableService
 from pyatv.interface import BaseService
+from pyatv.support import map_range
 from pyatv.support.http import HttpRequest, HttpResponse
 
 # pylint: disable=invalid-name
@@ -21,6 +24,18 @@ PIN_REQUIRED = 0x8
 PASSWORD_BIT = 0x80
 LEGACY_PAIRING_BIT = 0x200
 
+DBFS_MIN = -30.0
+DBFS_MAX = 0.0
+PERCENTAGE_MIN = 0.0
+PERCENTAGE_MAX = 100.0
+
+
+class AirPlayMajorVersion(Enum):
+    """Major AirPlay protocol version."""
+
+    AirPlayV1 = auto()
+    AirPlayV2 = auto()
+
 
 def _get_flags(properties: Mapping[str, str]) -> int:
     # Flags are either present via "sf" or "flags"
@@ -157,9 +172,11 @@ def encode_plist_body(data: Any):
     )
 
 
-def decode_plist_body(body: Union[str, bytes]) -> Any:
+def decode_plist_body(body: Union[str, bytes, Dict[Any, Any]]) -> Any:
     """Decode a binary plist payload."""
     try:
+        if isinstance(body, Dict):
+            return body
         return plistlib.loads(body if isinstance(body, bytes) else body.encode("utf-8"))
     except plistlib.InvalidFileException:
         return None
@@ -199,3 +216,56 @@ def log_response(logger, response: HttpResponse, message_prefix="") -> None:
             response.protocol,
             payload,
         )
+
+
+# TODO: I don't know how to properly detect if a receiver support AirPlay 2 or not,
+# so I'm guessing until I know better. The basic idea here is simple: the service
+# should have the "features" flag (either "features" or "ft") and have all 64 bits
+# present. Not sure if that is good enough, but will do for now.
+def get_protocol_version(service: BaseService) -> AirPlayMajorVersion:
+    """Return major AirPlay version supported by a service."""
+    features = service.properties.get("ft")
+    if not features:
+        features = service.properties.get("features", "")
+
+    # Ensure feature flag is valid and all flags present (i.e.
+    # 0xAABBCCDD,0xAABBCCDD, thus looking for ",")
+    if features.count(",") == 1:
+        return AirPlayMajorVersion.AirPlayV2
+    return AirPlayMajorVersion.AirPlayV1
+
+
+def update_service_details(service: MutableService):
+    """Update AirPlay service according to that it supports."""
+    service.requires_password = is_password_required(service)
+
+    if service.properties.get("acl", "0") == "1":
+        # Access control might say that pairing is not possible, e.g. only devices
+        # belonging to the same home (not supported by pyatv)
+        service.pairing = PairingRequirement.Disabled
+    else:
+        service.pairing = get_pairing_requirement(service)
+
+
+def pct_to_dbfs(level: float) -> float:
+    """Convert percentage level to dBFS.
+
+    Used for volume levels in AirPlay.
+    """
+    # AirPlay uses -144.0 as muted volume, so re-map 0.0 to that
+    if math.isclose(level, 0.0):
+        return -144.0
+
+    # Map percentage to dBFS
+    return map_range(level, PERCENTAGE_MIN, PERCENTAGE_MAX, DBFS_MIN, DBFS_MAX)
+
+
+def dbfs_to_pct(level: float) -> float:
+    """Convert dBFS to percentage."""
+    # AirPlay uses -144.0 as "muted", but we treat everything below -30.0 as
+    # muted to be a bit defensive
+    if level < DBFS_MIN:
+        return PERCENTAGE_MIN
+
+    # Map dBFS to percentage
+    return map_range(level, DBFS_MIN, DBFS_MAX, PERCENTAGE_MIN, PERCENTAGE_MAX)
diff --git a/pyatv/protocols/companion/server_auth.py b/pyatv/protocols/companion/server_auth.py
index 950dcc20..a8368c87 100644
--- a/pyatv/protocols/companion/server_auth.py
+++ b/pyatv/protocols/companion/server_auth.py
@@ -118,7 +118,7 @@ class CompanionServerAuth(ABC):
         )
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
-        encrypted = chacha.encrypt(tlv, nounce="PV-Msg02".encode())
+        encrypted = chacha.encrypt(tlv, nonce="PV-Msg02".encode())
 
         tlv = write_tlv(
             {
@@ -182,7 +182,7 @@ class CompanionServerAuth(ABC):
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
         decrypted_tlv_bytes = chacha.decrypt(
-            pairing_data[TlvValue.EncryptedData], nounce="PS-Msg05".encode()
+            pairing_data[TlvValue.EncryptedData], nonce="PS-Msg05".encode()
         )
 
         _LOGGER.debug("MSG5 EncryptedData=%s", read_tlv(decrypted_tlv_bytes))
@@ -209,7 +209,7 @@ class CompanionServerAuth(ABC):
         tlv = write_tlv(tlv)
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
-        encrypted = chacha.encrypt(tlv, nounce="PS-Msg06".encode())
+        encrypted = chacha.encrypt(tlv, nonce="PS-Msg06".encode())
 
         tlv = write_tlv({TlvValue.SeqNo: b"\x06", TlvValue.EncryptedData: encrypted})
 
diff --git a/pyatv/protocols/mrp/server_auth.py b/pyatv/protocols/mrp/server_auth.py
index 71415a62..c096b127 100644
--- a/pyatv/protocols/mrp/server_auth.py
+++ b/pyatv/protocols/mrp/server_auth.py
@@ -148,7 +148,7 @@ class MrpServerAuth(ABC):
         )
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
-        encrypted = chacha.encrypt(tlv, nounce="PV-Msg02".encode())
+        encrypted = chacha.encrypt(tlv, nonce="PV-Msg02".encode())
 
         msg = messages.crypto_pairing(
             {
@@ -228,7 +228,7 @@ class MrpServerAuth(ABC):
         )
 
         chacha = chacha20.Chacha20Cipher(session_key, session_key)
-        encrypted = chacha.encrypt(tlv, nounce="PS-Msg06".encode())
+        encrypted = chacha.encrypt(tlv, nonce="PS-Msg06".encode())
 
         msg = messages.crypto_pairing(
             {TlvValue.SeqNo: b"\x06", TlvValue.EncryptedData: encrypted}
diff --git a/pyatv/protocols/raop/__init__.py b/pyatv/protocols/raop/__init__.py
index 13883244..14212260 100644
--- a/pyatv/protocols/raop/__init__.py
+++ b/pyatv/protocols/raop/__init__.py
@@ -3,11 +3,10 @@
 import asyncio
 import io
 import logging
-import math
 from typing import Any, Dict, Generator, Mapping, Optional, Set, Tuple, Union, cast
 
 from pyatv import const, exceptions
-from pyatv.auth.hap_pairing import AuthenticationType, parse_credentials
+from pyatv.auth.hap_pairing import AuthenticationType
 from pyatv.const import (
     DeviceModel,
     FeatureName,
@@ -45,17 +44,20 @@ from pyatv.interface import (
     RemoteControl,
     Stream,
 )
-from pyatv.protocols.airplay import service_info as airplay_service_info
+from pyatv.protocols.airplay.auth import extract_credentials
 from pyatv.protocols.airplay.pairing import AirPlayPairingHandler
-from pyatv.protocols.airplay.utils import AirPlayFlags, parse_features
-from pyatv.protocols.raop.audio_source import AudioSource, open_source
-from pyatv.protocols.raop.raop import (
-    PlaybackInfo,
-    RaopClient,
-    RaopContext,
-    RaopListener,
+from pyatv.protocols.airplay.utils import (
+    AirPlayFlags,
+    AirPlayMajorVersion,
+    dbfs_to_pct,
+    get_protocol_version,
+    parse_features,
+    pct_to_dbfs,
+    update_service_details,
 )
-from pyatv.support import map_range
+from pyatv.protocols.raop.audio_source import AudioSource, open_source
+from pyatv.protocols.raop.protocols import StreamContext, airplayv1, airplayv2
+from pyatv.protocols.raop.stream_client import PlaybackInfo, RaopListener, StreamClient
 from pyatv.support.collections import dict_merge
 from pyatv.support.device_info import lookup_model
 from pyatv.support.http import ClientSessionManager, HttpConnection, http_connect
@@ -66,11 +68,6 @@ _LOGGER = logging.getLogger(__name__)
 
 INITIAL_VOLUME = 33.0  # Percent
 
-DBFS_MIN = -30.0
-DBFS_MAX = 0.0
-PERCENTAGE_MIN = 0.0
-PERCENTAGE_MAX = 100.0
-
 
 class RaopPushUpdater(AbstractPushUpdater):
     """Implementation of push update support for RAOP."""
@@ -119,20 +116,20 @@ class RaopPlaybackManager:
         self._is_acquired: bool = False
         self._address: str = address
         self._port: int = port
-        self._context: RaopContext = RaopContext()
+        self._context: StreamContext = StreamContext()
         self._connection: Optional[HttpConnection] = None
         self._rtsp: Optional[RtspSession] = None
-        self._raop: Optional[RaopClient] = None
+        self._stream_client: Optional[StreamClient] = None
 
     @property
-    def context(self) -> RaopContext:
+    def context(self) -> StreamContext:
         """Return RTSP context if a session is active."""
         return self._context
 
     @property
-    def raop(self) -> Optional[RaopClient]:
-        """Return RAOP client if a session is active."""
-        return self._raop
+    def stream_client(self) -> Optional[StreamClient]:
+        """Return stream client if a session is active."""
+        return self._stream_client
 
     def acquire(self) -> None:
         """Acquire playback manager for playback."""
@@ -141,23 +138,36 @@ class RaopPlaybackManager:
 
         self._is_acquired = True
 
-    async def setup(self) -> Tuple[RaopClient, RtspSession, RaopContext]:
+    async def setup(self, service: BaseService) -> Tuple[StreamClient, StreamContext]:
         """Set up a session or return active if it exists."""
-        if self._raop and self._rtsp and self._context:
-            return self._raop, self._rtsp, self._context
+        if self._stream_client and self._rtsp and self._context:
+            return self._stream_client, self._context
 
         self._connection = await http_connect(self._address, self._port)
         self._rtsp = RtspSession(self._connection)
-        self._raop = RaopClient(self._rtsp, self._context)
-        return self._raop, self._rtsp, self._context
+
+        protocol_version = get_protocol_version(service)
+        _LOGGER.debug("Using AirPlay version %s", protocol_version)
+
+        protocol_class = (
+            airplayv1.AirPlayV1
+            if protocol_version == AirPlayMajorVersion.AirPlayV1
+            else airplayv2.AirPlayV2
+        )
+
+        self._stream_client = StreamClient(
+            self._rtsp, self._context, protocol_class(self._context, self._rtsp)
+        )
+        return self._stream_client, self._context
 
     async def teardown(self) -> None:
         """Tear down and disconnect current session."""
-        if self._raop:
-            self._raop.close()
+        if self._stream_client:
+            self._stream_client.close()
         if self._connection:
+            self._connection.close()
             self._connection = None
-        self._raop = None
+        self._stream_client = None
         self._context.reset()
         self._rtsp = None
         self._connection = None
@@ -228,7 +238,7 @@ class RaopFeatures(Features):
             return FeatureInfo(FeatureState.Available)
 
         if feature_name in [FeatureName.Stop, FeatureName.Pause]:
-            is_streaming = self.playback_manager.raop is not None
+            is_streaming = self.playback_manager.stream_client is not None
             return FeatureInfo(
                 FeatureState.Available if is_streaming else FeatureState.Unavailable
             )
@@ -262,7 +272,7 @@ class RaopAudio(Audio):
         volume = cast(float, message.value)
 
         _LOGGER.debug("Protocol %s changed volume to %f", message.protocol.name, volume)
-        self.playback_manager.context.volume = RaopAudio._pct_to_dbfs(volume)
+        self.playback_manager.context.volume = pct_to_dbfs(volume)
 
     @property
     def has_changed_volume(self) -> bool:
@@ -276,18 +286,12 @@ class RaopAudio(Audio):
         if vol is None:
             return INITIAL_VOLUME
 
-        # AirPlay uses -144.0 as "muted", but we treat everything below -30.0 as
-        # muted to be a bit defensive
-        if vol < DBFS_MIN:
-            return PERCENTAGE_MIN
-
-        # Map dBFS to percentage
-        return map_range(vol, DBFS_MIN, DBFS_MAX, PERCENTAGE_MIN, PERCENTAGE_MAX)
+        return dbfs_to_pct(vol)
 
     async def set_volume(self, level: float) -> None:
         """Change current volume level."""
-        raop = self.playback_manager.raop
-        dbfs_volume = RaopAudio._pct_to_dbfs(level)
+        raop = self.playback_manager.stream_client
+        dbfs_volume = pct_to_dbfs(level)
 
         if raop:
             await raop.set_volume(dbfs_volume)
@@ -304,15 +308,6 @@ class RaopAudio(Audio):
         """Decrease volume by one step."""
         await self.set_volume(max(self.volume - 5.0, 0.0))
 
-    @staticmethod
-    def _pct_to_dbfs(level: float) -> float:
-        # AirPlay uses -144.0 as muted volume, so re-map 0.0 to that
-        if math.isclose(level, 0.0):
-            return -144.0
-
-        # Map percentage to dBFS
-        return map_range(level, PERCENTAGE_MIN, PERCENTAGE_MAX, DBFS_MIN, DBFS_MAX)
-
 
 class RaopStream(Stream):
     """Implementation of stream functionality."""
@@ -349,9 +344,9 @@ class RaopStream(Stream):
             Audio, Metadata, PushUpdater, RemoteControl
         )
         try:
-            client, _, context = await self.playback_manager.setup()
-            client.credentials = parse_credentials(self.core.service.credentials)
-            client.password = self.core.service.password
+            client, context = await self.playback_manager.setup(self.core.service)
+            context.credentials = extract_credentials(self.core.service)
+            context.password = self.core.service.password
 
             client.listener = self.listener
             await client.initialize(self.core.service.properties)
@@ -373,6 +368,7 @@ class RaopStream(Stream):
             # If the user didn't change volume level prior to streaming, try to extract
             # volume level from device (if supported). Otherwise set the default level
             # in pyatv.
+            volume = None
             if not self.audio.has_changed_volume and "initialVolume" in client.info:
                 initial_volume = client.info["initialVolume"]
                 if not isinstance(initial_volume, float):
@@ -382,9 +378,15 @@ class RaopStream(Stream):
                     )
                 context.volume = initial_volume
             else:
-                await self.audio.set_volume(self.audio.volume)
-
-            await client.send_audio(audio_file, file_metadata)
+                # Try to set volume. If it fails, defer to setting it once
+                # streaming has started.
+                try:
+                    await self.audio.set_volume(self.audio.volume)
+                except Exception as ex:
+                    _LOGGER.debug("Failed to set volume (%s), delaying call", ex)
+                    volume = self.audio.volume
+
+            await client.send_audio(audio_file, file_metadata, volume=volume)
         finally:
             takeover_release()
             if audio_file:
@@ -404,13 +406,13 @@ class RaopRemoteControl(RemoteControl):
     # gives a better experience in Home Assistant.
     async def pause(self) -> None:
         """Press key pause."""
-        if self.playback_manager.raop:
-            self.playback_manager.raop.stop()
+        if self.playback_manager.stream_client:
+            self.playback_manager.stream_client.stop()
 
     async def stop(self) -> None:
         """Press key stop."""
-        if self.playback_manager.raop:
-            self.playback_manager.raop.stop()
+        if self.playback_manager.stream_client:
+            self.playback_manager.stream_client.stop()
 
     async def volume_up(self) -> None:
         """Press key volume up."""
@@ -489,7 +491,7 @@ async def service_info(
         service.pairing = PairingRequirement.Disabled
     else:
         # Same behavior as for AirPlay expected, so re-using that here
-        await airplay_service_info(service, devinfo, services)
+        update_service_details(service)
 
 
 def setup(  # pylint: disable=too-many-locals
diff --git a/pyatv/protocols/raop/alac.py b/pyatv/protocols/raop/alac.py
new file mode 100644
index 00000000..a9d94a91
--- /dev/null
+++ b/pyatv/protocols/raop/alac.py
@@ -0,0 +1,51 @@
+"""Dummy implementation of ALAC.
+
+This is not by any means a proper implementation of ALAC, it merely serves as an
+abstraction for creating ALAC frames compatible with the most common use case in
+pyatv. It also simplifies decoding of frames in tests.
+
+Some assumptions here:
+
+* Endianness of samples are swapped from whatever is passed
+* Samples are 16 bit/2 bytes
+* Number of samples in a frame is derived from frame size
+* End tag (111) is present at the end of a frame
+"""
+from bitarray import bitarray
+
+
+def encode(frames: bytes) -> bytes:
+    """Encode audio samples into an ALAC frame."""
+    # ALAC frame with raw data. Not so pretty but will work for now until a
+    # proper ALAC encoder is added.
+    # This assumes stereo mode. Basically reversed from here:
+    # https://github.com/libav/libav/blob/c4642788e83b0858bca449f9b6e71ddb015dfa5d/libavcodec/alac.c#L407
+    audio = bitarray()
+    audio.extend("001" + (19 * "0") + "1")
+
+    # Due to endianness, bytes must be sweapped here. Maybe format can be changed so
+    # we don't have to?
+    for i in range(0, len(frames), 2):
+        audio.frombytes(bytes([frames[i + 1], frames[i]]))
+
+    audio.extend("111")  # End tag (libav is picky about this)
+
+    return audio.tobytes()
+
+
+def decode(frame: bytes) -> bytes:
+    """Decode audio sameples from an ALAC frame."""
+    buffer = bitarray()
+    buffer.frombytes(frame)
+
+    # Audio starts 23 bits in and last bits must be removed because of the end tag
+    buffer = buffer[23:-9]
+
+    # Change byte order of samples to match encode
+    for i in range(0, len(buffer), 16):
+        lower_sample = buffer[i : i + 8]
+        upper_sample = buffer[i + 8 : i + 16]
+        buffer[i : i + 8] = upper_sample
+        buffer[i + 8 : i + 16] = lower_sample
+
+    return buffer.tobytes()
diff --git a/pyatv/protocols/raop/protocols/__init__.py b/pyatv/protocols/raop/protocols/__init__.py
new file mode 100644
index 00000000..34346cac
--- /dev/null
+++ b/pyatv/protocols/raop/protocols/__init__.py
@@ -0,0 +1,90 @@
+"""Base classes used by streaming protocols."""
+from abc import ABC, abstractmethod
+import asyncio
+from random import randrange
+from typing import Optional, Tuple
+
+from pyatv.auth.hap_pairing import NO_CREDENTIALS, HapCredentials
+from pyatv.protocols.raop import timing
+from pyatv.support.rtsp import FRAMES_PER_PACKET
+
+
+class StreamContext:
+    """Data used for one RAOP session."""
+
+    def __init__(self) -> None:
+        """Initialize a new StreamContext."""
+        self.credentials: HapCredentials = NO_CREDENTIALS
+        self.password: Optional[str] = None
+
+        self.sample_rate: int = 44100
+        self.channels: int = 2
+        self.bytes_per_channel: int = 2
+        self.latency = 22050 + self.sample_rate
+
+        self.rtpseq: int = 0
+        self.start_ts = 0
+        self.head_ts = 0
+        self.padding_sent: int = 0
+
+        self.server_port: int = 0
+        self.event_port: int = 0
+        self.control_port: int = 0
+        self.timing_port: int = 0
+        self.rtsp_session: int = 0
+
+        self.volume: Optional[float] = None
+
+    def reset(self) -> None:
+        """Reset seasion.
+
+        Must be done when sample rate changes.
+        """
+        self.rtpseq = randrange(2**16)
+        self.start_ts = timing.ntp2ts(timing.ntp_now(), self.sample_rate)
+        self.head_ts = self.start_ts
+        self.latency = 22050 + self.sample_rate
+        self.padding_sent = 0
+
+    @property
+    def rtptime(self) -> int:
+        """Current RTP time with latency."""
+        return self.head_ts - (self.start_ts - self.latency)
+
+    @property
+    def position(self) -> float:
+        """Current position in stream (seconds with fraction)."""
+        # Do not consider latency here (so do not use rtptime)
+        return timing.ts2ms(self.head_ts - self.start_ts, self.sample_rate) / 1000.0
+
+    @property
+    def frame_size(self) -> int:
+        """Size of a single audio frame."""
+        return self.channels * self.bytes_per_channel
+
+    @property
+    def packet_size(self) -> int:
+        """Size of a full audio packet."""
+        return FRAMES_PER_PACKET * self.frame_size
+
+
+class StreamProtocol(ABC):
+    """Base interface for a streaming protocol."""
+
+    @abstractmethod
+    async def setup(self, timing_client_port: int, control_client_port: int) -> None:
+        """To setup connection prior to starting to stream."""
+
+    @abstractmethod
+    def teardown(self) -> None:
+        """Teardown resources allocated by setup efter streaming finished."""
+
+    @abstractmethod
+    async def start_feedback(self) -> None:
+        """Start to send feedback (if supported and required)."""
+
+    @abstractmethod
+    async def send_audio_packet(
+        self, transport: asyncio.DatagramTransport, rtp_header: bytes, audio: bytes
+    ) -> Tuple[int, bytes]:
+        """Send audio packet to receiver."""
diff --git a/pyatv/protocols/raop/protocols/airplayv1.py b/pyatv/protocols/raop/protocols/airplayv1.py
new file mode 100644
index 00000000..983aebfb
--- /dev/null
+++ b/pyatv/protocols/raop/protocols/airplayv1.py
@@ -0,0 +1,109 @@
+"""Implementation of AirPlay v1 protocol logic."""
+import asyncio
+import logging
+from typing import List, Mapping, Optional, Tuple
+
+from pyatv import exceptions
+from pyatv.protocols.airplay.auth import pair_verify
+from pyatv.protocols.raop.protocols import StreamContext, StreamProtocol
+from pyatv.support.rtsp import RtspSession
+
+_LOGGER = logging.getLogger(__name__)
+
+KEEP_ALIVE_INTERVAL = 25  # Seconds
+
+
+def parse_transport(transport: str) -> Tuple[List[str], Mapping[str, str]]:
+    """Parse Transport header in SETUP response."""
+    params = []
+    options = {}
+    for option in transport.split(";"):
+        if "=" in option:
+            key, value = option.split("=", maxsplit=1)
+            options[key] = value
+        else:
+            params.append(option)
+    return params, options
+
+
+class AirPlayV1(StreamProtocol):
+    """Stream protocol used for AirPlay v1 support."""
+
+    def __init__(self, context: StreamContext, rtsp: RtspSession) -> None:
+        """Initialize a new AirPlayV1 instance."""
+        super().__init__()
+        self.context = context
+        self.rtsp = rtsp
+        self._keep_alive_task: Optional[asyncio.Future] = None
+
+    async def setup(self, timing_client_port: int, control_client_port: int) -> None:
+        """To setup connection prior to starting to stream."""
+        verifier = pair_verify(self.context.credentials, self.rtsp.connection)
+        await verifier.verify_credentials()
+
+        await self.rtsp.announce(
+            self.context.bytes_per_channel,
+            self.context.channels,
+            self.context.sample_rate,
+            self.context.password,
+        )
+
+        resp = await self.rtsp.setup(
+            headers={
+                "Transport": (
+                    "RTP/AVP/UDP;unicast;interleaved=0-1;mode=record;"
+                    f"control_port={control_client_port};"
+                    f"timing_port={timing_client_port}"
+                )
+            }
+        )
+        _, options = parse_transport(resp.headers["Transport"])
+        self.context.timing_port = int(options.get("timing_port", 0))
+        self.context.control_port = int(options["control_port"])
+        self.context.rtsp_session = int(resp.headers["Session"])
+        self.context.server_port = int(options["server_port"])
+
+        _LOGGER.debug(
+            "Remote ports: control=%d, timing=%d, server=%d",
+            self.context.control_port,
+            self.context.timing_port,
+            self.context.server_port,
+        )
+
+    def teardown(self) -> None:
+        """Teardown resources allocated by setup efter streaming finished."""
+        if self._keep_alive_task:
+            self._keep_alive_task.cancel()
+            self._keep_alive_task = None
+
+    async def start_feedback(self) -> None:
+        """Start to send feedback (if supported and required)."""
+        feedback = await self.rtsp.feedback(allow_error=True)
+        if feedback.code == 200:
+            self._keep_alive_task = asyncio.ensure_future(self._send_keep_alive())
+        else:
+            _LOGGER.debug("Keep-alive not supported, not starting task")
+
+    async def _send_keep_alive(self) -> None:
+        _LOGGER.debug("Starting keep-alive task")
+
+        while True:
+            try:
+                await asyncio.sleep(KEEP_ALIVE_INTERVAL)
+
+                _LOGGER.debug("Sending keep-alive feedback")
+                await self.rtsp.feedback()
+            except asyncio.CancelledError:
+                break
+            except exceptions.ProtocolError:
+                _LOGGER.exception("feedback failed")
+
+        _LOGGER.debug("Feedback task finished")
+
+    async def send_audio_packet(
+        self, transport: asyncio.DatagramTransport, rtp_header: bytes, audio: bytes
+    ) -> Tuple[int, bytes]:
+        """Send audio packet to receiver."""
+        packet = rtp_header + audio
+        transport.sendto(packet)
+        return self.context.rtpseq, packet
diff --git a/pyatv/protocols/raop/protocols/airplayv2.py b/pyatv/protocols/raop/protocols/airplayv2.py
new file mode 100644
index 00000000..38009a25
--- /dev/null
+++ b/pyatv/protocols/raop/protocols/airplayv2.py
@@ -0,0 +1,163 @@
+"""Implementation of AirPlay v2 protocol logic."""
+import asyncio
+import logging
+from typing import Optional, Tuple
+from uuid import uuid4
+
+from pyatv.auth.hap_channel import setup_channel
+from pyatv.protocols.airplay.auth import verify_connection
+from pyatv.protocols.airplay.channels import EventChannel
+from pyatv.protocols.raop.protocols import StreamContext, StreamProtocol
+from pyatv.support.chacha20 import Chacha20Cipher
+from pyatv.support.http import decode_bplist_from_body
+from pyatv.support.rtsp import RtspSession
+
+_LOGGER = logging.getLogger(__name__)
+
+EVENTS_SALT = "Events-Salt"
+EVENTS_WRITE_INFO = "Events-Write-Encryption-Key"
+EVENTS_READ_INFO = "Events-Read-Encryption-Key"
+
+FEEDBACK_INTERVAL = 2.0  # Seconds
+
+
+class AirPlayV2(StreamProtocol):
+    """Stream protocol used for AirPlay v1 support."""
+
+    def __init__(self, context: StreamContext, rtsp: RtspSession) -> None:
+        """Initialize a new AirPlayV2 instance."""
+        super().__init__()
+        self.context = context
+        self.rtsp = rtsp
+        self.event_channel: Optional[asyncio.BaseTransport] = None
+        self._cipher: Optional[Chacha20Cipher] = None
+        self._feedback_task: Optional[asyncio.Task] = None
+
+    async def setup(self, timing_client_port: int, control_client_port: int) -> None:
+        """To setup connection prior to starting to stream."""
+        verifier = await verify_connection(
+            self.context.credentials, self.rtsp.connection
+        )
+
+        setup_resp = await self.rtsp.setup(
+            body={
+                "deviceID": "AA:BB:CC:DD:EE:FF",
+                "sessionUUID": str(uuid4()).upper(),
+                "timingPort": timing_client_port,
+                "timingProtocol": "NTP",
+                "sourceVersion": "409.16",
+            }
+        )
+        resp = decode_bplist_from_body(setup_resp)
+        _LOGGER.debug("Setup response body: %s", resp)
+
+        event_port = resp.get("eventPort", 0)
+
+        # This is a bit hacky, but I noticed that airplay2-receiver seems to set up the
+        # event channel some time after responding with the port used for it. So the
+        # connect call fails. By doing some retries here, it will work after some
+        # attempts. So keeping that here for compatibility reason.
+        retries = 5
+        transport = None
+        while transport is None:
+            try:
+                transport, _ = await setup_channel(
+                    EventChannel,
+                    verifier,
+                    self.rtsp.connection.remote_ip,
+                    event_port,
+                    EVENTS_SALT,
+                    EVENTS_READ_INFO,
+                    EVENTS_WRITE_INFO,
+                )
+            except ConnectionRefusedError:
+                retries -= 1
+                if retries == 0:
+                    raise
+
+                _LOGGER.debug("Connect failed, retrying")
+                await asyncio.sleep(1.0)
+
+        self.event_channel = transport
+
+        # Ok, so this is not really correct. I believe the shared secret should be used
+        # as base for the shared key, but it's hard to get hold of that here
+        # (abstractions). It doesn't really matter what the key is (could be hardcoded)
+        # as it's merely a security feature. For the sake of it, derive a key from event
+        # parameters, it won't hurt (and the key will be different every time).
+        out_key, _ = verifier.encryption_keys(
+            EVENTS_SALT, EVENTS_WRITE_INFO, EVENTS_READ_INFO
+        )
+        shared_secret = out_key[0:32]
+
+        setup_resp = await self.rtsp.setup(
+            body={
+                "streams": [
+                    {
+                        "audioFormat": 0x40000,
+                        "audioMode": "default",
+                        "controlPort": control_client_port,
+                        "ct": 2,  # Channels?
+                        "isMedia": True,
+                        "latencyMax": 88200,
+                        "latencyMin": 11025,
+                        "shk": shared_secret,
+                        "spf": 352,  # Samples Per Frame
+                        "sr": 44100,  # Sample rate
+                        "type": 0x60,
+                        "supportsDynamicStreamID": False,
+                        "streamConnectionID": self.rtsp.session_id,
+                    }
+                ]
+            }
+        )
+        resp = decode_bplist_from_body(setup_resp)
+        _LOGGER.debug("Setup stream response: %s", resp)
+
+        stream = resp["streams"][0]
+
+        self.context.control_port = stream["controlPort"]
+        self.context.server_port = stream["dataPort"]
+
+        self._cipher = Chacha20Cipher(shared_secret, shared_secret)
+
+    def teardown(self) -> None:
+        """Teardown resources allocated by setup efter streaming finished."""
+        if self._feedback_task:
+            self._feedback_task.cancel()
+            self._feedback_task = None
+        if self.event_channel:
+            self.event_channel.close()
+            self.event_channel = None
+
+    async def start_feedback(self) -> None:
+        """Start to send feedback (if supported and required)."""
+        if self._feedback_task is None:
+            self._feedback_task = asyncio.create_task(self._feedback_task_loop())
+
+    async def _feedback_task_loop(self) -> None:
+        _LOGGER.debug("Starting feedback task")
+        # TODO: Better end condition here to not risk infinite runs?
+        while True:
+            try:
+                await self.rtsp.feedback()
+            except Exception:
+                _LOGGER.exception("feedback failed")
+            await asyncio.sleep(FEEDBACK_INTERVAL)
+
+    async def send_audio_packet(
+        self, transport: asyncio.DatagramTransport, rtp_header: bytes, audio: bytes
+    ) -> Tuple[int, bytes]:
+        """Send audio packet to receiver."""
+        # TODO: This part is extremely sub-optimized. Should at least use a memoryview
+        # and do in-place operations to avoid copying memory left and right.
+        if self._cipher:
+            nonce = self._cipher.out_nonce
+            aad = rtp_header[4:12]
+            audio = self._cipher.encrypt(audio, nonce=nonce, aad=aad)
+
+        packet = rtp_header + audio + nonce
+
+        transport.sendto(packet)
+
+        return self.context.rtpseq, packet
diff --git a/pyatv/protocols/raop/raop.py b/pyatv/protocols/raop/stream_client.py
similarity index 79%
rename from pyatv/protocols/raop/raop.py
rename to pyatv/protocols/raop/stream_client.py
index 6ac520dd..57378053 100644
--- a/pyatv/protocols/raop/raop.py
+++ b/pyatv/protocols/raop/stream_client.py
@@ -1,18 +1,19 @@
-"""Support for RAOP (AirPlay v1)."""
+"""Client for AirPlay audio streaming.
+
+This is a "generic" client designed around how AirPlay works in regards to streaming.
+The client uses an underlying "protocol" for protocol specific bits, i.e. to support
+AirPlay v1 and/or v2. This is mainly for code re-use purposes.
+"""
 from abc import ABC, abstractmethod
 import asyncio
 import logging
-from random import randrange
 from time import perf_counter
-from typing import Any, Dict, List, Mapping, NamedTuple, Optional, Tuple, cast
+from typing import Any, Dict, Mapping, NamedTuple, Optional, Tuple, cast
 import weakref
 
-from bitarray import bitarray
-
 from pyatv import exceptions
-from pyatv.auth.hap_pairing import HapCredentials
-from pyatv.protocols.airplay.auth import pair_verify
-from pyatv.protocols.raop import timing
+from pyatv.protocols.airplay.utils import pct_to_dbfs
+from pyatv.protocols.raop import alac, timing
 from pyatv.protocols.raop.audio_source import AudioSource
 from pyatv.protocols.raop.fifo import PacketFifo
 from pyatv.protocols.raop.packets import (
@@ -28,6 +29,7 @@ from pyatv.protocols.raop.parsers import (
     get_encryption_types,
     get_metadata_types,
 )
+from pyatv.protocols.raop.protocols import StreamContext, StreamProtocol
 from pyatv.support import log_binary
 from pyatv.support.metadata import EMPTY_METADATA, MediaMetadata
 from pyatv.support.rtsp import FRAMES_PER_PACKET, RtspSession
@@ -40,74 +42,17 @@ MAX_PACKETS_COMPENSATE = 3
 # We should store this many packets in case retransmission is requested
 PACKET_BACKLOG_SIZE = 1000
 
-KEEP_ALIVE_INTERVAL = 25  # Seconds
-
 # Number of "too slow to keep up" warnings to suppress before warning about them
 SLOW_WARNING_THRESHOLD = 5
 
 # Metadata used when no metadata is present
 MISSING_METADATA = MediaMetadata(
-    title="Streaming with pyatv", artist="pyatv", album="RAOP", duration=0.0
+    title="Streaming with pyatv", artist="pyatv", album="AirPlay", duration=0.0
 )
 
 SUPPORTED_ENCRYPTIONS = EncryptionType.Unencrypted | EncryptionType.MFiSAP
 
 
-class RaopContext:
-    """Data used for one RAOP session."""
-
-    def __init__(self) -> None:
-        """Initialize a new RaopContext."""
-        self.sample_rate: int = 44100
-        self.channels: int = 2
-        self.bytes_per_channel: int = 2
-        self.latency = 22050 + self.sample_rate
-
-        self.rtpseq: int = 0
-        self.start_ts = 0
-        self.head_ts = 0
-        self.padding_sent: int = 0
-
-        self.server_port: int = 0
-        self.control_port: int = 0
-        self.timing_port: int = 0
-        self.rtsp_session: int = 0
-
-        self.volume: Optional[float] = None
-
-    def reset(self) -> None:
-        """Reset seasion.
-
-        Must be done when sample rate changes.
-        """
-        self.rtpseq = randrange(2**16)
-        self.start_ts = timing.ntp2ts(timing.ntp_now(), self.sample_rate)
-        self.head_ts = self.start_ts
-        self.latency = 22050 + self.sample_rate
-        self.padding_sent = 0
-
-    @property
-    def rtptime(self) -> int:
-        """Current RTP time with latency."""
-        return self.head_ts - (self.start_ts - self.latency)
-
-    @property
-    def position(self) -> float:
-        """Current position in stream (seconds with fraction)."""
-        # Do not consider latency here (so do not use rtptime)
-        return timing.ts2ms(self.head_ts - self.start_ts, self.sample_rate) / 1000.0
-
-    @property
-    def frame_size(self) -> int:
-        """Size of a single audio frame."""
-        return self.channels * self.bytes_per_channel
-
-    @property
-    def packet_size(self) -> int:
-        """Size of a full audio packet."""
-        return FRAMES_PER_PACKET * self.frame_size
-
-
 class PlaybackInfo(NamedTuple):
     """Information for what is currently playing."""
 
@@ -118,7 +63,7 @@ class PlaybackInfo(NamedTuple):
 class ControlClient(asyncio.Protocol):
     """Control client responsible for e.g. sync packets."""
 
-    def __init__(self, context: RaopContext, packet_backlog: PacketFifo):
+    def __init__(self, context: StreamContext, packet_backlog: PacketFifo):
         """Initialize a new ControlClient."""
         self.transport = None
         self.context = context
@@ -160,7 +105,7 @@ class ControlClient(asyncio.Protocol):
 
         first_packet = True
         current_time = timing.ts2ntp(self.context.head_ts, self.context.sample_rate)
-        while True:
+        while self.transport is not None:
             current_sec, current_frac = timing.ntp2parts(current_time)
             packet = SyncPacket.encode(
                 0x90 if first_packet else 0x80,
@@ -226,7 +171,7 @@ class ControlClient(asyncio.Protocol):
     @staticmethod
     def error_received(exc):
         """Handle a connection error."""
-        _LOGGER.error("Comtrol connection error: %s", exc)
+        _LOGGER.error("Control connection error: %s", exc)
 
     def connection_lost(self, exc):
         """Handle that connection was lost."""
@@ -309,19 +254,6 @@ class AudioProtocol(asyncio.Protocol):
         _LOGGER.debug("Audio connection lost (%s)", exc)
 
 
-def parse_transport(transport: str) -> Tuple[List[str], Mapping[str, str]]:
-    """Parse Transport header in SETUP response."""
-    params = []
-    options = {}
-    for option in transport.split(";"):
-        if "=" in option:
-            key, value = option.split("=", maxsplit=1)
-            options[key] = value
-        else:
-            params.append(option)
-    return params, options
-
-
 class RaopListener(ABC):
     """Listener interface for RAOP state changes."""
 
@@ -334,31 +266,30 @@ class RaopListener(ABC):
         """Media stopped playing."""
 
 
-class RaopClient:
-    """Simple RAOP client to stream audio."""
+class StreamClient:
+    """Simple client to stream audio."""
 
     def __init__(
         self,
         rtsp: RtspSession,
-        context: RaopContext,
+        context: StreamContext,
+        protocol: StreamProtocol,
     ):
-        """Initialize a new RaopClient instance."""
+        """Initialize a new StreamClient instance."""
         self.loop = asyncio.get_event_loop()
         self.rtsp: RtspSession = rtsp
-        self.context: RaopContext = context
-        self.credentials: Optional[HapCredentials] = None
-        self.password: Optional[str] = None
+        self.context: StreamContext = context
         self.control_client: Optional[ControlClient] = None
         self.timing_client: Optional[TimingClient] = None
         self._packet_backlog: PacketFifo = PacketFifo(PACKET_BACKLOG_SIZE)
         self._encryption_types: EncryptionType = EncryptionType.Unknown
         self._metadata_types: MetadataType = MetadataType.NotSupported
         self._metadata: MediaMetadata = EMPTY_METADATA
-        self._keep_alive_task: Optional[asyncio.Future] = None
         self._listener: Optional[weakref.ReferenceType[Any]] = None
         self._info: Dict[str, object] = {}
         self._properties: Mapping[str, str] = {}
         self._is_playing: bool = False
+        self._protocol: StreamProtocol = protocol
 
     @property
     def listener(self):
@@ -390,28 +321,11 @@ class RaopClient:
 
     def close(self):
         """Close session and free up resources."""
+        self._protocol.teardown()
         if self.control_client:
             self.control_client.close()
         if self.timing_client:
             self.timing_client.close()
-        if self._keep_alive_task:
-            self._keep_alive_task.cancel()
-
-    async def _send_keep_alive(self):
-        _LOGGER.debug("Starting keep-alive task")
-
-        while True:
-            try:
-                await asyncio.sleep(KEEP_ALIVE_INTERVAL)
-
-                _LOGGER.debug("Sending keep-alive feedback")
-                await self.rtsp.feedback()
-            except asyncio.CancelledError:
-                break
-            except exceptions.ProtocolError:
-                _LOGGER.exception("feedback failed")
-
-        _LOGGER.debug("Feedback task finished")
 
     async def initialize(self, properties: Mapping[str, str]):
         """Initialize the session."""
@@ -453,8 +367,12 @@ class RaopClient:
         self._info.update(await self.rtsp.info())
         _LOGGER.debug("Updated info parameters to: %s", self.info)
 
+        # Handle some special authentication cases
+        if self._requires_auth_setup:
+            await self.rtsp.auth_setup()
+
         # Set up the streaming session
-        await self._setup_session()
+        await self._protocol.setup(self.timing_client.port, self.control_client.port)
 
     def _update_output_properties(self, properties: Mapping[str, str]) -> None:
         (
@@ -469,42 +387,6 @@ class RaopClient:
             self.context.bytes_per_channel * 8,
         )
 
-    async def _setup_session(self):
-        if self._requires_auth_setup:
-            await self.rtsp.auth_setup()
-
-        verifier = pair_verify(self.credentials, self.rtsp.connection)
-        await verifier.verify_credentials()
-
-        await self.rtsp.announce(
-            self.context.bytes_per_channel,
-            self.context.channels,
-            self.context.sample_rate,
-            self.password,
-        )
-
-        resp = await self.rtsp.setup(
-            headers={
-                "Transport": (
-                    "RTP/AVP/UDP;unicast;interleaved=0-1;mode=record;"
-                    f"control_port={self.control_client.port};"
-                    f"timing_port={self.timing_client.port}"
-                )
-            }
-        )
-        _, options = parse_transport(resp.headers["Transport"])
-        self.context.timing_port = int(options.get("timing_port", 0))
-        self.context.control_port = int(options["control_port"])
-        self.context.rtsp_session = resp.headers["Session"]
-        self.context.server_port = int(options["server_port"])
-
-        _LOGGER.debug(
-            "Remote ports: control=%d, timing=%d, server=%d",
-            self.context.control_port,
-            self.context.timing_port,
-            self.context.server_port,
-        )
-
     @property
     def _requires_auth_setup(self):
         # Do auth-setup if MFiSAP encryption is supported by receiver. Also,
@@ -528,7 +410,11 @@ class RaopClient:
         self.context.volume = volume
 
     async def send_audio(  # pylint: disable=too-many-branches
-        self, wave_file: AudioSource, metadata: MediaMetadata = EMPTY_METADATA
+        self,
+        wave_file: AudioSource,
+        metadata: MediaMetadata = EMPTY_METADATA,
+        /,
+        volume: Optional[float] = None
     ):
         """Send an audio stream to the device."""
         if self.control_client is None or self.timing_client is None:
@@ -566,18 +452,16 @@ class RaopClient:
                 )
 
             # Start keep-alive task to ensure connection is not closed by remote device
-            feedback = await self.rtsp.feedback(allow_error=True)
-            if feedback.code == 200:
-                self._keep_alive_task = asyncio.ensure_future(self._send_keep_alive())
-            else:
-                _LOGGER.debug("Keep-alive not supported, not starting task")
+            await self._protocol.start_feedback()
 
             listener = self.listener
             if listener:
                 listener.playing(self.playback_info)
 
             # Start playback
-            await self.rtsp.record(
+            await self.rtsp.record()
+
+            await self.rtsp.flush(
                 headers={
                     "Range": "npt=0-",
                     "Session": self.context.rtsp_session,
@@ -587,6 +471,9 @@ class RaopClient:
                 }
             )
 
+            if volume:
+                await self.set_volume(pct_to_dbfs(volume))
+
             await self._stream_data(wave_file, transport)
         except (  # pylint: disable=try-except-raise
             exceptions.ProtocolError,
@@ -603,10 +490,8 @@ class RaopClient:
                 # more audio files. Refactor when support for that is added.
                 await self.rtsp.teardown(self.context.rtsp_session)
                 transport.close()
-            if self._keep_alive_task:
-                self._keep_alive_task.cancel()
-                self._keep_alive_task = None
-            self.control_client.stop()
+            self._protocol.teardown()
+            self.close()
 
             listener = self.listener
             if listener:
@@ -725,21 +610,17 @@ class RaopClient:
             self.rtsp.session_id,
         )
 
-        # ALAC frame with raw data. Not so pretty but will work for now until a
-        # proper ALAC encoder is added.
-        audio = bitarray("00" + str(self.context.channels - 1) + 19 * "0" + "1")
-        for i in range(0, len(frames), 2):
-            audio.frombytes(bytes([frames[i + 1], frames[i]]))
+        audio = alac.encode(frames)
 
         if transport.is_closing():
             _LOGGER.warning("Connection closed while streaming audio")
             return 0
 
-        packet = header + audio.tobytes()
-
-        # Add packet to backlog before sending
-        self._packet_backlog[self.context.rtpseq] = packet
-        transport.sendto(packet)
+        # Send packet and add it to backlog
+        rtpseq, packet = await self._protocol.send_audio_packet(
+            transport, header, audio
+        )
+        self._packet_backlog[rtpseq] = packet
 
         self.context.rtpseq = (self.context.rtpseq + 1) % (2**16)
         self.context.head_ts += int(len(frames) / self.context.frame_size)
diff --git a/pyatv/scripts/atvproxy.py b/pyatv/scripts/atvproxy.py
index 465f1239..98afe387 100755
--- a/pyatv/scripts/atvproxy.py
+++ b/pyatv/scripts/atvproxy.py
@@ -22,12 +22,7 @@ from pyatv.auth.server_auth import SERVER_IDENTIFIER
 from pyatv.const import Protocol
 from pyatv.core import MutableService, mdns
 from pyatv.protocols.airplay import extract_credentials, verify_connection
-from pyatv.protocols.airplay.channels import (
-    BaseDataStreamChannel,
-    BaseEventChannel,
-    DataStreamMessage,
-)
-from pyatv.protocols.airplay.remote_control import (
+from pyatv.protocols.airplay.ap2_session import (
     DATASTREAM_INPUT_INFO,
     DATASTREAM_OUTPUT_INFO,
     DATASTREAM_SALT,
@@ -35,6 +30,11 @@ from pyatv.protocols.airplay.remote_control import (
     EVENTS_SALT,
     EVENTS_WRITE_INFO,
 )
+from pyatv.protocols.airplay.channels import (
+    BaseDataStreamChannel,
+    BaseEventChannel,
+    DataStreamMessage,
+)
 from pyatv.protocols.airplay.server_auth import BaseAirPlayServerAuth
 from pyatv.protocols.airplay.utils import (
     decode_plist_body,
diff --git a/pyatv/scripts/atvremote.py b/pyatv/scripts/atvremote.py
index 7b0934d2..9eed1224 100644
--- a/pyatv/scripts/atvremote.py
+++ b/pyatv/scripts/atvremote.py
@@ -443,6 +443,13 @@ async def cli_handler(loop):
         default=False,
     )
 
+    parser.add_argument(
+        "--service-properties",
+        dest="service_properties",
+        default="",
+        help="manual MDNS properties",
+    )
+
     parser.add_argument(
         "--raop-password",
         help="optional password for raop",
@@ -480,6 +487,8 @@ async def cli_handler(loop):
     args = parser.parse_args()
     if args.manual and isinstance(args.id, list):
         parser.error("--manual only supports one identifier to --id")
+    if not args.manual and args.service_properties:
+        parser.error("--service-properties only allowed with --manual")
 
     loglevel = logging.WARNING
     if args.verbose:
@@ -556,8 +565,20 @@ async def _autodiscover_device(args, loop):
 
 
 def _manual_device(args):
+    properties = {}
+    if args.service_properties:
+        # Service properties are specified according to Xvar1=val1Xvar2=var2"
+        # where X is any character not in any variable or value. Examples would be
+        # "":name=test:type=something" or ",name=test,type=something" (both are
+        # equal). No proper error checking here!
+        split_char = args.service_properties[0]
+        properties = dict(
+            var.split("=", maxsplit=1)
+            for var in args.service_properties[1:].split(split_char)
+        )
+
     config = AppleTV(IPv4Address(args.address), args.name)
-    service = ManualService(args.id, args.protocol, args.port, {})
+    service = ManualService(args.id, args.protocol, args.port, properties)
     service.credentials = getattr(args, f"{args.protocol.name.lower()}_credentials")
     service.password = args.raop_password
     config.add_service(service)
diff --git a/pyatv/support/chacha20.py b/pyatv/support/chacha20.py
index 547b1448..287f7093 100644
--- a/pyatv/support/chacha20.py
+++ b/pyatv/support/chacha20.py
@@ -1,4 +1,6 @@
 """Transparent encryption layer using Chacha20_Poly1305."""
+from typing import Optional
+
 from chacha20poly1305_reuseable import ChaCha20Poly1305Reusable as ChaCha20Poly1305
 
 NONCE_LENGTH = 12
@@ -15,30 +17,48 @@ class Chacha20Cipher:
         self._in_counter = 0
         self._nonce_length = nonce_length
 
-    def encrypt(self, data, nounce=None, aad=None):
-        """Encrypt data with counter or specified nounce."""
-        if nounce is None:
-            nounce = self._out_counter.to_bytes(
-                length=self._nonce_length, byteorder="little"
-            )
+    @property
+    def out_nonce(self) -> bytes:
+        """Return next encrypt nonce.
+
+        This is the nonce that will be used by encrypt in the _next_ call if no custom
+        nonce is specified.
+        """
+        return self._out_counter.to_bytes(length=self._nonce_length, byteorder="little")
+
+    @property
+    def in_nonce(self) -> bytes:
+        """Return next decrypt nonce.
+
+        This is the nonce that will be used by decrypt in the _next_ call if no custom
+        nonce is specified.
+        """
+        return self._in_counter.to_bytes(length=self._nonce_length, byteorder="little")
+
+    def encrypt(
+        self, data: bytes, nonce: Optional[bytes] = None, aad: Optional[bytes] = None
+    ) -> bytes:
+        """Encrypt data with counter or specified nonce."""
+        if nonce is None:
+            nonce = self.out_nonce
             self._out_counter += 1
 
-        if len(nounce) < NONCE_LENGTH:
-            nounce = b"\x00" * (NONCE_LENGTH - len(nounce)) + nounce
+        if len(nonce) < NONCE_LENGTH:
+            nonce = b"\x00" * (NONCE_LENGTH - len(nonce)) + nonce
 
-        return self._enc_out.encrypt(nounce, data, aad)
+        return self._enc_out.encrypt(nonce, data, aad)
 
-    def decrypt(self, data, nounce=None, aad=None):
-        """Decrypt data with counter or specified nounce."""
-        if nounce is None:
-            nounce = self._in_counter.to_bytes(
-                length=self._nonce_length, byteorder="little"
-            )
+    def decrypt(
+        self, data: bytes, nonce: Optional[bytes] = None, aad: Optional[bytes] = None
+    ) -> bytes:
+        """Decrypt data with counter or specified nonce."""
+        if nonce is None:
+            nonce = self.in_nonce
             self._in_counter += 1
 
-        if len(nounce) < NONCE_LENGTH:
-            nounce = b"\x00" * (NONCE_LENGTH - len(nounce)) + nounce
+        if len(nonce) < NONCE_LENGTH:
+            nonce = b"\x00" * (NONCE_LENGTH - len(nonce)) + nonce
 
-        decrypted = self._enc_in.decrypt(nounce, data, aad)
+        decrypted = self._enc_in.decrypt(nonce, data, aad)
 
         return bytes(decrypted)
diff --git a/pyatv/support/http.py b/pyatv/support/http.py
index 5691cac9..8e9a3977 100644
--- a/pyatv/support/http.py
+++ b/pyatv/support/http.py
@@ -4,9 +4,20 @@ import asyncio
 from collections import deque
 import logging
 import pathlib
+import plistlib
 from queue import Queue
 import re
-from typing import Callable, Dict, Mapping, NamedTuple, Optional, Tuple, Union, cast
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Mapping,
+    NamedTuple,
+    Optional,
+    Tuple,
+    Union,
+    cast,
+)
 
 from aiohttp import ClientSession, web
 from aiohttp.web import middleware
@@ -77,7 +88,7 @@ class HttpResponse(NamedTuple):
     code: int
     message: str
     headers: Mapping[str, str]
-    body: Union[str, bytes]
+    body: Union[str, bytes, dict]
 
 
 class HttpRequest(NamedTuple):
@@ -102,8 +113,8 @@ def _parse_http_message(
     """Parse HTTP response."""
     try:
         header_str, body = message.split(b"\r\n\r\n", maxsplit=1)
-    except ValueError as ex:
-        raise ValueError("missing end lines") from ex
+    except ValueError:
+        return None, CaseInsensitiveDict(), b"", message
     headers = header_str.decode("utf-8").split("\r\n")
 
     msg_headers = CaseInsensitiveDict(_key_value(line) for line in headers[1:] if line)
@@ -145,7 +156,12 @@ def format_response(response: HttpResponse) -> bytes:
 
     body = response.body or b""
     if body:
-        body = body.encode("utf-8") if isinstance(body, str) else body
+        if isinstance(body, str):
+            body = body.encode("utf-8")
+        elif isinstance(body, dict):
+            body = plistlib.dumps(
+                body, fmt=plistlib.FMT_BINARY  # pylint: disable=no-member
+            )
         output += f"Content-Length: {len(body)}\r\n"
 
     return output.encode("utf-8") + b"\r\n" + body
@@ -154,7 +170,7 @@ def format_response(response: HttpResponse) -> bytes:
 def parse_response(response: bytes) -> Tuple[Optional[HttpResponse], bytes]:
     """Parse HTTP response."""
     first_line, msg_headers, msg_body, rest = _parse_http_message(response)
-    if not first_line:
+    if first_line is None:
         return None, rest
 
     # <method> <path> <protocol>/<version
@@ -202,6 +218,17 @@ def parse_request(request: bytes) -> Tuple[Optional[HttpRequest], bytes]:
     )
 
 
+def decode_bplist_from_body(response: HttpResponse) -> Dict[str, Any]:
+    """Decode a binary property list in a response."""
+    if not isinstance(response.body, (bytes, str)):
+        raise exceptions.ProtocolError(
+            f"expected bytes or str but got {type(response.body).__name__}"
+        )
+
+    body = response.body
+    return plistlib.loads(body if isinstance(body, bytes) else body.encode("utf-8"))
+
+
 class ClientSessionManager:
     """Manages an aiohttp ClientSession instance."""
 
diff --git a/pyatv/support/rtsp.py b/pyatv/support/rtsp.py
index 6ce1caf3..d769da0d 100644
--- a/pyatv/support/rtsp.py
+++ b/pyatv/support/rtsp.py
@@ -13,7 +13,7 @@ from typing import Any, Dict, Mapping, NamedTuple, Optional, Tuple, Union
 import async_timeout
 
 from pyatv.protocols.dmap import tags
-from pyatv.support.http import HttpConnection, HttpResponse
+from pyatv.support.http import HttpConnection, HttpResponse, decode_bplist_from_body
 from pyatv.support.metadata import MediaMetadata
 
 _LOGGER = logging.getLogger(__name__)
@@ -48,6 +48,8 @@ CURVE25519_PUB_KEY = (
     b"\x5d\x8c\x01\x2a\x0c\x7e\x1d\x4e"
 )
 
+BPLIST_CONTENT_TYPE = "application/x-apple-binary-plist"
+
 
 class DigestInfo(NamedTuple):
     """
@@ -105,12 +107,7 @@ class RtspSession:
             _LOGGER.debug("Device does not support /info")
             return {}
 
-        body = (
-            device_info.body
-            if isinstance(device_info.body, bytes)
-            else device_info.body.encode("utf-8")
-        )
-        return plistlib.loads(body)
+        return decode_bplist_from_body(device_info)
 
     async def auth_setup(self) -> HttpResponse:
         """Send auth-setup message."""
@@ -173,7 +170,7 @@ class RtspSession:
     async def setup(
         self,
         headers: Optional[Dict[str, Any]] = None,
-        body: Optional[Union[str, bytes]] = None,
+        body: Optional[Union[str, bytes, dict]] = None,
     ) -> HttpResponse:
         """Send SETUP message."""
         return await self.exchange("SETUP", headers=headers, body=body)
@@ -186,6 +183,14 @@ class RtspSession:
         """Send RECORD message."""
         return await self.exchange("RECORD", headers=headers, body=body)
 
+    async def flush(
+        self,
+        headers: Optional[Dict[str, Any]] = None,
+        body: Optional[Union[str, bytes]] = None,
+    ) -> HttpResponse:
+        """Send RECORD message."""
+        return await self.exchange("FLUSH", headers=headers, body=body)
+
     async def set_parameter(self, parameter: str, value: str) -> HttpResponse:
         """Send SET_PARAMETER message."""
         return await self.exchange(
@@ -234,7 +239,7 @@ class RtspSession:
         uri: Optional[str] = None,
         content_type: Optional[str] = None,
         headers: Optional[Mapping[str, object]] = None,
-        body: Optional[Union[str, bytes]] = None,
+        body: Optional[Union[str, bytes, dict]] = None,
         allow_error: bool = False,
         protocol: str = "RTSP/1.0",
     ) -> HttpResponse:
@@ -258,6 +263,13 @@ class RtspSession:
         if headers:
             hdrs.update(headers)
 
+        # If body is a dict, assume that payload should be sent as a binary plist
+        if isinstance(body, dict):
+            hdrs["Content-Type"] = BPLIST_CONTENT_TYPE
+            body = plistlib.dumps(
+                body, fmt=plistlib.FMT_BINARY  # pylint: disable=no-member
+            )
+
         # Map an asyncio Event to current CSeq and make the request
         self.requests[cseq] = (asyncio.Event(), None)
         resp = await self.connection.send_and_receive(


Test patch:
diff --git a/tests/fake_device/raop.py b/tests/fake_device/raop.py
index 30a331c9..650f43c9 100644
--- a/tests/fake_device/raop.py
+++ b/tests/fake_device/raop.py
@@ -1,5 +1,6 @@
 """Fake RAOP device for tests."""
 import asyncio
+from enum import IntFlag, auto
 from functools import wraps
 from hashlib import md5
 import logging
@@ -9,12 +10,11 @@ import string
 from types import SimpleNamespace
 from typing import Dict, Optional, cast
 
-from bitarray import bitarray
-
 from pyatv.protocols.dmap import parser
 from pyatv.protocols.dmap.tag_definitions import lookup_tag
+from pyatv.protocols.raop import alac
 from pyatv.protocols.raop.packets import RetransmitReqeust, RtpHeader, SyncPacket
-from pyatv.protocols.raop.raop import parse_transport
+from pyatv.protocols.raop.protocols.airplayv1 import parse_transport
 from pyatv.support.http import (
     BasicHttpServer,
     HttpRequest,
@@ -34,10 +34,42 @@ DIGEST_PAYLOAD = (
 REALM = "raop"
 
 
+class RaopServiceFlags(IntFlag):
+    """Flags used to alter fake service behavior."""
+
+    EMPTY = auto()
+    """Empty service flag."""
+
+    INFO_SUPPORTED = auto()
+    """If /info endpoint is supported."""
+
+    AUTH_REQUIRED = auto()
+    """If authentication is required or not."""
+
+    SUPPORTS_RETRANSMISSION = auto()
+    """If audio packet retransmission is supported."""
+
+    INITIAL_AUDIO_LEVEL = auto()
+    """If initial audio level is supported."""
+
+    FEEDBACK_SUPPORTED = auto()
+    """If feedback is supported."""
+
+    DELAYED_SET_VOLUME = auto()
+    """Fail to set volume prior to start start.
+
+    Some devices (at least Sonos) seems to fail when setting volume prior to starting
+    a stream. This flag mimcs that behavior.
+    """
+
+
 def requires_auth(method):
     @wraps(method)
     def _impl(self, request: HttpRequest, *args, **kwargs):
-        if self.state.auth_required and not self.state.auth_setup_performed:
+        if (
+            self.state.is_supported(RaopServiceFlags.AUTH_REQUIRED)
+            and not self.state.auth_setup_performed
+        ):
             return HttpResponse(
                 "RTSP", "1.0", 403, "Forbidden", {"CSeq": request.headers["CSeq"]}, b""
             )
@@ -93,38 +125,38 @@ def verify_password(method):
     return _impl
 
 
-def alac_decode(data: bytes) -> bytes:
-    """Decode an ALAC frame and return raw audio data."""
-    buffer = bitarray()
-    buffer.frombytes(data)
-
-    # Audio starts 23 bits in and last bit must be removed to keep byte alignment,
-    # otherwise an additional \x00 will be present at the end
-    return buffer[23:-1].tobytes()
-
-
 class FakeRaopState:
     """Internal state for RAOP service."""
 
     def __init__(self):
+        self.flags: RaopServiceFlags = (
+            RaopServiceFlags.INFO_SUPPORTED | RaopServiceFlags.FEEDBACK_SUPPORTED
+        )
         self.metadata = SimpleNamespace(title=None, artist=None, album=None)
         self.audio_packets: Dict[int, bytes] = {}  # seqo -> raw audio
         self.initial_audio_packet: Optional[int] = None
         self.password: Optional[str] = None
         self.nonce: Optional[str] = None
-        self.auth_required: bool = False
         self.auth_setup_performed: bool = False
-        self.supports_retransmissions: bool = True
-        self.supports_feedback: bool = True
         self.feedback_packets_received: int = 0
         self.sync_packets_received: int = 0
         self.drop_packets: int = 0
         self.control_port: int = 0
         self.remote_address: Optional[str] = None
         self.volume: float = INITIAL_VOLUME
-        self.initial_audio_level_supported: bool = False
-        self.info_supported: bool = True
         self.teardown_called: bool = False
+        self.streaming_started: bool = False
+
+    def is_supported(self, flag: RaopServiceFlags) -> bool:
+        """Return if a feature is supported."""
+        return flag in self.flags
+
+    def set_flag_state(self, flag: RaopServiceFlags, enabled: bool) -> None:
+        """Set if a feature is supported or not."""
+        if enabled:
+            self.flags |= flag
+        else:
+            self.flags &= ~flag
 
     @property
     def raw_audio(self) -> bytes:
@@ -148,6 +180,10 @@ class FakeRaopState:
             self.initial_audio_packet = seqno
         self.audio_packets[seqno] = audio_data
 
+    def reset_streaming(self) -> None:
+        self.metadata = SimpleNamespace(title=None, artist=None, album=None)
+        self.streaming_started = False
+
 
 class AudioReceiver(asyncio.Protocol):
     """Protocol used to receive audio packets."""
@@ -192,16 +228,16 @@ class AudioReceiver(asyncio.Protocol):
                 )
                 self.state.drop_packets -= 1
 
-                if self.state.supports_retransmissions:
+                if self.state.is_supported(RaopServiceFlags.SUPPORTS_RETRANSMISSION):
                     _LOGGER.debug("Requesting to retransmit seqno %d", header.seqno)
                     self._request_retransmit(
                         data, (self.state.remote_address, self.state.control_port)
                     )
             else:
-                self.state.add_audio_packet(header.seqno, alac_decode(data[12:]))
+                self.state.add_audio_packet(header.seqno, alac.decode(data[12:]))
         elif packet_type == 0x56:  # Retransmission
             original_packet = data[4:]  # Remove retransmission header
-            self.state.add_audio_packet(header.seqno, alac_decode(original_packet[12:]))
+            self.state.add_audio_packet(header.seqno, alac.decode(original_packet[12:]))
         else:
             _LOGGER.debug("Unhandled packet type: %d", packet_type)
 
@@ -324,6 +360,7 @@ class FakeRaopService(HttpSimpleRouter):
         self.add_route("POST", "/auth-setup", self.handle_auth_setup)
         self.add_route("GET", "/info", self.handle_info)
         self.add_route("TEARDOWN", "rtsp://*", self.handle_teardown)
+        self.add_route("FLUSH", "rtsp://*", self.handle_flush)
 
     async def start(self, start_web_server: bool):
         """Start the fake RAOP service."""
@@ -388,7 +425,7 @@ class FakeRaopService(HttpSimpleRouter):
         _LOGGER.debug("Received SETUP: %s", request)
         _, options = parse_transport(request.headers["Transport"])
         self.state.control_port = int(options["control_port"])
-        self.state.metadata = SimpleNamespace(title=None, artist=None, album=None)
+        self.state.reset_streaming()
         self._audio_receiver.reset()
         headers = {
             "Transport": (
@@ -413,8 +450,22 @@ class FakeRaopService(HttpSimpleRouter):
             self.state.metadata.artist = parser.first(tags, "mlit", "asar")
             self.state.metadata.album = parser.first(tags, "mlit", "asal")
         elif request.body.startswith("volume:"):
-            self.state.volume = float(request.body.split(" ", maxsplit=1)[1])
-            _LOGGER.debug("Changing volume to %f", self.state.volume)
+            # If delayed set volume is active, then streaming must have started
+            if (
+                self.state.is_supported(RaopServiceFlags.DELAYED_SET_VOLUME)
+                and not self.state.streaming_started
+            ):
+                return HttpResponse(
+                    "RTSP",
+                    "1.0",
+                    500,
+                    "Not supported here",
+                    {"CSeq": request.headers["CSeq"]},
+                    b"",
+                )
+            else:
+                self.state.volume = float(request.body.split(" ", maxsplit=1)[1])
+                _LOGGER.debug("Changing volume to %f", self.state.volume)
         else:
             return HttpResponse(
                 "RTSP",
@@ -434,7 +485,7 @@ class FakeRaopService(HttpSimpleRouter):
         """Handle incoming feedback request."""
         _LOGGER.debug("Received feedback: %s", request)
         self.state.feedback_packets_received += 1
-        if not self.state.supports_feedback:
+        if not self.state.is_supported(RaopServiceFlags.FEEDBACK_SUPPORTED):
             return HttpResponse(
                 "RTSP",
                 "1.0",
@@ -456,6 +507,16 @@ class FakeRaopService(HttpSimpleRouter):
             "RTSP", "1.0", 200, "OK", {"CSeq": request.headers["CSeq"]}, b""
         )
 
+    @requires_auth
+    @verify_password
+    def handle_flush(self, request: HttpRequest) -> Optional[HttpResponse]:
+        """Handle incoming FLUSH request."""
+        _LOGGER.debug("Received FLUSH: %s", request)
+        self.state.streaming_started = True
+        return HttpResponse(
+            "RTSP", "1.0", 200, "OK", {"CSeq": request.headers["CSeq"]}, b""
+        )
+
     @verify_password
     def handle_auth_setup(self, request: HttpRequest) -> Optional[HttpResponse]:
         """Handle incoming auth-setup request."""
@@ -472,7 +533,7 @@ class FakeRaopService(HttpSimpleRouter):
 
     def handle_info(self, request: HttpRequest) -> Optional[HttpResponse]:
         """Handle incoming info request."""
-        if not self.state.info_supported:
+        if not self.state.is_supported(RaopServiceFlags.INFO_SUPPORTED):
             return HttpResponse(
                 "RTSP",
                 "1.0",
@@ -484,7 +545,7 @@ class FakeRaopService(HttpSimpleRouter):
                 b"",
             )
         info = {}
-        if self.state.initial_audio_level_supported:
+        if self.state.is_supported(RaopServiceFlags.INITIAL_AUDIO_LEVEL):
             info["initialVolume"] = self.state.volume
         return HttpResponse(
             "RTSP",
@@ -515,7 +576,7 @@ class FakeRaopUseCases:
 
     def retransmissions_enabled(self, enabled: bool) -> None:
         """Enable or disable retransmissions."""
-        self.state.supports_retransmissions = enabled
+        self.state.set_flag_state(RaopServiceFlags.SUPPORTS_RETRANSMISSION, enabled)
 
     def drop_n_packets(self, packets: int) -> None:
         """Make fake device drop packets and trigger retransmission (if supported)."""
@@ -523,15 +584,15 @@ class FakeRaopUseCases:
 
     def feedback_enabled(self, enabled: bool) -> None:
         """Enable or disable support for /feedback endpoint."""
-        self.state.supports_feedback = enabled
+        self.state.set_flag_state(RaopServiceFlags.FEEDBACK_SUPPORTED, enabled)
 
     def initial_audio_level_supported(self, supported: bool) -> None:
         """Initial audio level reported in info from device."""
-        self.state.initial_audio_level_supported = supported
+        self.state.set_flag_state(RaopServiceFlags.INITIAL_AUDIO_LEVEL, supported)
 
     def require_auth(self, is_required: bool) -> None:
         """Enable or disable requirement to perform authentication."""
-        self.state.auth_required = is_required
+        self.state.set_flag_state(RaopServiceFlags.AUTH_REQUIRED, is_required)
 
     def password(self, new_password: Optional[str]) -> None:
         """Set a new password. Pass in None to remove the password."""
@@ -540,4 +601,10 @@ class FakeRaopUseCases:
 
     def supports_info(self, is_supported: bool) -> None:
         """State if /info is supported or not."""
-        self.state.info_supported = is_supported
+        self.state.set_flag_state(RaopServiceFlags.INFO_SUPPORTED, is_supported)
+
+    def delayed_set_volume(self, delayed_set_volume) -> None:
+        """Enable or disable delayed set volume."""
+        self.state.set_flag_state(
+            RaopServiceFlags.DELAYED_SET_VOLUME, delayed_set_volume
+        )
diff --git a/tests/protocols/airplay/test_airplay_interface.py b/tests/protocols/airplay/test_airplay_interface.py
index fb9dc8c6..33862c4c 100644
--- a/tests/protocols/airplay/test_airplay_interface.py
+++ b/tests/protocols/airplay/test_airplay_interface.py
@@ -2,9 +2,9 @@
 
 import pytest
 
-from pyatv.conf import ManualService
-from pyatv.const import FeatureName, FeatureState, Protocol
+from pyatv.const import FeatureName, FeatureState
 from pyatv.protocols.airplay import AirPlayFeatures
+from pyatv.protocols.airplay.utils import parse_features
 
 # AirPlayFeatures
 
@@ -18,6 +18,5 @@ from pyatv.protocols.airplay import AirPlayFeatures
     ],
 )
 def test_feature_play_url(flags, expected_state):
-    service = ManualService("id", Protocol.AirPlay, 0, {"features": flags})
-    features = AirPlayFeatures(service)
+    features = AirPlayFeatures(parse_features(flags))
     assert features.get_feature(FeatureName.PlayUrl).state == expected_state
diff --git a/tests/protocols/airplay/test_utils.py b/tests/protocols/airplay/test_utils.py
index cadb3f3f..f92d2ca8 100644
--- a/tests/protocols/airplay/test_utils.py
+++ b/tests/protocols/airplay/test_utils.py
@@ -10,7 +10,9 @@ from pyatv.const import PairingRequirement, Protocol
 from pyatv.core import MutableService
 from pyatv.protocols.airplay.utils import (
     AirPlayFlags,
+    AirPlayMajorVersion,
     get_pairing_requirement,
+    get_protocol_version,
     is_password_required,
     is_remote_control_supported,
     parse_features,
@@ -110,3 +112,19 @@ async def test_get_pairing_requirement(props, expected_req):
 def test_is_remote_control_supported(props, credentials, expected_supported):
     service = MutableService("id", Protocol.AirPlay, 0, props)
     assert is_remote_control_supported(service, credentials) == expected_supported
+
+
+@pytest.mark.parametrize(
+    "props, expected_version",
+    [
+        # Used by RAOP
+        ({"ft": "0xAABBCCDD"}, AirPlayMajorVersion.AirPlayV1),
+        ({"ft": "0xAABBCCDD,0xAABBCCDD"}, AirPlayMajorVersion.AirPlayV2),
+        # User by AirPlay
+        ({"features": "0xAABBCCDD"}, AirPlayMajorVersion.AirPlayV1),
+        ({"features": "0xAABBCCDD,0xAABBCCDD"}, AirPlayMajorVersion.AirPlayV2),
+    ],
+)
+def test_get_protocol_version(props, expected_version):
+    service = MutableService("id", Protocol.AirPlay, 0, props)
+    assert get_protocol_version(service) == expected_version
diff --git a/tests/protocols/raop/test_alac.py b/tests/protocols/raop/test_alac.py
new file mode 100644
index 00000000..420463df
--- /dev/null
+++ b/tests/protocols/raop/test_alac.py
@@ -0,0 +1,10 @@
+"""Unit tests for pyatv.protocols.raop.alac."""
+from pyatv.protocols.raop import alac
+
+AUDIO_FRAMES = b"\x01\x02\x03\x04"
+
+
+def test_encode_and_decode():
+    encoded = alac.encode(AUDIO_FRAMES)
+    decoded = alac.decode(encoded)
+    assert decoded == AUDIO_FRAMES
diff --git a/tests/protocols/raop/test_raop.py b/tests/protocols/raop/test_raop.py
index 9ae7abf5..aab8b601 100644
--- a/tests/protocols/raop/test_raop.py
+++ b/tests/protocols/raop/test_raop.py
@@ -7,7 +7,6 @@ import pytest
 from pyatv.const import DeviceModel, PairingRequirement, Protocol
 from pyatv.core import MutableService, mdns
 from pyatv.interface import DeviceInfo
-from pyatv.protocols.airplay.utils import AirPlayFlags
 from pyatv.protocols.raop import device_info, scan, service_info
 
 RAOP_SERVICE = "_raop._tcp.local"
diff --git a/tests/protocols/raop/test_raop_functional.py b/tests/protocols/raop/test_raop_functional.py
index ab22e36f..043932cc 100644
--- a/tests/protocols/raop/test_raop_functional.py
+++ b/tests/protocols/raop/test_raop_functional.py
@@ -19,6 +19,7 @@ from pyatv import connect, exceptions
 from pyatv.const import DeviceState, FeatureName, FeatureState, MediaType, Protocol
 from pyatv.exceptions import AuthenticationError
 from pyatv.interface import FeatureInfo, MediaMetadata, Playing, PushListener
+from pyatv.protocols.airplay.utils import dbfs_to_pct
 
 from tests.utils import data_path, stub_sleep, until
 
@@ -597,3 +598,18 @@ async def test_stream_metadata_from_http(
     assert raop_state.metadata.artist == "postlund"
     assert raop_state.metadata.album == "raop"
     assert raop_state.metadata.title == "pyatv"
+
+
+@pytest.mark.parametrize("raop_properties", [({"et": "0", "md": "0"})])
+async def test_stream_volume_set_after_stream_start(
+    raop_client, raop_state, raop_usecase
+):
+    raop_usecase.delayed_set_volume(True)
+
+    volume = 9
+
+    await raop_client.audio.set_volume(volume)
+    await raop_client.stream.stream_file(data_path("audio_1_packet_metadata.wav"))
+
+    assert math.isclose(raop_client.audio.volume, volume)
+    assert math.isclose(dbfs_to_pct(raop_state.volume), volume)
diff --git a/tests/support/test_http.py b/tests/support/test_http.py
index 1bc58c25..6feb6869 100644
--- a/tests/support/test_http.py
+++ b/tests/support/test_http.py
@@ -37,10 +37,7 @@ def test_parse_response_ok_first_line():
 
 
 def test_parse_response_missing_ending():
-    with pytest.raises(ValueError) as exc:
-        parse_response(b"HTTP/1.0 200 OK\r\n")
-
-    assert "missing end lines" in str(exc)
+    assert parse_response(b"HTTP/1.0 200 OK\r\n") == (None, b"HTTP/1.0 200 OK\r\n")
 
 
 def test_parse_response_headers():
diff --git a/tests/support/test_knock.py b/tests/support/test_knock.py
index efaacfad..0c403410 100644
--- a/tests/support/test_knock.py
+++ b/tests/support/test_knock.py
@@ -99,7 +99,7 @@ async def test_abort_knock_unhandled_exception(event_loop, caplog):
     start = time.monotonic()
     with patch("pyatv.support.knock.asyncio.open_connection", side_effect=ValueError):
         task = await knocker("127.0.0.1", [1, 2, 3, 4], event_loop, timeout=3)
-        # For unknown exceptions we want to still raise them
+        # For unknown exceptions we want to s'till raise them
         with pytest.raises(ValueError):
             await task
     end = time.monotonic()