ump_ng_autopost/

lib.rs

//! [`AutoPoster`] allos an application to asynchronously post messages to a
//! ump-ng server (at regular intervals).

mod err;

use std::{
  sync::{Arc, Weak},
  thread,
  time::{Duration, Instant}
};

use ump_ng::{Client, WeakClient};

use parking_lot::{Condvar, Mutex};

pub use err::Error;


/// Post callback outcome.
pub enum Proc<P> {
  /// Post and reset timer.
  PostReset(P),

  /// Post and remove (i.e. don't rerun, unless reregistered).
  PostRemove(P),

  /// Remove without posting
  Remove
}

/// Internal record used to keep track of a poster.
struct Record<I, P> {
  ident: I,
  dur: Duration,
  at: Instant,
  cb: Box<dyn FnMut() -> Proc<P> + Send>
}


struct Inner<I, P> {
  shutdown: bool,
  dirty: bool,
  posters: Vec<Record<I, P>>
}

impl<I, P> Default for Inner<I, P> {
  fn default() -> Self {
    Self {
      shutdown: false,
      dirty: false,
      posters: Vec::new()
    }
  }
}

impl<I, P> Inner<I, P> {
  pub fn new() -> Mutex<Self> {
    Mutex::new(Self::default())
  }
}


struct Shared<I, P> {
  inner: Mutex<Inner<I, P>>,
  signal: Condvar
}

impl<I, P> Default for Shared<I, P> {
  fn default() -> Self {
    Self {
      inner: Inner::new(),
      signal: Condvar::new()
    }
  }
}

impl<I, P> Shared<I, P> {
  fn new() -> Arc<Self> {
    Arc::new(Self::default())
  }
}


/// Representation of a running auto-poster.
///
/// Once this is dropped, its worker thread will be terminated.
pub struct AutoPoster<I, P> {
  sh: Weak<Shared<I, P>>,
  jh: Option<thread::JoinHandle<Result<(), Error>>>
}

impl<I, P> AutoPoster<I, P>
where
  I: Send + 'static
{
  /// Create a new `AutoPoster` instance.
  ///
  /// # Errors
  /// If the backgound thread can not be spawned, `std::io::Error` is returned.
  pub fn new<S, R, E>(
    clnt: &Client<P, S, R, E>
  ) -> Result<Self, std::io::Error>
  where
    P: Send + 'static,
    S: Send + 'static,
    R: Send + 'static,
    E: Send + 'static
  {
    let sh = Shared::new();

    // ToDo: autoclone
    let shc = Arc::clone(&sh);
    let wclnt = clnt.weak();
    let jh = thread::Builder::new()
      .name("ump-ng autoposter".into())
      .spawn(move || autoposter_thread(shc, wclnt))?;
    Ok(Self {
      sh: Arc::downgrade(&sh),
      jh: Some(jh)
    })
  }

  /// Return a [`AutoPostCtrl`] that can be used to add/remove new autoposters.
  #[must_use]
  pub fn controller(&self) -> AutoPostCtrl<I, P> {
    AutoPostCtrl {
      sh: Weak::clone(&self.sh)
    }
  }

  pub const fn take_jh(
    &mut self
  ) -> Option<thread::JoinHandle<Result<(), Error>>> {
    self.jh.take()
  }
}

impl<I, P> Drop for AutoPoster<I, P> {
  #[allow(clippy::significant_drop_tightening)]
  fn drop(&mut self) {
    if let Some(sh) = Weak::upgrade(&self.sh) {
      // Tell background thread to terminate
      let mut inner = sh.inner.lock();
      inner.shutdown = true;
      sh.signal.notify_one();
    }

    if let Some(jh) = self.jh.take() {
      let _ = jh.join();
    }
  }
}


/// Auto Post Controller.
///
/// Used to add/remove new poster callbacks.
pub struct AutoPostCtrl<I, P> {
  sh: Weak<Shared<I, P>>
}

impl<I, P> Clone for AutoPostCtrl<I, P> {
  fn clone(&self) -> Self {
    Self {
      sh: Weak::clone(&self.sh)
    }
  }
}

impl<I, P> AutoPostCtrl<I, P>
where
  I: PartialEq
{
  /// Add/update an auto poster.
  ///
  /// `ident` is a unique identifier used to reference a specific auto-poster.
  /// `dur` is the amount of time to wait before triggering the post.
  /// `cb` is a callback closure that will be called to request the message to
  /// be posted.
  ///
  /// The role of the closure is to return a [`Proc`]
  ///
  /// # Errors
  /// [`Error::NotRunning`] means the background thread has been terminated.
  #[allow(clippy::significant_drop_tightening)]
  pub fn set(
    &self,
    ident: I,
    dur: Duration,
    cb: impl FnMut() -> Proc<P> + Send + 'static
  ) -> Result<(), Error> {
    let Some(sh) = Weak::upgrade(&self.sh) else {
      return Err(Error::NotRunning);
    };

    let mut inner = sh.inner.lock();

    // If ident already exists, then update existing record.  Otherwise just
    // att it to the list.
    if let Some(idx) = inner.posters.iter().position(|x| x.ident == ident) {
      // SAFETY: We just found the entry
      let n = unsafe { inner.posters.get_unchecked_mut(idx) };
      n.dur = dur;
      n.cb = Box::new(cb);
      n.at = Instant::now() + dur;
    } else {
      inner.posters.push(Record {
        ident,
        dur,
        at: Instant::now() + dur,
        cb: Box::new(cb)
      });
    }
    inner.dirty = true;
    sh.signal.notify_one();

    Ok(())
  }

  /// Check if an identifier is currently in use.
  ///
  /// # Errors
  /// [`Error::NotRunning`] means the background thread has been terminated.
  pub fn have(&self, ident: &I) -> Result<bool, Error> {
    Weak::upgrade(&self.sh).map_or(Err(Error::NotRunning), |sh| {
      let inner = sh.inner.lock();
      Ok(inner.posters.iter().any(|x| x.ident == *ident))
    })
  }

  /// Remove an auto-poster instance.
  ///
  /// # Errors
  /// [`Error::NotRunning`] means the background thread has been terminated.
  #[allow(clippy::significant_drop_tightening)]
  pub fn remove(&self, ident: &I) -> Result<(), Error> {
    let Some(sh) = Weak::upgrade(&self.sh) else {
      return Err(Error::NotRunning);
    };

    let mut inner = sh.inner.lock();
    if let Some(idx) = inner.posters.iter().position(|x| x.ident == *ident) {
      inner.posters.swap_remove(idx);
      inner.dirty = true;
      sh.signal.notify_one();
    }

    Ok(())
  }
}


#[allow(clippy::significant_drop_tightening)]
#[allow(clippy::needless_pass_by_value)]
fn autoposter_thread<I, P, S, R, E>(
  sh: Arc<Shared<I, P>>,
  wclnt: WeakClient<P, S, R, E>
) -> Result<(), Error>
where
  P: Send + 'static,
  S: Send + 'static,
  R: Send + 'static,
  E: Send + 'static
{
  let mut inner = sh.inner.lock();

  loop {
    if inner.shutdown {
      //println!("shutdown");
      break Ok(());
    }

    // If the list of posters is "dirty" then sort it
    if inner.dirty {
      // Sort list so that nearest-in-time are later in the list
      inner.posters.sort_unstable_by(|a, b| b.at.cmp(&a.at));
      inner.dirty = false;
    }

    // Take next one off the list
    if let Some(mut n) = inner.posters.pop() {
      if Instant::now() >= n.at {
        match (n.cb)() {
          Proc::PostReset(msg) => {
            let Some(clnt) = wclnt.upgrade() else {
              //println!("client disappeared");
              break Err(Error::ClientDisappeared);
            };
            if clnt.post(msg).is_err() {
              //println!("server disappeared");
              break Err(Error::ServerDisappeared);
            }

            // Reset timer and push the node back on to the list
            n.at = Instant::now() + n.dur;
            inner.posters.push(n);

            // Mark list as dirty
            inner.dirty = true;
          }
          Proc::PostRemove(msg) => {
            let Some(clnt) = wclnt.upgrade() else {
              //println!("client disappeared");
              break Err(Error::ClientDisappeared);
            };
            if clnt.post(msg).is_err() {
              //println!("server disappeared");
              break Err(Error::ServerDisappeared);
            }
          }
          Proc::Remove => {
            // Do nothing
          }
        }
      } else {
        let at = n.at;
        inner.posters.push(n);
        //println!("Wait until: {at:?}");
        sh.signal.wait_until(&mut inner, at);
      }
    } else {
      // Wait forever
      //println!("Wait forever");
      sh.signal.wait(&mut inner);
    }
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :