righ_ovsdb/
client.rs

1//! TCP/Unix socket based OVSDB client.
2use std::collections::HashMap;
3use std::path::Path;
4
5use futures::{stream::StreamExt, SinkExt};
6use serde::de::DeserializeOwned;
7use tokio::{
8    io::{AsyncReadExt, AsyncWriteExt},
9    net::TcpStream,
10    sync::{
11        mpsc::{self, error::SendError},
12        oneshot::{self, error::RecvError},
13    },
14    task::JoinHandle,
15};
16use tokio_util::codec::Framed;
17
18use super::{protocol, schema::Schema};
19use crate::protocol::{
20    method::{
21        EchoParams, EchoResult, GetSchemaParams, ListDbsResult, Method, Operation, TransactParams,
22    },
23    Message, Request, Response,
24};
25
26/// Internal synchronization failure
27#[derive(Debug)]
28pub struct SynchronizationError(String);
29
30impl std::fmt::Display for SynchronizationError {
31    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
32        write!(f, "Synchronication failure: {}", self.0)
33    }
34}
35
36impl std::error::Error for SynchronizationError {}
37
38impl From<SendError<ClientCommand>> for SynchronizationError {
39    fn from(err: SendError<ClientCommand>) -> Self {
40        Self(err.to_string())
41    }
42}
43
44impl From<SendError<ClientRequest>> for SynchronizationError {
45    fn from(err: SendError<ClientRequest>) -> Self {
46        Self(err.to_string())
47    }
48}
49
50impl From<RecvError> for SynchronizationError {
51    fn from(err: RecvError) -> Self {
52        Self(err.to_string())
53    }
54}
55
56/// The error type for operations performed by the [Client].
57#[non_exhaustive]
58#[derive(thiserror::Error, Debug)]
59pub enum ClientError {
60    /// An internal error occurred during synchronization.
61    #[error("Failed to deliver command")]
62    Internal(#[from] SynchronizationError),
63    /// An error occurred trying to establish a connection with the OVSDB server.
64    #[error("Failed to establish connection with the server")]
65    ConnectionFailed(#[source] std::io::Error),
66    /// An error occurred while trying to shutdown the client main loop.
67    #[error("Failed to shutdown client")]
68    ShutdownError(#[from] tokio::task::JoinError),
69    /// A client method was executed, but the client is not connected to OVSDB.
70    #[error("Client thread not active")]
71    NotRunning,
72    /// A response was received from the OVSDB server that could not be processed.
73    #[error("Unexpected result received in response object")]
74    UnexpectedResult,
75    /// An error was encountered while processing send/receive with the OVSDB server.
76    #[error("An error occurred when communicating with the server")]
77    CommunicationFailure(#[from] protocol::CodecError),
78    /// A low-level OVSDB error was encountered.
79    #[error("OVSDB error")]
80    OvsdbError(#[from] crate::Error),
81}
82
83#[derive(Debug)]
84struct ClientRequest {
85    tx: oneshot::Sender<protocol::Response>,
86    request: Request,
87}
88
89#[derive(Clone, Copy, Debug)]
90enum ClientCommand {
91    Shutdown,
92}
93
94/// An OVSDB client, used to interact with an OVSDB database server.
95///
96/// The client is a thin wrapper around the various methods available in the OVSDB protocol.
97/// Instantiating a client is done through one of the `connect_` methods.
98///
99/// # Examples
100///
101/// ```rust,no_run
102/// use std::path::Path;
103///
104/// use ovsdb::Client;
105///
106/// # tokio_test::block_on(async {
107/// let client = Client::connect_unix(Path::new("/var/run/openvswitch/db.sock"))
108///     .await
109///     .unwrap();
110/// # })
111/// ```
112#[derive(Debug)]
113pub struct Client {
114    request_sender: Option<mpsc::Sender<ClientRequest>>,
115    command_sender: Option<mpsc::Sender<ClientCommand>>,
116    handle: JoinHandle<Result<(), ClientError>>,
117}
118
119impl Client {
120    fn new(
121        request_sender: mpsc::Sender<ClientRequest>,
122        command_sender: mpsc::Sender<ClientCommand>,
123        handle: JoinHandle<Result<(), ClientError>>,
124    ) -> Self {
125        Self {
126            request_sender: Some(request_sender),
127            command_sender: Some(command_sender),
128            handle,
129        }
130    }
131
132    /// Expose the client start for any kinds of Stream data not only TCP/Unix client.
133    ///
134    /// The client is a thin wrapper around the various methods available in the OVSDB protocol.
135    /// Instantiating a client is done through one of the `connect_` methods.
136    ///
137    pub async fn start<T>(stream: T) -> Result<Self, ClientError>
138    where
139        T: AsyncWriteExt + AsyncReadExt + Send + 'static,
140    {
141        let (requests_tx, requests_rx) = mpsc::channel(32);
142        let (commands_tx, commands_rx) = mpsc::channel(32);
143
144        let handle =
145            { tokio::spawn(async move { client_main(requests_rx, commands_rx, stream).await }) };
146
147        Ok(Client::new(requests_tx, commands_tx, handle))
148    }
149
150    /// Connect to an OVSDB server via TCP socket.
151    ///
152    /// # Examples
153    ///
154    /// ```rust,no_run
155    /// use ovsdb::client::Client;
156    ///
157    /// # tokio_test::block_on(async {
158    /// let client = Client::connect_tcp("127.0.0.1:6641")
159    ///     .await
160    ///     .unwrap();
161    /// # })
162    /// ```
163    pub async fn connect_tcp<T>(server_addr: T) -> Result<Self, ClientError>
164    where
165        T: AsRef<str> + tokio::net::ToSocketAddrs,
166    {
167        let stream = TcpStream::connect(server_addr)
168            .await
169            .map_err(ClientError::ConnectionFailed)?;
170        Client::start(stream).await
171    }
172
173    /// Connect to an OVSDB server via UNIX domain socket.
174    ///
175    /// # Examples
176    ///
177    /// ```rust,no_run
178    /// use std::path::Path;
179    ///
180    /// use ovsdb::client::Client;
181    ///
182    /// # tokio_test::block_on(async {
183    /// let client = Client::connect_unix(Path::new("/var/run/openvswitch/db.sock"))
184    ///     .await
185    ///     .unwrap();
186    /// # })
187    /// ```
188    #[cfg(not(any(target_os = "windows")))]
189    pub async fn connect_unix(socket: &Path) -> Result<Self, ClientError> {
190        use tokio::net::UnixStream;
191        let stream = UnixStream::connect(socket)
192            .await
193            .map_err(ClientError::ConnectionFailed)?;
194        Client::start(stream).await
195    }
196
197    /// Disconnect from the OVSDB server and stop processing messages.
198    ///
199    /// # Examples
200    ///
201    /// ```rust,no_run
202    /// use std::path::Path;
203    ///
204    /// use ovsdb::Client;
205    ///
206    /// # tokio_test::block_on(async {
207    /// let client = Client::connect_unix(Path::new("/var/run/openvswitch/db.sock"))
208    ///     .await
209    ///     .unwrap();
210    ///
211    /// // Perform OVSDB operations
212    ///
213    /// client.stop().await.unwrap();
214    /// # })
215    pub async fn stop(mut self) -> Result<(), ClientError> {
216        if let Some(sender) = self.command_sender.take() {
217            sender
218                .send(ClientCommand::Shutdown)
219                .await
220                .map_err(|e| ClientError::Internal(e.into()))?;
221            drop(sender);
222        };
223        if let Some(sender) = self.request_sender.take() {
224            drop(sender);
225        }
226
227        self.handle.await?
228    }
229
230    /// Execute a raw OVSDB request, receiving a raw response.
231    ///
232    /// Under normal circumstances, this method should only be called internally, with clients
233    /// preferring one of the purpose-built methods (ie. [`Client::echo`]).  However, if for some reason
234    /// those methods are insufficient, raw requests can be made to the database.
235    ///
236    pub async fn execute<T>(&self, request: Request) -> Result<Option<T>, ClientError>
237    where
238        T: DeserializeOwned,
239    {
240        let (tx, rx) = oneshot::channel();
241
242        match &self.request_sender {
243            Some(s) => {
244                s.send(ClientRequest { tx, request })
245                    .await
246                    .map_err(|e| ClientError::Internal(e.into()))?;
247                let res = rx.await.map_err(|e| ClientError::Internal(e.into()))?;
248                let r: Option<T> = res.result()?;
249                Ok(r)
250            }
251            None => Err(ClientError::NotRunning),
252        }
253    }
254
255    /// Issues an `echo` request to the OVSDB server.
256    ///
257    /// On success, the arguments to the request are returned as the result.
258    ///
259    /// ```rust,no_run
260    /// use std::path::Path;
261    ///
262    /// use ovsdb::client::Client;
263    ///
264    /// # tokio_test::block_on(async {
265    /// let client = Client::connect_unix(Path::new("/var/run/openvswitch/db.sock"))
266    ///     .await
267    ///     .unwrap();
268    ///
269    /// let args = vec!["Hello", "OVSDB"];
270    /// let result = client.echo(args.clone()).await.unwrap();
271    /// assert_eq!(*result, args);
272    /// # })
273    /// ```
274    pub async fn echo<T, I>(&self, args: T) -> Result<EchoResult, ClientError>
275    where
276        T: IntoIterator<Item = I> + Send,
277        I: Into<String> + std::fmt::Debug,
278    {
279        match self
280            .execute(crate::protocol::Request::new(
281                None,
282                Method::Echo,
283                Some(Box::new(EchoParams::new(args))),
284            ))
285            .await?
286        {
287            Some(data) => Ok(data),
288            None => Err(ClientError::UnexpectedResult),
289        }
290    }
291
292    /// Issues a `list_dbs` request to the OVSDB server.
293    ///
294    /// On success, a list of databases supported by the server are returned.
295    ///
296    /// ```rust,no_run
297    /// use std::path::Path;
298    ///
299    /// use ovsdb::Client;
300    ///
301    /// # tokio_test::block_on(async {
302    /// let client = Client::connect_unix(Path::new("/var/run/openvswitch/db.sock"))
303    ///     .await
304    ///     .unwrap();
305    ///
306    /// let dbs = client.list_databases().await.unwrap();
307    /// println!("available databases: {:#?}", dbs);
308    /// # })
309    /// ```
310    pub async fn list_databases(&self) -> Result<ListDbsResult, ClientError> {
311        match self
312            .execute(crate::protocol::Request::new(
313                None,
314                Method::ListDatabases,
315                None,
316            ))
317            .await?
318        {
319            Some(data) => Ok(data),
320            None => Err(ClientError::UnexpectedResult),
321        }
322    }
323
324    /// Issues a `get_schema` request to the OVSDB server.
325    ///
326    /// On success, a [Schema] instance is returned matching the OVSDB schema for the specified
327    /// database.
328    ///
329    /// ```rust,no_run
330    /// use std::path::Path;
331    ///
332    /// use ovsdb::Client;
333    ///
334    /// # tokio_test::block_on(async {
335    /// let client = Client::connect_unix(Path::new("/var/run/openvswitch/db.sock"))
336    ///     .await
337    ///     .unwrap();
338    ///
339    /// let schema = client.get_schema("Open_vSwitch").await.unwrap();
340    /// println!("Open_vSwitch schema: {:#?}", schema);
341    /// # })
342    /// ```
343    pub async fn get_schema<S>(&self, database: S) -> Result<Schema, ClientError>
344    where
345        S: Into<String>,
346    {
347        match self
348            .execute(crate::protocol::Request::new(
349                None,
350                Method::GetSchema,
351                Some(Box::new(GetSchemaParams::new(database))),
352            ))
353            .await?
354        {
355            Some(data) => Ok(data),
356            None => Err(ClientError::UnexpectedResult),
357        }
358    }
359
360    /// Issues a `transact` request to the OVSDB server.
361    ///
362    /// TODO
363    pub async fn transact<S, T>(
364        &self,
365        database: S,
366        operations: Vec<Operation>,
367    ) -> Result<T, ClientError>
368    where
369        S: Into<String>,
370        T: DeserializeOwned,
371    {
372        match self
373            .execute(crate::protocol::Request::new(
374                None,
375                Method::Transact,
376                Some(Box::new(TransactParams::new(database, operations))),
377            ))
378            .await?
379        {
380            Some(data) => Ok(data),
381            None => Err(ClientError::UnexpectedResult),
382        }
383    }
384}
385
386async fn client_main<T>(
387    mut requests: mpsc::Receiver<ClientRequest>,
388    mut commands: mpsc::Receiver<ClientCommand>,
389    stream: T,
390) -> Result<(), ClientError>
391where
392    T: AsyncReadExt + AsyncWriteExt,
393{
394    let (mut writer, mut reader) = Framed::new(stream, protocol::Codec::new()).split();
395    let mut channels: HashMap<String, oneshot::Sender<protocol::Response>> = HashMap::new();
396
397    loop {
398        tokio::select! {
399            Some(req) = requests.recv() => {
400                let request = req.request;
401                if let Some(id) = request.id() {
402                    channels.insert(id.clone(), req.tx);
403                }
404                // writer.send(request.into()).await?;
405                writer.send(request.into()).await?;
406            },
407            Some(cmd) = commands.recv() => {
408                match cmd {
409                    ClientCommand::Shutdown => {
410                        writer.close().await?;
411                        // todo!()
412                        // writer.
413                        // writer.shutdown().await?;
414                    }
415                }
416            }
417            Some(msg) = reader.next() => {
418                match msg {
419                    Ok(protocol::Message::Response(res)) => {
420                        if let Some(id) = res.id() {
421                            if let Some(tx) = channels.remove(id) {
422                                let _ = tx.send(res);
423                            }
424                        }
425                    },
426                    Ok(protocol::Message::Request(req)) => {
427                        if let Method::Echo = req.method() {
428                            let echo_uuid = String::from("echo");
429                            let response = Response::new(Some(echo_uuid), Some(serde_json::json!([])), None);
430                            let message = Message::Response(response);
431                            writer.send(message).await?;
432                        }
433                    },
434                    Err(_e) => todo!()
435                }
436            },
437            else => {
438                break;
439            }
440        }
441    }
442
443    Ok(())
444}
righ_ovsdb/client.rs

righ_ovsdb/
client.rs
