ump-ng 0.2.1

Micro message passing library for threads/tasks communication.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

//! Allow a thread/task, crossing sync/async boundaries in either direction, to
//! deliver an expected piece of data to another thread/task, with
//! notification.
//!
//! These are simple channels used to deliver data from one endpoint to
//! another, where the receiver will block until data has been delivered.

use crate::err::Error;

#[derive(Clone, Default)]
pub enum RCtxState {
  #[default]
  Queued,
  Active
}

/// Object used to reply to requests passed to the server.
pub struct ReplyContext<T, E>(swctx::SetCtx<T, RCtxState, E>);

impl<T, E> ReplyContext<T, E> {
  /// Send a reply back to originating client.
  ///
  /// # Example
  /// ```
  /// use std::thread;
  /// use ump_ng::{channel, MsgType};
  ///
  /// let (server, client) = channel::<(), String, String, ()>();
  /// let server_thread = thread::spawn(move || {
  ///   let MsgType::Request(data, rctx) = server.wait().unwrap() else {
  ///     panic!("Not Request type");
  ///   };
  ///   let reply = format!("Hello, {}!", data);
  ///   rctx.reply(reply).unwrap();
  /// });
  /// let msg = String::from("Client");
  /// let reply = client.req(msg).unwrap();
  /// assert_eq!(reply, "Hello, Client!");
  /// server_thread.join().unwrap();
  /// ```
  ///
  /// # Errors
  /// If the originating caller is no longer waiting for a reply (i.e. was
  /// dropped) [`Error::OriginDisappeared`] is returned.
  ///
  /// # Semantics
  /// This call is safe to make after the server context has been released.
  pub fn reply(self, data: T) -> Result<(), Error<E>> {
    self.0.set(data)?;
    Ok(())
  }

  /// Return an error to originating client.
  /// This will cause the calling client to return an error.  The error passed
  /// in the `err` parameter will be wrapped in a `Error::App(err)`.
  ///
  /// # Example
  ///
  /// ```
  /// use std::thread;
  /// use ump_ng::{channel, Error, MsgType};
  ///
  /// #[derive(Debug, PartialEq)]
  /// enum MyError {
  ///   SomeError(String)
  /// }
  ///
  /// let (server, client) = channel::<String, String, String, MyError>();
  /// let server_thread = thread::spawn(move || {
  ///   let MsgType::Request(_, rctx) = server.wait().unwrap() else {
  ///     panic!("Unexpected message type");
  ///   };
  ///   rctx.fail(MyError::SomeError("failed".to_string())).unwrap();
  /// });
  /// let msg = String::from("Client");
  /// let reply = client.req(msg);
  /// match reply {
  ///   Err(Error::App(MyError::SomeError(s))) => {
  ///     assert_eq!(s, "failed");
  ///   }
  ///   _ => {
  ///     panic!("Unexpected return value");
  ///   }
  /// }
  /// server_thread.join().unwrap();
  /// ```
  ///
  /// # Errors
  /// [`Error::OriginDisappeared`] indicates that the originating caller is no
  /// longer waiting for a reply (i.e. it was dropped).
  ///
  /// # Semantics
  /// This call is safe to make after the server context has been released.
  pub fn fail(self, err: E) -> Result<(), Error<E>> {
    self.0.fail(err)?;
    Ok(())
  }
}

impl<T, E> TryFrom<swctx::SetCtx<T, RCtxState, E>> for ReplyContext<T, E> {
  type Error = Error<E>;

  /// Convert a `SetCtx` into a `ReplyContext`.
  ///
  /// Sets the `SetCtx`'s stat to _Active_.
  fn try_from(
    sctx: swctx::SetCtx<T, RCtxState, E>
  ) -> Result<Self, Self::Error> {
    let _ = sctx.set_state(RCtxState::Active);
    Ok(Self(sctx))
  }
}

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