Trait ClassifyExt 

pub trait ClassifyExt: Stream<Item = Result<OvpnMessage, Error>> + Sized {
    // Provided method
    fn classify(self) -> Classified<Self> { ... }
}

Extension trait that adds .classify() to any stream of Result<OvpnMessage, io::Error>.

Example

use anyhow::Context;
use tokio::net::TcpStream;
use tokio_util::codec::Framed;
use futures::{SinkExt, StreamExt};
use openvpn_mgmt_codec::{OvpnCodec, OvpnCommand};

let stream = TcpStream::connect("127.0.0.1:7505").await?;
let mut framed = Framed::new(stream, OvpnCodec::new());

// Send a command and read the response with a timeout.
framed.send(OvpnCommand::Pid).await?;
let response = tokio::time::timeout(
    std::time::Duration::from_secs(5),
    framed.next(),
).await
 .context("stream ended")?;

println!("got: {response:?}");

Reconnection with backoff

use tokio::net::TcpStream;
use tokio_util::codec::Framed;
use futures::StreamExt;
use openvpn_mgmt_codec::{OvpnCodec, OvpnMessage};

let mut backoff = std::time::Duration::from_secs(1);
loop {
    match TcpStream::connect("127.0.0.1:7505").await {
        Ok(stream) => {
            backoff = std::time::Duration::from_secs(1); // reset
            let mut framed = Framed::new(stream, OvpnCodec::new());
            while let Some(msg) = framed.next().await {
                match msg {
                    Ok(msg) => println!("{msg:?}"),
                    Err(error) => { eprintln!("decode error: {error}"); break; }
                }
            }
            eprintln!("connection closed, reconnecting...");
        }
        Err(error) => {
            eprintln!("connect failed: {error}, retrying in {backoff:?}");
        }
    }
    tokio::time::sleep(backoff).await;
    backoff = (backoff * 2).min(std::time::Duration::from_secs(30));
}

Detecting connection loss via >FATAL:

use tokio::net::TcpStream;
use tokio_util::codec::Framed;
use futures::StreamExt;
use openvpn_mgmt_codec::{OvpnCodec, OvpnMessage, Notification};

let stream = TcpStream::connect("127.0.0.1:7505").await?;
let mut framed = Framed::new(stream, OvpnCodec::new());

while let Some(msg) = framed.next().await {
    match msg? {
        OvpnMessage::Notification(Notification::Fatal { message }) => {
            eprintln!("OpenVPN fatal: {message}");
            // Trigger graceful shutdown / reconnection.
            break;
        }
        other => println!("{other:?}"),
    }
}
// Stream ended — either FATAL or the daemon closed the connection.
// In both cases, you should reconnect (see reconnection example above).

Provided Methods

fn classify(self) -> Classified<Self>

Classify each OvpnMessage into a ManagementEvent, splitting notifications from command responses.

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety".

Implementors

impl<S: Stream<Item = Result<OvpnMessage, Error>>> ClassifyExt for S