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
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

use std::{
  ops::ControlFlow,
  thread::{self, JoinHandle}
};

use super::{channel, Client, MsgType, ReplyContext, Server};

pub trait Handler<P, S, R, E, RV> {
  /// Optional initialization callback.
  ///
  /// This is called on the dispatcher thread before the main message
  /// processing loop is entered.
  #[allow(unused_variables)]
  fn init(&mut self, weak_client: ump_ng::WeakClient<P, S, R, E>) {}

  /// Post message processing callback.
  ///
  /// The callback must return `ControlFlow::Continue(())` to keep the
  /// dispatcher loop going.  Returning `ControlFlow::Break(RV)` will cause the
  /// dispatcher loop to abort and returns the value in `RV` from the thread.
  fn post(&mut self, msg: P) -> ControlFlow<RV, ()>;

  /// Request message processing callback.
  ///
  /// The callback must return `ControlFlow::Continue(())` to keep the
  /// dispatcher loop going.  Returning `ControlFlow::Break(RV)` will cause the
  /// dispatcher loop to abort and returns the value in `RV` from the thread.
  fn req(&mut self, msg: S, rctx: ReplyContext<R, E>) -> ControlFlow<RV, ()>;

  /// Optional termination callback.
  ///
  /// This is called on the dispatcher thread just after the main message
  /// processing loop has been terminbated.
  fn term(&mut self, rv: Option<RV>) -> Option<RV> {
    rv
  }
}

/// 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).
pub fn spawn<P, S, R, E, RV>(
  mut handler: impl Handler<P, S, R, E, RV> + Send + 'static
) -> (Client<P, S, R, E>, JoinHandle<Option<RV>>)
where
  P: 'static + Send,
  S: 'static + Send,
  R: 'static + Send,
  E: 'static + Send,
  RV: 'static + Send
{
  let (server, client) = channel();
  let weak_client = client.weak();
  let jh = thread::spawn(move || {
    handler.init(weak_client);
    let ret = loop {
      match server.wait() {
        Ok(msg) => match msg {
          MsgType::Put(m) => match handler.post(m) {
            ControlFlow::Continue(_) => {}
            ControlFlow::Break(rv) => break Some(rv)
          },
          MsgType::Request(m, rctx) => match handler.req(m, rctx) {
            ControlFlow::Continue(_) => {}
            ControlFlow::Break(rv) => break Some(rv)
          }
        },
        Err(_) => break None
      }
    };
    handler.term(ret)
  });

  (client, jh)
}


/// Spawn a thread to run a pre-initialized handler.
///
/// It is assumed that the caller has initialized the handler, thus its
/// `init()` method will not be called.
///
/// ```
/// use std::ops::ControlFlow;
/// use ump_ng_server::{ump_ng, spawn_thread_preinit, ThreadedHandler,
///   ReplyContext};
/// let (server, client) = ump_ng::channel::<(), (), (), ()>();
///
/// struct MyHandler {
///   wclnt: ump_ng::WeakClient<(), (), (), ()>
/// }
/// impl ThreadedHandler<(), (), (), (), ()> for MyHandler {
///   fn post(&mut self, _: ()) -> ControlFlow<(), ()> {
///     ControlFlow::Continue(())
///   }
///   fn req(&mut self, _: (), rctx: ReplyContext<(), ()>)
///     -> ControlFlow<(), ()> {
///     ControlFlow::Continue(())
///   }
/// }
/// let handler = MyHandler {
///   wclnt: client.weak()
/// };
/// let jh = spawn_thread_preinit(server, handler);
///
/// // drop client to force dispatch loop to terminate
/// drop(client);
///
/// jh.join();
/// ```
pub fn spawn_preinit<P, S, R, E, RV>(
  server: Server<P, S, R, E>,
  mut handler: impl Handler<P, S, R, E, RV> + Send + 'static
) -> JoinHandle<Option<RV>>
where
  P: 'static + Send,
  S: 'static + Send,
  R: 'static + Send,
  E: 'static + Send,
  RV: 'static + Send
{
  thread::spawn(move || {
    let ret = loop {
      match server.wait() {
        Ok(msg) => match msg {
          MsgType::Put(m) => match handler.post(m) {
            ControlFlow::Continue(_) => {}
            ControlFlow::Break(rv) => break Some(rv)
          },
          MsgType::Request(m, rctx) => match handler.req(m, rctx) {
            ControlFlow::Continue(_) => {}
            ControlFlow::Break(rv) => break Some(rv)
          }
        },
        Err(_) => break None
      }
    };
    handler.term(ret)
  })
}

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