reed/cover-theif
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: go back to using channels for the services

Browse Source 
embassy-sync RpcService didn't support async very well
This commit is contained in:
reed
2026-10-06 17:01:48 -05:00
parent cbcb5b0b5f
commit a45cb6efbd
4 changed files with 220 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config.rs
+76 -40
View File
@@ -1,11 +1,13 @@
use defmt::error;
use embassy_embedded_hal::adapter::BlockingAsync;
use embassy_executor::Spawner;
use embassy_sync::{blocking_mutex::raw::CriticalSectionRawMutex, rpc_service::RpcService};
use embassy_sync::{blocking_mutex::raw::CriticalSectionRawMutex, channel::Channel};
use esp_hal::peripherals::FLASH;
use esp_radio::wifi::sta::StationConfig;
use esp_storage::FlashStorage;
use heapless::String;
use int_enum::IntEnum;
use oneshot::Sender;
use postcard::to_slice;
use sequential_storage::{
cache::{Cache, Uncached},
@@ -14,35 +16,28 @@ use sequential_storage::{
use serde::{Deserialize, Serialize};
use thiserror::Error;
pub static CONFIG_SERVICE: RpcService<CriticalSectionRawMutex, ConfigService, 64> =
RpcService::new();
pub async fn start_config(spawner: Spawner, flash: FLASH<'static>) {
let storage = MapStorage::<ConfigKey, _, _>::new(
BlockingAsync::new(FlashStorage::new(flash)),
const { MapConfig::new(0x3F_0000..0x40_0000) },
Cache::new_uncached(),
);
pub struct ConfigService {
storage: MapStorage<
ConfigKey,
BlockingAsync<FlashStorage<'static>>,
Cache<Uncached, Uncached, Uncached, ConfigKey>,
>,
data_buffer: [u8; 256],
spawner.spawn(handler(storage).unwrap());
}
impl ConfigService {
pub async fn set_wifi_creds(&mut self, value: WifiCredentials) -> Result<(), ConfigError> {
let mut buf = [0u8; 256];
to_slice(&value, &mut buf)?;
pub async fn set_wifi_creds(value: WifiCredentials) -> Result<(), ConfigError> {
let (send, recv) = oneshot::channel();
CONFIG_CHANNEL
.send(ConfigOperationMessage::SetWifiCreds(value, send))
.await;
recv.await?
}
self.storage
.store_item(&mut self.data_buffer, &ConfigKey::Wifi, &buf)
.await
.map_err(|e| ConfigError::FlashError(e))?;
static CONFIG_CHANNEL: Channel<CriticalSectionRawMutex, ConfigOperationMessage, 1> = Channel::new();
// TODO: send the change to the wifi service
// WIFI_SERVICE
// .call(|s| s.set_station_config(esp_radio::wifi::Config::Station(value.into())))
// .await;
Ok(())
}
enum ConfigOperationMessage {
SetWifiCreds(WifiCredentials, Sender<Result<(), ConfigError>>),
}
#[derive(Debug, Deserialize, Serialize)]
@@ -59,24 +54,50 @@ impl From<WifiCredentials> for StationConfig {
}
}
pub async fn start_config(spawner: Spawner, flash: FLASH<'static>) {
let storage = MapStorage::<ConfigKey, _, _>::new(
BlockingAsync::new(FlashStorage::new(flash)),
const { MapConfig::new(0x3F_0000..0x40_0000) },
Cache::new_uncached(),
);
let service = ConfigService {
storage: storage,
data_buffer: [0u8; 256],
// TODO: refactor this when try blocks are implemented
#[embassy_executor::task]
async fn handler(
mut storage: MapStorage<
ConfigKey,
BlockingAsync<FlashStorage<'static>>,
Cache<Uncached, Uncached, Uncached, ConfigKey>,
>,
) -> ! {
let resp_fail = |_| {
// TODO: make tracing work so if this code is reached you know where the error originated
error!("failed to respond");
};
spawner.spawn(handler(service).unwrap());
}
let mut data_buffer = [0u8; 256];
#[embassy_executor::task]
async fn handler(mut service: ConfigService) -> ! {
CONFIG_SERVICE.run(&mut service).await
loop {
let operation = CONFIG_CHANNEL.receive().await;
match operation {
ConfigOperationMessage::SetWifiCreds(value, resp) => {
let mut buf = [0u8; 256];
if let Err(err) = to_slice(&value, &mut buf) {
let _ = resp.send(Err(err.into())).map_err(resp_fail);
continue;
};
if let Err(err) = storage
.store_item(&mut data_buffer, &ConfigKey::Wifi, &buf)
.await
{
let _ = resp.send(Err(err.into())).map_err(resp_fail);
continue;
};
// TODO: send the change to the wifi service
// WIFI_SERVICE
// .call(|s| s.set_station_config(esp_radio::wifi::Config::Station(value.into())))
// .await;
let _ = resp.send(Ok(())).map_err(resp_fail);
}
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, IntEnum)]
@@ -122,4 +143,19 @@ pub enum ConfigError {
#[error("Error storing data on the flash: {0:?}")]
FlashError(sequential_storage::Error<esp_storage::FlashStorageError>),
#[error("Error receiving response: {0}")]
RecvError(oneshot::RecvError),
}
impl From<sequential_storage::Error<esp_storage::FlashStorageError>> for ConfigError {
fn from(v: sequential_storage::Error<esp_storage::FlashStorageError>) -> Self {
Self::FlashError(v)
}
}
impl From<oneshot::RecvError> for ConfigError {
fn from(v: oneshot::RecvError) -> Self {
Self::RecvError(v)
}
}
src/wifi.rs
+67 -49
View File
@@ -1,36 +1,15 @@
use defmt::{error, info};
use embassy_executor::Spawner;
use embassy_futures::select::{Either, select};
use embassy_net::{Runner, Stack};
use embassy_sync::{
blocking_mutex::raw::CriticalSectionRawMutex, mutex::Mutex, rpc_service::RpcService,
};
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
use embassy_sync::channel::Channel;
use embassy_sync::signal::Signal;
use embassy_time::{Duration, Timer};
use esp_hal::peripherals::WIFI;
use esp_radio::wifi::sta::StationConfig;
use esp_radio::wifi::{ControllerConfig, Interface, WifiController, WifiError};
pub static WIFI_SERVICE: RpcService<CriticalSectionRawMutex, WifiService, 64> = RpcService::new();
pub struct WifiService {
pub net_stack: Stack<'static>,
wifi_controller: &'static Mutex<CriticalSectionRawMutex, WifiController<'static>>,
}
impl WifiService {
pub async fn set_station_config(
&self,
config: esp_radio::wifi::Config,
) -> Result<(), WifiError> {
let mut controller = self.wifi_controller.lock().await;
controller.set_config(&config)?;
Ok(())
}
pub fn set_dhcp_config(&self, config: embassy_net::DhcpConfig) {
self.net_stack
.set_config_v4(embassy_net::ConfigV4::Dhcp(config));
}
}
use esp_radio::wifi::{ControllerConfig, Interface, WifiController};
use oneshot::{RecvError, Sender};
pub fn start_wifi(spawner: Spawner, wifi: WIFI<'static>) -> Stack<'static> {
let wifi_config = esp_radio::wifi::Config::Station(StationConfig::default());
@@ -58,50 +37,89 @@ pub fn start_wifi(spawner: Spawner, wifi: WIFI<'static>) -> Stack<'static> {
seed,
);
let service = WifiService {
net_stack: net_stack,
wifi_controller: crate::mk_static!(Mutex<CriticalSectionRawMutex, WifiController<'static>>, Mutex::new(wifi_controller)),
};
spawner.spawn(net_handler(net_runner).unwrap());
spawner.spawn(connection_handler(service.wifi_controller).unwrap());
spawner.spawn(handler(service).unwrap());
spawner.spawn(connection_handler(wifi_controller).unwrap());
spawner.spawn(handler(net_stack).unwrap());
spawner.spawn(ip_logger(net_stack).unwrap());
net_stack
}
#[embassy_executor::task]
async fn handler(mut service: WifiService) -> ! {
WIFI_SERVICE.run(&mut service).await
pub async fn set_station_config(config: esp_radio::wifi::Config) -> Result<(), RecvError> {
let (send, recv) = oneshot::channel();
WIFI_CHANNEL
.send(WifiOperationMessage::SetStationConfig(config, send))
.await;
recv.await
}
pub async fn set_dhcp_config(config: embassy_net::DhcpConfig) -> Result<(), RecvError> {
let (send, recv) = oneshot::channel();
WIFI_CHANNEL
.send(WifiOperationMessage::SetDhcpConfig(config, send))
.await;
recv.await
}
static WIFI_CHANNEL: Channel<CriticalSectionRawMutex, WifiOperationMessage, 1> = Channel::new();
enum WifiOperationMessage {
SetStationConfig(esp_radio::wifi::Config, Sender<()>),
SetDhcpConfig(embassy_net::DhcpConfig, Sender<()>),
}
#[embassy_executor::task]
async fn connection_handler(
controller: &'static Mutex<CriticalSectionRawMutex, WifiController<'static>>,
) -> ! {
async fn handler(net_stack: Stack<'static>) -> ! {
loop {
match WIFI_CHANNEL.receive().await {
WifiOperationMessage::SetStationConfig(config, resp) => {
WIFI_HANDLER_RPC.signal(config);
if let Err(_) = resp.send(()) {
// TODO: find a way to not hard-code symbols in a string
error!("WifiOperationMessage::SetStationConfig failed to respond")
};
}
WifiOperationMessage::SetDhcpConfig(config, resp) => {
net_stack.set_config_v4(embassy_net::ConfigV4::Dhcp(config));
if let Err(_) = resp.send(()) {
// TODO: find a way to not hard-code symbols in a string
error!("WifiOperationMessage::SetDhcpConfig failed to respond")
};
}
}
}
}
static WIFI_HANDLER_RPC: Signal<CriticalSectionRawMutex, esp_radio::wifi::Config> = Signal::new();
#[embassy_executor::task]
async fn connection_handler(mut controller: WifiController<'static>) -> ! {
async fn connection(controller: &mut WifiController<'static>) {
info!("Connecting to WiFi...");
match controller.lock().await.connect_async().await {
match controller.connect_async().await {
Ok(info) => {
info!("Wifi connected to {:?}", info);
// wait until we're no longer connected
// TODO: reimplement the wait for disconnect
// let info = controller
// .lock()
// .await
// .wait_for_disconnect_async()
// .await
// .ok();
// info!("Disconnected: {:?}", info);
let info = controller.wait_for_disconnect_async().await.ok();
info!("Disconnected: {:?}", info);
}
Err(e) => {
error!("Failed to connect to wifi: {:?}", e);
}
}
}
loop {
if let Either::First(req) =
select(WIFI_HANDLER_RPC.wait(), connection(&mut controller)).await
{
if let Some(err) = controller.set_config(&req).err() {
error!("Wifi controller failed to set a new config: {:?}", err)
}
}
}
}
#[embassy_executor::task]