GKaszewski/k-frame
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

per-source polling, initial client state, webhook, preview, client tracking

Browse Source 
- per-source poll intervals: spawn task per source with own interval,
  manager re-checks sources every 30s for add/remove
- initial screen update on TCP connect: send layout + widget states
- client tracking: ClientRegistry port, GET /api/clients, dashboard list
- webhook adapter: POST /api/webhook/{source_id} feeds data into projection
- widget preview: GET /api/widgets/{id}/preview returns current state
- serve SPA from Axum: ServeDir + index.html fallback via KFRAME_SPA_DIR
- layout builder delete confirmation with AlertDialog
- form validation: required fields disable save button
- guide page at /guide
- fix architecture: ClientDto to api-types, ClientRegistry + WidgetStateReader
  ports in domain, DataProjection has internal Mutex, no adapter cross-deps
- ESP32: full screen clear on layout change (stale pixel fix)
This commit is contained in:
Gabriel Kaszewski
2026-06-19 00:42:31 +02:00
parent 26ebfad3a2
commit 1d7b5324d6
39 changed files with 1232 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/adapters/http-api/Cargo.toml
View File

@@ -19,3 +19,4 @@ tower.workspace = true
serde_json.workspace = true
config-memory.workspace = true
tcp-server.workspace = true
application.workspace = true

43
crates/adapters/http-api/src/lib.rs
View File

@@ -1,43 +1,72 @@
mod routes;
use axum::Router;
use domain::{ConfigRepository, EventPublisher};
use domain::{BroadcastPort, ClientRegistry, ConfigRepository, EventPublisher, WidgetStateReader};
use std::sync::Arc;
use tower_http::cors::CorsLayer;
use tower_http::services::{ServeDir, ServeFile};
pub struct AppState<C, E> {
pub struct AppState<C, E, W, B, R> {
pub config: Arc<C>,
pub events: Arc<E>,
pub widget_states: Arc<W>,
pub broadcaster: Arc<B>,
pub clients: Arc<R>,
pub spa_dir: Option<String>,
}
impl<C, E> Clone for AppState<C, E> {
impl<C, E, W, B, R> Clone for AppState<C, E, W, B, R> {
fn clone(&self) -> Self {
Self {
config: self.config.clone(),
events: self.events.clone(),
widget_states: self.widget_states.clone(),
broadcaster: self.broadcaster.clone(),
clients: self.clients.clone(),
spa_dir: self.spa_dir.clone(),
}
}
}
pub fn router<C, E>(state: AppState<C, E>) -> Router
pub fn router<C, E, W, B, R>(state: AppState<C, E, W, B, R>) -> Router
where
C: ConfigRepository + Send + Sync + 'static,
C::Error: std::fmt::Debug + Send,
E: EventPublisher + Send + Sync + 'static,
E::Error: std::fmt::Debug + Send,
W: WidgetStateReader + Send + Sync + 'static,
B: BroadcastPort + Send + Sync + 'static,
B::Error: std::fmt::Debug + Send,
R: ClientRegistry + Send + Sync + 'static,
{
Router::new()
let spa_dir = state.spa_dir.clone();
let app = Router::new()
.nest("/api", routes::api_routes())
.layer(CorsLayer::permissive())
.with_state(state)
.with_state(state);
if let Some(dir) = spa_dir {
let index = format!("{dir}/index.html");
app.fallback_service(ServeDir::new(&dir).fallback(ServeFile::new(index)))
} else {
app
}
}
pub async fn serve<C, E>(addr: &str, state: AppState<C, E>) -> Result<(), std::io::Error>
pub async fn serve<C, E, W, B, R>(
addr: &str,
state: AppState<C, E, W, B, R>,
) -> Result<(), std::io::Error>
where
C: ConfigRepository + Send + Sync + 'static,
C::Error: std::fmt::Debug + Send,
E: EventPublisher + Send + Sync + 'static,
E::Error: std::fmt::Debug + Send,
W: WidgetStateReader + Send + Sync + 'static,
B: BroadcastPort + Send + Sync + 'static,
B::Error: std::fmt::Debug + Send,
R: ClientRegistry + Send + Sync + 'static,
{
let app = router(state);
let listener = tokio::net::TcpListener::bind(addr).await?;

25
crates/adapters/http-api/src/routes/clients.rs Normal file
View File

@@ -0,0 +1,25 @@
use crate::AppState;
use api_types::ClientDto;
use axum::extract::State;
use axum::response::Json;
use domain::{ClientRegistry, ConfigRepository, EventPublisher};
type S<C, E, W, B, R> = State<AppState<C, E, W, B, R>>;
pub async fn list_clients<C, E, W, B, R>(State(state): S<C, E, W, B, R>) -> Json<Vec<ClientDto>>
where
C: ConfigRepository,
C::Error: std::fmt::Debug,
E: EventPublisher,
E::Error: std::fmt::Debug,
R: ClientRegistry,
{
Json(
state
.clients
.list_clients()
.iter()
.map(ClientDto::from)
.collect(),
)
}

22
crates/adapters/http-api/src/routes/data_sources.rs
View File

@@ -8,10 +8,10 @@ use axum::{
};
use domain::{ConfigRepository, EventPublisher};
type S<C, E> = State<AppState<C, E>>;
type S<C, E, W, B, R> = State<AppState<C, E, W, B, R>>;
pub async fn list_data_sources<C, E>(
State(state): S<C, E>,
pub async fn list_data_sources<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
) -> Result<Json<Vec<DataSourceDto>>, StatusCode>
where
C: ConfigRepository,
@@ -27,8 +27,8 @@ where
Ok(Json(sources.iter().map(DataSourceDto::from).collect()))
}
pub async fn get_data_source<C, E>(
State(state): S<C, E>,
pub async fn get_data_source<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(id): Path<u16>,
) -> Result<Json<DataSourceDto>, StatusCode>
where
@@ -48,8 +48,8 @@ where
}
}
pub async fn create_data_source<C, E>(
State(state): S<C, E>,
pub async fn create_data_source<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Json(body): Json<DataSourceDto>,
) -> Result<StatusCode, (StatusCode, String)>
where
@@ -68,8 +68,8 @@ where
Ok(StatusCode::CREATED)
}
pub async fn update_data_source<C, E>(
State(state): S<C, E>,
pub async fn update_data_source<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(_id): Path<u16>,
Json(body): Json<DataSourceDto>,
) -> Result<StatusCode, (StatusCode, String)>
@@ -89,8 +89,8 @@ where
Ok(StatusCode::OK)
}
pub async fn delete_data_source<C, E>(
State(state): S<C, E>,
pub async fn delete_data_source<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(id): Path<u16>,
) -> Result<StatusCode, StatusCode>
where

10
crates/adapters/http-api/src/routes/layout.rs
View File

@@ -4,9 +4,11 @@ use application::ConfigService;
use axum::{extract::State, http::StatusCode, response::Json};
use domain::{ConfigRepository, EventPublisher};
type S<C, E> = State<AppState<C, E>>;
type S<C, E, W, B, R> = State<AppState<C, E, W, B, R>>;
pub async fn get_layout<C, E>(State(state): S<C, E>) -> Result<Json<Option<LayoutDto>>, StatusCode>
pub async fn get_layout<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
) -> Result<Json<Option<LayoutDto>>, StatusCode>
where
C: ConfigRepository,
C::Error: std::fmt::Debug,
@@ -21,8 +23,8 @@ where
Ok(Json(layout.as_ref().map(LayoutDto::from)))
}
pub async fn update_layout<C, E>(
State(state): S<C, E>,
pub async fn update_layout<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Json(body): Json<LayoutDto>,
) -> Result<StatusCode, (StatusCode, String)>
where

51
crates/adapters/http-api/src/routes/mod.rs
View File

@@ -1,53 +1,74 @@
mod clients;
mod data_sources;
mod layout;
mod presets;
mod webhook;
mod widgets;
use crate::AppState;
use axum::Router;
use axum::routing::{get, post};
use domain::{ConfigRepository, EventPublisher};
use domain::{BroadcastPort, ClientRegistry, ConfigRepository, EventPublisher, WidgetStateReader};
pub fn api_routes<C, E>() -> Router<AppState<C, E>>
pub fn api_routes<C, E, W, B, R>() -> Router<AppState<C, E, W, B, R>>
where
C: ConfigRepository + Send + Sync + 'static,
C::Error: std::fmt::Debug + Send,
E: EventPublisher + Send + Sync + 'static,
E::Error: std::fmt::Debug + Send,
W: WidgetStateReader + Send + Sync + 'static,
B: BroadcastPort + Send + Sync + 'static,
B::Error: std::fmt::Debug + Send,
R: ClientRegistry + Send + Sync + 'static,
{
Router::new()
.route(
"/widgets",
get(widgets::list_widgets::<C, E>).post(widgets::create_widget::<C, E>),
get(widgets::list_widgets::<C, E, W, B, R>)
.post(widgets::create_widget::<C, E, W, B, R>),
)
.route(
"/widgets/{id}",
get(widgets::get_widget::<C, E>)
.put(widgets::update_widget::<C, E>)
.delete(widgets::delete_widget::<C, E>),
get(widgets::get_widget::<C, E, W, B, R>)
.put(widgets::update_widget::<C, E, W, B, R>)
.delete(widgets::delete_widget::<C, E, W, B, R>),
)
.route(
"/widgets/{id}/preview",
get(widgets::preview_widget::<C, E, W, B, R>),
)
.route(
"/data-sources",
get(data_sources::list_data_sources::<C, E>)
.post(data_sources::create_data_source::<C, E>),
get(data_sources::list_data_sources::<C, E, W, B, R>)
.post(data_sources::create_data_source::<C, E, W, B, R>),
)
.route(
"/data-sources/{id}",
get(data_sources::get_data_source::<C, E>)
.put(data_sources::update_data_source::<C, E>)
.delete(data_sources::delete_data_source::<C, E>),
get(data_sources::get_data_source::<C, E, W, B, R>)
.put(data_sources::update_data_source::<C, E, W, B, R>)
.delete(data_sources::delete_data_source::<C, E, W, B, R>),
)
.route(
"/layout",
get(layout::get_layout::<C, E>).put(layout::update_layout::<C, E>),
get(layout::get_layout::<C, E, W, B, R>).put(layout::update_layout::<C, E, W, B, R>),
)
.route(
"/presets",
get(presets::list_presets::<C, E>).post(presets::create_preset::<C, E>),
get(presets::list_presets::<C, E, W, B, R>)
.post(presets::create_preset::<C, E, W, B, R>),
)
.route(
"/presets/{id}",
get(presets::get_preset::<C, E>).delete(presets::delete_preset::<C, E>),
get(presets::get_preset::<C, E, W, B, R>)
.delete(presets::delete_preset::<C, E, W, B, R>),
)
.route(
"/presets/{id}/load",
post(presets::load_preset::<C, E, W, B, R>),
)
.route("/clients", get(clients::list_clients::<C, E, W, B, R>))
.route(
"/webhook/{source_id}",
post(webhook::receive_webhook::<C, E, W, B, R>),
)
.route("/presets/{id}/load", post(presets::load_preset::<C, E>))
}

22
crates/adapters/http-api/src/routes/presets.rs
View File

@@ -8,9 +8,11 @@ use axum::{
};
use domain::{ConfigRepository, EventPublisher};
type S<C, E> = State<AppState<C, E>>;
type S<C, E, W, B, R> = State<AppState<C, E, W, B, R>>;
pub async fn list_presets<C, E>(State(state): S<C, E>) -> Result<Json<Vec<PresetDto>>, StatusCode>
pub async fn list_presets<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
) -> Result<Json<Vec<PresetDto>>, StatusCode>
where
C: ConfigRepository,
C::Error: std::fmt::Debug,
@@ -25,8 +27,8 @@ where
Ok(Json(presets.iter().map(PresetDto::from).collect()))
}
pub async fn get_preset<C, E>(
State(state): S<C, E>,
pub async fn get_preset<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(id): Path<u16>,
) -> Result<Json<PresetDto>, StatusCode>
where
@@ -46,8 +48,8 @@ where
}
}
pub async fn create_preset<C, E>(
State(state): S<C, E>,
pub async fn create_preset<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Json(body): Json<CreatePresetDto>,
) -> Result<StatusCode, (StatusCode, String)>
where
@@ -66,8 +68,8 @@ where
Ok(StatusCode::CREATED)
}
pub async fn delete_preset<C, E>(
State(state): S<C, E>,
pub async fn delete_preset<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(id): Path<u16>,
) -> Result<StatusCode, StatusCode>
where
@@ -83,8 +85,8 @@ where
Ok(StatusCode::NO_CONTENT)
}
pub async fn load_preset<C, E>(
State(state): S<C, E>,
pub async fn load_preset<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(id): Path<u16>,
) -> Result<StatusCode, (StatusCode, String)>
where

81
crates/adapters/http-api/src/routes/webhook.rs Normal file
View File

@@ -0,0 +1,81 @@
use crate::AppState;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::Json;
use domain::{BroadcastPort, ConfigRepository, EventPublisher, WidgetStateReader};
type S<C, E, W, B, R> = State<AppState<C, E, W, B, R>>;
pub async fn receive_webhook<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(source_id): Path<u16>,
Json(body): Json<serde_json::Value>,
) -> Result<StatusCode, (StatusCode, String)>
where
C: ConfigRepository,
C::Error: std::fmt::Debug,
E: EventPublisher,
E::Error: std::fmt::Debug,
W: WidgetStateReader,
B: BroadcastPort,
B::Error: std::fmt::Debug,
{
let source = state
.config
.get_data_source(source_id)
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:?}")))?
.ok_or((StatusCode::NOT_FOUND, "data source not found".into()))?;
if source.source_type != domain::DataSourceType::Webhook {
return Err((
StatusCode::BAD_REQUEST,
"data source is not a webhook type".into(),
));
}
let raw = json_to_domain_value(body);
let widgets = state
.config
.list_widgets()
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:?}")))?;
let layout = state
.config
.get_layout()
.await
.map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, format!("{e:?}")))?;
let changed = state
.widget_states
.apply_raw_data(source_id, &raw, &widgets)
.await;
if !changed.is_empty()
&& let Some(l) = &layout
{
let _ = state.broadcaster.push_screen_update(l, &changed).await;
}
Ok(StatusCode::OK)
}
fn json_to_domain_value(json: serde_json::Value) -> domain::Value {
match json {
serde_json::Value::Null => domain::Value::Null,
serde_json::Value::Bool(b) => domain::Value::Bool(b),
serde_json::Value::Number(n) => domain::Value::Number(n.as_f64().unwrap_or(0.0)),
serde_json::Value::String(s) => domain::Value::String(s),
serde_json::Value::Array(arr) => {
domain::Value::Array(arr.into_iter().map(json_to_domain_value).collect())
}
serde_json::Value::Object(obj) => {
let map = obj
.into_iter()
.map(|(k, v)| (k, json_to_domain_value(v)))
.collect();
domain::Value::Object(map)
}
}
}

67
crates/adapters/http-api/src/routes/widgets.rs
View File

@@ -6,11 +6,13 @@ use axum::{
http::StatusCode,
response::Json,
};
use domain::{ConfigRepository, EventPublisher};
use domain::{ConfigRepository, EventPublisher, WidgetStateReader};
type S<C, E> = State<AppState<C, E>>;
type S<C, E, W, B, R> = State<AppState<C, E, W, B, R>>;
pub async fn list_widgets<C, E>(State(state): S<C, E>) -> Result<Json<Vec<WidgetDto>>, StatusCode>
pub async fn list_widgets<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
) -> Result<Json<Vec<WidgetDto>>, StatusCode>
where
C: ConfigRepository,
C::Error: std::fmt::Debug,
@@ -25,8 +27,8 @@ where
Ok(Json(widgets.iter().map(WidgetDto::from).collect()))
}
pub async fn get_widget<C, E>(
State(state): S<C, E>,
pub async fn get_widget<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(id): Path<u16>,
) -> Result<Json<WidgetDto>, StatusCode>
where
@@ -46,8 +48,8 @@ where
}
}
pub async fn create_widget<C, E>(
State(state): S<C, E>,
pub async fn create_widget<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Json(body): Json<CreateWidgetDto>,
) -> Result<StatusCode, (StatusCode, String)>
where
@@ -66,8 +68,8 @@ where
Ok(StatusCode::CREATED)
}
pub async fn update_widget<C, E>(
State(state): S<C, E>,
pub async fn update_widget<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(_id): Path<u16>,
Json(body): Json<CreateWidgetDto>,
) -> Result<StatusCode, (StatusCode, String)>
@@ -87,8 +89,8 @@ where
Ok(StatusCode::OK)
}
pub async fn delete_widget<C, E>(
State(state): S<C, E>,
pub async fn delete_widget<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(id): Path<u16>,
) -> Result<StatusCode, StatusCode>
where
@@ -103,3 +105,46 @@ where
.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
Ok(StatusCode::NO_CONTENT)
}
pub async fn preview_widget<C, E, W, B, R>(
State(state): S<C, E, W, B, R>,
Path(id): Path<u16>,
) -> Result<Json<serde_json::Value>, StatusCode>
where
C: ConfigRepository,
C::Error: std::fmt::Debug,
E: EventPublisher,
E::Error: std::fmt::Debug,
W: WidgetStateReader,
{
match state.widget_states.get_widget_state(id).await {
Some(ws) => {
let map: serde_json::Map<String, serde_json::Value> = ws
.data
.iter()
.map(|(k, v)| (k.clone(), domain_value_to_json(v)))
.collect();
Ok(Json(serde_json::Value::Object(map)))
}
None => Err(StatusCode::NOT_FOUND),
}
}
fn domain_value_to_json(v: &domain::Value) -> serde_json::Value {
match v {
domain::Value::Null => serde_json::Value::Null,
domain::Value::Bool(b) => serde_json::Value::Bool(*b),
domain::Value::Number(n) => serde_json::json!(n),
domain::Value::String(s) => serde_json::Value::String(s.clone()),
domain::Value::Array(arr) => {
serde_json::Value::Array(arr.iter().map(domain_value_to_json).collect())
}
domain::Value::Object(obj) => {
let map = obj
.iter()
.map(|(k, v)| (k.clone(), domain_value_to_json(v)))
.collect();
serde_json::Value::Object(map)
}
}
}

14
crates/adapters/http-api/tests/api_tests.rs
View File

@@ -1,15 +1,21 @@
use application::DataProjection;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use config_memory::MemoryConfigStore;
use http_api::{AppState, router};
use std::sync::Arc;
use tcp_server::TcpEventBus;
use tcp_server::{ClientTracker, TcpBroadcaster, TcpEventBus};
use tower::ServiceExt;
fn test_app() -> axum::Router {
let config = Arc::new(MemoryConfigStore::new());
let events = Arc::new(TcpEventBus::new(16));
let state = AppState { config, events };
let state = AppState {
config: Arc::new(MemoryConfigStore::new()),
events: Arc::new(TcpEventBus::new(16)),
widget_states: Arc::new(DataProjection::new()),
broadcaster: Arc::new(TcpBroadcaster::new(16)),
clients: Arc::new(ClientTracker::new()),
spa_dir: None,
};
router(state)
}

48
crates/adapters/tcp-server/src/client_tracker.rs Normal file
View File

@@ -0,0 +1,48 @@
use domain::{ClientRegistry, ConnectedClient};
use std::net::SocketAddr;
use std::sync::Mutex;
use std::time::SystemTime;
#[derive(Default)]
pub struct ClientTracker {
clients: Mutex<Vec<ConnectedClient>>,
}
impl ClientTracker {
pub fn new() -> Self {
Self::default()
}
pub fn add(&self, addr: SocketAddr) {
let info = ConnectedClient {
addr: addr.to_string(),
connected_at: SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.unwrap_or_default()
.as_secs(),
};
self.clients.lock().unwrap().push(info);
}
pub fn remove(&self, addr: SocketAddr) {
let addr_str = addr.to_string();
self.clients.lock().unwrap().retain(|c| c.addr != addr_str);
}
}
impl ClientRegistry for ClientTracker {
fn add_client(&self, addr: &str, connected_at: u64) {
self.clients.lock().unwrap().push(ConnectedClient {
addr: addr.to_string(),
connected_at,
});
}
fn remove_client(&self, addr: &str) {
self.clients.lock().unwrap().retain(|c| c.addr != addr);
}
fn list_clients(&self) -> Vec<ConnectedClient> {
self.clients.lock().unwrap().clone()
}
}

2
crates/adapters/tcp-server/src/lib.rs
View File

@@ -1,9 +1,11 @@
mod broadcaster;
mod client_tracker;
mod error;
mod event_bus;
mod server;
pub use broadcaster::TcpBroadcaster;
pub use client_tracker::ClientTracker;
pub use error::TcpServerError;
pub use event_bus::TcpEventBus;
pub use server::run_tcp_server;

82
crates/adapters/tcp-server/src/server.rs
View File

@@ -1,15 +1,27 @@
use crate::broadcaster::TcpBroadcaster;
use crate::client_tracker::ClientTracker;
use crate::error::TcpServerError;
use domain::{ConfigRepository, WidgetStateReader};
use protocol::{ServerMessage, WidgetDescriptor, WireDisplayHint, WireLayoutNode, encode};
use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::sync::broadcast;
use tracing::{info, warn};
use tracing::{error, info, warn};
pub async fn run_tcp_server(
use crate::broadcaster::TcpBroadcaster;
pub async fn run_tcp_server<C, W>(
addr: &str,
broadcaster: Arc<TcpBroadcaster>,
) -> Result<(), TcpServerError> {
tracker: Arc<ClientTracker>,
config: Arc<C>,
widget_states: Arc<W>,
) -> Result<(), TcpServerError>
where
C: ConfigRepository + Send + Sync + 'static,
C::Error: std::fmt::Debug + Send,
W: WidgetStateReader + Send + Sync + 'static,
{
let listener = TcpListener::bind(addr).await.map_err(TcpServerError::Io)?;
info!(addr, "TCP server listening");
@@ -17,9 +29,21 @@ pub async fn run_tcp_server(
let (mut socket, peer) = listener.accept().await.map_err(TcpServerError::Io)?;
info!(%peer, "client connected");
tracker.add(peer);
let tracker = tracker.clone();
let mut rx = broadcaster.subscribe();
let initial_frame = build_initial_frame(&*config, &*widget_states).await;
tokio::spawn(async move {
if let Some(frame) = initial_frame
&& socket.write_all(&frame).await.is_err()
{
info!(%peer, "client disconnected during initial send");
tracker.remove(peer);
return;
}
loop {
match rx.recv().await {
Ok(frame) => {
@@ -34,6 +58,56 @@ pub async fn run_tcp_server(
}
}
}
tracker.remove(peer);
});
}
}
async fn build_initial_frame<C, W>(config: &C, widget_states: &W) -> Option<Vec<u8>>
where
C: ConfigRepository,
C::Error: std::fmt::Debug,
W: WidgetStateReader,
{
let layout = match config.get_layout().await {
Ok(Some(l)) => l,
Ok(None) => return None,
Err(e) => {
error!(error = ?e, "failed to fetch layout for initial send");
return None;
}
};
let widgets = match config.list_widgets().await {
Ok(w) => w,
Err(e) => {
error!(error = ?e, "failed to fetch widgets for initial send");
return None;
}
};
let wire_layout: WireLayoutNode = (&layout.root).into();
let mut wire_widgets = Vec::new();
for w in &widgets {
if let Some(s) = widget_states.get_widget_state(w.id).await {
wire_widgets.push(WidgetDescriptor {
id: w.id,
display_hint: WireDisplayHint::IconValue,
state: (&s).into(),
});
}
}
let msg = ServerMessage::ScreenUpdate {
layout: wire_layout,
widgets: wire_widgets,
};
match encode(&msg) {
Ok(frame) => Some(frame),
Err(e) => {
error!(error = %e, "failed to encode initial screen update");
None
}
}
}

17
crates/api-types/src/client.rs Normal file
View File

@@ -0,0 +1,17 @@
use domain::ConnectedClient;
use serde::Serialize;
#[derive(Serialize)]
pub struct ClientDto {
pub addr: String,
pub connected_at: u64,
}
impl From<&ConnectedClient> for ClientDto {
fn from(c: &ConnectedClient) -> Self {
Self {
addr: c.addr.clone(),
connected_at: c.connected_at,
}
}
}

2
crates/api-types/src/lib.rs
View File

@@ -1,8 +1,10 @@
pub mod client;
pub mod data_source;
pub mod layout;
pub mod preset;
pub mod widget;
pub use client::ClientDto;
pub use data_source::DataSourceDto;
pub use layout::{LayoutChildDto, LayoutDto, LayoutNodeDto, SizingDto};
pub use preset::{CreatePresetDto, PresetDto};

1
crates/application/Cargo.toml
View File

@@ -6,6 +6,7 @@ edition = "2024"
[dependencies]
domain.workspace = true
thiserror.workspace = true
tokio.workspace = true
[dev-dependencies]
tokio = { workspace = true }

43
crates/application/src/data_projection.rs
View File

@@ -1,9 +1,17 @@
use domain::{DataSourceId, Value, WidgetConfig, WidgetId, WidgetState};
use domain::{DataSourceId, Value, WidgetConfig, WidgetId, WidgetState, WidgetStateReader};
use std::collections::HashMap;
use tokio::sync::Mutex;
#[derive(Default)]
pub struct DataProjection {
current: HashMap<WidgetId, WidgetState>,
current: Mutex<HashMap<WidgetId, WidgetState>>,
}
impl Default for DataProjection {
fn default() -> Self {
Self {
current: Mutex::new(HashMap::new()),
}
}
}
impl DataProjection {
@@ -11,16 +19,17 @@ impl DataProjection {
Self::default()
}
pub fn get_state(&self, widget_id: WidgetId) -> Option<&WidgetState> {
self.current.get(&widget_id)
pub async fn get_state(&self, widget_id: WidgetId) -> Option<WidgetState> {
self.current.lock().await.get(&widget_id).cloned()
}
pub fn apply_poll_result(
&mut self,
pub async fn apply_poll_result(
&self,
data_source_id: DataSourceId,
raw: &Value,
widget_configs: &[WidgetConfig],
) -> Vec<(WidgetId, WidgetState)> {
let mut current = self.current.lock().await;
let mut changed = Vec::new();
for config in widget_configs {
@@ -30,13 +39,12 @@ impl DataProjection {
let new_state = config.extract(raw);
let is_changed = self
.current
let is_changed = current
.get(&config.id)
.is_none_or(|prev| *prev != new_state);
if is_changed {
self.current.insert(config.id, new_state.clone());
current.insert(config.id, new_state.clone());
changed.push((config.id, new_state));
}
}
@@ -44,3 +52,18 @@ impl DataProjection {
changed
}
}
impl WidgetStateReader for DataProjection {
async fn get_widget_state(&self, id: WidgetId) -> Option<WidgetState> {
self.get_state(id).await
}
async fn apply_raw_data(
&self,
source_id: u16,
raw: &Value,
widgets: &[WidgetConfig],
) -> Vec<(WidgetId, WidgetState)> {
self.apply_poll_result(source_id, raw, widgets).await
}
}

50
crates/application/tests/data_projection_tests.rs
View File

@@ -1,5 +1,5 @@
use application::DataProjection;
use domain::{DisplayHint, KeyMapping, Value, WidgetConfig, WidgetId, WidgetState};
use domain::{DisplayHint, KeyMapping, Value, WidgetConfig};
use std::collections::BTreeMap;
fn weather_widget() -> WidgetConfig {
@@ -28,12 +28,14 @@ fn weather_response(temp: f64) -> Value {
]))
}
#[test]
fn apply_poll_result_detects_new_widget_state() {
let mut projection = DataProjection::new();
#[tokio::test]
async fn apply_poll_result_detects_new_widget_state() {
let projection = DataProjection::new();
let widgets = vec![weather_widget()];
let changed = projection.apply_poll_result(10, &weather_response(5.4), &widgets);
let changed = projection
.apply_poll_result(10, &weather_response(5.4), &widgets)
.await;
assert_eq!(changed.len(), 1);
assert_eq!(changed[0].0, 1);
@@ -43,24 +45,32 @@ fn apply_poll_result_detects_new_widget_state() {
);
}
#[test]
fn apply_poll_result_returns_empty_when_nothing_changed() {
let mut projection = DataProjection::new();
#[tokio::test]
async fn apply_poll_result_returns_empty_when_nothing_changed() {
let projection = DataProjection::new();
let widgets = vec![weather_widget()];
projection.apply_poll_result(10, &weather_response(5.4), &widgets);
let changed = projection.apply_poll_result(10, &weather_response(5.4), &widgets);
projection
.apply_poll_result(10, &weather_response(5.4), &widgets)
.await;
let changed = projection
.apply_poll_result(10, &weather_response(5.4), &widgets)
.await;
assert!(changed.is_empty());
}
#[test]
fn apply_poll_result_detects_changed_value() {
let mut projection = DataProjection::new();
#[tokio::test]
async fn apply_poll_result_detects_changed_value() {
let projection = DataProjection::new();
let widgets = vec![weather_widget()];
projection.apply_poll_result(10, &weather_response(5.4), &widgets);
let changed = projection.apply_poll_result(10, &weather_response(6.1), &widgets);
projection
.apply_poll_result(10, &weather_response(5.4), &widgets)
.await;
let changed = projection
.apply_poll_result(10, &weather_response(6.1), &widgets)
.await;
assert_eq!(changed.len(), 1);
assert_eq!(
@@ -69,9 +79,9 @@ fn apply_poll_result_detects_changed_value() {
);
}
#[test]
fn apply_poll_result_only_updates_widgets_bound_to_source() {
let mut projection = DataProjection::new();
#[tokio::test]
async fn apply_poll_result_only_updates_widgets_bound_to_source() {
let projection = DataProjection::new();
let widgets = vec![
weather_widget(),
WidgetConfig::new(
@@ -86,7 +96,9 @@ fn apply_poll_result_only_updates_widgets_bound_to_source() {
),
];
let changed = projection.apply_poll_result(10, &weather_response(5.4), &widgets);
let changed = projection
.apply_poll_result(10, &weather_response(5.4), &widgets)
.await;
assert_eq!(changed.len(), 1);
assert_eq!(changed[0].0, 1);

2
crates/bootstrap/src/config.rs
View File

@@ -5,6 +5,7 @@ pub struct ServerConfig {
pub tcp_addr: String,
pub http_addr: String,
pub poll_interval_secs: u64,
pub spa_dir: Option<String>,
}
impl ServerConfig {
@@ -18,6 +19,7 @@ impl ServerConfig {
.ok()
.and_then(|v| v.parse().ok())
.unwrap_or(5),
spa_dir: env::var("KFRAME_SPA_DIR").ok(),
}
}
}

14
crates/bootstrap/src/event_handler.rs
View File

@@ -3,14 +3,13 @@ use config_sqlite::SqliteConfigStore;
use domain::{BroadcastPort, ConfigRepository, DomainEvent};
use std::sync::Arc;
use tcp_server::{TcpBroadcaster, TcpEventBus};
use tokio::sync::Mutex;
use tracing::{error, info, warn};
pub async fn run(
event_bus: Arc<TcpEventBus>,
config: Arc<SqliteConfigStore>,
broadcaster: Arc<TcpBroadcaster>,
projection: Arc<Mutex<DataProjection>>,
projection: Arc<DataProjection>,
) {
let mut rx = event_bus.subscribe();
@@ -25,11 +24,12 @@ pub async fn run(
}
};
let proj = projection.lock().await;
let widget_states: Vec<_> = widgets
.iter()
.filter_map(|w| proj.get_state(w.id).map(|s| (w.id, s.clone())))
.collect();
let mut widget_states = Vec::new();
for w in &widgets {
if let Some(s) = projection.get_state(w.id).await {
widget_states.push((w.id, s));
}
}
if let Err(e) = broadcaster
.push_screen_update(&layout, &widget_states)

15
crates/bootstrap/src/main.rs
View File

@@ -7,8 +7,7 @@ use application::DataProjection;
use config_sqlite::SqliteConfigStore;
use http_api::AppState;
use std::sync::Arc;
use tcp_server::{TcpBroadcaster, TcpEventBus, run_tcp_server};
use tokio::sync::Mutex;
use tcp_server::{ClientTracker, TcpBroadcaster, TcpEventBus, run_tcp_server};
use tracing::{error, info};
#[tokio::main]
@@ -29,12 +28,16 @@ async fn main() -> Result<()> {
let event_bus = Arc::new(TcpEventBus::new(64));
let broadcaster = Arc::new(TcpBroadcaster::new(64));
let projection = Arc::new(Mutex::new(DataProjection::new()));
let projection = Arc::new(DataProjection::new());
let tracker = Arc::new(ClientTracker::new());
let tcp_addr = cfg.tcp_addr.clone();
let tcp_bc = broadcaster.clone();
let tcp_tracker = tracker.clone();
let tcp_config = config_store.clone();
let tcp_proj = projection.clone();
tokio::spawn(async move {
if let Err(e) = run_tcp_server(&tcp_addr, tcp_bc).await {
if let Err(e) = run_tcp_server(&tcp_addr, tcp_bc, tcp_tracker, tcp_config, tcp_proj).await {
error!(error = %e, "tcp server failed");
}
});
@@ -44,6 +47,10 @@ async fn main() -> Result<()> {
let http_state = AppState {
config: config_store.clone(),
events: event_bus.clone(),
widget_states: projection.clone(),
broadcaster: broadcaster.clone(),
clients: tracker.clone(),
spa_dir: cfg.spa_dir,
};
tokio::spawn(async move {
if let Err(e) = http_api::serve(&http_addr, http_state).await {

146
crates/bootstrap/src/polling.rs
View File

@@ -7,71 +7,135 @@ use domain::{
use http_json::HttpJsonAdapter;
use media_adapter::MediaAdapter;
use rss_adapter::RssAdapter;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tcp_server::TcpBroadcaster;
use tokio::sync::Mutex;
use tokio::task::JoinHandle;
use tracing::{debug, info, warn};
const SOURCE_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
pub async fn run(
config: Arc<SqliteConfigStore>,
broadcaster: Arc<TcpBroadcaster>,
projection: Arc<Mutex<DataProjection>>,
poll_interval_secs: u64,
projection: Arc<DataProjection>,
_poll_interval_secs: u64,
) -> Result<()> {
let http_adapter = HttpJsonAdapter::new();
let media_adapter = MediaAdapter::new();
let rss_adapter = RssAdapter::new();
let interval = Duration::from_secs(poll_interval_secs);
let http_adapter = Arc::new(HttpJsonAdapter::new());
let media_adapter = Arc::new(MediaAdapter::new());
let rss_adapter = Arc::new(RssAdapter::new());
info!(interval_secs = poll_interval_secs, "polling loop started");
let mut running: HashMap<u16, JoinHandle<()>> = HashMap::new();
info!("polling manager started");
loop {
tokio::time::sleep(interval).await;
let sources = config
.list_data_sources()
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let widgets = config
.list_widgets()
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
let layout = config
.get_layout()
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
if sources.is_empty() || widgets.is_empty() {
debug!("no sources or widgets configured, skipping poll");
continue;
}
let current_ids: Vec<u16> = sources.iter().map(|s| s.id).collect();
let mut all_changed: Vec<(u16, WidgetState)> = Vec::new();
running.retain(|id, handle| {
if !current_ids.contains(id) {
info!(source_id = id, "stopping poll for removed source");
handle.abort();
false
} else {
true
}
});
for source in &sources {
let result =
match poll_source(&http_adapter, &media_adapter, &rss_adapter, source).await {
Ok(v) => v,
Err(e) => {
warn!(source = %source.name, error = %e, "poll failed");
continue;
}
};
if source.source_type == DataSourceType::Webhook {
continue;
}
if running.contains_key(&source.id) {
continue;
}
let mut proj = projection.lock().await;
let changed = proj.apply_poll_result(source.id, &result, &widgets);
all_changed.extend(changed);
let source_id = source.id;
let source = source.clone();
let config = config.clone();
let broadcaster = broadcaster.clone();
let projection = projection.clone();
let http = http_adapter.clone();
let media = media_adapter.clone();
let rss = rss_adapter.clone();
info!(
source_id = source.id,
name = %source.name,
interval_secs = source.poll_interval.as_secs(),
"starting poll task"
);
let handle = tokio::spawn(async move {
poll_loop(source, config, broadcaster, projection, http, media, rss).await;
});
running.insert(source_id, handle);
}
if !all_changed.is_empty() {
if let Some(l) = &layout {
broadcaster
.push_screen_update(l, &all_changed)
.await
.map_err(|e| anyhow::anyhow!("{e}"))?;
if running.is_empty() {
debug!("no pollable sources, waiting");
}
tokio::time::sleep(SOURCE_REFRESH_INTERVAL).await;
}
}
async fn poll_loop(
source: DataSource,
config: Arc<SqliteConfigStore>,
broadcaster: Arc<TcpBroadcaster>,
projection: Arc<DataProjection>,
http_adapter: Arc<HttpJsonAdapter>,
media_adapter: Arc<MediaAdapter>,
rss_adapter: Arc<RssAdapter>,
) {
let interval = source.poll_interval;
loop {
tokio::time::sleep(interval).await;
let result = match poll_source(&http_adapter, &media_adapter, &rss_adapter, &source).await {
Ok(v) => v,
Err(e) => {
warn!(source = %source.name, error = %e, "poll failed");
continue;
}
info!(count = all_changed.len(), "pushed widget updates");
};
let widgets = match config.list_widgets().await {
Ok(w) => w,
Err(e) => {
warn!(error = %e, "failed to fetch widgets");
continue;
}
};
let layout = match config.get_layout().await {
Ok(l) => l,
Err(e) => {
warn!(error = %e, "failed to fetch layout");
continue;
}
};
let changed: Vec<(u16, WidgetState)> = projection
.apply_poll_result(source.id, &result, &widgets)
.await;
if !changed.is_empty() {
if let Some(l) = &layout
&& let Err(e) = broadcaster.push_screen_update(l, &changed).await
{
warn!(error = %e, "failed to push update");
}
info!(source = %source.name, count = changed.len(), "pushed widget updates");
}
}
}

3
crates/client-esp32/src/tasks/render.rs
View File

@@ -21,9 +21,10 @@ pub fn run(
loop {
match rx.recv_timeout(RENDER_POLL_INTERVAL) {
Ok(msg) => {
let is_screen_update = matches!(msg, ServerMessage::ScreenUpdate { .. });
let repaints = app.handle_message(msg);
if !repaints.is_empty() && first_update {
if !repaints.is_empty() && (first_update || is_screen_update) {
display.fill_background(SCREEN).unwrap();
first_update = false;
}

5
crates/domain/src/lib.rs
View File

@@ -10,7 +10,10 @@ pub use entities::{
LayoutPreset, LayoutPresetId, WidgetConfig, WidgetId,
};
pub use events::DomainEvent;
pub use ports::{BroadcastPort, ConfigRepository, DataSourcePort, EventPublisher};
pub use ports::{
BroadcastPort, ClientRegistry, ConfigRepository, ConnectedClient, DataSourcePort,
EventPublisher, WidgetStateReader,
};
pub use value_objects::{
ContainerNode, Direction, DisplayHint, KeyMapping, Layout, LayoutChild, LayoutNode,
LayoutValidationError, Sizing, Value, WidgetError, WidgetState,

11
crates/domain/src/ports/client_registry.rs Normal file
View File

@@ -0,0 +1,11 @@
#[derive(Clone)]
pub struct ConnectedClient {
pub addr: String,
pub connected_at: u64,
}
pub trait ClientRegistry {
fn add_client(&self, addr: &str, connected_at: u64);
fn remove_client(&self, addr: &str);
fn list_clients(&self) -> Vec<ConnectedClient>;
}

4
crates/domain/src/ports/mod.rs
View File

@@ -1,9 +1,13 @@
mod broadcast;
mod client_registry;
mod config_repository;
mod data_source_port;
mod event;
mod widget_state_reader;
pub use broadcast::BroadcastPort;
pub use client_registry::{ClientRegistry, ConnectedClient};
pub use config_repository::ConfigRepository;
pub use data_source_port::DataSourcePort;
pub use event::EventPublisher;
pub use widget_state_reader::WidgetStateReader;

14
crates/domain/src/ports/widget_state_reader.rs Normal file
View File

@@ -0,0 +1,14 @@
use crate::entities::WidgetId;
use crate::value_objects::WidgetState;
use std::future::Future;
pub trait WidgetStateReader {
fn get_widget_state(&self, id: WidgetId) -> impl Future<Output = Option<WidgetState>> + Send;
fn apply_raw_data(
&self,
source_id: u16,
raw: &crate::value_objects::Value,
widgets: &[crate::entities::WidgetConfig],
) -> impl Future<Output = Vec<(WidgetId, WidgetState)>> + Send;
}