Compare commits
No commits in common. "57938c154116f7a7375d77c237d91401b95a057c" and "9c20e780d02dea6ede51ace2ebcba033d5fbd8e3" have entirely different histories.
57938c1541
...
9c20e780d0
6 changed files with 34 additions and 162 deletions
|
@ -20,7 +20,6 @@ kxio = { workspace = true }
|
||||||
|
|
||||||
# boilerplate
|
# boilerplate
|
||||||
derive_more = { workspace = true }
|
derive_more = { workspace = true }
|
||||||
derive-with = { workspace = true }
|
|
||||||
|
|
||||||
# Actors
|
# Actors
|
||||||
actix = { workspace = true }
|
actix = { workspace = true }
|
||||||
|
@ -28,8 +27,6 @@ actix = { workspace = true }
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
# Testing
|
# Testing
|
||||||
# assert2 = { workspace = true }
|
# assert2 = { workspace = true }
|
||||||
test-log = { workspace = true }
|
|
||||||
tokio = { workspace = true }
|
|
||||||
|
|
||||||
[lints.clippy]
|
[lints.clippy]
|
||||||
nursery = { level = "warn", priority = -1 }
|
nursery = { level = "warn", priority = -1 }
|
||||||
|
|
|
@ -1,9 +1,5 @@
|
||||||
//
|
//
|
||||||
#[cfg(test)]
|
|
||||||
mod tests;
|
|
||||||
|
|
||||||
use actix::prelude::*;
|
use actix::prelude::*;
|
||||||
use derive_more::Constructor;
|
|
||||||
use git_next_actor_macros::message;
|
use git_next_actor_macros::message;
|
||||||
use git_next_config as config;
|
use git_next_config as config;
|
||||||
use git_next_config::server::{ServerConfig, ServerStorage, Webhook};
|
use git_next_config::server::{ServerConfig, ServerStorage, Webhook};
|
||||||
|
@ -13,31 +9,18 @@ use git_next_git::{Generation, RepoDetails};
|
||||||
use git_next_repo_actor::{messages::CloneRepo, RepoActor};
|
use git_next_repo_actor::{messages::CloneRepo, RepoActor};
|
||||||
use git_next_webhook_actor as webhook;
|
use git_next_webhook_actor as webhook;
|
||||||
use kxio::{fs::FileSystem, network::Network};
|
use kxio::{fs::FileSystem, network::Network};
|
||||||
use std::{
|
use std::path::PathBuf;
|
||||||
net::SocketAddr,
|
use tracing::{error, info, warn};
|
||||||
path::PathBuf,
|
|
||||||
sync::{Arc, RwLock},
|
|
||||||
};
|
|
||||||
use tracing::{error, info};
|
|
||||||
use webhook::{AddWebhookRecipient, ShutdownWebhook, WebhookActor, WebhookRouter};
|
use webhook::{AddWebhookRecipient, ShutdownWebhook, WebhookActor, WebhookRouter};
|
||||||
|
|
||||||
pub use git_next_git::repository::{factory::real as repository_factory, RepositoryFactory};
|
pub use git_next_git::repository::{factory::real as repository_factory, RepositoryFactory};
|
||||||
|
|
||||||
message!(ReceiveServerConfig: ServerConfig: "Notification of newly loaded server configuration.
|
message!(ReceiveServerConfig: ServerConfig: "Notification of newly loaded server configuration.
|
||||||
|
|
||||||
|
|
||||||
This message will prompt the `git-next` server to stop and restart all repo-actors.
|
This message will prompt the `git-next` server to stop and restart all repo-actors.
|
||||||
|
|
||||||
Contains the new server configuration.");
|
Contains the new server configuration.");
|
||||||
|
|
||||||
#[derive(Clone, Debug, PartialEq, Eq, Constructor)]
|
|
||||||
pub struct ValidServerConfig {
|
|
||||||
server_config: ServerConfig,
|
|
||||||
socket_address: SocketAddr,
|
|
||||||
server_storage: ServerStorage,
|
|
||||||
}
|
|
||||||
message!(ReceiveValidServerConfig: ValidServerConfig: "Notification of validated server configuration.");
|
|
||||||
|
|
||||||
#[derive(Debug, derive_more::Display, derive_more::From)]
|
#[derive(Debug, derive_more::Display, derive_more::From)]
|
||||||
pub enum Error {
|
pub enum Error {
|
||||||
#[display("Failed to create data directories")]
|
#[display("Failed to create data directories")]
|
||||||
|
@ -54,8 +37,6 @@ pub enum Error {
|
||||||
}
|
}
|
||||||
type Result<T> = core::result::Result<T, Error>;
|
type Result<T> = core::result::Result<T, Error>;
|
||||||
|
|
||||||
#[derive(derive_with::With)]
|
|
||||||
#[with(message_log)]
|
|
||||||
pub struct Server {
|
pub struct Server {
|
||||||
generation: Generation,
|
generation: Generation,
|
||||||
webhook: Option<Addr<WebhookActor>>,
|
webhook: Option<Addr<WebhookActor>>,
|
||||||
|
@ -63,9 +44,6 @@ pub struct Server {
|
||||||
net: Network,
|
net: Network,
|
||||||
repository_factory: Box<dyn RepositoryFactory>,
|
repository_factory: Box<dyn RepositoryFactory>,
|
||||||
sleep_duration: std::time::Duration,
|
sleep_duration: std::time::Duration,
|
||||||
|
|
||||||
// testing
|
|
||||||
message_log: Option<Arc<RwLock<Vec<String>>>>,
|
|
||||||
}
|
}
|
||||||
impl Actor for Server {
|
impl Actor for Server {
|
||||||
type Context = Context<Self>;
|
type Context = Context<Self>;
|
||||||
|
@ -81,59 +59,53 @@ impl Handler<FileUpdated> for Server {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
};
|
};
|
||||||
self.do_send(ReceiveServerConfig::new(server_config), ctx);
|
ctx.notify(ReceiveServerConfig::new(server_config));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
impl Handler<ReceiveServerConfig> for Server {
|
impl Handler<ReceiveServerConfig> for Server {
|
||||||
type Result = ();
|
type Result = ();
|
||||||
|
|
||||||
fn handle(&mut self, msg: ReceiveServerConfig, ctx: &mut Self::Context) -> Self::Result {
|
#[allow(clippy::cognitive_complexity)] // TODO: (#75) reduce complexity
|
||||||
tracing::info!("recieved server config");
|
fn handle(&mut self, msg: ReceiveServerConfig, _ctx: &mut Self::Context) -> Self::Result {
|
||||||
let Ok(socket_addr) = msg.http() else {
|
let Ok(socket_addr) = msg.http() else {
|
||||||
error!("Unable to parse http.addr");
|
warn!("Unable to parse http.addr");
|
||||||
return;
|
return;
|
||||||
};
|
};
|
||||||
|
|
||||||
let Some(server_storage) = self.server_storage(&msg) else {
|
|
||||||
error!("Server storage not available");
|
|
||||||
return;
|
|
||||||
};
|
|
||||||
|
|
||||||
if msg.webhook().base_url().ends_with('/') {
|
|
||||||
error!("webhook.url must not end with a '/'");
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
|
|
||||||
self.do_send(
|
|
||||||
ReceiveValidServerConfig::new(ValidServerConfig::new(
|
|
||||||
msg.0,
|
|
||||||
socket_addr,
|
|
||||||
server_storage,
|
|
||||||
)),
|
|
||||||
ctx,
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
impl Handler<ReceiveValidServerConfig> for Server {
|
|
||||||
type Result = ();
|
|
||||||
|
|
||||||
fn handle(&mut self, msg: ReceiveValidServerConfig, _ctx: &mut Self::Context) -> Self::Result {
|
|
||||||
let ValidServerConfig {
|
|
||||||
server_config,
|
|
||||||
socket_address,
|
|
||||||
server_storage,
|
|
||||||
} = msg.0;
|
|
||||||
if let Some(webhook) = self.webhook.take() {
|
if let Some(webhook) = self.webhook.take() {
|
||||||
webhook.do_send(ShutdownWebhook);
|
webhook.do_send(ShutdownWebhook);
|
||||||
}
|
}
|
||||||
self.generation.inc();
|
self.generation.inc();
|
||||||
|
let server_config = msg;
|
||||||
|
// Server Storage
|
||||||
|
let dir = server_config.storage().path();
|
||||||
|
if !dir.exists() {
|
||||||
|
info!(?dir, "server storage doesn't exist - creating it");
|
||||||
|
if let Err(err) = self.fs.dir_create(dir) {
|
||||||
|
error!(?err, ?dir, "Failed to create server storage");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let Ok(canon) = dir.canonicalize() else {
|
||||||
|
error!(?dir, "Failed to confirm server storage");
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
info!(dir = ?canon, "server storage");
|
||||||
|
|
||||||
|
// Forge directories in Server Storage
|
||||||
|
let server_storage = server_config.storage();
|
||||||
|
if let Err(err) = self.create_forge_data_directories(&server_config, dir) {
|
||||||
|
error!(?err, "Failure creating forge storage");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
// Webhook Server
|
// Webhook Server
|
||||||
info!("Starting Webhook Server...");
|
info!("Starting Webhook Server...");
|
||||||
let webhook_router = WebhookRouter::default().start();
|
let webhook_router = WebhookRouter::default().start();
|
||||||
let webhook = server_config.webhook();
|
let webhook = server_config.webhook();
|
||||||
|
|
||||||
// Forge Actors
|
// Forge Actors
|
||||||
for (forge_alias, forge_config) in server_config.forges() {
|
for (forge_alias, forge_config) in server_config.forges() {
|
||||||
self.create_forge_repos(forge_config, forge_alias.clone(), &server_storage, webhook)
|
self.create_forge_repos(forge_config, forge_alias.clone(), server_storage, webhook)
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|a| self.start_actor(a))
|
.map(|a| self.start_actor(a))
|
||||||
.map(|(repo_alias, addr)| {
|
.map(|(repo_alias, addr)| {
|
||||||
|
@ -141,7 +113,8 @@ impl Handler<ReceiveValidServerConfig> for Server {
|
||||||
})
|
})
|
||||||
.for_each(|msg| webhook_router.do_send(msg));
|
.for_each(|msg| webhook_router.do_send(msg));
|
||||||
}
|
}
|
||||||
let webhook = WebhookActor::new(socket_address, webhook_router.recipient()).start();
|
|
||||||
|
let webhook = WebhookActor::new(socket_addr, webhook_router.recipient()).start();
|
||||||
self.webhook.replace(webhook);
|
self.webhook.replace(webhook);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -160,7 +133,6 @@ impl Server {
|
||||||
net,
|
net,
|
||||||
repository_factory: repo,
|
repository_factory: repo,
|
||||||
sleep_duration,
|
sleep_duration,
|
||||||
message_log: None,
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
fn create_forge_data_directories(
|
fn create_forge_data_directories(
|
||||||
|
@ -275,43 +247,4 @@ impl Server {
|
||||||
info!("Started");
|
info!("Started");
|
||||||
(repo_alias, addr)
|
(repo_alias, addr)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn server_storage(&self, server_config: &ReceiveServerConfig) -> Option<ServerStorage> {
|
|
||||||
let server_storage = server_config.storage().clone();
|
|
||||||
let dir = server_storage.path();
|
|
||||||
if !dir.exists() {
|
|
||||||
if let Err(err) = self.fs.dir_create(dir) {
|
|
||||||
error!(?err, ?dir, "Failed to create server storage");
|
|
||||||
return None;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
let Ok(canon) = dir.canonicalize() else {
|
|
||||||
error!(?dir, "Failed to confirm server storage");
|
|
||||||
return None;
|
|
||||||
};
|
|
||||||
if let Err(err) = self.create_forge_data_directories(server_config, &canon) {
|
|
||||||
error!(?err, "Failure creating forge storage");
|
|
||||||
return None;
|
|
||||||
}
|
|
||||||
Some(server_storage)
|
|
||||||
}
|
|
||||||
|
|
||||||
fn do_send<M>(&mut self, msg: M, _ctx: &mut <Self as actix::Actor>::Context)
|
|
||||||
where
|
|
||||||
M: actix::Message + Send + 'static + std::fmt::Debug,
|
|
||||||
Self: actix::Handler<M>,
|
|
||||||
<M as actix::Message>::Result: Send,
|
|
||||||
{
|
|
||||||
tracing::info!(?msg, "send");
|
|
||||||
if let Some(message_log) = &self.message_log {
|
|
||||||
let log_message = format!("send: {:?}", msg);
|
|
||||||
tracing::debug!(log_message);
|
|
||||||
if let Ok(mut log) = message_log.write() {
|
|
||||||
log.push(log_message);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
#[cfg(not(test))]
|
|
||||||
_ctx.address().do_send(msg);
|
|
||||||
tracing::info!("sent");
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,7 +0,0 @@
|
||||||
pub fn a_filesystem() -> kxio::fs::FileSystem {
|
|
||||||
kxio::fs::temp().unwrap_or_else(|e| panic!("{}", e))
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn a_network() -> kxio::network::MockNetwork {
|
|
||||||
kxio::network::MockNetwork::new()
|
|
||||||
}
|
|
|
@ -1,3 +0,0 @@
|
||||||
mod receive_server_config;
|
|
||||||
|
|
||||||
mod given;
|
|
|
@ -1,49 +0,0 @@
|
||||||
//
|
|
||||||
use crate::{tests::given, ReceiveServerConfig, Server};
|
|
||||||
use actix::prelude::*;
|
|
||||||
use git_next_config::server::{Http, ServerConfig, ServerStorage, Webhook};
|
|
||||||
use std::{
|
|
||||||
collections::BTreeMap,
|
|
||||||
sync::{Arc, RwLock},
|
|
||||||
};
|
|
||||||
|
|
||||||
#[test_log::test(actix::test)]
|
|
||||||
async fn when_webhook_url_has_trailing_slash_should_not_send() {
|
|
||||||
//given
|
|
||||||
// parameters
|
|
||||||
let fs = given::a_filesystem();
|
|
||||||
let net = given::a_network();
|
|
||||||
let repo = git_next_git::repository::factory::mock();
|
|
||||||
let duration = std::time::Duration::from_millis(1);
|
|
||||||
|
|
||||||
// sut
|
|
||||||
let server = Server::new(fs.clone(), net.into(), repo, duration);
|
|
||||||
|
|
||||||
// collaborators
|
|
||||||
let http = Http::new("0.0.0.0".to_string(), 80);
|
|
||||||
let webhook = Webhook::new("http://localhost/".to_string()); // With trailing slash
|
|
||||||
let server_storage = ServerStorage::new((fs.base()).to_path_buf());
|
|
||||||
let repos = BTreeMap::default();
|
|
||||||
|
|
||||||
// debugging
|
|
||||||
let message_log: Arc<RwLock<Vec<String>>> = Arc::new(RwLock::new(vec![]));
|
|
||||||
let server = server.with_message_log(Some(message_log.clone()));
|
|
||||||
|
|
||||||
//when
|
|
||||||
server
|
|
||||||
.start()
|
|
||||||
.do_send(ReceiveServerConfig::new(ServerConfig::new(
|
|
||||||
http,
|
|
||||||
webhook,
|
|
||||||
server_storage,
|
|
||||||
repos,
|
|
||||||
)));
|
|
||||||
tokio::time::sleep(std::time::Duration::from_millis(1)).await;
|
|
||||||
|
|
||||||
//then
|
|
||||||
// INFO: assert that ReceiveValidServerConfig is NOT sent
|
|
||||||
tracing::debug!(?message_log, "");
|
|
||||||
assert!(message_log.read().iter().any(|log| !log
|
|
||||||
.iter()
|
|
||||||
.any(|line| line != "send: ReceiveValidServerConfig")));
|
|
||||||
}
|
|
|
@ -40,6 +40,7 @@ pub fn start(
|
||||||
|
|
||||||
info!("Starting Server...");
|
info!("Starting Server...");
|
||||||
let execution = async move {
|
let execution = async move {
|
||||||
|
//-
|
||||||
let server = Server::new(fs.clone(), net.clone(), repo, sleep_duration).start();
|
let server = Server::new(fs.clone(), net.clone(), repo, sleep_duration).start();
|
||||||
server.do_send(FileUpdated);
|
server.do_send(FileUpdated);
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue