octoturge
9dff600160
The only way to give a gateway a printer was to edit printers.toml and restart it. A printer added from the console now arrives as a provision_printer message, is checked by actually connecting to it (so a mistyped access code is reported to the operator, not discovered later), and only then is saved to the gateway's own SQLite file and joined to the running fleet - no restart. remove_printer undoes it. Both are answered with a command_result carrying the request id. The access code exists only on the gateway: the cloud relays it once and keeps nothing, which is the platform's rule for printer credentials. The store file is created owner-only; the message type's Debug and the fleet entry's Debug redact the credential; and an unparseable message from the cloud is no longer logged with its payload, because the message that fails to parse is exactly the one carrying an access code. The cloud may not choose a certificate path or serial port on the gateway's disk, so provision_printer cannot carry them. Startup now reads printers.toml (strict, as before; a missing file is fine) and the store (a bad stored entry is skipped, not fatal), so the gateway comes up with its fleet with the API unreachable. Discovery treats stored printers as already configured. rusqlite (MIT, bundled SQLite) passes cargo deny.
612 lines
29 KiB
Rust
612 lines
29 KiB
Rust
//! The cloud connection: a WebSocket to `continuum-backend`'s `/ws/edge/v1`.
|
|
//!
|
|
//! This is where the proxy earns its name. Until now it connected, said
|
|
//! hello, and sent heartbeats into the void while the printer fleet lived in
|
|
//! a completely separate part of the program with nothing routing between
|
|
//! them. Both directions are wired up now:
|
|
//!
|
|
//! - **Up:** every `TELEMETRY_INTERVAL`, one snapshot per printer, batched
|
|
//! into a single `Telemetry` message.
|
|
//! - **Down:** a `Command` from the cloud is looked up by printer id,
|
|
//! executed, and answered with a `CommandResult`.
|
|
//!
|
|
//! - **Out of band:** anything that needs to reach the cloud from outside
|
|
//! this loop goes through an mpsc channel (`outbound`), which is how a
|
|
//! command that takes minutes reports its outcome after the loop has
|
|
//! already answered for it.
|
|
//!
|
|
//! Still deliberately the simplest version of a reconnecting client: a fixed
|
|
//! retry delay rather than a backoff curve. Telemetry produced while the
|
|
//! uplink is down is not buffered, and that is now a considered decision
|
|
//! rather than a gap - the cloud stores only each printer's latest state and
|
|
//! has no table for a history, and a Bambu's state is level-triggered
|
|
//! (`FINISH` persists until the next print starts), so the first snapshot
|
|
//! after a reconnect carries the outcome. Replaying a backlog of five-second
|
|
//! snapshots into a row that holds one value would churn it with stale
|
|
//! readings for no consumer. See PLAN.md for what would actually be worth
|
|
//! buffering.
|
|
|
|
mod download;
|
|
mod webrtc;
|
|
pub mod protocol;
|
|
|
|
pub use protocol::{ClientMessage, DiscoveredPrinter, PrinterCommand, PrinterTelemetry, ServerMessage};
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use futures_util::{SinkExt, StreamExt};
|
|
use tokio::sync::Mutex;
|
|
use tokio_tungstenite::tungstenite::Message;
|
|
use tokio_tungstenite::tungstenite::handshake::client::{Request, generate_key};
|
|
use tracing::{info, warn};
|
|
|
|
use crate::config::Config;
|
|
use crate::provision;
|
|
use crate::store::Store;
|
|
use protocol::ProvisionPrinter;
|
|
use crate::printer::{GenericPrinter, PrinterHandle};
|
|
|
|
/// The websocket, once connected. Named because `connect` has to hand it to
|
|
/// `serve` and the full type is three layers of generics.
|
|
type WsStream = tokio_tungstenite::WebSocketStream<
|
|
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
|
|
>;
|
|
|
|
/// First retry delay after a failed connection, doubling up to
|
|
/// `RETRY_DELAY_MAX`.
|
|
const RETRY_DELAY: Duration = Duration::from_secs(5);
|
|
/// Ceiling for the backoff.
|
|
///
|
|
/// A gateway cannot tell a cloud that is briefly restarting from a token the
|
|
/// cloud will never accept - both are a failed connection here - so a flat
|
|
/// 5s retry meant a permanently misconfigured gateway knocking twelve times a
|
|
/// minute, forever, which is both pointless and the shape of something worth
|
|
/// rate-limiting at the other end. A minute is short enough that an outage
|
|
/// resolves without anyone touching the gateway, which matters when the
|
|
/// gateway is in someone else's workshop.
|
|
const RETRY_DELAY_MAX: Duration = Duration::from_secs(60);
|
|
/// Depth of the out-of-band outbound queue.
|
|
///
|
|
/// Bounded, because an unbounded queue facing a cloud that may be down for
|
|
/// hours is just a slow memory leak. Only out-of-band messages go through it,
|
|
/// one per long-running command, so 64 is far more than a farm generates
|
|
/// between reconnects. A sender waits when it is full rather than dropping,
|
|
/// which is the right trade for a message whose whole purpose is to explain a
|
|
/// failure.
|
|
const OUTBOUND_CAPACITY: usize = 64;
|
|
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(15);
|
|
/// Faster than the heartbeat, slower than the printers push. A Bambu on v2
|
|
/// emits a full status roughly every second; forwarding at that rate would be
|
|
/// a lot of traffic to describe a machine whose state changes on the scale of
|
|
/// a layer, so this samples the cached report instead of chasing every push.
|
|
const TELEMETRY_INTERVAL: Duration = Duration::from_secs(5);
|
|
|
|
/// The printer fleet, shared between `main` (which owns it and keeps every
|
|
/// connection alive) and this module (which reads status and dispatches
|
|
/// commands).
|
|
///
|
|
/// `Arc` is shared ownership - several places hold the same value and the
|
|
/// last one to let go drops it, roughly how every TS object reference
|
|
/// already behaves. `Mutex` is what makes mutation across tasks safe: Rust
|
|
/// will not let two tasks hold `&mut` to the same data, so a lock is how you
|
|
/// take turns. `tokio::sync::Mutex` rather than `std::sync::Mutex`
|
|
/// specifically because the code below holds the lock across an `.await`,
|
|
/// which the std one is not designed for.
|
|
///
|
|
/// One lock for the whole `Vec` is coarse: a slow command to one printer
|
|
/// stalls telemetry for all of them. Fine at the handful-of-printers scale
|
|
/// this runs at now, and the shape to revisit (a lock per printer, or an
|
|
/// actor task per printer owning its own handle) before a farm gets large.
|
|
pub type SharedFleet = Arc<Mutex<Vec<PrinterHandle>>>;
|
|
|
|
/// The `Host` header for a websocket URL: authority without the scheme.
|
|
///
|
|
/// Needed because building the upgrade request by hand (to attach the edge
|
|
/// token) means tokio-tungstenite no longer derives the handshake headers
|
|
/// from a URL for us. An error rather than a guess if the URL has no
|
|
/// authority: a `Host` of the wrong thing fails the handshake with something
|
|
/// far less obvious than this does.
|
|
fn host_header(url: &str) -> anyhow::Result<String> {
|
|
let authority = url
|
|
.split_once("://")
|
|
.map(|(_, rest)| rest)
|
|
.unwrap_or(url)
|
|
.split(['/', '?', '#'])
|
|
.next()
|
|
.unwrap_or("");
|
|
|
|
if authority.is_empty() {
|
|
anyhow::bail!("the uplink URL has no host to connect to: {url}");
|
|
}
|
|
Ok(authority.to_string())
|
|
}
|
|
|
|
/// The queue a task elsewhere puts a message on to have the uplink send it.
|
|
///
|
|
/// Created by the caller rather than inside [`run`] so that a task which is
|
|
/// not the uplink - printer discovery, today - can hold the sender without
|
|
/// the uplink having to know what those tasks are. The property that matters
|
|
/// is unchanged either way: it outlives any one connection, so a message
|
|
/// produced while the socket is down waits for the next socket instead of
|
|
/// being dropped, and the sender living for the whole program is what stops
|
|
/// `recv()` from ever returning `None`.
|
|
pub fn channel() -> (
|
|
tokio::sync::mpsc::Sender<ClientMessage>,
|
|
tokio::sync::mpsc::Receiver<ClientMessage>,
|
|
) {
|
|
tokio::sync::mpsc::channel(OUTBOUND_CAPACITY)
|
|
}
|
|
|
|
/// Runs forever: connect, serve the connection until it drops, wait, retry.
|
|
pub async fn run(
|
|
config: Config,
|
|
fleet: SharedFleet,
|
|
store: Option<Arc<Store>>,
|
|
outbound: tokio::sync::mpsc::Sender<ClientMessage>,
|
|
mut inbox: tokio::sync::mpsc::Receiver<ClientMessage>,
|
|
) {
|
|
let mut delay = RETRY_DELAY;
|
|
|
|
loop {
|
|
info!(url = %config.uplink_url, "connecting to cloud uplink");
|
|
|
|
match connect(&config).await {
|
|
Ok(ws_stream) => {
|
|
// Reset on a completed handshake, which is the only
|
|
// unambiguous evidence that the cloud is reachable and this
|
|
// gateway's token is accepted - and the reason `connect` is
|
|
// split out at all. Resetting after `serve` returns instead
|
|
// would not distinguish a connection that lived for an hour
|
|
// from one that failed to open.
|
|
delay = RETRY_DELAY;
|
|
if let Err(err) = serve(ws_stream, &config, &fleet, store.as_ref(), &outbound, &mut inbox).await {
|
|
warn!(?err, "uplink connection dropped, reconnecting in {delay:?}");
|
|
}
|
|
}
|
|
Err(err) => warn!(?err, "uplink connection failed, retrying in {delay:?}"),
|
|
}
|
|
|
|
tokio::time::sleep(delay).await;
|
|
delay = (delay * 2).min(RETRY_DELAY_MAX);
|
|
}
|
|
}
|
|
|
|
/// Opens the websocket. Separate from `serve` so `run` can tell a failed
|
|
/// handshake from a connection that opened and later dropped; they want
|
|
/// different retry behaviour.
|
|
async fn connect(config: &Config) -> anyhow::Result<WsStream> {
|
|
// The edge token goes on the upgrade request, not in the first message.
|
|
// The backend verifies it before the socket exists, so a bad token is an
|
|
// HTTP 401 here rather than a connection that opens and is then closed -
|
|
// and the credential stays out of the URL, where it would land in every
|
|
// access and proxy log along the way. `Authorization` is the header
|
|
// those tools already know to redact.
|
|
let request = Request::builder()
|
|
.uri(&config.uplink_url)
|
|
.header("Authorization", format!("Bearer {}", config.gateway_token))
|
|
// tokio-tungstenite fills these in for a plain URL, but not once we
|
|
// hand it a request we built ourselves - and a handshake missing any
|
|
// of them is rejected by the server rather than ignored.
|
|
.header("Host", host_header(&config.uplink_url)?)
|
|
.header("Connection", "Upgrade")
|
|
.header("Upgrade", "websocket")
|
|
.header("Sec-WebSocket-Version", "13")
|
|
.header("Sec-WebSocket-Key", generate_key())
|
|
.body(())?;
|
|
|
|
let (ws_stream, _response) = tokio_tungstenite::connect_async(request).await?;
|
|
Ok(ws_stream)
|
|
}
|
|
|
|
/// Says hello, then pumps the connection until it drops.
|
|
async fn serve(
|
|
ws_stream: WsStream,
|
|
config: &Config,
|
|
fleet: &SharedFleet,
|
|
store: Option<&Arc<Store>>,
|
|
outbound: &tokio::sync::mpsc::Sender<ClientMessage>,
|
|
inbox: &mut tokio::sync::mpsc::Receiver<ClientMessage>,
|
|
) -> anyhow::Result<()> {
|
|
let (mut write, mut read) = ws_stream.split();
|
|
|
|
let hello = ClientMessage::Hello {
|
|
gateway_id: config.gateway_id.clone(),
|
|
farm_id: config.farm_id.clone(),
|
|
stream_base_url: config.stream_base_url.clone(),
|
|
};
|
|
write.send(Message::Text(serde_json::to_string(&hello)?)).await?;
|
|
|
|
let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
|
|
let mut telemetry = tokio::time::interval(TELEMETRY_INTERVAL);
|
|
let mut seq: u64 = 0;
|
|
|
|
loop {
|
|
tokio::select! {
|
|
_ = heartbeat.tick() => {
|
|
seq += 1;
|
|
let msg = ClientMessage::Heartbeat { seq };
|
|
write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
|
|
}
|
|
|
|
_ = telemetry.tick() => {
|
|
// Read first, then ask for the next one. A v1 Bambu answers
|
|
// `refresh_status` over MQTT some time after this returns, so
|
|
// collecting first means every tick reports the previous
|
|
// tick's refresh - one interval of lag, rather than a
|
|
// guaranteed miss from reading before the answer could
|
|
// possibly have arrived.
|
|
let printers = collect_telemetry(fleet).await;
|
|
refresh_fleet(fleet).await;
|
|
// Nothing to say is not worth a frame. Happens whenever no
|
|
// printer has reported yet - every vendor but Bambu is still
|
|
// a stub, so a Prusa-only fleet is silent here by design.
|
|
if !printers.is_empty() {
|
|
let msg = ClientMessage::Telemetry { printers };
|
|
write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
|
|
}
|
|
}
|
|
|
|
// Whatever a task elsewhere wants said. Taken from the queue
|
|
// before it is written, so a send that fails here loses the
|
|
// message - the connection is going down anyway, and the
|
|
// alternative (peek, write, then pop) buys little for a message
|
|
// that is already best-effort diagnostics.
|
|
Some(msg) = inbox.recv() => {
|
|
write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
|
|
}
|
|
|
|
frame = read.next() => {
|
|
match frame {
|
|
Some(Ok(Message::Text(text))) => {
|
|
match serde_json::from_str::<ServerMessage>(&text) {
|
|
Ok(ServerMessage::Command { command_id, printer_id, command }) => {
|
|
// `start_print` is the one command that cannot
|
|
// be answered inline. See `spawn_long_command`.
|
|
let msg = if command.runs_for_minutes() {
|
|
spawn_long_command(
|
|
fleet.clone(),
|
|
printer_id.clone(),
|
|
command,
|
|
command_id.clone(),
|
|
outbound.clone(),
|
|
);
|
|
ClientMessage::CommandResult {
|
|
command_id, ok: true, error: None,
|
|
}
|
|
} else {
|
|
match dispatch(fleet, &printer_id, command).await {
|
|
Ok(()) => ClientMessage::CommandResult {
|
|
command_id, ok: true, error: None,
|
|
},
|
|
Err(err) => {
|
|
warn!(%printer_id, %err, "command failed");
|
|
ClientMessage::CommandResult {
|
|
command_id, ok: false, error: Some(err),
|
|
}
|
|
}
|
|
}
|
|
};
|
|
write.send(Message::Text(serde_json::to_string(&msg)?)).await?;
|
|
}
|
|
// Retrying will not help: the token is wrong, not
|
|
// the network. Logged at error level and returned
|
|
// so it is obvious which of the two it is.
|
|
Ok(ServerMessage::HelloError { message }) => {
|
|
tracing::error!(%message, "cloud rejected this gateway's handshake - check CONTINUUM_GATEWAY_TOKEN");
|
|
return Ok(());
|
|
}
|
|
// Spawned, never awaited inline: go2rtc gathers
|
|
// ICE candidates before answering, which
|
|
// includes a STUN round trip, so this takes
|
|
// seconds. Awaiting it here would stall
|
|
// telemetry and heartbeats for the whole farm
|
|
// while one person opened one camera.
|
|
Ok(ServerMessage::WebrtcOffer { session_id, camera_stream, sdp }) => {
|
|
spawn_webrtc_negotiation(
|
|
config.go2rtc_api_listen.clone(),
|
|
session_id,
|
|
camera_stream,
|
|
sdp,
|
|
outbound.clone(),
|
|
);
|
|
}
|
|
// Spawned: connecting to the new printer can take
|
|
// many seconds, and awaiting it here would stall
|
|
// heartbeats and telemetry for the whole farm.
|
|
Ok(ServerMessage::ProvisionPrinter { request_id, printer }) => {
|
|
spawn_provision(fleet.clone(), store.cloned(), request_id, Some(printer), outbound.clone());
|
|
}
|
|
Ok(ServerMessage::RemovePrinter { request_id, printer_id }) => {
|
|
spawn_remove(fleet.clone(), store.cloned(), request_id, printer_id, outbound.clone());
|
|
}
|
|
Ok(other) => info!(?other, "received from cloud"),
|
|
// Loudly. This used to be an `if let Ok(..)` with
|
|
// no else, which is the same silent-drop bug that
|
|
// hid a real `print.push_status` failure for
|
|
// weeks - and here it would mean commands
|
|
// vanishing with the cloud never being told.
|
|
// The payload is NOT logged. A `provision_printer` message
|
|
// carries an access code, and one that failed to parse is
|
|
// exactly the message that would end up here.
|
|
Err(err) => warn!(%err, bytes = text.len(), "unparseable message from cloud, ignoring"),
|
|
}
|
|
}
|
|
Some(Ok(Message::Close(_))) | None => return Ok(()),
|
|
Some(Err(err)) => return Err(err.into()),
|
|
_ => {}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Asks every printer to report in full, for the next tick to pick up. Failure
|
|
/// is logged and otherwise ignored: a printer that cannot be asked right now
|
|
/// is a printer whose telemetry goes stale, not a reason to tear down the
|
|
/// cloud connection for the whole farm.
|
|
async fn refresh_fleet(fleet: &SharedFleet) {
|
|
let fleet = fleet.lock().await;
|
|
for printer in fleet.iter() {
|
|
if let Err(err) = printer.refresh_status().await {
|
|
warn!(printer = %printer.base().id, %err, "failed to request a status refresh");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// One snapshot per printer that has something to report. Translates Bambu's
|
|
/// `print.push_status` into the vendor-neutral shape the cloud consumes - see
|
|
/// `PrinterTelemetry` for why this isn't just forwarded raw.
|
|
async fn collect_telemetry(fleet: &SharedFleet) -> Vec<PrinterTelemetry> {
|
|
let fleet = fleet.lock().await;
|
|
let mut out = Vec::new();
|
|
|
|
for printer in fleet.iter() {
|
|
let base = printer.base();
|
|
let Some(report) = printer.print_status().await else {
|
|
continue;
|
|
};
|
|
|
|
out.push(PrinterTelemetry {
|
|
printer_id: base.id.clone(),
|
|
name: base.name.clone(),
|
|
state: report.gcode_state.clone(),
|
|
percent: report.mc_percent,
|
|
stage: report.stg_cur,
|
|
layer: report.layer_num,
|
|
total_layers: report.total_layer_num,
|
|
remaining_minutes: report.mc_remaining_time,
|
|
// An idle printer reports `""` rather than omitting the field;
|
|
// empty-string-as-absent is worth normalising here so the cloud
|
|
// doesn't have to know that.
|
|
job_name: report.subtask_name.clone().filter(|s| !s.is_empty()),
|
|
nozzle_temp: report.nozzle_temper,
|
|
nozzle_target: report.nozzle_target_temper,
|
|
bed_temp: report.bed_temper,
|
|
bed_target: report.bed_target_temper,
|
|
print_error: report.print_error,
|
|
active_alerts: report.hms.as_ref().map(|h| h.len()),
|
|
// Derived from the same reported URL `camera::sources` builds
|
|
// the go2rtc config from, so a printer is advertised as
|
|
// streamable here only if it is also actually being served
|
|
// there. Asking `camera` rather than keeping a second copy of
|
|
// the rule means the two cannot drift apart.
|
|
camera_stream: crate::camera::source_for(
|
|
printer,
|
|
report.ipcam.as_ref().and_then(|c| c.rtsp_url.as_deref()),
|
|
)
|
|
.map(|source| source.stream_name),
|
|
});
|
|
}
|
|
|
|
out
|
|
}
|
|
|
|
/// Runs a command that takes minutes, off this connection's task, and answers
|
|
/// the cloud immediately.
|
|
///
|
|
/// Everything else here is an MQTT publish that returns in milliseconds, so
|
|
/// awaiting it inline is fine. `start_print` is not: it downloads tens of
|
|
/// megabytes and then transfers them to a printer over FTPS at a measured
|
|
/// 0.2 MiB/s, which is about four minutes for a 50 MB project. Awaited
|
|
/// inline, that blocks this whole `select!` loop - no heartbeats and, far
|
|
/// worse, **no telemetry** for the duration.
|
|
///
|
|
/// No telemetry is the one that makes this mandatory rather than tidy. The
|
|
/// cloud derives a job's state from the telemetry that follows a start, so a
|
|
/// gateway that goes silent for four minutes during every job start is
|
|
/// withholding exactly the information the job depends on, while the backend
|
|
/// is also timing the command out after ten seconds and concluding the start
|
|
/// failed.
|
|
///
|
|
/// So the `ok` this produces means "accepted, and working on it" - which is
|
|
/// all the cloud ever trusted a start reply to mean anyway. *That* a failure
|
|
/// after this point happened is still noticed by the cloud on its own: the
|
|
/// job never reaches `printing` and the backend reaps it. What used to be
|
|
/// missing was *why*, which lived only in this gateway's log - on a machine
|
|
/// in someone else's workshop.
|
|
///
|
|
/// So a late failure now goes back over the outbound queue as a second
|
|
/// `CommandResult` for the same `command_id`. Be clear about what the backend
|
|
/// does with it: that command is long since resolved, so it is not awaiting
|
|
/// an answer and the result does not change the job's fate. It is logged with
|
|
/// its reason, next to the gateway and farm that produced it, which is the
|
|
/// difference between "job failed" and "job failed: filament mapping for
|
|
/// plate 1 names a slot the AMS does not have". Attaching it to the job row
|
|
/// needs a column `print_jobs` does not have yet.
|
|
///
|
|
/// Nothing is sent on success - the telemetry that follows says so better
|
|
/// than a message could, and a result for an already-resolved command is
|
|
/// noise unless it carries news.
|
|
fn spawn_long_command(
|
|
fleet: SharedFleet,
|
|
printer_id: String,
|
|
command: PrinterCommand,
|
|
command_id: String,
|
|
outbound: tokio::sync::mpsc::Sender<ClientMessage>,
|
|
) {
|
|
tokio::spawn(async move {
|
|
match dispatch(&fleet, &printer_id, command).await {
|
|
Ok(()) => info!(%printer_id, "long-running command finished"),
|
|
Err(err) => {
|
|
// Still logged locally, and still at error level: the queue
|
|
// only reaches a cloud that is up, and this is the copy that
|
|
// survives a gateway with no internet.
|
|
tracing::error!(
|
|
%printer_id, %err,
|
|
"long-running command failed after the cloud was already told it was accepted"
|
|
);
|
|
let late = ClientMessage::CommandResult {
|
|
command_id,
|
|
ok: false,
|
|
error: Some(err),
|
|
};
|
|
// `send` waits for room rather than dropping, and only fails
|
|
// if the receiver is gone - which means `run` has returned,
|
|
// i.e. the process is shutting down and there is nobody left
|
|
// to tell.
|
|
if outbound.send(late).await.is_err() {
|
|
warn!(%printer_id, "uplink is shutting down; the failure reason was not reported");
|
|
}
|
|
}
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Adds a printer off the connection's task and always reports how it went:
|
|
/// the console is holding a form open waiting for exactly this.
|
|
fn spawn_provision(
|
|
fleet: SharedFleet,
|
|
store: Option<Arc<Store>>,
|
|
request_id: String,
|
|
request: Option<ProvisionPrinter>,
|
|
outbound: tokio::sync::mpsc::Sender<ClientMessage>,
|
|
) {
|
|
tokio::spawn(async move {
|
|
let result = match (store, request) {
|
|
(Some(store), Some(request)) => provision::add(&fleet, &store, request).await,
|
|
(None, _) => Err("this gateway has no local store, so it cannot add printers".to_string()),
|
|
(_, None) => Err("no printer in the request".to_string()),
|
|
};
|
|
report(request_id, result, &outbound).await;
|
|
});
|
|
}
|
|
|
|
fn spawn_remove(
|
|
fleet: SharedFleet,
|
|
store: Option<Arc<Store>>,
|
|
request_id: String,
|
|
printer_id: String,
|
|
outbound: tokio::sync::mpsc::Sender<ClientMessage>,
|
|
) {
|
|
tokio::spawn(async move {
|
|
let result = match store {
|
|
Some(store) => provision::remove(&fleet, &store, &printer_id).await,
|
|
None => Err("this gateway has no local store".to_string()),
|
|
};
|
|
report(request_id, result, &outbound).await;
|
|
});
|
|
}
|
|
|
|
async fn report(request_id: String, result: Result<(), String>, outbound: &tokio::sync::mpsc::Sender<ClientMessage>) {
|
|
let msg = match result {
|
|
Ok(()) => ClientMessage::CommandResult { command_id: request_id, ok: true, error: None },
|
|
Err(error) => {
|
|
warn!(%error, "printer request failed");
|
|
ClientMessage::CommandResult { command_id: request_id, ok: false, error: Some(error) }
|
|
}
|
|
};
|
|
if outbound.send(msg).await.is_err() {
|
|
warn!("uplink is shutting down; the result was not reported");
|
|
}
|
|
}
|
|
|
|
/// Answers one WebRTC offer, off this connection's task.
|
|
///
|
|
/// Always answers, success or failure: the cloud is holding a browser's
|
|
/// request open waiting for this, and a silent failure there is a viewer
|
|
/// watching a video element that never loads until something times out.
|
|
fn spawn_webrtc_negotiation(
|
|
api_listen: String,
|
|
session_id: String,
|
|
camera_stream: String,
|
|
offer_sdp: String,
|
|
outbound: tokio::sync::mpsc::Sender<ClientMessage>,
|
|
) {
|
|
tokio::spawn(async move {
|
|
let reply = match webrtc::negotiate(&api_listen, &camera_stream, &offer_sdp).await {
|
|
Ok(sdp) => ClientMessage::WebrtcAnswer { session_id, sdp },
|
|
Err(err) => {
|
|
// Logged as well as returned. The cloud gets the reason, but
|
|
// this is the copy that survives a gateway whose uplink is
|
|
// about to drop.
|
|
warn!(%camera_stream, %err, "could not negotiate a camera stream");
|
|
ClientMessage::WebrtcError { session_id, message: err.to_string() }
|
|
}
|
|
};
|
|
if outbound.send(reply).await.is_err() {
|
|
warn!(%camera_stream, "uplink is shutting down; the viewer gets no answer");
|
|
}
|
|
});
|
|
}
|
|
|
|
/// Finds the printer by our own id and runs the command. The error is a
|
|
/// `String` rather than a typed error because its only destination is a
|
|
/// `CommandResult` going back to the cloud as text.
|
|
async fn dispatch(fleet: &SharedFleet, printer_id: &str, command: PrinterCommand) -> Result<(), String> {
|
|
// Fetched before the fleet lock is taken: a download is bounded by
|
|
// minutes of network time, and holding the lock across it would stall
|
|
// telemetry for every other printer. The transfer to the printer that
|
|
// follows has to hold it (it needs `&mut` access), but that is no reason
|
|
// to hold it through this part too. A `start_print` for a printer that
|
|
// does not exist is therefore found out after the download rather than
|
|
// before; the lookup below is cheap, so it is done first for that case.
|
|
let downloaded = match &command {
|
|
PrinterCommand::StartPrint { url, .. } => {
|
|
if !fleet.lock().await.iter().any(|p| p.base().id == printer_id) {
|
|
return Err(format!("no printer with id '{printer_id}' in this gateway's fleet"));
|
|
}
|
|
Some(
|
|
download::fetch_job_file(url)
|
|
.await
|
|
.map_err(|err| format!("download failed (nothing was sent to the printer): {err}"))?,
|
|
)
|
|
}
|
|
_ => None,
|
|
};
|
|
|
|
let mut fleet = fleet.lock().await;
|
|
let printer = fleet
|
|
.iter_mut()
|
|
.find(|p| p.base().id == printer_id)
|
|
.ok_or_else(|| format!("no printer with id '{printer_id}' in this gateway's fleet"))?;
|
|
|
|
info!(%printer_id, ?command, "dispatching cloud command");
|
|
|
|
match command {
|
|
PrinterCommand::SetFanSpeed { percent } => {
|
|
printer.set_fan_speed(percent).await.map_err(|err| err.to_string())
|
|
}
|
|
PrinterCommand::SendGcodeLine { line } => {
|
|
printer.send_gcode_line(line).await.map_err(|err| err.to_string())
|
|
}
|
|
PrinterCommand::ControlPrint { control } => {
|
|
printer.control_print(control).await.map_err(|err| err.to_string())
|
|
}
|
|
PrinterCommand::StartPrint { job_id, file_name, .. } => {
|
|
let bytes = downloaded.ok_or_else(|| "internal error: start_print reached dispatch with no downloaded file".to_string())?;
|
|
let byte_count = bytes.len();
|
|
info!(%printer_id, %job_id, %file_name, byte_count, "job file downloaded, sending to printer");
|
|
// One call covers deriving the filament mapping, the FTPS
|
|
// transfer, the truncation check and the start, and reports them
|
|
// all as one error - so the stage here is "printer", not finer.
|
|
// The printer's own message says which part (mapping, upload of
|
|
// the file, or start) it was.
|
|
printer
|
|
.upload_project_file(file_name, bytes)
|
|
.await
|
|
.map_err(|err| format!("printer upload/start failed (download of {byte_count} bytes succeeded): {err}"))
|
|
}
|
|
}
|
|
}
|