WIP: dispatch NotifyUser messages to server actor
This commit is contained in:
parent
022e6bcbb0
commit
4a0e31081a
17 changed files with 72 additions and 30 deletions
|
@ -2,7 +2,7 @@
|
||||||
use actix::prelude::*;
|
use actix::prelude::*;
|
||||||
use tracing::Instrument as _;
|
use tracing::Instrument as _;
|
||||||
|
|
||||||
use crate as actor;
|
use crate::{self as actor};
|
||||||
|
|
||||||
impl Handler<actor::messages::LoadConfigFromRepo> for actor::RepoActor {
|
impl Handler<actor::messages::LoadConfigFromRepo> for actor::RepoActor {
|
||||||
type Result = ();
|
type Result = ();
|
||||||
|
@ -26,8 +26,10 @@ impl Handler<actor::messages::LoadConfigFromRepo> for actor::RepoActor {
|
||||||
actor::logger(&log, "send: LoadedConfig");
|
actor::logger(&log, "send: LoadedConfig");
|
||||||
addr.do_send(actor::messages::ReceiveRepoConfig::new(repo_config))
|
addr.do_send(actor::messages::ReceiveRepoConfig::new(repo_config))
|
||||||
}
|
}
|
||||||
Err(err) => tracing::warn!(?err, "Failed to load config"),
|
Err(err) => {
|
||||||
// TODO: (#95) should notify user
|
tracing::warn!(?err, "Failed to load config");
|
||||||
|
// self.notify_user(UserNotification::RepoConfigLoadFailure(err))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
.in_current_span()
|
.in_current_span()
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
//
|
//
|
||||||
use crate as actor;
|
use crate::{self as actor, UserNotification};
|
||||||
use actix::prelude::*;
|
use actix::prelude::*;
|
||||||
use git_next_git as git;
|
use git_next_git as git;
|
||||||
|
|
||||||
|
@ -33,7 +33,7 @@ impl Handler<actor::messages::ReceiveCIStatus> for actor::RepoActor {
|
||||||
}
|
}
|
||||||
git::forge::commit::Status::Fail => {
|
git::forge::commit::Status::Fail => {
|
||||||
tracing::warn!("Checks have failed");
|
tracing::warn!("Checks have failed");
|
||||||
// TODO: (#95) test: repo with next ahead of main and failing CI should notify user
|
self.notify_user(UserNotification::CICheckFailed(next));
|
||||||
actor::delay_send(
|
actor::delay_send(
|
||||||
addr,
|
addr,
|
||||||
sleep_duration,
|
sleep_duration,
|
||||||
|
|
|
@ -2,6 +2,7 @@ mod branch;
|
||||||
pub mod handlers;
|
pub mod handlers;
|
||||||
mod load;
|
mod load;
|
||||||
pub mod messages;
|
pub mod messages;
|
||||||
|
mod user_notification;
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests;
|
mod tests;
|
||||||
|
@ -13,6 +14,8 @@ use actix::prelude::*;
|
||||||
use derive_more::Deref;
|
use derive_more::Deref;
|
||||||
use git_next_config as config;
|
use git_next_config as config;
|
||||||
use git_next_git as git;
|
use git_next_git as git;
|
||||||
|
use messages::NotifyUser;
|
||||||
|
pub use user_notification::UserNotification;
|
||||||
|
|
||||||
use kxio::network::Network;
|
use kxio::network::Network;
|
||||||
use tracing::{info, warn, Instrument};
|
use tracing::{info, warn, Instrument};
|
||||||
|
@ -48,8 +51,10 @@ pub struct RepoActor {
|
||||||
net: Network,
|
net: Network,
|
||||||
forge: Box<dyn git::ForgeLike>,
|
forge: Box<dyn git::ForgeLike>,
|
||||||
log: Option<RepoActorLog>,
|
log: Option<RepoActorLog>,
|
||||||
|
notify_user_recipient: Option<Recipient<NotifyUser>>,
|
||||||
}
|
}
|
||||||
impl RepoActor {
|
impl RepoActor {
|
||||||
|
#[allow(clippy::too_many_arguments)]
|
||||||
pub fn new(
|
pub fn new(
|
||||||
repo_details: git::RepoDetails,
|
repo_details: git::RepoDetails,
|
||||||
forge: Box<dyn git::ForgeLike>,
|
forge: Box<dyn git::ForgeLike>,
|
||||||
|
@ -58,6 +63,7 @@ impl RepoActor {
|
||||||
net: Network,
|
net: Network,
|
||||||
repository_factory: Box<dyn git::repository::RepositoryFactory>,
|
repository_factory: Box<dyn git::repository::RepositoryFactory>,
|
||||||
sleep_duration: std::time::Duration,
|
sleep_duration: std::time::Duration,
|
||||||
|
notify_user_recipient: Option<Recipient<NotifyUser>>,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
let message_token = messages::MessageToken::default();
|
let message_token = messages::MessageToken::default();
|
||||||
Self {
|
Self {
|
||||||
|
@ -76,6 +82,13 @@ impl RepoActor {
|
||||||
net,
|
net,
|
||||||
sleep_duration,
|
sleep_duration,
|
||||||
log: None,
|
log: None,
|
||||||
|
notify_user_recipient,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn notify_user(&mut self, user_notification: UserNotification) {
|
||||||
|
if let Some(recipient) = &self.notify_user_recipient {
|
||||||
|
recipient.do_send(NotifyUser::from(user_notification));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -6,6 +6,8 @@ use git_next_actor_macros::message;
|
||||||
use git_next_config as config;
|
use git_next_config as config;
|
||||||
use git_next_git as git;
|
use git_next_git as git;
|
||||||
|
|
||||||
|
use crate::UserNotification;
|
||||||
|
|
||||||
message!(LoadConfigFromRepo: "Request to load the `git-next.toml` from the git repo.");
|
message!(LoadConfigFromRepo: "Request to load the `git-next.toml` from the git repo.");
|
||||||
message!(CloneRepo: "Request to clone (or open) the git repo.");
|
message!(CloneRepo: "Request to clone (or open) the git repo.");
|
||||||
message!(ReceiveRepoConfig: config::RepoConfig: r#"Notification that the `git-next.toml` file has been loaded from the repo and parsed.
|
message!(ReceiveRepoConfig: config::RepoConfig: r#"Notification that the `git-next.toml` file has been loaded from the repo and parsed.
|
||||||
|
@ -63,3 +65,4 @@ Contains a tuple of the commit that was checked (the tip of the `next` branch) a
|
||||||
message!(AdvanceNext: (git::Commit, Vec<git::Commit>): "Request to advance the `next` branch on to the next commit on the `dev branch."); // next commit and the dev commit history
|
message!(AdvanceNext: (git::Commit, Vec<git::Commit>): "Request to advance the `next` branch on to the next commit on the `dev branch."); // next commit and the dev commit history
|
||||||
message!(AdvanceMain: git::Commit: "Request to advance the `main` branch on to same commit as the `next` branch."); // next commit
|
message!(AdvanceMain: git::Commit: "Request to advance the `main` branch on to same commit as the `next` branch."); // next commit
|
||||||
message!(WebhookNotification: config::webhook::forge_notification::ForgeNotification: "Notification of a webhook message from the forge.");
|
message!(WebhookNotification: config::webhook::forge_notification::ForgeNotification: "Notification of a webhook message from the forge.");
|
||||||
|
message!(NotifyUser: UserNotification: "Request to send the message payload to the notification webhook");
|
||||||
|
|
|
@ -209,6 +209,7 @@ pub fn a_repo_actor(
|
||||||
net,
|
net,
|
||||||
repository_factory,
|
repository_factory,
|
||||||
std::time::Duration::from_nanos(1),
|
std::time::Duration::from_nanos(1),
|
||||||
|
None,
|
||||||
)
|
)
|
||||||
.with_log(actors_log),
|
.with_log(actors_log),
|
||||||
log,
|
log,
|
||||||
|
|
7
crates/repo-actor/src/user_notification.rs
Normal file
7
crates/repo-actor/src/user_notification.rs
Normal file
|
@ -0,0 +1,7 @@
|
||||||
|
use git_next_git::Commit;
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||||
|
pub enum UserNotification {
|
||||||
|
CICheckFailed(Commit),
|
||||||
|
RepoConfigLoadFailure(String),
|
||||||
|
}
|
|
@ -3,9 +3,9 @@ use actix::prelude::*;
|
||||||
use git_next_config::server::ServerConfig;
|
use git_next_config::server::ServerConfig;
|
||||||
use git_next_file_watcher_actor::FileUpdated;
|
use git_next_file_watcher_actor::FileUpdated;
|
||||||
|
|
||||||
use crate::{messages::ReceiveServerConfig, Server};
|
use crate::{messages::ReceiveServerConfig, ServerActor};
|
||||||
|
|
||||||
impl Handler<FileUpdated> for Server {
|
impl Handler<FileUpdated> for ServerActor {
|
||||||
type Result = ();
|
type Result = ();
|
||||||
|
|
||||||
fn handle(&mut self, _msg: FileUpdated, ctx: &mut Self::Context) -> Self::Result {
|
fn handle(&mut self, _msg: FileUpdated, ctx: &mut Self::Context) -> Self::Result {
|
||||||
|
|
|
@ -1,4 +1,5 @@
|
||||||
mod file_updated;
|
mod file_updated;
|
||||||
|
mod notify_user;
|
||||||
mod receive_server_config;
|
mod receive_server_config;
|
||||||
mod receive_valid_server_config;
|
mod receive_valid_server_config;
|
||||||
mod shutdown;
|
mod shutdown;
|
||||||
|
|
|
@ -1,12 +1,14 @@
|
||||||
//
|
//
|
||||||
use actix::prelude::*;
|
use actix::prelude::*;
|
||||||
|
use git_next_repo_actor::messages::NotifyUser;
|
||||||
|
|
||||||
use crate::{messages::NotifyUser, WebhookActor};
|
use crate::ServerActor;
|
||||||
|
|
||||||
impl Handler<NotifyUser> for WebhookActor {
|
impl Handler<NotifyUser> for ServerActor {
|
||||||
type Result = ();
|
type Result = ();
|
||||||
|
|
||||||
fn handle(&mut self, _msg: NotifyUser, _ctx: &mut Self::Context) -> Self::Result {
|
fn handle(&mut self, _msg: NotifyUser, _ctx: &mut Self::Context) -> Self::Result {
|
||||||
// TODO: (#95) should notify user
|
// TODO: (#95) should notify user
|
||||||
|
// send post to notification webhook url
|
||||||
}
|
}
|
||||||
}
|
}
|
|
@ -2,10 +2,10 @@ use actix::prelude::*;
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
messages::{ReceiveServerConfig, ReceiveValidServerConfig, ValidServerConfig},
|
messages::{ReceiveServerConfig, ReceiveValidServerConfig, ValidServerConfig},
|
||||||
Server,
|
ServerActor,
|
||||||
};
|
};
|
||||||
|
|
||||||
impl Handler<ReceiveServerConfig> for Server {
|
impl Handler<ReceiveServerConfig> for ServerActor {
|
||||||
type Result = ();
|
type Result = ();
|
||||||
|
|
||||||
#[allow(clippy::cognitive_complexity)]
|
#[allow(clippy::cognitive_complexity)]
|
||||||
|
|
|
@ -3,13 +3,13 @@ use git_next_webhook_actor::{AddWebhookRecipient, ShutdownWebhook, WebhookActor,
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
messages::{ReceiveValidServerConfig, ValidServerConfig},
|
messages::{ReceiveValidServerConfig, ValidServerConfig},
|
||||||
Server,
|
ServerActor,
|
||||||
};
|
};
|
||||||
|
|
||||||
impl Handler<ReceiveValidServerConfig> for Server {
|
impl Handler<ReceiveValidServerConfig> for ServerActor {
|
||||||
type Result = ();
|
type Result = ();
|
||||||
|
|
||||||
fn handle(&mut self, msg: ReceiveValidServerConfig, _ctx: &mut Self::Context) -> Self::Result {
|
fn handle(&mut self, msg: ReceiveValidServerConfig, ctx: &mut Self::Context) -> Self::Result {
|
||||||
let ValidServerConfig {
|
let ValidServerConfig {
|
||||||
server_config,
|
server_config,
|
||||||
socket_address,
|
socket_address,
|
||||||
|
@ -26,7 +26,13 @@ impl Handler<ReceiveValidServerConfig> for Server {
|
||||||
// Forge Actors
|
// Forge Actors
|
||||||
for (forge_alias, forge_config) in server_config.forges() {
|
for (forge_alias, forge_config) in server_config.forges() {
|
||||||
let repo_actors = self
|
let repo_actors = self
|
||||||
.create_forge_repos(forge_config, forge_alias.clone(), &server_storage, webhook)
|
.create_forge_repos(
|
||||||
|
forge_config,
|
||||||
|
forge_alias.clone(),
|
||||||
|
&server_storage,
|
||||||
|
webhook,
|
||||||
|
ctx.address().recipient(),
|
||||||
|
)
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|a| self.start_actor(a))
|
.map(|a| self.start_actor(a))
|
||||||
.collect::<Vec<_>>();
|
.collect::<Vec<_>>();
|
||||||
|
|
|
@ -2,9 +2,9 @@
|
||||||
use actix::prelude::*;
|
use actix::prelude::*;
|
||||||
use git_next_webhook_actor::ShutdownWebhook;
|
use git_next_webhook_actor::ShutdownWebhook;
|
||||||
|
|
||||||
use crate::{messages::Shutdown, Server};
|
use crate::{messages::Shutdown, ServerActor};
|
||||||
|
|
||||||
impl Handler<Shutdown> for Server {
|
impl Handler<Shutdown> for ServerActor {
|
||||||
type Result = ();
|
type Result = ();
|
||||||
|
|
||||||
fn handle(&mut self, _msg: Shutdown, _ctx: &mut Self::Context) -> Self::Result {
|
fn handle(&mut self, _msg: Shutdown, _ctx: &mut Self::Context) -> Self::Result {
|
||||||
|
|
|
@ -10,6 +10,7 @@ use git_next_config as config;
|
||||||
use git_next_config::server::{InboundWebhook, ServerConfig, ServerStorage};
|
use git_next_config::server::{InboundWebhook, ServerConfig, ServerStorage};
|
||||||
use git_next_config::{ForgeAlias, ForgeConfig, GitDir, RepoAlias, ServerRepoConfig};
|
use git_next_config::{ForgeAlias, ForgeConfig, GitDir, RepoAlias, ServerRepoConfig};
|
||||||
use git_next_git::{Generation, RepoDetails};
|
use git_next_git::{Generation, RepoDetails};
|
||||||
|
use git_next_repo_actor::messages::NotifyUser;
|
||||||
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};
|
||||||
|
@ -43,7 +44,7 @@ type Result<T> = core::result::Result<T, Error>;
|
||||||
|
|
||||||
#[derive(derive_with::With)]
|
#[derive(derive_with::With)]
|
||||||
#[with(message_log)]
|
#[with(message_log)]
|
||||||
pub struct Server {
|
pub struct ServerActor {
|
||||||
generation: Generation,
|
generation: Generation,
|
||||||
webhook: Option<Addr<WebhookActor>>,
|
webhook: Option<Addr<WebhookActor>>,
|
||||||
fs: FileSystem,
|
fs: FileSystem,
|
||||||
|
@ -55,11 +56,11 @@ pub struct Server {
|
||||||
// testing
|
// testing
|
||||||
message_log: Option<Arc<RwLock<Vec<String>>>>,
|
message_log: Option<Arc<RwLock<Vec<String>>>>,
|
||||||
}
|
}
|
||||||
impl Actor for Server {
|
impl Actor for ServerActor {
|
||||||
type Context = Context<Self>;
|
type Context = Context<Self>;
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Server {
|
impl ServerActor {
|
||||||
pub fn new(
|
pub fn new(
|
||||||
fs: FileSystem,
|
fs: FileSystem,
|
||||||
net: Network,
|
net: Network,
|
||||||
|
@ -105,6 +106,7 @@ impl Server {
|
||||||
forge_name: ForgeAlias,
|
forge_name: ForgeAlias,
|
||||||
server_storage: &ServerStorage,
|
server_storage: &ServerStorage,
|
||||||
webhook: &InboundWebhook,
|
webhook: &InboundWebhook,
|
||||||
|
notify_user_recipient: Recipient<NotifyUser>,
|
||||||
) -> Vec<(ForgeAlias, RepoAlias, RepoActor)> {
|
) -> Vec<(ForgeAlias, RepoAlias, RepoActor)> {
|
||||||
let span =
|
let span =
|
||||||
tracing::info_span!("create_forge_repos", name = %forge_name, config = %forge_config);
|
tracing::info_span!("create_forge_repos", name = %forge_name, config = %forge_config);
|
||||||
|
@ -114,7 +116,11 @@ impl Server {
|
||||||
let mut repos = vec![];
|
let mut repos = vec![];
|
||||||
let creator = self.create_actor(forge_name, forge_config.clone(), server_storage, webhook);
|
let creator = self.create_actor(forge_name, forge_config.clone(), server_storage, webhook);
|
||||||
for (repo_alias, server_repo_config) in forge_config.repos() {
|
for (repo_alias, server_repo_config) in forge_config.repos() {
|
||||||
let forge_repo = creator((repo_alias, server_repo_config));
|
let forge_repo = creator((
|
||||||
|
repo_alias,
|
||||||
|
server_repo_config,
|
||||||
|
notify_user_recipient.clone(),
|
||||||
|
));
|
||||||
info!(
|
info!(
|
||||||
alias = %forge_repo.1,
|
alias = %forge_repo.1,
|
||||||
"Created Repo"
|
"Created Repo"
|
||||||
|
@ -130,14 +136,17 @@ impl Server {
|
||||||
forge_config: ForgeConfig,
|
forge_config: ForgeConfig,
|
||||||
server_storage: &ServerStorage,
|
server_storage: &ServerStorage,
|
||||||
webhook: &InboundWebhook,
|
webhook: &InboundWebhook,
|
||||||
) -> impl Fn((RepoAlias, &ServerRepoConfig)) -> (ForgeAlias, RepoAlias, RepoActor) {
|
) -> impl Fn(
|
||||||
|
(RepoAlias, &ServerRepoConfig, Recipient<NotifyUser>),
|
||||||
|
) -> (ForgeAlias, RepoAlias, RepoActor) {
|
||||||
let server_storage = server_storage.clone();
|
let server_storage = server_storage.clone();
|
||||||
let webhook = webhook.clone();
|
let webhook = webhook.clone();
|
||||||
let net = self.net.clone();
|
let net = self.net.clone();
|
||||||
let repository_factory = self.repository_factory.duplicate();
|
let repository_factory = self.repository_factory.duplicate();
|
||||||
let generation = self.generation;
|
let generation = self.generation;
|
||||||
let sleep_duration = self.sleep_duration;
|
let sleep_duration = self.sleep_duration;
|
||||||
move |(repo_alias, server_repo_config)| {
|
// let notify_user_recipient = server_addr.recipient();
|
||||||
|
move |(repo_alias, server_repo_config, notify_user_recipient)| {
|
||||||
let span = tracing::info_span!("create_actor", alias = %repo_alias, config = %server_repo_config);
|
let span = tracing::info_span!("create_actor", alias = %repo_alias, config = %server_repo_config);
|
||||||
let _guard = span.enter();
|
let _guard = span.enter();
|
||||||
info!("Creating Repo");
|
info!("Creating Repo");
|
||||||
|
@ -173,6 +182,7 @@ impl Server {
|
||||||
net.clone(),
|
net.clone(),
|
||||||
repository_factory.duplicate(),
|
repository_factory.duplicate(),
|
||||||
sleep_duration,
|
sleep_duration,
|
||||||
|
Some(notify_user_recipient),
|
||||||
);
|
);
|
||||||
(forge_name.clone(), repo_alias, actor)
|
(forge_name.clone(), repo_alias, actor)
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,5 +1,5 @@
|
||||||
//
|
//
|
||||||
use crate::{tests::given, ReceiveServerConfig, Server};
|
use crate::{tests::given, ReceiveServerConfig, ServerActor};
|
||||||
use actix::prelude::*;
|
use actix::prelude::*;
|
||||||
use git_next_config::server::{Http, InboundWebhook, Notification, ServerConfig, ServerStorage};
|
use git_next_config::server::{Http, InboundWebhook, Notification, ServerConfig, ServerStorage};
|
||||||
use std::{
|
use std::{
|
||||||
|
@ -17,7 +17,7 @@ async fn when_webhook_url_has_trailing_slash_should_not_send() {
|
||||||
let duration = std::time::Duration::from_millis(1);
|
let duration = std::time::Duration::from_millis(1);
|
||||||
|
|
||||||
// sut
|
// sut
|
||||||
let server = Server::new(fs.clone(), net.into(), repo, duration);
|
let server = ServerActor::new(fs.clone(), net.into(), repo, duration);
|
||||||
|
|
||||||
// collaborators
|
// collaborators
|
||||||
let http = Http::new("0.0.0.0".to_string(), 80);
|
let http = Http::new("0.0.0.0".to_string(), 80);
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
//
|
//
|
||||||
use actix::prelude::*;
|
use actix::prelude::*;
|
||||||
use git_next_file_watcher_actor::{FileUpdated, FileWatcher};
|
use git_next_file_watcher_actor::{FileUpdated, FileWatcher};
|
||||||
use git_next_server_actor::Server;
|
use git_next_server_actor::ServerActor;
|
||||||
use kxio::{fs::FileSystem, network::Network};
|
use kxio::{fs::FileSystem, network::Network};
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
use tracing::{error, info, level_filters::LevelFilter};
|
use tracing::{error, info, level_filters::LevelFilter};
|
||||||
|
@ -40,7 +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 = ServerActor::new(fs.clone(), net.clone(), repo, sleep_duration).start();
|
||||||
server.do_send(FileUpdated);
|
server.do_send(FileUpdated);
|
||||||
|
|
||||||
info!("Starting File Watcher...");
|
info!("Starting File Watcher...");
|
||||||
|
|
|
@ -1,2 +1 @@
|
||||||
mod notify_user;
|
|
||||||
mod shutdown_webhook;
|
mod shutdown_webhook;
|
||||||
|
|
|
@ -2,5 +2,3 @@
|
||||||
use git_next_actor_macros::message;
|
use git_next_actor_macros::message;
|
||||||
|
|
||||||
message!(ShutdownWebhook: "Request to shutdown the Webhook actor");
|
message!(ShutdownWebhook: "Request to shutdown the Webhook actor");
|
||||||
|
|
||||||
message!(NotifyUser: String: "Request to send the message payload to the notification webhook");
|
|
||||||
|
|
Loading…
Reference in a new issue