Jędrzej Stuczyński
818405982d
* Initial commit of the new dashboard code. * Periodically grabbing topology json * Pulling file saving out into its own module * Ignoring downloaded topology file * Moved everything public into a public folder * Refreshing the mixmining report * Mounting static files from /public * Including mixminiming report grabber * Leaving the route in place to pick up later. It's not used right now. * Removing json download from git * Ignoring topology download * Moving recurrent jobs in to a jobs module * Adding websocket dependencies * Starting to get client/server websocket functionality running. * Fixing unused imports * Separating client and server functionality a bit more cleanly * WIP to sketch out the ws client and server a bit more * Initial metrics broadcaster * Import fixup * Spawning rocket in tokio task * Removed outdated comment * removed the js file Co-authored-by: Dave <futurechimp@users.noreply.github.com>
67 lines
2.0 KiB
Rust
67 lines
2.0 KiB
Rust
use futures_util::{SinkExt, StreamExt};
|
|
use log::*;
|
|
use std::{io::Error as IoError, net::SocketAddr};
|
|
use tokio::net::{TcpListener, TcpStream};
|
|
use tokio::sync::broadcast;
|
|
use tokio_tungstenite::accept_async;
|
|
use tokio_tungstenite::tungstenite::Message;
|
|
|
|
pub struct DashboardWebsocketServer {
|
|
sender: broadcast::Sender<Message>,
|
|
addr: String,
|
|
}
|
|
|
|
impl DashboardWebsocketServer {
|
|
pub fn new(port: u16, sender: broadcast::Sender<Message>) -> DashboardWebsocketServer {
|
|
let addr = format!("[::]:{}", port);
|
|
DashboardWebsocketServer { addr, sender }
|
|
}
|
|
|
|
pub async fn start(self) -> Result<(), IoError> {
|
|
let try_socket = TcpListener::bind(&self.addr).await;
|
|
|
|
let mut listener = try_socket?;
|
|
info!("starting to listen on {}", self.addr);
|
|
while let Ok((stream, addr)) = listener.accept().await {
|
|
tokio::spawn(Self::handle_connection(
|
|
stream,
|
|
addr,
|
|
self.sender.subscribe(),
|
|
));
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn handle_connection(
|
|
stream: TcpStream,
|
|
addr: SocketAddr,
|
|
mut receiver: broadcast::Receiver<Message>,
|
|
) {
|
|
let mut ws_stream = match accept_async(stream).await {
|
|
Ok(ws_stream) => ws_stream,
|
|
Err(err) => {
|
|
warn!(
|
|
"error while performing the websocket handshake with {} - {:?}",
|
|
addr, err
|
|
);
|
|
return;
|
|
}
|
|
};
|
|
|
|
info!("client connected from {}", addr);
|
|
while let Some(message) = receiver.next().await {
|
|
let message = message.expect("the websocket broadcaster is dead!");
|
|
if let Err(err) = ws_stream.send(message).await {
|
|
warn!(
|
|
"failed to send subscribed message back to client ({}) - {}",
|
|
addr, err
|
|
);
|
|
return;
|
|
} else {
|
|
info!("sent message to {}", addr)
|
|
}
|
|
}
|
|
}
|
|
}
|