Handle L2CAP info dynamically (#28)

* Add feature and MTU fields in L2CAP manager constructor
* Add register/unregister API for fixed channels

.github/workflows/python-build-test.yml

@@ -29,11 +29,11 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        python -m pip install ".[test,development,documentation]"
+        python -m pip install ".[build,test,development,documentation]"
     - name: Test with pytest
       run: |
         pytest
     - name: Build
       run: |
         inv build
-        inv mkdocs
+        inv build.mkdocs

.github/workflows/python-publish.yml

@@ -30,7 +30,7 @@ jobs:
     - name: Build package
       run: python -m build
     - name: Publish package to PyPI
-      if: github.event_name == 'push' && startsWith(github.ref, 'refs/tags')
+      if: github.event_name == 'release' && startsWith(github.ref, 'refs/tags')
       uses: pypa/gh-action-pypi-publish@release/v1
       with:
         user: __token__

apps/console.py

@@ -122,6 +122,8 @@ class ConsoleApp:
                 },
                 'read': LiveCompleter(self.known_attributes),
                 'write': LiveCompleter(self.known_attributes),
+                'subscribe': LiveCompleter(self.known_attributes),
+                'unsubscribe': LiveCompleter(self.known_attributes),
                 'quit': None,
                 'exit': None
             })
@@ -331,7 +333,7 @@ class ConsoleApp:
 
         await self.show_attributes(attributes)
 
-    def find_attribute(self, param):
+    def find_characteristic(self, param):
         parts = param.split('.')
         if len(parts) == 2:
             service_uuid = UUID(parts[0]) if parts[0] != '*' else None
@@ -344,7 +346,10 @@ class ConsoleApp:
         elif len(parts) == 1:
             if parts[0].startswith('#'):
                 attribute_handle = int(f'{parts[0][1:]}', 16)
-                return attribute_handle
+                for service in self.connected_peer.services:
+                    for characteristic in service.characteristics:
+                        if characteristic.handle == attribute_handle:
+                            return characteristic
 
     async def command(self, command):
         try:
@@ -457,13 +462,13 @@ class ConsoleApp:
             self.show_error('invalid syntax', 'expected read <attribute>')
             return
 
-        attribute = self.find_attribute(params[0])
-        if attribute is None:
+        characteristic = self.find_characteristic(params[0])
+        if characteristic is None:
             self.show_error('no such characteristic')
             return
 
-        value = await self.connected_peer.read_value(attribute)
-        self.append_to_output(f'VALUE: {value}')
+        value = await characteristic.read_value()
+        self.append_to_output(f'VALUE: 0x{value.hex()}')
 
     async def do_write(self, params):
         if not self.connected_peer:
@@ -482,21 +487,48 @@ class ConsoleApp:
             except ValueError:
                 value = str.encode(params[1])  # must be a string
 
-        attribute = self.find_attribute(params[0])
-        if attribute is None:
+        characteristic = self.find_characteristic(params[0])
+        if characteristic is None:
             self.show_error('no such characteristic')
             return
 
         # use write with response if supported
-        with_response = (
-            (attribute.properties & Characteristic.WRITE)
-            if hasattr(attribute, "properties")
-            else False
+        with_response = characteristic.properties & Characteristic.WRITE
+        await characteristic.write_value(value, with_response=with_response)
+
+    async def do_subscribe(self, params):
+        if not self.connected_peer:
+            self.show_error('not connected')
+            return
+
+        if len(params) != 1:
+            self.show_error('invalid syntax', 'expected subscribe <attribute>')
+            return
+
+        characteristic = self.find_characteristic(params[0])
+        if characteristic is None:
+            self.show_error('no such characteristic')
+            return
+
+        await characteristic.subscribe(
+            lambda value: self.append_to_output(f"{characteristic} VALUE: 0x{value.hex()}"),
         )
 
-        await self.connected_peer.write_value(
-            attribute, value, with_response=with_response
-        )
+    async def do_unsubscribe(self, params):
+        if not self.connected_peer:
+            self.show_error('not connected')
+            return
+
+        if len(params) != 1:
+            self.show_error('invalid syntax', 'expected subscribe <attribute>')
+            return
+
+        characteristic = self.find_characteristic(params[0])
+        if characteristic is None:
+            self.show_error('no such characteristic')
+            return
+
+        await characteristic.unsubscribe()
 
     async def do_exit(self, params):
         self.ui.exit()

bumble/device.py

@@ -123,6 +123,9 @@ class Peer:
     async def subscribe(self, characteristic, subscriber=None):
         return await self.gatt_client.subscribe(characteristic, subscriber)
 
+    async def unsubscribe(self, characteristic, subscriber=None):
+        return await self.gatt_client.unsubscribe(characteristic, subscriber)
+
     async def read_value(self, attribute):
         return await self.gatt_client.read_value(attribute)
 
@@ -312,6 +315,8 @@ class DeviceConfiguration:
         self.le_simultaneous_enabled  = True
         self.classic_sc_enabled       = True
         self.classic_ssp_enabled      = True
+        self.connectable              = True
+        self.discoverable             = True
         self.advertising_data = bytes(
             AdvertisingData([(AdvertisingData.COMPLETE_LOCAL_NAME, bytes(self.name, 'utf-8'))])
         )
@@ -330,6 +335,8 @@ class DeviceConfiguration:
         self.le_simultaneous_enabled  = config.get('le_simultaneous_enabled', self.le_simultaneous_enabled)
         self.classic_sc_enabled       = config.get('classic_sc_enabled', self.classic_sc_enabled)
         self.classic_ssp_enabled      = config.get('classic_ssp_enabled', self.classic_ssp_enabled)
+        self.connectable              = config.get('connectable', self.connectable)
+        self.discoverable             = config.get('discoverable', self.discoverable)
 
         # Load or synthesize an IRK
         irk = config.get('irk')
@@ -443,7 +450,8 @@ class Device(CompositeEventEmitter):
         self.command_timeout          = 10  # seconds
         self.gatt_server              = gatt_server.Server(self)
         self.sdp_server               = sdp.Server(self)
-        self.l2cap_channel_manager    = l2cap.ChannelManager()
+        self.l2cap_channel_manager    = l2cap.ChannelManager(
+            [l2cap.L2CAP_Information_Request.EXTENDED_FEATURE_FIXED_CHANNELS])
         self.advertisement_data       = {}
         self.scanning                 = False
         self.discovering              = False
@@ -451,8 +459,6 @@ class Device(CompositeEventEmitter):
         self.disconnecting            = False
         self.connections              = {}  # Connections, by connection handle
         self.classic_enabled          = False
-        self.discoverable             = False
-        self.connectable              = False
         self.inquiry_response         = None
         self.address_resolver         = None
 
@@ -473,6 +479,8 @@ class Device(CompositeEventEmitter):
         self.le_simultaneous_enabled  = config.le_simultaneous_enabled
         self.classic_ssp_enabled      = config.classic_ssp_enabled
         self.classic_sc_enabled       = config.classic_sc_enabled
+        self.discoverable             = config.discoverable
+        self.connectable              = config.connectable
 
         # If a name is passed, override the name from the config
         if name:
@@ -487,6 +495,8 @@ class Device(CompositeEventEmitter):
         # Setup SMP
         # TODO: allow using a public address
         self.smp_manager = smp.Manager(self, self.random_address)
+        self.l2cap_channel_manager.register_fixed_channel(
+            smp.SMP_CID, self.on_smp_pdu)
 
         # Register the SDP server with the L2CAP Channel Manager
         self.sdp_server.register(self.l2cap_channel_manager)
@@ -494,6 +504,7 @@ class Device(CompositeEventEmitter):
         # Add a GAP Service if requested
         if generic_access_service:
             self.gatt_server.add_service(GenericAccessService(self.name))
+        self.l2cap_channel_manager.register_fixed_channel(ATT_CID, self.on_gatt_pdu)
 
         # Forward some events
         setup_event_forwarding(self.gatt_server, self, 'characteristic_subscription')
@@ -620,6 +631,8 @@ class Device(CompositeEventEmitter):
                 HCI_Write_Secure_Connections_Host_Support_Command(
                     secure_connections_host_support=int(self.classic_sc_enabled))
             )
+            await self.set_connectable(self.connectable)
+            await self.set_discoverable(self.discoverable)
 
         # Let the SMP manager know about the address
         # TODO: allow using a public address
@@ -1494,7 +1507,6 @@ class Device(CompositeEventEmitter):
     def on_pairing_failure(self, connection, reason):
         connection.emit('pairing_failure', reason)
 
-    @host_event_handler
     @with_connection_from_handle
     def on_gatt_pdu(self, connection, pdu):
         # Parse the L2CAP payload into an ATT PDU object
@@ -1513,7 +1525,6 @@ class Device(CompositeEventEmitter):
                 return
             connection.gatt_server.on_gatt_pdu(connection, att_pdu)
 
-    @host_event_handler
     @with_connection_from_handle
     def on_smp_pdu(self, connection, pdu):
         self.smp_manager.on_smp_pdu(connection, pdu)

bumble/gatt_client.py

@@ -110,6 +110,9 @@ class CharacteristicProxy(AttributeProxy):
     async def subscribe(self, subscriber=None):
         return await self.client.subscribe(self, subscriber)
 
+    async def unsubscribe(self, subscriber=None):
+        return await self.client.unsubscribe(self, subscriber)
+
     def __str__(self):
         return f'Characteristic(handle=0x{self.handle:04X}, uuid={self.uuid}, properties={Characteristic.properties_as_string(self.properties)})'
 
@@ -544,10 +547,36 @@ class Client:
         for subscriber_set in subscriber_sets:
             if subscriber is not None:
                 subscriber_set.add(subscriber)
-            subscriber_set.add(lambda value: characteristic.emit('update', self.connection, value))
+            # Add the characteristic as a subscriber, which will result in the characteristic
+            # emitting an 'update' event when a notification or indication is received
+            subscriber_set.add(characteristic)
 
         await self.write_value(cccd, struct.pack('<H', bits), with_response=True)
 
+    async def unsubscribe(self, characteristic, subscriber=None):
+        # If we haven't already discovered the descriptors for this characteristic, do it now
+        if not characteristic.descriptors_discovered:
+            await self.discover_descriptors(characteristic)
+
+        # Look for the CCCD descriptor
+        cccd = characteristic.get_descriptor(GATT_CLIENT_CHARACTERISTIC_CONFIGURATION_DESCRIPTOR)
+        if not cccd:
+            logger.warning('unsubscribing from characteristic with no CCCD descriptor')
+            return
+
+        if subscriber is not None:
+            # Remove matching subscriber from subscriber sets
+            for subscriber_set in (self.notification_subscribers, self.indication_subscribers):
+                subscribers = subscriber_set.get(characteristic.handle, [])
+                if subscriber in subscribers:
+                    subscribers.remove(subscriber)
+        else:
+            # Remove all subscribers for this attribute from the sets!
+            self.notification_subscribers.pop(characteristic.handle, None)
+            self.indication_subscribers.pop(characteristic.handle, None)
+
+        await self.write_value(cccd, b'\x00\x00', with_response=True)
+
     async def read_value(self, attribute, no_long_read=False):
         '''
         See Vol 3, Part G - 4.8.1 Read Characteristic Value
@@ -714,7 +743,10 @@ class Client:
         if not subscribers:
             logger.warning('!!! received notification with no subscriber')
         for subscriber in subscribers:
-            subscriber(notification.attribute_value)
+            if callable(subscriber):
+                subscriber(notification.attribute_value)
+            else:
+                subscriber.emit('update', notification.attribute_value)
 
     def on_att_handle_value_indication(self, indication):
         # Call all subscribers
@@ -722,7 +754,10 @@ class Client:
         if not subscribers:
             logger.warning('!!! received indication with no subscriber')
         for subscriber in subscribers:
-            subscriber(indication.attribute_value)
+            if callable(subscriber):
+                subscriber(indication.attribute_value)
+            else:
+                subscriber.emit('update', indication.attribute_value)
 
         # Confirm that we received the indication
         self.send_confirmation(ATT_Handle_Value_Confirmation())

bumble/host.py

@@ -56,13 +56,7 @@ class Connection:
 
     def on_acl_pdu(self, pdu):
         l2cap_pdu = L2CAP_PDU.from_bytes(pdu)
-
-        if l2cap_pdu.cid == ATT_CID:
-            self.host.on_gatt_pdu(self, l2cap_pdu.payload)
-        elif l2cap_pdu.cid == SMP_CID:
-            self.host.on_smp_pdu(self, l2cap_pdu.payload)
-        else:
-            self.host.on_l2cap_pdu(self, l2cap_pdu.cid, l2cap_pdu.payload)
+        self.host.on_l2cap_pdu(self, l2cap_pdu.cid, l2cap_pdu.payload)
 
 
 # -----------------------------------------------------------------------------
@@ -299,12 +293,6 @@ class Host(EventEmitter):
         if connection := self.connections.get(packet.connection_handle):
             connection.on_hci_acl_data_packet(packet)
 
-    def on_gatt_pdu(self, connection, pdu):
-        self.emit('gatt_pdu', connection.handle, pdu)
-
-    def on_smp_pdu(self, connection, pdu):
-        self.emit('smp_pdu', connection.handle, pdu)
-
     def on_l2cap_pdu(self, connection, cid, pdu):
         self.emit('l2cap_pdu', connection.handle, cid, pdu)
 

bumble/l2cap.py

@@ -20,11 +20,11 @@ import logging
 import struct
 
 from colors import color
+from pyee import EventEmitter
 
 from .core import BT_CENTRAL_ROLE, InvalidStateError, ProtocolError
 from .hci import (HCI_LE_Connection_Update_Command, HCI_Object, key_with_value,
                   name_or_number)
-from .utils import EventEmitter
 
 # -----------------------------------------------------------------------------
 # Logging
@@ -414,6 +414,18 @@ class L2CAP_Information_Request(L2CAP_Control_Frame):
     EXTENDED_FEATURES_SUPPORTED = 0x0002
     FIXED_CHANNELS_SUPPORTED    = 0x0003
 
+    EXTENDED_FEATURE_FLOW_MODE_CONTROL                 = 0x0001
+    EXTENDED_FEATURE_RETRANSMISSION_MODE               = 0x0002
+    EXTENDED_FEATURE_BIDIRECTIONAL_QOS                 = 0x0004
+    EXTENDED_FEATURE_ENHANCED_RETRANSMISSION_MODE      = 0x0008
+    EXTENDED_FEATURE_STREAMING_MODE                    = 0x0010
+    EXTENDED_FEATURE_FCS_OPTION                        = 0x0020
+    EXTENDED_FEATURE_EXTENDED_FLOW_SPEC                = 0x0040
+    EXTENDED_FEATURE_FIXED_CHANNELS                    = 0x0080
+    EXTENDED_FEATURE_EXTENDED_WINDOW_SIZE              = 0x0100
+    EXTENDED_FEATURE_UNICAST_CONNECTIONLESS_DATA       = 0x0200
+    EXTENDED_FEATURE_ENHANCED_CREDIT_BASE_FLOW_CONTROL = 0x0400
+
     INFO_TYPE_NAMES = {
         CONNECTIONLESS_MTU:          'CONNECTIONLESS_MTU',
         EXTENDED_FEATURES_SUPPORTED: 'EXTENDED_FEATURES_SUPPORTED',
@@ -817,11 +829,16 @@ class Channel(EventEmitter):
 
 # -----------------------------------------------------------------------------
 class ChannelManager:
-    def __init__(self):
-        self.host        = None
-        self.channels    = {}  # Channels, mapped by connection and cid
-        self.identifiers = {}  # Incrementing identifier values by connection
-        self.servers     = {}  # Servers accepting connections, by PSM
+    def __init__(self, extended_features=None, connectionless_mtu=1024):
+        self.host               = None
+        self.channels           = {}  # Channels, mapped by connection and cid
+        # Fixed channel handlers, mapped by cid
+        self.fixed_channels     = {
+            L2CAP_SIGNALING_CID: None, L2CAP_LE_SIGNALING_CID: None}
+        self.identifiers        = {}  # Incrementing identifier values by connection
+        self.servers            = {}  # Servers accepting connections, by PSM
+        self.extended_features  = [] if extended_features is None else extended_features
+        self.connectionless_mtu = connectionless_mtu
 
     def find_channel(self, connection_handle, cid):
         if connection_channels := self.channels.get(connection_handle):
@@ -840,6 +857,13 @@ class ChannelManager:
         identifier = (self.identifiers.setdefault(connection.handle, 0) + 1) % 256
         self.identifiers[connection.handle] = identifier
         return identifier
+    
+    def register_fixed_channel(self, cid, handler):
+        self.fixed_channels[cid] = handler
+    
+    def deregister_fixed_channel(self, cid):
+        if cid in self.fixed_channels:
+            del self.fixed_channels[cid]
 
     def register_server(self, psm, server):
         self.servers[psm] = server
@@ -855,6 +879,8 @@ class ChannelManager:
             control_frame = L2CAP_Control_Frame.from_bytes(pdu)
 
             self.on_control_frame(connection, cid, control_frame)
+        elif cid in self.fixed_channels:
+            self.fixed_channels[cid](connection.handle, pdu)
         else:
             if (channel := self.find_channel(connection.handle, cid)) is None:
                 logger.warn(color(f'channel not found for 0x{connection.handle:04X}:{cid}', 'red'))
@@ -999,13 +1025,13 @@ class ChannelManager:
     def on_l2cap_information_request(self, connection, cid, request):
         if request.info_type == L2CAP_Information_Request.CONNECTIONLESS_MTU:
             result = L2CAP_Information_Response.SUCCESS
-            data = struct.pack('<H', 1024)  # TODO: don't use a fixed value
+            data = self.connectionless_mtu.to_bytes(2, 'little')
         elif request.info_type == L2CAP_Information_Request.EXTENDED_FEATURES_SUPPORTED:
             result = L2CAP_Information_Response.SUCCESS
-            data = bytes.fromhex('00000000')  # TODO: don't use a fixed value
+            data = sum(self.extended_features).to_bytes(4, 'little')
         elif request.info_type == L2CAP_Information_Request.FIXED_CHANNELS_SUPPORTED:
             result = L2CAP_Information_Response.SUCCESS
-            data = bytes.fromhex('FFFFFFFFFFFFFFFF')  # TODO: don't use a fixed value
+            data = sum(1 << cid for cid in self.fixed_channels).to_bytes(8, 'little')
         else:
             result = L2CAP_Information_Request.NO_SUPPORTED
 

bumble/rfcomm.py

@@ -17,9 +17,10 @@
 # -----------------------------------------------------------------------------
 import logging
 import asyncio
-from colors import color
 
-from .utils import EventEmitter
+from colors import color
+from pyee import EventEmitter
+
 from .core import InvalidStateError, ProtocolError, ConnectionError
 
 # -----------------------------------------------------------------------------

docs/mkdocs/src/apps_and_tools/console.md

@@ -7,10 +7,12 @@ The Console app is an interactive text user interface that offers a number of fu
 
   * scanning
   * advertising
-  * connecting to devices
+  * connecting to and disconnecting from devices
   * changing connection parameters
+  * enabling encryption
   * discovering GATT services and characteristics
-  * read & write GATT characteristics
+  * reading and writing GATT characteristics
+  * subscribing to and unsubscribing from GATT characteristics
 
 The console user interface has 3 main panes:
 

setup.cfg

@@ -57,12 +57,13 @@ console_scripts =
     bumble-link-relay = bumble.apps.link_relay.link_relay:main
 
 [options.extras_require]
+build = 
+    build >= 0.7
 test =
     pytest >= 6.2
     pytest-asyncio >= 0.17
 development =
     invoke >= 1.4
-    build >= 0.7
     nox >= 2022
 documentation =
     mkdocs >= 1.2.3

tasks.py

@@ -23,35 +23,52 @@ ROOT_DIR = os.path.dirname(os.path.realpath(__file__))
 
 ns = Collection()
 
+# Building
 build_tasks = Collection()
-ns.add_collection(build_tasks, name='build')
-
+ns.add_collection(build_tasks, name="build")
 
 @task
-def build(ctx):
-    ctx.run('python -m build')
+def build(ctx, install=False):
+    if install:
+        ctx.run('python -m pip install .[build]')
 
-build_tasks.add_task(build, default=True, name='build')
+    ctx.run("python -m build")
 
+build_tasks.add_task(build, default=True)
+
+@task
+def release_build(ctx):
+    build(ctx, install=True)
+
+build_tasks.add_task(release_build, name="release")
+
+@task
+def mkdocs(ctx):
+    ctx.run("mkdocs build -f docs/mkdocs/mkdocs.yml")
+
+build_tasks.add_task(mkdocs, name="mkdocs")
+
+# Testing
 test_tasks = Collection()
-ns.add_collection(test_tasks, name='test')
+ns.add_collection(test_tasks, name="test")
 
 @task
-def test(ctx, filter=None, junit=False):
+def test(ctx, filter=None, junit=False, install=False):
+    # Install the package before running the tests
+    if install:
+        ctx.run("python -m pip install .[test]")
+
     args = ""
     if junit:
         args += "--junit-xml test-results.xml"
     if filter is not None:
         args += " -k '{}'".format(filter)
-    ctx.run('python -m pytest {} {}'
-            .format(os.path.join(ROOT_DIR, "tests"), args))
-
-test_tasks.add_task(test, name='test', default=True)
+    ctx.run("python -m pytest {} {}".format(os.path.join(ROOT_DIR, "tests"), args))
 
+test_tasks.add_task(test, default=True)
 
 @task
-def mkdocs(ctx):
-    ctx.run('mkdocs build -f docs/mkdocs/mkdocs.yml')
+def release_test(ctx):
+    test(ctx, install=True)
 
-
-ns.add_task(mkdocs)
+test_tasks.add_task(release_test, name="release")

tests/gatt_test.py

@@ -419,10 +419,12 @@ async def test_subscribe_notify():
     assert(len(c) == 1)
     c3 = c[0]
 
+    c1._called = False
     c1._last_update = None
 
-    def on_c1_update(connection, value):
-        c1._last_update = (connection, value)
+    def on_c1_update(value):
+        c1._called = True
+        c1._last_update = value
 
     c1.on('update', on_c1_update)
     await peer.subscribe(c1)
@@ -434,44 +436,73 @@ async def test_subscribe_notify():
     assert(not characteristic1._last_subscription[2])
     await server.indicate_subscribers(characteristic1)
     await async_barrier()
-    assert(c1._last_update is None)
+    assert(not c1._called)
     await server.notify_subscribers(characteristic1)
     await async_barrier()
-    assert(c1._last_update is not None)
-    assert(c1._last_update[1] == characteristic1.value)
+    assert(c1._called)
+    assert(c1._last_update == characteristic1.value)
 
+    c1._called = False
+    await peer.unsubscribe(c1)
+    await server.notify_subscribers(characteristic1)
+    assert(not c1._called)
+
+    c2._called = False
     c2._last_update = None
 
     def on_c2_update(value):
-        c2._last_update = (connection, value)
+        c2._called = True
+        c2._last_update = value
 
     await peer.subscribe(c2, on_c2_update)
     await async_barrier()
     await server.notify_subscriber(characteristic2._last_subscription[0], characteristic2)
     await async_barrier()
-    assert(c2._last_update is None)
+    assert(not c2._called)
     await server.indicate_subscriber(characteristic2._last_subscription[0], characteristic2)
     await async_barrier()
-    assert(c2._last_update is not None)
-    assert(c2._last_update[1] == characteristic2.value)
+    assert(c2._called)
+    assert(c2._last_update == characteristic2.value)
 
-    c3._last_update = None
+    c2._called = False
+    await peer.unsubscribe(c2, on_c2_update)
+    await server.indicate_subscriber(characteristic2._last_subscription[0], characteristic2)
+    await async_barrier()
+    assert(not c2._called)
 
-    def on_c3_update(connection, value):
-        c3._last_update = (connection, value)
+    def on_c3_update(value):
+        c3._called = True
+        c3._last_update = value
+
+    def on_c3_update_2(value):
+        c3._called_2 = True
+        c3._last_update_2 = value
 
     c3.on('update', on_c3_update)
-    await peer.subscribe(c3)
+    await peer.subscribe(c3, on_c3_update_2)
     await async_barrier()
     await server.notify_subscriber(characteristic3._last_subscription[0], characteristic3)
     await async_barrier()
-    assert(c3._last_update is not None)
-    assert(c3._last_update[1] == characteristic3.value)
+    assert(c3._called)
+    assert(c3._last_update == characteristic3.value)
+    assert(c3._called_2)
+    assert(c3._last_update_2 == characteristic3.value)
     characteristic3.value = bytes([1, 2, 3])
     await server.indicate_subscriber(characteristic3._last_subscription[0], characteristic3)
     await async_barrier()
-    assert(c3._last_update is not None)
-    assert(c3._last_update[1] == characteristic3.value)
+    assert(c3._called)
+    assert(c3._last_update == characteristic3.value)
+    assert(c3._called_2)
+    assert(c3._last_update_2 == characteristic3.value)
+
+    c3._called = False
+    c3._called_2 = False
+    await peer.unsubscribe(c3)
+    await server.notify_subscriber(characteristic3._last_subscription[0], characteristic3)
+    await server.indicate_subscriber(characteristic3._last_subscription[0], characteristic3)
+    await async_barrier()
+    assert(not c3._called)
+    assert(not c3._called_2)
 
 
 # -----------------------------------------------------------------------------