Compare commits

..

2 Commits

Author SHA1 Message Date
0cd2d364aa break out network_change_listener 2024-07-25 21:01:40 +02:00
f34c4609a5 exit listener 2024-07-25 20:46:30 +02:00
4 changed files with 126 additions and 55 deletions

43
src/exit_listener.rs Normal file
View File

@ -0,0 +1,43 @@
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use tokio::{
self, signal,
sync::{futures::Notified, Notify},
};
pub struct ExitListener {
should_exit: Arc<AtomicBool>,
notify: Arc<Notify>,
}
impl ExitListener {
pub fn new() -> Self {
let this = Self {
should_exit: Arc::new(AtomicBool::new(false)),
notify: Arc::new(Notify::new()),
};
let should_exit = this.should_exit.clone();
let notify = this.notify.clone();
tokio::spawn(async move {
signal::ctrl_c()
.await
.expect("Failed to install CTRL+C signal handler");
should_exit.store(true, Ordering::SeqCst);
notify.notify_one();
});
this
}
pub fn notified(&self) -> Notified<'_> {
self.notify.notified()
}
pub fn should_exit(&self) -> bool {
self.should_exit.load(Ordering::SeqCst)
}
}

View File

@ -2,16 +2,20 @@
mod cloudflare;
mod config;
mod exit_listener;
mod logging;
mod message_handler;
mod network_change_listener;
mod public_ip;
mod tests;
pub mod utils;
pub use cloudflare::CloudflareClient;
pub use config::{get_config_path, read_config, Config};
pub use exit_listener::ExitListener;
pub use logging::init_logger;
pub use message_handler::MessageHandler;
pub use network_change_listener::NetworkChangeListener;
pub use public_ip::get_current_public_ipv4;
pub const PROGRAM_NAME: &'static str = "dynip-cloudflare";

View File

@ -1,36 +1,14 @@
// SPDX: BSD-2-Clause
use futures::future::{self, Either};
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use futures::stream::StreamExt;
use log::{error, info};
use netlink_sys::{AsyncSocket, SocketAddr};
use rtnetlink::new_connection;
use scopeguard::defer;
use tokio::time;
use dynip_cloudflare::{
utils::{self, duration_to_string},
CloudflareClient, MessageHandler, MAX_ERORS_IN_ROW_DEFAULT,
utils, CloudflareClient, ExitListener, MessageHandler, NetworkChangeListener,
MAX_ERORS_IN_ROW_DEFAULT,
};
use scopeguard::defer;
use tokio::{signal, sync::Notify, time};
const RTNLGRP_LINK: u32 = 1;
const RTNLGRP_IPV4_IFADDR: u32 = 5;
const fn nl_mgrp(group: u32) -> u32 {
if group > 31 {
panic!("use netlink_sys::Socket::add_membership() for this group");
}
if group == 0 {
0
} else {
1 << (group - 1)
}
}
#[tokio::main]
async fn main() {
@ -38,19 +16,7 @@ async fn main() {
defer! {
log::logger().flush();
}
let should_exit = Arc::new(AtomicBool::new(false));
let notify = Arc::new(Notify::new());
let should_exit_clone = should_exit.clone();
let notify_clone = notify.clone();
tokio::spawn(async move {
signal::ctrl_c()
.await
.expect("Failed to install CTRL+C signal handler");
should_exit_clone.store(true, Ordering::SeqCst);
notify_clone.notify_one();
});
let exit_listener = ExitListener::new();
let config = if let Some(aux) = utils::get_config().await {
aux
@ -67,18 +33,16 @@ async fn main() {
}
};
let (mut conn, mut _handle, mut messages) = new_connection().unwrap();
let groups = nl_mgrp(RTNLGRP_LINK) | nl_mgrp(RTNLGRP_IPV4_IFADDR);
let addr = SocketAddr::new(0, groups);
if let Err(e) = conn.socket_mut().socket_mut().bind(&addr) {
error!("Failed to bind to socket: {:?}", &e);
let mut network_change_listener = match NetworkChangeListener::new() {
Some(aux) => {
info!("Listening for IPv4 address changes and interface connect/disconnect events...");
aux
}
None => {
error!("Failed to initialize networkchangelistener");
return;
}
tokio::spawn(conn);
info!("Listening for IPv4 address changes and interface connect/disconnect events...");
};
let mut message_handler = MessageHandler::new(
&mut cloudflare,
@ -91,22 +55,22 @@ async fn main() {
interval
});
while !should_exit.load(Ordering::SeqCst) {
while !exit_listener.should_exit() {
let tick_future = match interval.as_mut() {
Some(interval) => Either::Left(interval.tick()),
None => Either::Right(future::pending::<tokio::time::Instant>()),
};
tokio::select! {
_ = notify.notified() => break,
_ = exit_listener.notified() => break,
_ = tick_future => {
if let Some(duration) = config.max_duration.as_ref() {
let duration_string = duration_to_string(duration);
let duration_string = utils::duration_to_string(duration);
let log_string = format!("{} has passed since last check, checking...", duration_string.trim());
message_handler.log_and_check(Some(&log_string), Option::<&&str>::None).await;
}
}
message = messages.next() => {
message = network_change_listener.next_message() => {
if let Some((message, _)) = message {
if let Some(interval) = interval.as_mut() {
interval.reset();

View File

@ -0,0 +1,60 @@
use futures::{
channel::mpsc::UnboundedReceiver,
stream::{Next, StreamExt},
};
use log::error;
use netlink_packet_core::NetlinkMessage;
use netlink_packet_route::RouteNetlinkMessage;
use netlink_sys::{AsyncSocket, SocketAddr};
use rtnetlink::new_connection;
use tokio::task::JoinHandle;
const RTNLGRP_LINK: u32 = 1;
const RTNLGRP_IPV4_IFADDR: u32 = 5;
const fn nl_mgrp(group: u32) -> u32 {
if group > 31 {
panic!("use netlink_sys::Socket::add_membership() for this group");
}
if group == 0 {
0
} else {
1 << (group - 1)
}
}
type Messages = UnboundedReceiver<(NetlinkMessage<RouteNetlinkMessage>, SocketAddr)>;
pub struct NetworkChangeListener {
messages: Messages,
thread_handle: JoinHandle<()>,
}
impl NetworkChangeListener {
pub fn new() -> Option<Self> {
let (mut conn, mut _handle, messages) = new_connection().ok()?;
let groups = nl_mgrp(RTNLGRP_LINK) | nl_mgrp(RTNLGRP_IPV4_IFADDR);
let addr = SocketAddr::new(0, groups);
if let Err(e) = conn.socket_mut().socket_mut().bind(&addr) {
error!("Failed to bind to socket: {:?}", &e);
return None;
}
let thread_handle = tokio::spawn(conn);
Some(Self {
messages,
thread_handle,
})
}
pub fn next_message(&mut self) -> Next<'_, Messages> {
self.messages.next()
}
}
impl Drop for NetworkChangeListener {
fn drop(&mut self) {
self.thread_handle.abort();
}
}