[package]
name = "ump-ng-server"
version = "0.4.0"
edition = "2021"
license = "0BSD"
categories = [ "concurrency", "asynchronous" ]
keywords = [ "channel", "threads", "sync", "message-passing" ]
repository = "https://repos.qrnch.tech/pub/ump-ng-server"
description = "Server message dispatch loop for ump-ng."
rust-version = "1.56"

[features]
default = ["tokio"]
tokio = ["dep:tokio", "dep:async-trait"]
tracing = ["dep:tracing"]
watchdog = ["dep:parking_lot"]

[dependencies]
async-trait = { version = "0.1.82", optional = true }
parking_lot = { version = "0.12.3", optional = true }


tokio = { version = "1.40.0", features = ["rt"], optional = true }
tracing = { version = "0.1.40", optional = true }
ump-ng = { version = "0.2.0" }

[dev-dependencies]
tokio-test = { version = "0.4.4" }

[package.metadata.docs.rs]
all-features = true
rustdoc-args = ["--cfg", "docsrs", "--generate-link-to-definition"]

# This is a configuration file for the bacon tool
#
# Bacon repository: https://github.com/Canop/bacon
# Complete help on configuration: https://dystroy.org/bacon/config/
# You can also check bacon's own bacon.toml file
#  as an example: https://github.com/Canop/bacon/blob/main/bacon.toml

# For information about clippy lints, see:
# https://github.com/rust-lang/rust-clippy/blob/master/README.md

#default_job = "check"
default_job = "clippy-all-pedantic"

[jobs.check]
command = ["cargo", "check", "--color", "always"]
need_stdout = false

[jobs.check-all]
command = ["cargo", "check", "--all-targets", "--color", "always"]
need_stdout = false

# Run clippy on the default target
[jobs.clippy]
command = [
    "cargo", "clippy",
    "--color", "always",
]
need_stdout = false

# Run clippy on all targets
# To disable some lints, you may change the job this way:
#    [jobs.clippy-all]
#    command = [
#        "cargo", "clippy",
#        "--all-targets",
#        "--color", "always",
#    	 "--",
#    	 "-A", "clippy::bool_to_int_with_if",
#    	 "-A", "clippy::collapsible_if",
#    	 "-A", "clippy::derive_partial_eq_without_eq",
#    ]
# need_stdout = false
[jobs.clippy-all]
command = [
    "cargo", "clippy",
    "--all-targets",
    "--color", "always",
]
need_stdout = false

[jobs.clippy-pedantic]
command = [
    "cargo", "clippy",
    "--color", "always",
    "--",
    "-Wclippy::all",
    "-Wclippy::pedantic",
    "-Wclippy::nursery",
    "-Wclippy::cargo"
]
need_stdout = false

[jobs.clippy-all-pedantic]
command = [
    "cargo", "clippy",
    "--all-targets",
    "--color", "always",
    "--",
    "-Wclippy::all",
    "-Wclippy::pedantic",
    "-Wclippy::nursery",
    "-Wclippy::cargo"
]
need_stdout = false

# This job lets you run
# - all tests: bacon test
# - a specific test: bacon test -- config::test_default_files
# - the tests of a package: bacon test -- -- -p config
[jobs.test]
command = [
    "cargo", "test", "--color", "always",
    "--", "--color", "always", # see https://github.com/Canop/bacon/issues/124
]
need_stdout = true

[jobs.doc]
command = ["cargo", "doc", "--color", "always", "--no-deps"]
need_stdout = false

# If the doc compiles, then it opens in your browser and bacon switches
# to the previous job
[jobs.doc-open]
command = ["cargo", "doc", "--color", "always", "--no-deps", "--open"]
need_stdout = false
on_success = "back" # so that we don't open the browser at each change

# You can run your application and have the result displayed in bacon,
# *if* it makes sense for this crate.
# Don't forget the `--color always` part or the errors won't be
# properly parsed.
# If your program never stops (eg a server), you may set `background`
# to false to have the cargo run output immediately displayed instead
# of waiting for program's end.
[jobs.run]
command = [
    "cargo", "run",
    "--color", "always",
    # put launch parameters for your program behind a `--` separator
]
need_stdout = true
allow_warnings = true
background = true

# This parameterized job runs the example of your choice, as soon
# as the code compiles.
# Call it as
#    bacon ex -- my-example
[jobs.ex]
command = ["cargo", "run", "--color", "always", "--example"]
need_stdout = true
allow_warnings = true

# You may define here keybindings that would be specific to
# a project, for example a shortcut to launch a specific job.
# Shortcuts to internal functions (scrolling, toggling, etc.)
# should go in your personal global prefs.toml file instead.
[keybindings]
# alt-m = "job:my-job"
c = "job:clippy-all" # comment this to have 'c' run clippy on only the default target

}

/// Launch a task that will process incoming messages from an ump-ng server
/// end-point.
///
/// See top module's documentation for an overview of the [dispatch
/// loop](crate#dispatch-loop).
///
/// # Errors
/// An application-defined error `E` is returned if the dispatch loop is
/// terminated by a handler returning `ControlFlow::Break(E)`.
#[allow(clippy::type_complexity)]
pub fn spawn<P, S, R, E, RV, F>(
  hbldr: impl FnOnce(&Client<P, S, R, E>) -> Result<F, E>
) -> Result<(Client<P, S, R, E>, JoinHandle<Option<RV>>), E>
where
  P: 'static + Send,
  S: 'static + Send,

    let ret = loop {
      match server.async_wait().await {
        Ok(msg) => {
          #[cfg(feature = "watchdog")]
          wdog.begin_process();

          let res = match msg {
            MsgType::Post(m) => handler.post(m).await,
            MsgType::Request(m, rctx) => handler.req(m, rctx).await
          };

          #[cfg(feature = "watchdog")]
          wdog.end_process();

          match res {
            ControlFlow::Continue(()) => {}
            ControlFlow::Break(rv) => break Some(rv)
          }
        }
        Err(_) => break None
      }
    };

}

/// Launch a thread that will process incoming messages from an ump-ng server
/// end-point.
///
/// See top module's documentation for an overview of the [dispatch
/// loop](crate#dispatch-loop).
///
/// # Errors
/// An application-defined error `E` is returned if the dispatch loop is
/// terminated by a handler returning `ControlFlow::Break(E)`.
#[allow(clippy::type_complexity)]
pub fn spawn<P, S, R, E, RV, F>(
  hbldr: impl FnOnce(&Client<P, S, R, E>) -> Result<F, E>
) -> Result<(Client<P, S, R, E>, JoinHandle<Option<RV>>), E>
where
  P: 'static + Send,
  S: 'static + Send,

    let ret = loop {
      match server.wait() {
        Ok(msg) => {
          #[cfg(feature = "watchdog")]
          wdog.begin_process();

          let res = match msg {
            MsgType::Post(m) => handler.post(m),
            MsgType::Request(m, rctx) => handler.req(m, rctx)
          };

          #[cfg(feature = "watchdog")]
          wdog.end_process();

          match res {
            ControlFlow::Continue(()) => {}
            ControlFlow::Break(rv) => break Some(rv)
          }
        }
        Err(_) => break None
      }
    };

}

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

pub fn run() -> WatchDog {
  let inner = Inner { state: State::Idle };

  let shared = Shared {
    inner: Mutex::new(inner),
    signal: Condvar::new()
  };

  let sh = Arc::new(shared);
  let shared = Arc::clone(&sh);
  let jh = std::thread::spawn(move || monitor_thread(&shared));

  WatchDog { sh, jh }
}

pub struct WatchDog {
  sh: Arc<Shared>,
  jh: JoinHandle<()>
}

impl WatchDog {
  pub(crate) fn begin_process(&self) {
    let mut g = self.sh.inner.lock();
    g.state = State::Processing {
      start_time: Instant::now()
    };
    self.sh.signal.notify_one();
    drop(g);
  }

  pub(crate) fn end_process(&self) {
    let mut g = self.sh.inner.lock();
    g.state = State::Idle;
    self.sh.signal.notify_one();
    drop(g);
  }

  pub(crate) fn kill(self) -> std::thread::Result<()> {
    let mut g = self.sh.inner.lock();
    g.state = State::Term;
    self.sh.signal.notify_one();
    drop(g);
    self.jh.join()
  }
}


#[allow(clippy::significant_drop_tightening)]
fn monitor_thread(sh: &Arc<Shared>) {
  let mut g = sh.inner.lock();
  loop {
    match g.state {
      State::Idle => {
        // Wait to be notified about a state change
        sh.signal.wait(&mut g);
      }

#[allow(dead_code)]
pub enum Request {
  Add(u32, u32),
  GetSignalState,
  Croak
}

#[derive(Debug, PartialEq, Eq)]
#[allow(dead_code)]
pub enum Reply {
  Sum(u64),
  SignalState(bool),
  OkIWillCroak
}

      }
      Put::Sleep(dur) => {
        self.sleep_count += 1;
        std::thread::sleep(dur);
        ControlFlow::Continue(())
      }
      Put::Croak => ControlFlow::Break(42),
      Put::CroakSleepCount => {
        ControlFlow::Break(u32::try_from(self.sleep_count).unwrap())
      }
    }
  }

  fn req(
    &mut self,
    msg: Request,
    rctx: ump_ng_server::ReplyContext<Reply, ()>
  ) -> ControlFlow<u32, ()> {
    match msg {
      Request::Add(a, b) => {
        rctx.reply(Reply::Sum(u64::from(a + b))).unwrap();
        ControlFlow::Continue(())
      }
      Request::GetSignalState => {
        rctx.reply(Reply::SignalState(self.did_signal)).unwrap();
        ControlFlow::Continue(())
      }
      Request::Croak => {

      }
      Put::Sleep(dur) => {
        self.sleep_count += 1;
        std::thread::sleep(dur);
        ControlFlow::Continue(())
      }
      Put::Croak => ControlFlow::Break(42),
      Put::CroakSleepCount => {
        ControlFlow::Break(u32::try_from(self.sleep_count).unwrap())
      }
    }
  }

  async fn req(
    &mut self,
    msg: Request,
    rctx: ump_ng_server::ReplyContext<Reply, ()>
  ) -> ControlFlow<u32, ()> {
    match msg {
      Request::Add(a, b) => {
        rctx.reply(Reply::Sum(u64::from(a + b))).unwrap();
        ControlFlow::Continue(())
      }
      Request::GetSignalState => {
        rctx.reply(Reply::SignalState(self.did_signal)).unwrap();
        ControlFlow::Continue(())
      }
      Request::Croak => {

# Change Log

⚠️  indicates a breaking change.

## [Unreleased]

[Details](/vdiff?from=ump-ng-server-0.4.0&to=trunk)

### Added

### Changed

### Removed

---

## [0.4.0] - 2024-09-10

[Details](/vdiff?from=ump-ng-server-0.3.0&to=ump-ng-server-0.4.0)

### Changed

- ⚠️ Updated to `ump-ng` to `0.2.0`.

---

## [0.3.0] - 2024-02-20

[Details](/vdiff?from=ump-ng-server-0.2.1&to=ump-ng-server-0.3.0)

Yanked 0.2.1 and rereleased as 0.3.0 due to breaking changes.

---

# ump-ng-server

The _ump-ng-server_ crate is a server message dispatch abstraction for
[ump-ng](https://repos.qrnch.tech/pub/ump-ng).


## Feature labels in documentation

The crate's documentation uses automatically generated feature labels, which
currently requires nightly featuers.  To build the documentation locally use:

```
$ RUSTFLAGS="--cfg docsrs" RUSTDOCFLAGS="--cfg docsrs" cargo +nightly doc --all-features

```


## Change log

The details of changes can always be found in the timeline, but for a
high-level view of changes between released versions there's a manually