2019-09-09 19:06:24 +02:00
|
|
|
use flodgatt::{
|
2020-01-08 15:51:25 +01:00
|
|
|
config, err,
|
2019-07-08 13:31:42 +02:00
|
|
|
parse_client_request::{sse, user, ws},
|
2019-10-03 06:34:41 +02:00
|
|
|
redis_to_client_stream::{self, ClientAgent},
|
2019-07-06 02:08:50 +02:00
|
|
|
};
|
2020-01-09 23:54:57 +01:00
|
|
|
use std::{collections::HashMap, env, fs, net, os::unix::fs::PermissionsExt};
|
|
|
|
use tokio::net::UnixListener;
|
2020-01-07 23:27:46 +01:00
|
|
|
use warp::{path, ws::Ws2, Filter};
|
2019-02-19 20:29:32 +01:00
|
|
|
|
2019-04-15 20:22:44 +02:00
|
|
|
fn main() {
|
2019-10-04 00:02:23 +02:00
|
|
|
dotenv::from_filename(
|
|
|
|
match env::var("ENV").ok().as_ref().map(String::as_str) {
|
|
|
|
Some("production") => ".env.production",
|
|
|
|
Some("development") | None => ".env",
|
|
|
|
Some(_) => err::die_with_msg("Unknown ENV variable specified.\n Valid options are: `production` or `development`."),
|
|
|
|
}).ok();
|
2019-10-09 02:35:26 +02:00
|
|
|
let env_vars_map: HashMap<_, _> = dotenv::vars().collect();
|
2020-01-06 03:58:18 +01:00
|
|
|
let env_vars = config::EnvVar::new(env_vars_map);
|
2019-10-04 00:02:23 +02:00
|
|
|
pretty_env_logger::init();
|
2020-01-06 03:58:18 +01:00
|
|
|
|
2020-01-10 23:56:19 +01:00
|
|
|
log::warn!(
|
2020-01-06 03:58:18 +01:00
|
|
|
"Flodgatt recognized the following environmental variables:{}",
|
|
|
|
env_vars.clone()
|
|
|
|
);
|
2019-10-04 00:02:23 +02:00
|
|
|
let redis_cfg = config::RedisConfig::from_env(env_vars.clone());
|
2019-10-09 02:35:26 +02:00
|
|
|
let cfg = config::DeploymentConfig::from_env(env_vars.clone());
|
|
|
|
|
2019-10-04 00:02:23 +02:00
|
|
|
let postgres_cfg = config::PostgresConfig::from_env(env_vars.clone());
|
2020-03-19 01:37:10 +01:00
|
|
|
let pg_pool = user::PgPool::new(postgres_cfg);
|
2019-10-04 00:02:23 +02:00
|
|
|
|
2020-03-19 01:37:10 +01:00
|
|
|
let client_agent_sse = ClientAgent::blank(redis_cfg, pg_pool.clone());
|
2019-07-08 13:31:42 +02:00
|
|
|
let client_agent_ws = client_agent_sse.clone_with_shared_receiver();
|
2019-05-01 00:41:13 +02:00
|
|
|
|
2020-01-10 23:56:19 +01:00
|
|
|
log::warn!("Streaming server initialized and ready to accept connections");
|
2019-07-10 04:20:11 +02:00
|
|
|
|
2019-07-06 02:08:50 +02:00
|
|
|
// Server Sent Events
|
2019-10-09 02:35:26 +02:00
|
|
|
let sse_update_interval = *cfg.ws_interval;
|
2020-01-10 21:45:16 +01:00
|
|
|
let sse_routes = sse::extract_user_or_reject(pg_pool.clone())
|
2019-09-05 03:48:29 +02:00
|
|
|
.and(warp::sse())
|
|
|
|
.map(
|
2020-03-19 01:37:10 +01:00
|
|
|
move |user: user::Subscription, sse_connection_to_client: warp::sse::Sse| {
|
2020-01-10 23:56:19 +01:00
|
|
|
log::info!("Incoming SSE request");
|
2019-09-05 03:48:29 +02:00
|
|
|
// Create a new ClientAgent
|
|
|
|
let mut client_agent = client_agent_sse.clone_with_shared_receiver();
|
2019-09-09 19:06:24 +02:00
|
|
|
// Assign ClientAgent to generate stream of updates for the user/timeline pair
|
|
|
|
client_agent.init_for_user(user);
|
2019-09-05 03:48:29 +02:00
|
|
|
// send the updates through the SSE connection
|
2019-10-04 00:02:23 +02:00
|
|
|
redis_to_client_stream::send_updates_to_sse(
|
|
|
|
client_agent,
|
|
|
|
sse_connection_to_client,
|
|
|
|
sse_update_interval,
|
|
|
|
)
|
2019-09-05 03:48:29 +02:00
|
|
|
},
|
|
|
|
)
|
|
|
|
.with(warp::reply::with::header("Connection", "keep-alive"))
|
2019-10-02 06:03:18 +02:00
|
|
|
.recover(err::handle_errors);
|
2019-04-28 23:28:57 +02:00
|
|
|
|
2019-07-06 02:08:50 +02:00
|
|
|
// WebSocket
|
2019-10-09 02:35:26 +02:00
|
|
|
let ws_update_interval = *cfg.ws_interval;
|
2020-03-19 01:37:10 +01:00
|
|
|
let websocket_routes = ws::extract_user_and_token_or_reject(pg_pool.clone())
|
2019-09-09 19:06:24 +02:00
|
|
|
.and(warp::ws::ws2())
|
2020-03-19 01:37:10 +01:00
|
|
|
.map(
|
|
|
|
move |user: user::Subscription, token: Option<String>, ws: Ws2| {
|
|
|
|
log::info!("Incoming websocket request");
|
|
|
|
// Create a new ClientAgent
|
|
|
|
let mut client_agent = client_agent_ws.clone_with_shared_receiver();
|
|
|
|
// Assign that agent to generate a stream of updates for the user/timeline pair
|
|
|
|
client_agent.init_for_user(user);
|
|
|
|
// send the updates through the WS connection (along with the User's access_token
|
|
|
|
// which is sent for security)
|
2019-09-11 06:13:45 +02:00
|
|
|
|
2020-03-19 01:37:10 +01:00
|
|
|
(
|
|
|
|
ws.on_upgrade(move |socket| {
|
|
|
|
redis_to_client_stream::send_updates_to_ws(
|
|
|
|
socket,
|
|
|
|
client_agent,
|
|
|
|
ws_update_interval,
|
|
|
|
)
|
|
|
|
}),
|
|
|
|
token.unwrap_or_else(String::new),
|
|
|
|
)
|
|
|
|
},
|
|
|
|
)
|
2019-07-04 16:57:15 +02:00
|
|
|
.map(|(reply, token)| warp::reply::with_header(reply, "sec-websocket-protocol", token));
|
2019-05-09 05:02:01 +02:00
|
|
|
|
2019-10-04 00:02:23 +02:00
|
|
|
let cors = warp::cors()
|
|
|
|
.allow_any_origin()
|
|
|
|
.allow_methods(cfg.cors.allowed_methods)
|
|
|
|
.allow_headers(cfg.cors.allowed_headers);
|
|
|
|
|
2020-01-07 23:27:46 +01:00
|
|
|
let health = warp::path!("api" / "v1" / "streaming" / "health").map(|| "OK");
|
|
|
|
|
2020-01-08 15:51:25 +01:00
|
|
|
if let Some(socket) = &*cfg.unix_socket {
|
2020-01-10 23:56:19 +01:00
|
|
|
log::warn!("Using Unix socket {}", socket);
|
2020-01-09 23:54:57 +01:00
|
|
|
fs::remove_file(socket).unwrap_or_default();
|
2020-01-08 15:51:25 +01:00
|
|
|
let incoming = UnixListener::bind(socket).unwrap().incoming();
|
2020-01-09 23:54:57 +01:00
|
|
|
|
|
|
|
fs::set_permissions(socket, PermissionsExt::from_mode(0o666)).unwrap();
|
|
|
|
|
2020-01-08 15:51:25 +01:00
|
|
|
warp::serve(health.or(websocket_routes.or(sse_routes).with(cors))).run_incoming(incoming);
|
2019-10-04 00:02:23 +02:00
|
|
|
} else {
|
2020-01-09 23:54:57 +01:00
|
|
|
let server_addr = net::SocketAddr::new(*cfg.address, cfg.port.0);
|
2020-01-07 23:27:46 +01:00
|
|
|
warp::serve(health.or(websocket_routes.or(sse_routes).with(cors))).run(server_addr);
|
2019-10-04 00:02:23 +02:00
|
|
|
}
|
2019-02-11 09:45:14 +01:00
|
|
|
}
|