Something went wrong. Try again.
Live STIB-MIVB departures on the Pebble Time 2. It covers the bus, tram, and metro network of Brussels
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340use stibble_server::{ api, bmc, irail, provider::{FeedFailure, FeedStage}, quota, state, storage::Store, villo,};
use std::sync::Arc;use std::time::Duration;
#[tokio::main]async fn main() { tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| "info".into()), ) .init(); let database = std::env::var("STIBBLE_DB").unwrap_or_else(|_| "/data/stibble.sqlite".into()); let extension = std::env::var("SPATIALITE_EXTENSION").expect("SPATIALITE_EXTENSION is required"); let store = Store::open(database, extension) .await .expect("open spatial store"); let key = std::env::var("BMC_PARTNER_KEY").expect("BMC_PARTNER_KEY is required"); let base = std::env::var("BMC_BASE") .unwrap_or_else(|_| "https://api-management-opendata-production.azure-api.net".into()); let quota = Arc::new(quota::Quota::new(500, 12000)); let client = Arc::new(bmc::Client::new(base, key, quota)); let (train_refresh_tx, train_refresh_rx) = tokio::sync::mpsc::channel(64); let app_state = state::AppState::new(store.clone(), train_refresh_tx);
let maintenance_store = store.clone(); tokio::spawn(async move { loop { if let Err(error) = maintenance_store.maintain(chrono::Utc::now()).await { tracing::warn!("storage maintenance failed: {error:#}"); } tokio::time::sleep(Duration::from_secs(3600)).await; } });
let liveboard_store = store.clone(); let liveboard_http = app_state.irail_http.clone(); let liveboard_quota = app_state.irail_quota.clone(); tokio::spawn(async move { if let Err(error) = irail::run_liveboard_worker( liveboard_store, liveboard_http, liveboard_quota, train_refresh_rx, ) .await { tracing::warn!("irail liveboard worker stopped: {error:#}"); } });
let s = store.clone(); let c = client.clone(); tokio::spawn(async move { loop { let result = match c.fetch_static().await { Ok(batch) => s.apply_static(batch).await.map_err(|source| FeedFailure { stage: FeedStage::Commit, source, }), Err(failure) => Err(failure), }; match result { Ok(()) => { tracing::info!("static data refreshed"); tokio::time::sleep(Duration::from_secs(24 * 3600)).await; } Err(failure) => { let message = format!("{:#}", failure.source); if let Err(error) = s .record_feed_failure( "bmc", "static", failure.stage, &message, chrono::Utc::now(), ) .await { tracing::warn!("static failure recording failed: {error:#}"); } tracing::warn!("static refresh failed: {failure}"); tokio::time::sleep(Duration::from_secs(300)).await; } } } });
let s = store.clone(); let c = client.clone(); tokio::spawn(async move { loop { let waiting = match c.fetch_waiting().await { Ok(batch) => s.apply_waiting(batch).await.map_err(|source| FeedFailure { stage: FeedStage::Commit, source, }), Err(failure) => Err(failure), }; match waiting { Ok(()) => tracing::info!("waiting data refreshed"), Err(failure) => { let message = format!("{:#}", failure.source); if let Err(error) = s .record_feed_failure( "bmc", "waiting", failure.stage, &message, chrono::Utc::now(), ) .await { tracing::warn!("waiting failure recording failed: {error:#}"); } tracing::warn!("waiting refresh failed: {failure}"); } } let positions = match c.fetch_positions().await { Ok(batch) => s.apply_positions(batch).await.map_err(|source| FeedFailure { stage: FeedStage::Commit, source, }), Err(failure) => Err(failure), }; match positions { Ok(()) => tracing::info!("positions data refreshed"), Err(failure) => { let message = format!("{:#}", failure.source); if let Err(error) = s .record_feed_failure( "bmc", "positions", failure.stage, &message, chrono::Utc::now(), ) .await { tracing::warn!("positions failure recording failed: {error:#}"); } tracing::warn!("positions refresh failed: {failure}"); } } tokio::time::sleep(Duration::from_secs(30)).await; } });
let s = store.clone(); let c = client.clone(); tokio::spawn(async move { loop { let result = match c.fetch_disruptions().await { Ok(batch) => s.apply_disruptions(batch).await.map_err(|source| FeedFailure { stage: FeedStage::Commit, source, }), Err(failure) => Err(failure), }; match result { Ok(()) => tracing::info!("disruptions data refreshed"), Err(failure) => { let message = format!("{:#}", failure.source); if let Err(error) = s .record_feed_failure( "bmc", "disruptions", failure.stage, &message, chrono::Utc::now(), ) .await { tracing::warn!("disruptions failure recording failed: {error:#}"); } tracing::warn!("disruptions refresh failed: {failure}"); } } tokio::time::sleep(Duration::from_secs(300)).await; } });
let irail_store = store.clone(); let irail_http = app_state.irail_http.clone(); let irail_quota = app_state.irail_quota.clone(); tokio::spawn(async move { loop { let result = match irail::refresh_stations(&irail_http, &irail_quota).await { Ok(batch) => irail_store .apply_train_stations(batch) .await .map_err(|source| FeedFailure { stage: FeedStage::Commit, source, }), Err(failure) => Err(failure), }; match result { Ok(()) => { tracing::info!("irail stations refreshed"); tokio::time::sleep(Duration::from_secs(24 * 3600)).await; } Err(failure) => { let message = format!("{:#}", failure.source); if let Err(error) = irail_store .record_feed_failure( "irail", "stations", failure.stage, &message, chrono::Utc::now(), ) .await { tracing::warn!("irail stations failure recording failed: {error:#}"); } tracing::warn!("irail stations refresh failed: {failure}"); tokio::time::sleep(Duration::from_secs(300)).await; } } } });
let villo_state = app_state.clone(); let villo_store = store.clone(); tokio::spawn(async move { loop { let result = match villo::fetch_discovery( &villo_state.irail_http, &villo_state.villo_quota, ) .await { Ok(batch) => villo_store .apply_villo_discovery(batch) .await .map_err(|source| FeedFailure { stage: FeedStage::Commit, source, }), Err(failure) => Err(failure), }; match result { Ok(()) => { tracing::info!("villo discovery refreshed"); tokio::time::sleep(Duration::from_secs(24 * 3600)).await; } Err(failure) => { let message = format!("{:#}", failure.source); if let Err(error) = villo_store .record_feed_failure( "villo", "discovery", failure.stage, &message, chrono::Utc::now(), ) .await { tracing::warn!("villo discovery failure recording failed: {error:#}"); } tracing::warn!("villo discovery refresh failed: {failure}"); tokio::time::sleep(Duration::from_secs(300)).await; } } } });
let villo_state = app_state.clone(); let villo_store = store.clone(); tokio::spawn(async move { loop { let status_url = match villo_store.villo_status_url().await { Ok(Some(status_url)) => status_url, Ok(None) => { tokio::time::sleep(Duration::from_secs(60)).await; continue; } Err(source) => { tracing::warn!("villo status URL read failed: {source:#}"); tokio::time::sleep(Duration::from_secs(60)).await; continue; } }; let result = match villo::fetch_status( &villo_state.irail_http, &villo_state.villo_quota, &status_url, ) .await { Ok(batch) => villo_store .apply_villo_status(batch) .await .map_err(|source| FeedFailure { stage: FeedStage::Commit, source, }), Err(failure) => Err(failure), }; match result { Ok(()) => tracing::info!("villo status refreshed"), Err(failure) => { let message = format!("{:#}", failure.source); if let Err(error) = villo_store .record_feed_failure( "villo", "status", failure.stage, &message, chrono::Utc::now(), ) .await { tracing::warn!("villo status failure recording failed: {error:#}"); } tracing::warn!("villo status refresh failed: {failure}"); } } tokio::time::sleep(Duration::from_secs(60)).await; } });
let addr = std::env::var("BIND_ADDR").unwrap_or_else(|_| "0.0.0.0:8080".into()); let listener = tokio::net::TcpListener::bind(&addr).await.expect("bind"); tracing::info!("listening on {addr}"); axum::serve(listener, api::router(app_state.clone())) .await .expect("serve");}