usecop/
node.rs

1// -----------------------------------------------------------------------------
2// This file is part of µSECoP.
3//
4// This program is free software; you can redistribute it and/or modify it under
5// the terms of the GNU General Public License as published by the Free Software
6// Foundation; either version 2 of the License, or (at your option) any later
7// version.
8//
9// This program is distributed in the hope that it will be useful, but WITHOUT
10// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
11// FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
12// details.
13//
14// You should have received a copy of the GNU General Public License along with
15// this program; if not, write to the Free Software Foundation, Inc.,
16// 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
17//
18// Module authors:
19//   Georg Brandl <g.brandl@fz-juelich.de>
20//   Enrico Faulhaber <enrico.faulhaber@frm2.tum.de>
21//
22// -----------------------------------------------------------------------------
23//
24//! SECoP node abstraction.
25
26use core2::io::Write;
27use crate::proto::{InMsg, OutMsg, Error, Specifier, Timestamp};
28use crate::{wire, http, ClientId, ModuleDescription, ToJson};
29
30
31#[derive(Clone, Copy, Default, PartialEq, Eq)]
32enum ClientState {
33    #[default]
34    Disconnected,
35    Initial,
36    WebSocket(bool),
37    Plain(bool),
38}
39
40impl ClientState {
41    fn activate(&mut self, yesno: bool) {
42        match self {
43            Self::WebSocket(active) | Self::Plain(active) => *active = yesno,
44            _ => ()
45        }
46    }
47}
48
49
50pub struct SecNode<T: Modules, const MAX_CLIENTS: ClientId = 8> {
51    state: [ClientState; MAX_CLIENTS],
52    equipment_id: &'static str,
53    description: &'static str,
54    modules: T,
55}
56
57impl<T: Modules, const MAX_CLIENTS: ClientId> SecNode<T, MAX_CLIENTS> {
58    pub fn new(equipment_id: &'static str, description: &'static str, modules: T) -> Self {
59        Self {
60            state: [ClientState::Disconnected; MAX_CLIENTS],
61            equipment_id,
62            description,
63            modules,
64        }
65    }
66
67    pub fn new_client(&mut self) -> Option<ClientId> {
68        self.state.iter().position(|&x| x == ClientState::Disconnected)
69                         .map(|i| { self.state[i] = ClientState::Initial; i })
70    }
71
72    pub fn client_connected(&mut self, client: ClientId) {
73        self.state[client] = ClientState::Initial;
74    }
75
76    pub fn client_finished(&mut self, client: ClientId) {
77        self.state[client] = ClientState::Disconnected;
78    }
79
80    pub fn process(&mut self, time: Timestamp, input: &mut [u8], id: ClientId,
81                   callback: impl FnMut(ClientId, &dyn Fn(&mut dyn Write)))
82        -> Result<usize, ()>
83    {
84
85        match self.state[id] {
86            ClientState::Initial if input.starts_with(b"GET") => match http::handle(input, id, callback) {
87                Ok(Some(used)) => {
88                    self.state[id] = ClientState::WebSocket(false);
89                    Ok(used)
90                }
91                Ok(None) => Ok(0),  // need more input
92                Err(_) => Err(()),
93            }
94            ClientState::Initial => {
95                self.state[id] = ClientState::Plain(false);
96                // recursive call for Plain case
97                self.process(time, input, id, callback)
98            }
99            ClientState::WebSocket(_) => match http::decode_ws(input) {
100                Err(_) => {
101                    self.state[id] = ClientState::Disconnected;
102                    Err(())
103                }
104                Ok(Some((msg, used))) => {
105                    self.process_msg(time, msg, id, callback);
106                    Ok(used)
107                }
108                Ok(None) => Ok(0)
109            }
110            ClientState::Plain(_) => match find_line(input) {
111                Some((msg, used)) => {
112                    self.process_msg(time, msg, id, callback);
113                    Ok(used)
114                }
115                None => Ok(0)
116            }
117            ClientState::Disconnected => Err(()),  // should not happen
118        }
119    }
120
121    fn process_msg(&mut self, time: Timestamp, msg: &mut [u8], id: ClientId,
122                   mut callback: impl FnMut(ClientId, &dyn Fn(&mut dyn Write))) {
123        let states = self.state;
124        let mut sender = Sender::new(id, &mut callback, &states);
125        let result = match InMsg::parse(msg) {
126            Ok(input_msg) => match input_msg {
127                InMsg::Change { spec, value } => match self.modules.by_name(spec.module) {
128                    Some(m) => return m.change(time, spec, value, sender),
129                    None => Error::no_module().spec_msg(wire::CHANGE, spec),
130                },
131                InMsg::Do { spec, arg } => match self.modules.by_name(spec.module) {
132                    Some(m) => return m.do_(time, spec, arg, sender),
133                    None => Error::no_module().spec_msg(wire::DO, spec),
134                },
135                InMsg::Read { spec } => match self.modules.by_name(spec.module) {
136                    Some(m) => return m.read(time, spec, sender),
137                    None => Error::no_module().spec_msg(wire::READ, spec),
138                },
139                InMsg::Activate { module } => {
140                    self.state[id].activate(true);
141                    sender.distribute_single = true;
142                    self.modules.for_each(|name, module| {
143                        module.poll(time, name, true, &mut sender);
144                    });
145                    OutMsg::Active { module }
146                },
147                InMsg::Deactivate { module } => {
148                    self.state[id].activate(false);
149                    OutMsg::Inactive { module }
150                },
151                InMsg::Ping { token } => OutMsg::Pong { token, value: (), time },
152                InMsg::Idn => OutMsg::IdnReply,
153                InMsg::Describe =>
154                    return self.modules.describe(self.equipment_id, self.description, sender),
155                InMsg::Help => OutMsg::Helping { message: "See https://sampleenvironment.org/secop" },
156            },
157            Err(errmsg) => errmsg,
158        };
159        sender.send(result)
160    }
161
162    pub fn poll(&mut self, time: Timestamp, mut callback: impl FnMut(ClientId, &dyn Fn(&mut dyn Write))) {
163        let states = self.state;
164        let mut sender = Sender::new(MAX_CLIENTS + 1, &mut callback, &states);
165        self.modules.for_each(|name, module| {
166            module.poll(time, name, false, &mut sender);
167        });
168    }
169}
170
171pub trait Modules {
172    fn count(&self) -> usize;
173    fn by_name(&mut self, name: &str) -> Option<&mut dyn Module>;
174    fn for_each(&mut self, f: impl FnMut(&str, &mut dyn Module));
175    fn describe(&self, eq_id: &str, desc: &str, sender: Sender);
176}
177
178pub trait Module {
179    /// Return the descriptive data for this module (a JSON object).
180    fn describe(&self) -> ModuleDescription;
181    /// Read a parameter and possibly emit an update message.
182    fn read(&mut self, time: Timestamp, spec: Specifier, reply: Sender);
183    /// Change a parameter and possibly emit an update message.
184    fn change(&mut self, time: Timestamp, spec: Specifier, value: &mut str, reply: Sender);
185    /// Execute a command.
186    fn do_(&mut self, time: Timestamp, spec: Specifier, arg: &mut str, reply: Sender);
187    /// Do a polling cycle.
188    fn poll(&mut self, time: Timestamp, name: &str, all: bool, reply: &mut Sender);
189}
190
191/// The internal state of a module needed by the framework.
192pub struct ModuleInternals {
193    pub description: &'static str,
194    pub pollinterval: f64,
195    pub lastpoll: f64,
196}
197
198impl ModuleInternals {
199    pub fn new(description: &'static str, pollinterval: f64) -> Self {
200        Self {
201            description,
202            pollinterval,
203            lastpoll: 0.0,
204        }
205    }
206}
207
208fn find_line(input: &mut [u8]) -> Option<(&mut [u8], usize)> {
209    input.iter().position(|&x| x == b'\n').map(|i| (&mut input[..i], i + 1))
210}
211
212/// A callback that we can call with the desired client ID, and it then calls
213/// the inner callback with a writer to let us write data.
214type SenderCallback<'a> = &'a mut dyn FnMut(ClientId, &dyn Fn(&mut dyn Write));
215
216pub struct Sender<'a> {
217    client: ClientId,
218    writer: SenderCallback<'a>,
219    states: &'a [ClientState],
220    pub distribute_single: bool,
221}
222
223impl Sender<'_> {
224    fn new<'a>(client: ClientId, writer: SenderCallback<'a>,
225               states: &'a [ClientState]) -> Sender<'a> {
226        Sender { client, writer, states, distribute_single: false }
227    }
228
229    pub fn send<V: ToJson>(&mut self, msg: OutMsg<'_, V>) {
230        if matches!(self.states[self.client], ClientState::WebSocket(_)) {
231            self.send_ws(self.client, &msg);
232        } else {
233            self.send_plain(self.client, &msg);
234        }
235    }
236
237    pub fn distribute<V: ToJson>(&mut self, msg: OutMsg<'_, V>) {
238        if self.distribute_single {
239            return self.send(msg);
240        }
241        for (sn, state) in self.states.iter().enumerate() {
242            match state {
243                ClientState::Plain(true) => self.send_plain(sn, &msg),
244                ClientState::WebSocket(true) => self.send_ws(sn, &msg),
245                _ => continue,
246            }
247        }
248    }
249
250    fn send_plain<V: ToJson>(&mut self, sn: ClientId, msg: &OutMsg<'_, V>) {
251        (self.writer)(sn, &|writer: &mut dyn Write| {
252            let _ = write!(writer, "{}", msg);
253        });
254    }
255
256    fn send_ws<V: ToJson>(&mut self, sn: ClientId, msg: &OutMsg<'_, V>) {
257        (self.writer)(sn, &|writer: &mut dyn Write| {
258            let mut writer = crate::http::WsWriterWrapper::new(writer);
259            let _ = write!(writer, "{}", msg);
260            let _ = writer.send_frame(true);
261        });
262    }
263}
usecop/node.rs

usecop/
node.rs
