2020-05-28 21:32:33 +00:00
|
|
|
use syndicate::{config, spaces, packets, ConnId};
|
2020-05-17 11:30:18 +00:00
|
|
|
use syndicate::peer::{Peer, ResultC2S};
|
2019-09-19 09:54:01 +00:00
|
|
|
|
2019-10-16 15:54:58 +00:00
|
|
|
use std::sync::{Mutex, Arc};
|
2020-05-17 11:30:18 +00:00
|
|
|
use futures::{SinkExt, StreamExt};
|
|
|
|
|
2020-05-18 12:33:36 +00:00
|
|
|
use tracing::{Level, error, info, trace};
|
|
|
|
use tracing_futures::Instrument;
|
|
|
|
|
2019-10-16 15:54:58 +00:00
|
|
|
use tokio::net::TcpListener;
|
2020-05-17 11:30:18 +00:00
|
|
|
use tokio::net::TcpStream;
|
|
|
|
use tokio_util::codec::Framed;
|
2018-12-09 13:28:01 +00:00
|
|
|
|
2020-05-17 11:30:18 +00:00
|
|
|
use tungstenite::Message;
|
|
|
|
|
2020-05-18 11:16:14 +00:00
|
|
|
use structopt::StructOpt; // for from_args in main
|
2020-05-17 11:30:18 +00:00
|
|
|
|
|
|
|
type UnitAsyncResult = Result<(), std::io::Error>;
|
|
|
|
|
2020-05-28 21:32:33 +00:00
|
|
|
fn message_error<E: std::fmt::Display>(e: E) -> packets::Error {
|
|
|
|
packets::Error::Message(e.to_string())
|
2020-05-17 11:30:18 +00:00
|
|
|
}
|
|
|
|
|
2020-05-28 21:32:33 +00:00
|
|
|
fn encode_message(p: packets::S2C) ->
|
|
|
|
Result<Message, packets::Error>
|
2020-05-17 11:30:18 +00:00
|
|
|
{
|
2020-05-24 19:18:48 +00:00
|
|
|
let mut bs = Vec::with_capacity(128);
|
2020-05-28 21:32:33 +00:00
|
|
|
preserves::ser::to_writer(&mut bs, &p)?;
|
2020-05-24 19:18:48 +00:00
|
|
|
Ok(Message::Binary(bs))
|
2020-05-17 11:30:18 +00:00
|
|
|
}
|
|
|
|
|
2020-05-28 21:32:33 +00:00
|
|
|
fn message_encoder(p: packets::S2C) -> futures::future::Ready<Result<Message, packets::Error>>
|
2020-05-17 11:30:18 +00:00
|
|
|
{
|
2020-05-28 21:32:33 +00:00
|
|
|
futures::future::ready(encode_message(p))
|
2020-05-17 11:30:18 +00:00
|
|
|
}
|
|
|
|
|
2020-05-28 21:32:33 +00:00
|
|
|
fn message_decoder(r: Result<Message, tungstenite::Error>) -> ResultC2S
|
2020-05-18 08:44:09 +00:00
|
|
|
{
|
2020-05-28 21:32:33 +00:00
|
|
|
loop {
|
|
|
|
return match r {
|
|
|
|
Ok(ref m) => match m {
|
|
|
|
Message::Text(_) =>
|
|
|
|
Err(preserves::error::syntax_error("Text websocket frames are not accepted")),
|
|
|
|
Message::Binary(ref bs) =>
|
|
|
|
Ok(preserves::de::from_bytes(bs)?),
|
|
|
|
Message::Ping(_) =>
|
|
|
|
continue, // pings are handled by tungstenite before we see them
|
|
|
|
Message::Pong(_) =>
|
|
|
|
continue, // unsolicited pongs are to be ignored
|
|
|
|
Message::Close(_) =>
|
|
|
|
Err(preserves::error::eof()),
|
2020-05-17 11:30:18 +00:00
|
|
|
}
|
2020-05-28 21:32:33 +00:00
|
|
|
Err(tungstenite::Error::Io(e)) =>
|
|
|
|
Err(e.into()),
|
|
|
|
Err(e) =>
|
|
|
|
Err(message_error(e)),
|
2020-05-17 11:30:18 +00:00
|
|
|
}
|
2020-05-28 21:32:33 +00:00
|
|
|
}
|
2020-05-17 11:30:18 +00:00
|
|
|
}
|
2019-09-19 09:54:01 +00:00
|
|
|
|
2020-05-17 11:30:18 +00:00
|
|
|
async fn run_connection(connid: ConnId,
|
|
|
|
mut stream: TcpStream,
|
2020-05-18 11:16:14 +00:00
|
|
|
spaces: Arc<Mutex<spaces::Spaces>>,
|
2020-05-18 14:46:41 +00:00
|
|
|
addr: std::net::SocketAddr,
|
2020-05-18 11:16:14 +00:00
|
|
|
config: config::ServerConfigRef) ->
|
2020-05-17 11:30:18 +00:00
|
|
|
UnitAsyncResult
|
|
|
|
{
|
|
|
|
let mut buf = [0; 1]; // peek at the first byte to see what kind of connection to expect
|
|
|
|
match stream.peek(&mut buf).await? {
|
|
|
|
1 => match buf[0] {
|
|
|
|
71 /* ASCII 'G' for "GET" */ => {
|
2020-05-18 14:46:41 +00:00
|
|
|
info!(protocol = display("websocket"), peer = debug(addr));
|
2020-05-17 11:30:18 +00:00
|
|
|
let s = tokio_tungstenite::accept_async(stream).await
|
|
|
|
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
|
|
|
|
let (o, i) = s.split();
|
2020-05-28 21:32:33 +00:00
|
|
|
let i = i.map(message_decoder);
|
|
|
|
let o = o.sink_map_err(message_error).with(message_encoder);
|
2020-05-17 11:30:18 +00:00
|
|
|
let mut p = Peer::new(connid, i, o);
|
2020-05-18 11:16:14 +00:00
|
|
|
p.run(spaces, &config).await?
|
2020-05-17 11:30:18 +00:00
|
|
|
},
|
|
|
|
_ => {
|
2020-05-18 14:46:41 +00:00
|
|
|
info!(protocol = display("raw"), peer = debug(addr));
|
2020-05-28 21:32:33 +00:00
|
|
|
let (o, i) = Framed::new(stream, packets::Codec::new()).split();
|
2020-05-17 11:30:18 +00:00
|
|
|
let mut p = Peer::new(connid, i, o);
|
2020-05-18 11:16:14 +00:00
|
|
|
p.run(spaces, &config).await?
|
2020-05-17 11:30:18 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
0 => return Err(std::io::Error::new(std::io::ErrorKind::UnexpectedEof,
|
|
|
|
"closed before starting")),
|
|
|
|
_ => unreachable!()
|
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
2020-05-18 12:33:36 +00:00
|
|
|
static NEXT_ID: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
|
|
|
|
|
2020-05-18 11:16:14 +00:00
|
|
|
async fn run_listener(spaces: Arc<Mutex<spaces::Spaces>>, port: u16, config: config::ServerConfigRef) ->
|
|
|
|
UnitAsyncResult
|
|
|
|
{
|
2019-10-15 15:58:52 +00:00
|
|
|
let mut listener = TcpListener::bind(format!("0.0.0.0:{}", port)).await?;
|
2019-09-19 09:54:01 +00:00
|
|
|
loop {
|
|
|
|
let (stream, addr) = listener.accept().await?;
|
2020-05-18 12:33:36 +00:00
|
|
|
let id = NEXT_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
2019-10-16 13:16:50 +00:00
|
|
|
let spaces = Arc::clone(&spaces);
|
2020-05-18 11:16:14 +00:00
|
|
|
let config = Arc::clone(&config);
|
|
|
|
if let Some(n) = config.recv_buffer_size { stream.set_recv_buffer_size(n)?; }
|
|
|
|
if let Some(n) = config.send_buffer_size { stream.set_send_buffer_size(n)?; }
|
2019-09-19 09:54:01 +00:00
|
|
|
tokio::spawn(async move {
|
2020-05-18 14:46:41 +00:00
|
|
|
match run_connection(id, stream, spaces, addr, config).await {
|
2020-05-18 12:33:36 +00:00
|
|
|
Ok(()) => info!("closed"),
|
|
|
|
Err(e) => info!(error = display(e), "closed"),
|
2019-09-19 09:54:01 +00:00
|
|
|
}
|
2020-05-18 12:33:36 +00:00
|
|
|
}.instrument(tracing::info_span!("connection", id)));
|
2019-09-17 19:28:31 +00:00
|
|
|
}
|
2018-12-09 13:28:01 +00:00
|
|
|
}
|
2020-05-17 11:30:18 +00:00
|
|
|
|
2020-05-18 08:44:57 +00:00
|
|
|
async fn periodic_tasks(spaces: Arc<Mutex<spaces::Spaces>>) -> UnitAsyncResult {
|
2020-05-18 12:33:36 +00:00
|
|
|
let interval = core::time::Duration::from_secs(10);
|
2020-05-18 09:36:44 +00:00
|
|
|
let mut delay = tokio::time::interval(interval);
|
2020-05-18 08:44:57 +00:00
|
|
|
loop {
|
|
|
|
delay.next().await.unwrap();
|
|
|
|
{
|
|
|
|
let mut spaces = spaces.lock().unwrap();
|
|
|
|
spaces.cleanup();
|
2020-05-18 12:33:36 +00:00
|
|
|
spaces.dump_stats(interval);
|
2020-05-18 08:44:57 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-05-17 11:30:18 +00:00
|
|
|
#[tokio::main]
|
|
|
|
async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
2020-05-18 12:33:36 +00:00
|
|
|
let filter = tracing_subscriber::filter::EnvFilter::from_default_env()
|
|
|
|
.add_directive(tracing_subscriber::filter::LevelFilter::INFO.into());
|
|
|
|
let subscriber = tracing_subscriber::FmtSubscriber::builder()
|
|
|
|
.with_ansi(true)
|
|
|
|
.with_max_level(Level::TRACE)
|
|
|
|
.with_env_filter(filter)
|
|
|
|
.finish();
|
|
|
|
tracing::subscriber::set_global_default(subscriber)
|
|
|
|
.expect("Could not set tracing global subscriber");
|
|
|
|
|
2020-05-18 11:16:14 +00:00
|
|
|
let config = Arc::new(config::ServerConfig::from_args());
|
2020-05-17 11:30:18 +00:00
|
|
|
|
|
|
|
let spaces = Arc::new(Mutex::new(spaces::Spaces::new()));
|
2020-05-18 08:44:35 +00:00
|
|
|
let mut daemons = Vec::new();
|
2020-05-17 11:30:18 +00:00
|
|
|
|
2020-05-18 08:44:57 +00:00
|
|
|
{
|
|
|
|
let spaces = Arc::clone(&spaces);
|
|
|
|
tokio::spawn(async move {
|
|
|
|
periodic_tasks(spaces).await
|
|
|
|
});
|
|
|
|
}
|
|
|
|
|
2020-05-18 12:33:36 +00:00
|
|
|
trace!("startup");
|
|
|
|
|
2020-06-01 08:02:39 +00:00
|
|
|
{
|
|
|
|
const BRIGHT_GREEN: &str = "\x1b[92m";
|
|
|
|
const GREEN: &str = "\x1b[32m";
|
|
|
|
const NORMAL: &str = "\x1b[0m";
|
|
|
|
const BRIGHT_YELLOW: &str = "\x1b[93m";
|
|
|
|
info!(r" {} __{}__{}__ {}", GREEN, BRIGHT_GREEN, GREEN, NORMAL);
|
|
|
|
info!(r" {} /{}_/ \_{}\ {}", GREEN, BRIGHT_GREEN, GREEN, NORMAL);
|
|
|
|
info!(r" {} / \__/ \ {} __ __", BRIGHT_GREEN, NORMAL);
|
|
|
|
info!(r" {}/{}\__/ \__/{}\{} _______ ______ ____/ /__________ / /____", GREEN, BRIGHT_GREEN, GREEN, NORMAL);
|
|
|
|
info!(r" {}\{}/ \__/ \{}/{} / ___/ / / / __ \/ __ / / ___/ __ \/ __/ _ \", GREEN, BRIGHT_GREEN, GREEN, NORMAL);
|
|
|
|
info!(r" {} \__/ \__/ {} _\_ \/ /_/ / / / / /_/ / / /__/ /_/ / /_/ __/", BRIGHT_GREEN, NORMAL);
|
|
|
|
info!(r" {} \_{}\__/{}_/ {} /____/\__, /_/ /_/\____/_/\___/\__/_/\__/\___/", GREEN, BRIGHT_GREEN, GREEN, NORMAL);
|
|
|
|
info!(r" /____/");
|
|
|
|
info!(r"");
|
|
|
|
info!(r" {}version {}{}", BRIGHT_YELLOW, env!("CARGO_PKG_VERSION"), NORMAL);
|
|
|
|
info!(r"");
|
|
|
|
info!(r" documentation & reference material: https://syndicate-lang.org/");
|
|
|
|
info!(r" source code & bug tracker: https://git.leastfixedpoint.com/");
|
|
|
|
info!(r"");
|
|
|
|
}
|
|
|
|
|
2020-05-18 11:16:14 +00:00
|
|
|
for port in config.ports.clone() {
|
2020-05-17 11:30:18 +00:00
|
|
|
let spaces = Arc::clone(&spaces);
|
2020-05-18 11:16:14 +00:00
|
|
|
let config = Arc::clone(&config);
|
2020-05-18 08:44:35 +00:00
|
|
|
daemons.push(tokio::spawn(async move {
|
2020-05-18 12:33:36 +00:00
|
|
|
info!(port, "listening");
|
2020-05-18 11:16:14 +00:00
|
|
|
match run_listener(spaces, port, config).await {
|
2020-05-17 11:30:18 +00:00
|
|
|
Ok(()) => (),
|
|
|
|
Err(e) => {
|
2020-05-18 12:33:36 +00:00
|
|
|
error!("{}", e);
|
2020-05-17 11:30:18 +00:00
|
|
|
std::process::exit(2)
|
|
|
|
}
|
|
|
|
}
|
2020-05-18 12:33:36 +00:00
|
|
|
}.instrument(tracing::info_span!("listener", port))));
|
2020-05-17 11:30:18 +00:00
|
|
|
}
|
|
|
|
|
2020-05-18 08:44:35 +00:00
|
|
|
futures::future::join_all(daemons).await;
|
2020-05-17 11:30:18 +00:00
|
|
|
Ok(())
|
|
|
|
}
|