Merge changes I0c822a81,I6838ecc1,I98c253ce into main (419ad59f) · Commits · e / os / android_packages_modules_Bluetooth

framework/tests/bumble/src/android/bluetooth/Host.kt

0 → 100644

+132 −0

Original line number	Original line	Diff line number	Diff line
			/*
			* Copyright (C) 2024 The Android Open Source Project
			*
			* Licensed under the Apache License, Version 2.0 (the "License");
			* you may not use this file except in compliance with the License.
			* You may obtain a copy of the License at
			*
			* http://www.apache.org/licenses/LICENSE-2.0
			*
			* Unless required by applicable law or agreed to in writing, software
			* distributed under the License is distributed on an "AS IS" BASIS,
			* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
			* See the License for the specific language governing permissions and
			* limitations under the License.
			*/

			package android.bluetooth

			import android.content.BroadcastReceiver
			import android.content.Context
			import android.content.Intent
			import android.content.IntentFilter
			import android.util.Log
			import com.google.common.truth.Truth
			import com.google.common.truth.Truth.assertThat
			import java.io.Closeable
			import kotlin.time.Duration.Companion.seconds
			import kotlinx.coroutines.CoroutineScope
			import kotlinx.coroutines.Dispatchers
			import kotlinx.coroutines.cancel
			import kotlinx.coroutines.channels.awaitClose
			import kotlinx.coroutines.channels.trySendBlocking
			import kotlinx.coroutines.flow.Flow
			import kotlinx.coroutines.flow.SharingStarted
			import kotlinx.coroutines.flow.callbackFlow
			import kotlinx.coroutines.flow.filter
			import kotlinx.coroutines.flow.first
			import kotlinx.coroutines.flow.shareIn
			import kotlinx.coroutines.launch
			import kotlinx.coroutines.runBlocking
			import kotlinx.coroutines.withTimeout

			@kotlinx.coroutines.ExperimentalCoroutinesApi
			public class Host(context: Context) : Closeable {
			private val TAG = "PandoraHost"

			private val flow: Flow<Intent>
			private val scope: CoroutineScope
			private val bluetoothManager = context.getSystemService(BluetoothManager::class.java)
			private val bluetoothAdapter = bluetoothManager!!.adapter

			init {
			scope = CoroutineScope(Dispatchers.Default.limitedParallelism(1))
			val intentFilter = IntentFilter()
			intentFilter.addAction(BluetoothDevice.ACTION_BOND_STATE_CHANGED)
			intentFilter.addAction(BluetoothDevice.ACTION_PAIRING_REQUEST)

			flow = intentFlow(context, intentFilter, scope).shareIn(scope, SharingStarted.Eagerly)
			}

			override fun close() {
			scope.cancel()
			}

			public fun createBondAndVerify(remoteDevice: BluetoothDevice) {
			Log.d(TAG, "createBondAndVerify: $remoteDevice")
			if (bluetoothAdapter.bondedDevices.contains(remoteDevice)) {
			Log.d(TAG, "createBondAndVerify: already bonded")
			return
			}

			runBlocking(scope.coroutineContext) {
			withTimeout(TIMEOUT) {
			Truth.assertThat(remoteDevice.createBond()).isTrue()
			flow
			.filter { it.getAction() == BluetoothDevice.ACTION_PAIRING_REQUEST }
			.filter { it.getBluetoothDeviceExtra() == remoteDevice }
			.first()

			remoteDevice.setPairingConfirmation(true)

			flow
			.filter { it.action == BluetoothDevice.ACTION_BOND_STATE_CHANGED }
			.filter { it.getBluetoothDeviceExtra() == remoteDevice }
			.filter {
			it.getIntExtra(BluetoothDevice.EXTRA_BOND_STATE, BluetoothAdapter.ERROR) ==
			BluetoothDevice.BOND_BONDED
			}
			.first()
			Log.d(TAG, "createBondAndVerify: bonded")
			}
			}
			}

			fun removeBondAndVerify(remoteDevice: BluetoothDevice) {
			Log.d(TAG, "removeBondAndVerify: $remoteDevice")
			runBlocking(scope.coroutineContext) {
			withTimeout(TIMEOUT) {
			assertThat(remoteDevice.removeBond()).isTrue()
			flow
			.filter { it.getAction() == BluetoothDevice.ACTION_BOND_STATE_CHANGED }
			.filter { it.getBluetoothDeviceExtra() == remoteDevice }
			.filter {
			it.getIntExtra(BluetoothDevice.EXTRA_BOND_STATE, BluetoothAdapter.ERROR) ==
			BluetoothDevice.BOND_NONE
			}
			Log.d(TAG, "removeBondAndVerify: done")
			}
			}
			}

			fun Intent.getBluetoothDeviceExtra(): BluetoothDevice =
			this.getParcelableExtra(BluetoothDevice.EXTRA_DEVICE, BluetoothDevice::class.java)!!

			@kotlinx.coroutines.ExperimentalCoroutinesApi
			fun intentFlow(context: Context, intentFilter: IntentFilter, scope: CoroutineScope) =
			callbackFlow {
			val broadcastReceiver: BroadcastReceiver =
			object : BroadcastReceiver() {
			override fun onReceive(context: Context, intent: Intent) {
			scope.launch { trySendBlocking(intent) }
			}
			}
			context.registerReceiver(broadcastReceiver, intentFilter)

			awaitClose { context.unregisterReceiver(broadcastReceiver) }
			}

			companion object {
			private val TIMEOUT = 10.seconds
			}
			}

framework/tests/bumble/src/android/bluetooth/RfcommTest.kt

+10 −138

Original line number	Original line	Diff line number	Diff line
	@@ -16,26 +16,18 @@
	package android.bluetooth		package android.bluetooth

	import android.Manifest		import android.Manifest
	import android.content.BroadcastReceiver
	import android.content.Context		import android.content.Context
	import android.content.Intent
	import android.content.IntentFilter
	import android.util.Log
	import androidx.test.core.app.ApplicationProvider		import androidx.test.core.app.ApplicationProvider
	import androidx.test.ext.junit.runners.AndroidJUnit4		import androidx.test.ext.junit.runners.AndroidJUnit4
	import androidx.test.platform.app.InstrumentationRegistry		import androidx.test.platform.app.InstrumentationRegistry
	import com.android.compatibility.common.util.AdoptShellPermissionsRule		import com.android.compatibility.common.util.AdoptShellPermissionsRule
	import com.google.common.truth.Truth		import com.google.common.truth.Truth
	import com.google.protobuf.ByteString		import com.google.protobuf.ByteString
	import io.grpc.stub.StreamObserver
	import java.time.Duration		import java.time.Duration
	import java.util.UUID		import java.util.UUID
	import java.util.concurrent.TimeUnit		import java.util.concurrent.TimeUnit
	import kotlinx.coroutines.*		import kotlinx.coroutines.*
	import kotlinx.coroutines.channels.*		import kotlinx.coroutines.channels.*
	import kotlinx.coroutines.flow.Flow
	import kotlinx.coroutines.flow.consumeAsFlow
	import kotlinx.coroutines.flow.first
	import org.junit.After		import org.junit.After
	import org.junit.Before		import org.junit.Before
	import org.junit.Rule		import org.junit.Rule
	@@ -44,69 +36,9 @@ import org.junit.runner.RunWith
	import pandora.RfcommProto		import pandora.RfcommProto
	import pandora.RfcommProto.ServerId		import pandora.RfcommProto.ServerId
	import pandora.RfcommProto.StartServerRequest		import pandora.RfcommProto.StartServerRequest
	import pandora.SecurityProto.PairingEvent
	import pandora.SecurityProto.PairingEventAnswer

	@kotlinx.coroutines.ExperimentalCoroutinesApi
	fun bondingFlow(context: Context, peer: BluetoothDevice, state: Int): Flow<Intent> {
	val channel = Channel<Intent>(Channel.UNLIMITED)
	val receiver: BroadcastReceiver =
	object : BroadcastReceiver() {
	override fun onReceive(context: Context, intent: Intent) {
	if (
	peer ==
	intent.getParcelableExtra(
	BluetoothDevice.EXTRA_DEVICE,
	BluetoothDevice::class.java
	)
	) {
	if (intent.getIntExtra(BluetoothDevice.EXTRA_BOND_STATE, -1) == state) {
	channel.trySendBlocking(intent)
	}
	}
	}
	}
	context.registerReceiver(receiver, IntentFilter(BluetoothDevice.ACTION_BOND_STATE_CHANGED))
	channel.invokeOnClose { context.unregisterReceiver(receiver) }
	return channel.consumeAsFlow()
	}

	class PairingResponder(
	private val mPeer: BluetoothDevice,
	private val mPairingEventIterator: Iterator<PairingEvent>,
	private val mPairingEventAnswerObserver: StreamObserver<PairingEventAnswer>
	) : BroadcastReceiver() {
	override fun onReceive(context: Context, intent: Intent) {
	when (intent.action) {
	BluetoothDevice.ACTION_PAIRING_REQUEST -> {
	if (
	mPeer ==
	intent.getParcelableExtra(
	BluetoothDevice.EXTRA_DEVICE,
	BluetoothDevice::class.java
	)
	) {
	if (
	BluetoothDevice.PAIRING_VARIANT_CONSENT ==
	intent.getIntExtra(BluetoothDevice.EXTRA_PAIRING_VARIANT, -1)
	) {
	mPeer.setPairingConfirmation(true)
	val pairingEvent: PairingEvent = mPairingEventIterator.next()
	Truth.assertThat(pairingEvent.hasJustWorks()).isTrue()
	mPairingEventAnswerObserver.onNext(
	PairingEventAnswer.newBuilder()
	.setEvent(pairingEvent)
	.setConfirm(true)
	.build()
	)
	}
	}
	}
	}
	}
	}

	@RunWith(AndroidJUnit4::class)		@RunWith(AndroidJUnit4::class)
			@kotlinx.coroutines.ExperimentalCoroutinesApi
	class RfcommTest {		class RfcommTest {
	private val mContext = ApplicationProvider.getApplicationContext<Context>()		private val mContext = ApplicationProvider.getApplicationContext<Context>()
	private val mManager = mContext.getSystemService(BluetoothManager::class.java)		private val mManager = mContext.getSystemService(BluetoothManager::class.java)
	@@ -126,62 +58,37 @@ class RfcommTest {
	@Rule @JvmField val mBumble = PandoraDevice()		@Rule @JvmField val mBumble = PandoraDevice()

	private lateinit var mBumbleDevice: BluetoothDevice		private lateinit var mBumbleDevice: BluetoothDevice
	private lateinit var mPairingResponder: PairingResponder		private lateinit var host: Host
	private lateinit var mPairingEventAnswerObserver: StreamObserver<PairingEventAnswer>
	private val mPairingEventStreamObserver: StreamObserverSpliterator<PairingEvent> =
	StreamObserverSpliterator()
	private var mConnectionCounter = 1		private var mConnectionCounter = 1

	@Before		@Before
	fun setUp() {		fun setUp() {
	mBumbleDevice = mBumble.remoteDevice		mBumbleDevice = mBumble.remoteDevice
	mPairingEventAnswerObserver =		host = Host(mContext)
	mBumble		host.createBondAndVerify(mBumbleDevice)
	.security()
	.withDeadlineAfter(GRPC_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)
	.onPairing(mPairingEventStreamObserver)

	val pairingFilter = IntentFilter(BluetoothDevice.ACTION_PAIRING_REQUEST)
	mPairingResponder =
	PairingResponder(
	mBumbleDevice,
	mPairingEventStreamObserver.iterator(),
	mPairingEventAnswerObserver
	)
	mContext.registerReceiver(mPairingResponder, pairingFilter)

	// TODO: Ideally we shouldn't need this, remove
	runBlocking { removeBondIfBonded(mBumbleDevice) }
	}		}

	@After		@After
	fun tearDown() {		fun tearDown() {
	mContext.unregisterReceiver(mPairingResponder)		if (mAdapter.bondedDevices.contains(mBumbleDevice)) {
			host.removeBondAndVerify(mBumbleDevice)
			}
			host.close()
	}		}

	@Test		@Test
	fun clientConnectToOpenServerSocketBondedInsecure() {		fun clientConnectToOpenServerSocketBondedInsecure() {
	startServer { serverId ->		startServer { serverId -> createConnectAcceptSocket(isSecure = false, serverId) }
	runBlocking { withTimeout(BOND_TIMEOUT.toMillis()) { bondDevice(mBumbleDevice) } }

	createConnectAcceptSocket(isSecure = false, serverId)
	}
	}		}

	@Test		@Test
	fun clientConnectToOpenServerSocketBondedSecure() {		fun clientConnectToOpenServerSocketBondedSecure() {
	startServer { serverId ->		startServer { serverId -> createConnectAcceptSocket(isSecure = true, serverId) }
	runBlocking { withTimeout(BOND_TIMEOUT.toMillis()) { bondDevice(mBumbleDevice) } }

	createConnectAcceptSocket(isSecure = true, serverId)
	}
	}		}

	@Test		@Test
	fun clientSendDataOverInsecureSocket() {		fun clientSendDataOverInsecureSocket() {
	startServer { serverId ->		startServer { serverId ->
	runBlocking { withTimeout(BOND_TIMEOUT.toMillis()) { bondDevice(mBumbleDevice) } }

	val (insecureSocket, connection) = createConnectAcceptSocket(isSecure = false, serverId)		val (insecureSocket, connection) = createConnectAcceptSocket(isSecure = false, serverId)
	val data: ByteArray = "Test data for clientSendDataOverInsecureSocket".toByteArray()		val data: ByteArray = "Test data for clientSendDataOverInsecureSocket".toByteArray()
	val socketOs = insecureSocket.outputStream		val socketOs = insecureSocket.outputStream
	@@ -199,8 +106,6 @@ class RfcommTest {
	@Test		@Test
	fun clientSendDataOverSecureSocket() {		fun clientSendDataOverSecureSocket() {
	startServer { serverId ->		startServer { serverId ->
	runBlocking { withTimeout(BOND_TIMEOUT.toMillis()) { bondDevice(mBumbleDevice) } }

	val (secureSocket, connection) = createConnectAcceptSocket(isSecure = true, serverId)		val (secureSocket, connection) = createConnectAcceptSocket(isSecure = true, serverId)
	val data: ByteArray = "Test data for clientSendDataOverSecureSocket".toByteArray()		val data: ByteArray = "Test data for clientSendDataOverSecureSocket".toByteArray()
	val socketOs = secureSocket.outputStream		val socketOs = secureSocket.outputStream
	@@ -218,8 +123,6 @@ class RfcommTest {
	@Test		@Test
	fun clientReceiveDataOverInsecureSocket() {		fun clientReceiveDataOverInsecureSocket() {
	startServer { serverId ->		startServer { serverId ->
	runBlocking { withTimeout(BOND_TIMEOUT.toMillis()) { bondDevice(mBumbleDevice) } }

	val (insecureSocket, connection) = createConnectAcceptSocket(isSecure = false, serverId)		val (insecureSocket, connection) = createConnectAcceptSocket(isSecure = false, serverId)
	val buffer = ByteArray(64)		val buffer = ByteArray(64)
	val socketIs = insecureSocket.inputStream		val socketIs = insecureSocket.inputStream
	@@ -238,8 +141,6 @@ class RfcommTest {
	@Test		@Test
	fun clientReceiveDataOverSecureSocket() {		fun clientReceiveDataOverSecureSocket() {
	startServer { serverId ->		startServer { serverId ->
	runBlocking { withTimeout(BOND_TIMEOUT.toMillis()) { bondDevice(mBumbleDevice) } }

	val (secureSocket, connection) = createConnectAcceptSocket(isSecure = true, serverId)		val (secureSocket, connection) = createConnectAcceptSocket(isSecure = true, serverId)
	val buffer = ByteArray(64)		val buffer = ByteArray(64)
	val socketIs = secureSocket.inputStream		val socketIs = secureSocket.inputStream
	@@ -297,33 +198,6 @@ class RfcommTest {
	return connectionResponse.connection		return connectionResponse.connection
	}		}

	@OptIn(ExperimentalCoroutinesApi::class)
	private suspend fun bondDevice(remoteDevice: BluetoothDevice) {
	if (mAdapter.bondedDevices.contains(remoteDevice)) {
	Log.d(TAG, "bondDevice(): The device is already bonded")
	return
	}

	val flow = bondingFlow(mContext, remoteDevice, BluetoothDevice.BOND_BONDED)

	Truth.assertThat(remoteDevice.createBond()).isTrue()

	flow.first()
	}

	@OptIn(ExperimentalCoroutinesApi::class)
	private suspend fun removeBondIfBonded(deviceToRemove: BluetoothDevice) {
	if (!mAdapter.bondedDevices.contains(deviceToRemove)) {
	Log.d(TAG, "removeBondIfBonded(): Tried to remove a device that isn't bonded")
	return
	}
	val flow = bondingFlow(mContext, deviceToRemove, BluetoothDevice.BOND_NONE)

	Truth.assertThat(deviceToRemove.removeBond()).isTrue()

	flow.first()
	}

	private fun startServer(		private fun startServer(
	name: String = TEST_SERVER_NAME,		name: String = TEST_SERVER_NAME,
	uuid: String = TEST_UUID,		uuid: String = TEST_UUID,
	@@ -341,14 +215,12 @@ class RfcommTest {
	.stopServer(		.stopServer(
	RfcommProto.StopServerRequest.newBuilder().setServer(response.server).build()		RfcommProto.StopServerRequest.newBuilder().setServer(response.server).build()
	)		)
	runBlocking { removeBondIfBonded(mBumbleDevice) }
	}		}
	}		}

	companion object {		companion object {
	private val TAG = RfcommTest::class.java.getSimpleName()		private val TAG = RfcommTest::class.java.getSimpleName()
	private val GRPC_TIMEOUT = Duration.ofSeconds(10)		private val GRPC_TIMEOUT = Duration.ofSeconds(10)
	private val BOND_TIMEOUT = Duration.ofSeconds(20)
	private const val TEST_UUID = "2ac5d8f1-f58d-48ac-a16b-cdeba0892d65"		private const val TEST_UUID = "2ac5d8f1-f58d-48ac-a16b-cdeba0892d65"
	private const val SERIAL_PORT_UUID = "00001101-0000-1000-8000-00805F9B34FB"		private const val SERIAL_PORT_UUID = "00001101-0000-1000-8000-00805F9B34FB"
	private const val TEST_SERVER_NAME = "RFCOMM Server"		private const val TEST_SERVER_NAME = "RFCOMM Server"

framework/tests/bumble/src/android/bluetooth/SdpClientTest.java

+11 −8

Original line number	Original line	Diff line number	Diff line
	@@ -13,7 +13,6 @@
	* See the License for the specific language governing permissions and		* See the License for the specific language governing permissions and
	* limitations under the License.		* limitations under the License.
	*/		*/

	package android.bluetooth;		package android.bluetooth;

	import static com.google.common.truth.Truth.assertThat;		import static com.google.common.truth.Truth.assertThat;
	@@ -38,6 +37,9 @@ import org.junit.runner.RunWith;

	import pandora.HostProto.ConnectRequest;		import pandora.HostProto.ConnectRequest;

			import java.util.Arrays;
			import java.util.List;

	/** Test cases for {@link ServiceDiscoveryManager}. */		/** Test cases for {@link ServiceDiscoveryManager}. */
	@RunWith(AndroidJUnit4.class)		@RunWith(AndroidJUnit4.class)
	public class SdpClientTest {		public class SdpClientTest {
	@@ -47,7 +49,7 @@ public class SdpClientTest {
	private final BluetoothManager mManager = mContext.getSystemService(BluetoothManager.class);		private final BluetoothManager mManager = mContext.getSystemService(BluetoothManager.class);
	private final BluetoothAdapter mAdapter = mManager.getAdapter();		private final BluetoothAdapter mAdapter = mManager.getAdapter();

	private SettableFuture<ParcelUuid[]> mFutureIntent;		private SettableFuture<List<ParcelUuid>> mFutureIntent;

	@Rule public final AdoptShellPermissionsRule mPermissionRule = new AdoptShellPermissionsRule();		@Rule public final AdoptShellPermissionsRule mPermissionRule = new AdoptShellPermissionsRule();

	@@ -61,7 +63,7 @@ public class SdpClientTest {
	ParcelUuid[] parcelUuids =		ParcelUuid[] parcelUuids =
	intent.getParcelableArrayExtra(		intent.getParcelableArrayExtra(
	BluetoothDevice.EXTRA_UUID, ParcelUuid.class);		BluetoothDevice.EXTRA_UUID, ParcelUuid.class);
	mFutureIntent.set(parcelUuids);		mFutureIntent.set(Arrays.asList(parcelUuids));
	}		}
	}		}
	};		};
	@@ -76,21 +78,22 @@ public class SdpClientTest {
	String local_addr = mAdapter.getAddress();		String local_addr = mAdapter.getAddress();
	byte[] local_bytes_addr = Utils.addressBytesFromString(local_addr);		byte[] local_bytes_addr = Utils.addressBytesFromString(local_addr);

	// Initiate connect from remote
	mBumble.hostBlocking()		mBumble.hostBlocking()
	.connect(		.connect(
	ConnectRequest.newBuilder()		ConnectRequest.newBuilder()
	.setAddress(ByteString.copyFrom(local_bytes_addr))		.setAddress(ByteString.copyFrom(local_bytes_addr))
	.build());		.build());

	// Get the remote device
	BluetoothDevice device = mBumble.getRemoteDevice();		BluetoothDevice device = mBumble.getRemoteDevice();

	// Execute service discovery procedure
	assertThat(device.fetchUuidsWithSdp()).isTrue();		assertThat(device.fetchUuidsWithSdp()).isTrue();

	ParcelUuid[] arr = mFutureIntent.get();		assertThat(mFutureIntent.get())
	assertThat(arr).asList().contains(BluetoothUuid.HFP);		.containsExactly(
			BluetoothUuid.HFP,
			BluetoothUuid.A2DP_SOURCE,
			BluetoothUuid.A2DP_SINK,
			BluetoothUuid.AVRCP);

	mContext.unregisterReceiver(mConnectionStateReceiver);		mContext.unregisterReceiver(mConnectionStateReceiver);
	}		}

framework/tests/bumble/src/bumble_server.py

+6 −2

Original line number	Original line	Diff line number	Diff line
	@@ -24,13 +24,15 @@ from bumble_experimental.asha import AshaService
	from bumble_experimental.dck import DckService		from bumble_experimental.dck import DckService
	from bumble_experimental.gatt import GATTService		from bumble_experimental.gatt import GATTService
	from bumble_experimental.rfcomm import RFCOMMService		from bumble_experimental.rfcomm import RFCOMMService
			from bumble_experimental.avrcp import AvrcpService

	from pandora_experimental.asha_grpc_aio import add_AshaServicer_to_server		from pandora_experimental.asha_grpc_aio import add_AshaServicer_to_server
	from pandora_experimental.dck_grpc_aio import add_DckServicer_to_server		from pandora_experimental.dck_grpc_aio import add_DckServicer_to_server
	from pandora_experimental.gatt_grpc_aio import add_GATTServicer_to_server		from pandora_experimental.gatt_grpc_aio import add_GATTServicer_to_server
	from pandora_experimental.rfcomm_grpc_aio import add_RFCOMMServicer_to_server		from pandora_experimental.rfcomm_grpc_aio import add_RFCOMMServicer_to_server
			from pandora_experimental.avrcp_grpc_aio import add_AVRCPServicer_to_server

	from typing import Dict, Any		from typing import Any, Dict

	BUMBLE_SERVER_GRPC_PORT = 7999		BUMBLE_SERVER_GRPC_PORT = 7999
	ROOTCANAL_PORT_CUTTLEFISH = 7300		ROOTCANAL_PORT_CUTTLEFISH = 7300
	@@ -68,7 +70,9 @@ def args_parser() -> argparse.ArgumentParser:
	return parser		return parser


	def register_experimental_services():		def register_experimental_services() -> None:
			bumble_server.register_servicer_hook(
			lambda bumble, _, server: add_AVRCPServicer_to_server(AvrcpService(bumble.device), server))
	bumble_server.register_servicer_hook(		bumble_server.register_servicer_hook(
	lambda bumble, _, server: add_AshaServicer_to_server(AshaService(bumble.device), server))		lambda bumble, _, server: add_AshaServicer_to_server(AshaService(bumble.device), server))
	bumble_server.register_servicer_hook(		bumble_server.register_servicer_hook(

pandora/server/bumble_experimental/avrcp.py

0 → 100644

+76 −0

Original line number	Original line	Diff line number	Diff line
			# Copyright 2024 Google LLC
			#
			# Licensed under the Apache License, Version 2.0 (the "License");
			# you may not use this file except in compliance with the License.
			# You may obtain a copy of the License at
			#
			# https://www.apache.org/licenses/LICENSE-2.0
			#
			# Unless required by applicable law or agreed to in writing, software
			# distributed under the License is distributed on an "AS IS" BASIS,
			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
			# See the License for the specific language governing permissions and
			# limitations under the License.

			from bumble.avdtp import Listener as AvdtpListener, MediaCodecCapabilities, AVDTP_AUDIO_MEDIA_TYPE
			from bumble.avrcp import Protocol as AvrcpProtocol, make_target_service_sdp_records, make_controller_service_sdp_records
			from bumble.a2dp import (A2DP_SBC_CODEC_TYPE, SBC_DUAL_CHANNEL_MODE, SBC_JOINT_STEREO_CHANNEL_MODE,
			SBC_LOUDNESS_ALLOCATION_METHOD, SBC_MONO_CHANNEL_MODE, SBC_SNR_ALLOCATION_METHOD,
			SBC_STEREO_CHANNEL_MODE, SbcMediaCodecInformation, make_audio_sink_service_sdp_records,
			make_audio_source_service_sdp_records)
			from bumble.device import Device
			from pandora_experimental.avrcp_grpc_aio import AVRCPServicer


			class AvrcpService(AVRCPServicer):
			device: Device

			def __init__(self, device: Device) -> None:
			super().__init__()
			self.device = device

			sdp_records = {
			0x00010002: make_audio_source_service_sdp_records(0x00010002), # A2DP Source
			0x00010003: make_audio_sink_service_sdp_records(0x00010003), # A2DP Sink
			0x00010004: make_controller_service_sdp_records(0x00010004), # AVRCP Controller
			0x00010005: make_target_service_sdp_records(0x00010005), # AVRCP Target
			}
			self.device.sdp_service_records.update(sdp_records)

			# Register AVDTP L2cap
			avdtp_listener = AvdtpListener.for_device(device)

			def on_avdtp_connection(server) -> None: # type: ignore
			server.add_sink(codec_capabilities()) # type: ignore

			avdtp_listener.on('connection', on_avdtp_connection) # type: ignore

			# Register AVRCP L2cap
			avrcp_protocol = AvrcpProtocol(delegate=None)
			avrcp_protocol.listen(device)


			def codec_capabilities() -> MediaCodecCapabilities:
			"""Codec capabilities for the Bumble sink devices."""

			return MediaCodecCapabilities(
			media_type=AVDTP_AUDIO_MEDIA_TYPE,
			media_codec_type=A2DP_SBC_CODEC_TYPE,
			media_codec_information=SbcMediaCodecInformation.from_lists(
			sampling_frequencies=[48000, 44100, 32000, 16000],
			channel_modes=[
			SBC_MONO_CHANNEL_MODE,
			SBC_DUAL_CHANNEL_MODE,
			SBC_STEREO_CHANNEL_MODE,
			SBC_JOINT_STEREO_CHANNEL_MODE,
			],
			block_lengths=[4, 8, 12, 16],
			subbands=[4, 8],
			allocation_methods=[
			SBC_LOUDNESS_ALLOCATION_METHOD,
			SBC_SNR_ALLOCATION_METHOD,
			],
			minimum_bitpool_value=2,
			maximum_bitpool_value=53,
			),
			)