Multiple improvements to the exchange and transport layers

examples/onoff_light/src/dev_att.rs

@@ -21,7 +21,7 @@ use rs_matter::error::{Error, ErrorCode};
 pub struct HardCodedDevAtt {}
 
 impl HardCodedDevAtt {
-    pub fn new() -> Self {
+    pub const fn new() -> Self {
         Self {}
     }
 }

examples/onoff_light/src/main.rs

@@ -19,23 +19,29 @@ use core::borrow::Borrow;
 use core::pin::pin;
 use std::net::UdpSocket;
 
-use embassy_futures::select::select3;
+use embassy_futures::select::{select, select4};
 
+use embassy_sync::blocking_mutex::raw::NoopRawMutex;
+use embassy_time::{Duration, Timer};
 use log::info;
 
 use rs_matter::core::{CommissioningData, Matter};
 use rs_matter::data_model::cluster_basic_information::BasicInfoConfig;
 use rs_matter::data_model::cluster_on_off;
+use rs_matter::data_model::core::IMBuffer;
 use rs_matter::data_model::device_types::DEV_TYPE_ON_OFF_LIGHT;
 use rs_matter::data_model::objects::*;
 use rs_matter::data_model::root_endpoint;
+use rs_matter::data_model::subscriptions::Subscriptions;
 use rs_matter::data_model::system_model::descriptor;
 use rs_matter::error::Error;
 use rs_matter::mdns::MdnsService;
 use rs_matter::persist::Psm;
+use rs_matter::respond::DefaultResponder;
 use rs_matter::secure_channel::spake2p::VerifierData;
-use rs_matter::transport::core::{PacketBuffers, MATTER_SOCKET_BIND_ADDR};
-use rs_matter::utils::select::EitherUnwrap;
+use rs_matter::transport::core::MATTER_SOCKET_BIND_ADDR;
+use rs_matter::utils::buf::PooledBuffers;
+use rs_matter::utils::select::Coalesce;
 use rs_matter::MATTER_PORT;
 
 mod dev_att;
@@ -49,7 +55,7 @@ fn main() -> Result<(), Error> {
         // e.g., an opt-level of "0" will require a several times' larger stack.
         //
         // Optimizing/lowering `rs-matter` memory consumption is an ongoing topic.
-        .stack_size(180 * 1024)
+        .stack_size(95 * 1024)
         .spawn(run)
         .unwrap();
 
@@ -62,9 +68,9 @@ fn run() -> Result<(), Error> {
     );
 
     info!(
-        "Matter memory: Matter={}, PacketBuffers={}",
+        "Matter memory: Matter={}B, IM Buffers={}B",
         core::mem::size_of::<Matter>(),
-        core::mem::size_of::<PacketBuffers>(),
+        core::mem::size_of::<PooledBuffers<10, NoopRawMutex, IMBuffer>>()
     );
 
     let dev_det = BasicInfoConfig {
@@ -81,57 +87,92 @@ fn run() -> Result<(), Error> {
 
     let dev_att = dev_att::HardCodedDevAtt::new();
 
-    // NOTE:
-    // For `no_std` environments, provide your own epoch and rand functions here
-    let epoch = rs_matter::utils::epoch::sys_epoch;
-    let rand = rs_matter::utils::rand::sys_rand;
-
     let matter = Matter::new(
-        // vid/pid should match those in the DAC
         &dev_det,
         &dev_att,
+        // NOTE:
+        // For `no_std` environments, provide your own epoch and rand functions here
         MdnsService::Builtin,
-        epoch,
-        rand,
+        rs_matter::utils::epoch::sys_epoch,
+        rs_matter::utils::rand::sys_rand,
         MATTER_PORT,
     );
 
+    matter.initialize_transport_buffers()?;
+
     info!("Matter initialized");
 
-    let handler = HandlerCompat(handler(&matter));
+    let buffers = PooledBuffers::<10, NoopRawMutex, _>::new(0);
+
+    info!("IM buffers initialized");
+
+    let mut mdns = pin!(run_mdns(&matter));
+
+    let on_off = cluster_on_off::OnOffCluster::new(*matter.borrow());
+
+    let subscriptions = Subscriptions::<3>::new();
+
+    // Assemble our Data Model handler by composing the predefined Root Endpoint handler with our custom On/Off clusters
+    let dm_handler = HandlerCompat(dm_handler(&matter, &on_off));
+
+    // Create a default responder capable of handling up to 3 subscriptions
+    // All other subscription requests will be turned down with "resource exhausted"
+    let responder = DefaultResponder::new(&matter, &buffers, &subscriptions, dm_handler);
+    info!(
+        "Responder memory: Responder={}B, Runner={}B",
+        core::mem::size_of_val(&responder),
+        core::mem::size_of_val(&responder.run::<4, 4>())
+    );
+
+    // Run the responder with up to 4 handlers (i.e. 4 exchanges can be handled simultenously)
+    // Clients trying to open more exchanges than the ones currently running will get "I'm busy, please try again later"
+    let mut respond = pin!(responder.run::<4, 4>());
+
+    // This is a sample code that simulates state changes triggered by the HAL
+    // Changes will be properly communicated to the Matter controllers and other Matter apps (i.e. Google Home, Alexa), thanks to subscriptions
+    let mut device = pin!(async {
+        loop {
+            Timer::after(Duration::from_secs(5)).await;
+
+            on_off.set(!on_off.get());
+            subscriptions.notify_changed();
+
+            info!("Lamp toggled");
+        }
+    });
 
     // NOTE:
     // When using a custom UDP stack (e.g. for `no_std` environments), replace with a UDP socket bind for your custom UDP stack
     // The returned socket should be splittable into two halves, where each half implements `UdpSend` and `UdpReceive` respectively
     let socket = async_io::Async::<UdpSocket>::bind(MATTER_SOCKET_BIND_ADDR)?;
 
-    let mut packet_buffers = PacketBuffers::new();
-    let mut runner = pin!(matter.run(
+    // Run the Matter and mDNS transports
+    let mut transport = pin!(matter.run(
         &socket,
         &socket,
-        &mut packet_buffers,
-        CommissioningData {
+        Some(CommissioningData {
             // TODO: Hard-coded for now
             verifier: VerifierData::new_with_pw(123456, *matter.borrow()),
             discriminator: 250,
-        },
-        &handler,
+        }),
     ));
 
-    let mut mdns_runner = pin!(run_mdns(&matter));
-
     // NOTE:
     // Replace with your own persister for e.g. `no_std` environments
     let mut psm = Psm::new(&matter, std::env::temp_dir().join("rs-matter"))?;
-    let mut psm_runner = pin!(psm.run());
-
-    let runner = select3(&mut runner, &mut mdns_runner, &mut psm_runner);
+    let mut persist = pin!(psm.run());
+
+    // Combine all async tasks in a single one
+    let all = select4(
+        &mut transport,
+        &mut mdns,
+        &mut persist,
+        select(&mut respond, &mut device).coalesce(),
+    );
 
     // NOTE:
     // Replace with a different executor for e.g. `no_std` environments
-    futures_lite::future::block_on(runner).unwrap()?;
-
-    Ok(())
+    futures_lite::future::block_on(all.coalesce())
 }
 
 const NODE: Node<'static> = Node {
@@ -146,7 +187,10 @@ const NODE: Node<'static> = Node {
     ],
 };
 
-fn handler<'a>(matter: &'a Matter<'a>) -> impl Metadata + NonBlockingHandler + 'a {
+fn dm_handler<'a>(
+    matter: &'a Matter<'a>,
+    on_off: &'a cluster_on_off::OnOffCluster,
+) -> impl Metadata + NonBlockingHandler + 'a {
     (
         NODE,
         root_endpoint::handler(0, matter)
@@ -155,11 +199,7 @@ fn handler<'a>(matter: &'a Matter<'a>) -> impl Metadata + NonBlockingHandler + '
                 descriptor::ID,
                 descriptor::DescriptorCluster::new(*matter.borrow()),
             )
-            .chain(
-                1,
-                cluster_on_off::ID,
-                cluster_on_off::OnOffCluster::new(*matter.borrow()),
-            ),
+            .chain(1, cluster_on_off::ID, on_off),
     )
 }
 

rs-matter/Cargo.toml

@@ -20,6 +20,7 @@ alloc = []
 openssl = ["alloc", "dep:openssl", "foreign-types", "hmac", "sha2"]
 mbedtls = ["alloc", "dep:mbedtls"]
 rustcrypto = ["alloc", "sha2", "hmac", "pbkdf2", "hkdf", "aes", "ccm", "p256", "elliptic-curve", "crypto-bigint", "x509-cert", "rand_core"]
+large-buffers = [] # TCP support
 
 [dependencies]
 rs-matter-macros = { version = "0.1", path = "../rs-matter-macros" }
@@ -45,6 +46,7 @@ domain = { version = "0.9", default-features = false, features = ["heapless"] }
 octseq = { version = "0.3", default-features = false }
 portable-atomic = "1"
 qrcodegen-no-heap = "1.8"
+scopeguard = "1"
 
 # crypto
 openssl = { version = "0.10", optional = true }