Struct Session 

pub struct Session { /* private fields */ }

A NETCONF session with pipelining and streaming support.

Architecture:

• Writer: the session holds the write half of the stream and a codec for encoding outgoing RPCs.
• Reader task: a background tokio task reads framed chunks from the read half, classifies them (<rpc-reply> vs <notification>), and routes them to the correct handler.
• Pipelining: rpc_send() writes an RPC and returns an RpcFuture without waiting for the reply. Multiple RPCs can be in flight simultaneously.
• Streaming: rpc_stream() writes an RPC and returns an RpcStream that yields raw bytes as they arrive. This avoids buffering the entire response for large payloads.

Implementations

impl Session

pub async fn connect( host: &str, port: u16, username: &str, password: &str, ) -> Result<Self>

Connect to a NETCONF server over SSH with password authentication.

Examples found in repository

examples/basic.rs (line 6)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn connect_with_config( host: &str, port: u16, username: &str, password: &str, config: Config, ) -> Result<Self>

Connect with custom configuration.

Examples found in repository

examples/disconnect_watcher.rs (line 17)

async fn main() -> netconf_rust::Result<()> {
    let config = Config::builder()
        .keepalive_interval(Duration::from_secs(10))
        .keepalive_max(3)
        .rpc_timeout(Duration::from_secs(30))
        .connect_timeout(Duration::from_secs(10))
        .nodelay(true)
        .build();

    let session = Arc::new(
        Session::connect_with_config("localhost", 830, "netconf", "netconf", config).await?,
    );

    println!("Connected (session {})", session.session_id());

    // Spawn a background task that fires when the session disconnects.
    // This is the pattern you'd use in a long-running service to clean up
    // sessions from a DashMap or similar structure.
    let watcher_session = Arc::clone(&session);
    let watcher = tokio::spawn(async move {
        let reason = watcher_session.disconnected().await;
        println!("Disconnect detected: {reason}");
        // In a real service you'd remove the session from your map here:
        //   sessions.remove(&uuid);
    });

    // Normal operations — the watcher runs in the background
    let config = session
        .get_config(netconf_rust::Datastore::Running, None)
        .await?;
    println!("Got config ({} bytes)", config.len());

    // You can also race an RPC against disconnect using select!
    tokio::select! {
        result = session.get_config(netconf_rust::Datastore::Running, None) => {
            match result {
                Ok(data) => println!("Got config again ({} bytes)", data.len()),
                Err(e) => eprintln!("RPC failed: {e}"),
            }
        }
        reason = session.disconnected() => {
            eprintln!("Connection lost while waiting for RPC: {reason}");
        }
    }

    // Graceful close — the watcher will fire with DisconnectReason::Eof
    // after the server acknowledges the close and drops the connection.
    session.close_session().await?;
    println!("Session closed");

    // Wait for the watcher to complete
    let _ = watcher.await;

    Ok(())
}

pub async fn from_stream<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>( stream: S, ) -> Result<Self>

Create a session from an existing stream (useful for testing).

pub async fn from_stream_with_config<S: AsyncRead + AsyncWrite + Unpin + Send + 'static>( stream: S, config: Config, ) -> Result<Self>

Create a session from an existing stream with custom configuration.

pub fn session_id(&self) -> u32

Examples found in repository

examples/disconnect_watcher.rs (line 20)

async fn main() -> netconf_rust::Result<()> {
    let config = Config::builder()
        .keepalive_interval(Duration::from_secs(10))
        .keepalive_max(3)
        .rpc_timeout(Duration::from_secs(30))
        .connect_timeout(Duration::from_secs(10))
        .nodelay(true)
        .build();

    let session = Arc::new(
        Session::connect_with_config("localhost", 830, "netconf", "netconf", config).await?,
    );

    println!("Connected (session {})", session.session_id());

    // Spawn a background task that fires when the session disconnects.
    // This is the pattern you'd use in a long-running service to clean up
    // sessions from a DashMap or similar structure.
    let watcher_session = Arc::clone(&session);
    let watcher = tokio::spawn(async move {
        let reason = watcher_session.disconnected().await;
        println!("Disconnect detected: {reason}");
        // In a real service you'd remove the session from your map here:
        //   sessions.remove(&uuid);
    });

    // Normal operations — the watcher runs in the background
    let config = session
        .get_config(netconf_rust::Datastore::Running, None)
        .await?;
    println!("Got config ({} bytes)", config.len());

    // You can also race an RPC against disconnect using select!
    tokio::select! {
        result = session.get_config(netconf_rust::Datastore::Running, None) => {
            match result {
                Ok(data) => println!("Got config again ({} bytes)", data.len()),
                Err(e) => eprintln!("RPC failed: {e}"),
            }
        }
        reason = session.disconnected() => {
            eprintln!("Connection lost while waiting for RPC: {reason}");
        }
    }

    // Graceful close — the watcher will fire with DisconnectReason::Eof
    // after the server acknowledges the close and drops the connection.
    session.close_session().await?;
    println!("Session closed");

    // Wait for the watcher to complete
    let _ = watcher.await;

    Ok(())
}

examples/basic.rs (line 8)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub fn server_capabilities(&self) -> &[String]

Examples found in repository

examples/basic.rs (line 10)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub fn framing_mode(&self) -> FramingMode

Examples found in repository

examples/basic.rs (line 9)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub fn state(&self) -> SessionState

Examples found in repository

examples/basic.rs (line 126)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub fn disconnected( &self, ) -> impl Future<Output = DisconnectReason> + Send + 'static

Returns a future that completes when the session disconnects.

Can be called multiple times — each call clones an internal watch::Receiver, so multiple tasks can independently await the same disconnect event. If the session is already disconnected when called, returns immediately.

Example

let reason = session.disconnected().await;
println!("session died: {reason}");

Examples found in repository

examples/disconnect_watcher.rs (line 27)

async fn main() -> netconf_rust::Result<()> {
    let config = Config::builder()
        .keepalive_interval(Duration::from_secs(10))
        .keepalive_max(3)
        .rpc_timeout(Duration::from_secs(30))
        .connect_timeout(Duration::from_secs(10))
        .nodelay(true)
        .build();

    let session = Arc::new(
        Session::connect_with_config("localhost", 830, "netconf", "netconf", config).await?,
    );

    println!("Connected (session {})", session.session_id());

    // Spawn a background task that fires when the session disconnects.
    // This is the pattern you'd use in a long-running service to clean up
    // sessions from a DashMap or similar structure.
    let watcher_session = Arc::clone(&session);
    let watcher = tokio::spawn(async move {
        let reason = watcher_session.disconnected().await;
        println!("Disconnect detected: {reason}");
        // In a real service you'd remove the session from your map here:
        //   sessions.remove(&uuid);
    });

    // Normal operations — the watcher runs in the background
    let config = session
        .get_config(netconf_rust::Datastore::Running, None)
        .await?;
    println!("Got config ({} bytes)", config.len());

    // You can also race an RPC against disconnect using select!
    tokio::select! {
        result = session.get_config(netconf_rust::Datastore::Running, None) => {
            match result {
                Ok(data) => println!("Got config again ({} bytes)", data.len()),
                Err(e) => eprintln!("RPC failed: {e}"),
            }
        }
        reason = session.disconnected() => {
            eprintln!("Connection lost while waiting for RPC: {reason}");
        }
    }

    // Graceful close — the watcher will fire with DisconnectReason::Eof
    // after the server acknowledges the close and drops the connection.
    session.close_session().await?;
    println!("Session closed");

    // Wait for the watcher to complete
    let _ = watcher.await;

    Ok(())
}

pub async fn rpc_send(&self, inner_xml: &str) -> Result<RpcFuture>

Send a raw RPC and return a future for the reply (pipelining).

This writes the RPC to the server immediately but does not wait for the reply. Call RpcFuture::response() to await the reply. Multiple RPCs can be pipelined by calling this repeatedly before awaiting any of them.

Examples found in repository

examples/basic.rs (line 83)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn rpc_raw(&self, inner_xml: &str) -> Result<RpcReply>

Send a raw RPC and wait for the reply.

Examples found in repository

examples/basic.rs (line 110)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn rpc_stream(&self, inner_xml: &str) -> Result<RpcStream>

Send a raw RPC and return a streaming response.

Unlike rpc_send() which buffers the entire response, this returns an RpcStream that yields raw XML bytes as individual chunks arrive from the server. The stream implements AsyncRead, so it can be piped directly into compression encoders, file writers, or any byte-oriented consumer.

Normal (buffered) and streaming RPCs can be freely interleaved on the same session — the reader task routes each message independently based on its message-id.

Example

// Pipeline a normal RPC alongside a streaming one
let edit_future = session.rpc_send("<edit-config>...</edit-config>").await?;
let mut config_stream = session.rpc_stream("<get-config><source><running/></source></get-config>").await?;

// Stream the large response
let mut buf = [0u8; 8192];
loop {
    let n = config_stream.read(&mut buf).await?;
    if n == 0 { break; }
    // process buf[..n]
}

// Collect the normal RPC reply
let edit_reply = edit_future.response().await?;

pub async fn get_config( &self, source: Datastore, filter: Option<&str>, ) -> Result<String>

Retrieve configuration from a datastore.

Examples found in repository

examples/disconnect_watcher.rs (line 35)

async fn main() -> netconf_rust::Result<()> {
    let config = Config::builder()
        .keepalive_interval(Duration::from_secs(10))
        .keepalive_max(3)
        .rpc_timeout(Duration::from_secs(30))
        .connect_timeout(Duration::from_secs(10))
        .nodelay(true)
        .build();

    let session = Arc::new(
        Session::connect_with_config("localhost", 830, "netconf", "netconf", config).await?,
    );

    println!("Connected (session {})", session.session_id());

    // Spawn a background task that fires when the session disconnects.
    // This is the pattern you'd use in a long-running service to clean up
    // sessions from a DashMap or similar structure.
    let watcher_session = Arc::clone(&session);
    let watcher = tokio::spawn(async move {
        let reason = watcher_session.disconnected().await;
        println!("Disconnect detected: {reason}");
        // In a real service you'd remove the session from your map here:
        //   sessions.remove(&uuid);
    });

    // Normal operations — the watcher runs in the background
    let config = session
        .get_config(netconf_rust::Datastore::Running, None)
        .await?;
    println!("Got config ({} bytes)", config.len());

    // You can also race an RPC against disconnect using select!
    tokio::select! {
        result = session.get_config(netconf_rust::Datastore::Running, None) => {
            match result {
                Ok(data) => println!("Got config again ({} bytes)", data.len()),
                Err(e) => eprintln!("RPC failed: {e}"),
            }
        }
        reason = session.disconnected() => {
            eprintln!("Connection lost while waiting for RPC: {reason}");
        }
    }

    // Graceful close — the watcher will fire with DisconnectReason::Eof
    // after the server acknowledges the close and drops the connection.
    session.close_session().await?;
    println!("Session closed");

    // Wait for the watcher to complete
    let _ = watcher.await;

    Ok(())
}

examples/basic.rs (line 13)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn get_config_payload( &self, source: Datastore, filter: Option<&str>, ) -> Result<DataPayload>

Retrieve configuration as a zero-copy DataPayload.

Same as get_config() but returns a DataPayload instead of String, avoiding a copy of the response body. Use payload.as_str() for a zero-copy &str view, or payload.reader() for streaming XML events.

Examples found in repository

examples/basic.rs (line 26)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn get_config_stream( &self, source: Datastore, filter: Option<&str>, ) -> Result<RpcStream>

Retrieve configuration from a datastore as a streaming response.

Same as get_config() but returns an RpcStream instead of buffering the entire response.

pub async fn get(&self, filter: Option<&str>) -> Result<String>

Retrieve running configuration and state data.

pub async fn get_payload(&self, filter: Option<&str>) -> Result<DataPayload>

Retrieve running configuration and state data as a zero-copy DataPayload.

Same as get() but returns a DataPayload instead of String, avoiding a copy of the response body.

pub async fn get_stream(&self, filter: Option<&str>) -> Result<RpcStream>

Retrieve running configuration and state data as a streaming response.

Same as get() but returns an RpcStream instead of buffering the entire response.

pub async fn edit_config(&self, target: Datastore, config: &str) -> Result<()>

Edit the configuration of a target datastore.

Examples found in repository

examples/basic.rs (lines 48-64)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn lock(&self, target: Datastore) -> Result<()>

Lock a datastore

Examples found in repository

examples/basic.rs (line 44)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn unlock(&self, target: Datastore) -> Result<()>

Unlock a datastore.

Examples found in repository

examples/basic.rs (line 77)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn commit(&self) -> Result<()>

Commit the candidate configuration to running.

Examples found in repository

examples/basic.rs (line 69)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn close_session(&self) -> Result<()>

Gracefully close the NETCONF session.

Examples found in repository

examples/disconnect_watcher.rs (line 54)

async fn main() -> netconf_rust::Result<()> {
    let config = Config::builder()
        .keepalive_interval(Duration::from_secs(10))
        .keepalive_max(3)
        .rpc_timeout(Duration::from_secs(30))
        .connect_timeout(Duration::from_secs(10))
        .nodelay(true)
        .build();

    let session = Arc::new(
        Session::connect_with_config("localhost", 830, "netconf", "netconf", config).await?,
    );

    println!("Connected (session {})", session.session_id());

    // Spawn a background task that fires when the session disconnects.
    // This is the pattern you'd use in a long-running service to clean up
    // sessions from a DashMap or similar structure.
    let watcher_session = Arc::clone(&session);
    let watcher = tokio::spawn(async move {
        let reason = watcher_session.disconnected().await;
        println!("Disconnect detected: {reason}");
        // In a real service you'd remove the session from your map here:
        //   sessions.remove(&uuid);
    });

    // Normal operations — the watcher runs in the background
    let config = session
        .get_config(netconf_rust::Datastore::Running, None)
        .await?;
    println!("Got config ({} bytes)", config.len());

    // You can also race an RPC against disconnect using select!
    tokio::select! {
        result = session.get_config(netconf_rust::Datastore::Running, None) => {
            match result {
                Ok(data) => println!("Got config again ({} bytes)", data.len()),
                Err(e) => eprintln!("RPC failed: {e}"),
            }
        }
        reason = session.disconnected() => {
            eprintln!("Connection lost while waiting for RPC: {reason}");
        }
    }

    // Graceful close — the watcher will fire with DisconnectReason::Eof
    // after the server acknowledges the close and drops the connection.
    session.close_session().await?;
    println!("Session closed");

    // Wait for the watcher to complete
    let _ = watcher.await;

    Ok(())
}

examples/basic.rs (line 125)

async fn main() -> netconf_rust::Result<()> {
    // --- Connect ---
    let session = Session::connect("localhost", 830, "netconf", "netconf").await?;

    println!("Session ID: {}", session.session_id());
    println!("Framing: {:?}", session.framing_mode());
    println!("Capabilities: {}", session.server_capabilities().join(", "));

    // --- Get full running config ---
    let config = session.get_config(Datastore::Running, None).await?;
    println!("\n=== Running config ===\n{config}");

    // --- Filtered get-config with subtree filter ---
    let interfaces = session
        .get_config(
            Datastore::Running,
            Some(r#"<interfaces xmlns="urn:ietf:params:xml:ns:yang:ietf-interfaces"/>"#),
        )
        .await?;
    println!("\n=== Interfaces ===\n{interfaces}");

    // --- Zero-copy DataPayload ---
    let payload = session.get_config_payload(Datastore::Running, None).await?;
    println!("\n=== Config payload size: {} bytes ===", payload.len());

    // Stream XML events directly from the payload's bytes
    let mut reader = payload.reader();
    let mut element_count = 0;
    loop {
        match reader.read_event() {
            Ok(quick_xml::events::Event::Start(_)) => element_count += 1,
            Ok(quick_xml::events::Event::Eof) => break,
            _ => {}
        }
    }
    println!("Total XML elements: {element_count}");

    // --- Edit config with candidate commit ---
    // Note: NACM may deny access to certain data models depending on the user's
    // permissions. We use the netconf-server model which the netconf user can write.
    session.lock(Datastore::Candidate).await?;
    println!("\n=== Locked candidate ===");

    match session
        .edit_config(
            Datastore::Candidate,
            r#"
        <netconf-server xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-server">
            <listen>
                <endpoint>
                    <name>default-ssh</name>
                    <ssh>
                        <tcp-server-parameters>
                            <local-address>0.0.0.0</local-address>
                        </tcp-server-parameters>
                    </ssh>
                </endpoint>
            </listen>
        </netconf-server>
    "#,
        )
        .await
    {
        Ok(()) => {
            println!("Edited candidate config");
            session.commit().await?;
            println!("Committed");
        }
        Err(e) => {
            eprintln!("edit-config failed (NACM?): {e}");
        }
    }

    session.unlock(Datastore::Candidate).await?;
    println!("Unlocked candidate");

    // --- Pipelining: send 3 RPCs before awaiting any replies ---
    println!("\n=== Pipelining ===");
    let fut1 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut2 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;
    let fut3 = session
        .rpc_send("<get-config><source><running/></source></get-config>")
        .await?;

    println!(
        "Sent 3 RPCs (message-ids: {}, {}, {})",
        fut1.message_id(),
        fut2.message_id(),
        fut3.message_id()
    );

    let reply1 = fut1.response().await?;
    let reply2 = fut2.response().await?;
    let reply3 = fut3.response().await?;
    println!(
        "Got 3 replies (message-ids: {}, {}, {})",
        reply1.message_id, reply2.message_id, reply3.message_id
    );

    // --- Raw RPC for operations not covered by convenience methods ---
    println!("\n=== Raw RPC ===");
    let reply = session
        .rpc_raw("<get-schema><identifier>ietf-interfaces</identifier></get-schema>")
        .await?;
    match reply.body {
        netconf_rust::RpcReplyBody::Data(payload) => {
            println!("Got schema ({} bytes)", payload.len())
        }
        netconf_rust::RpcReplyBody::Ok => println!("ok"),
        netconf_rust::RpcReplyBody::Error(errors) => {
            for e in &errors {
                eprintln!("Error: {}: {}", e.error_tag, e.error_message);
            }
        }
    }

    // --- Graceful close ---
    session.close_session().await?;
    println!("\nSession closed (state: {})", session.state());

    Ok(())
}

pub async fn close(self) -> Result<()>

Gracefully close the NETCONF session and shut down the transport.

Sends a <close-session/> RPC, shuts down the write half of the stream, and drops the session (aborting the reader task and releasing the SSH handle). Prefer this over close_session() when you are done with the session entirely.

pub async fn kill_session(&self, session_id: u32) -> Result<()>

Force-close another NETCONF session.

pub fn with_timeout(&self, timeout: Duration) -> SessionWithTimeout<'_>

Return a wrapper that applies timeout to every RPC sent through it, overriding the session-level rpc_timeout.

pub fn pending_rpc_count(&self) -> usize

Number of RPCs that have been sent but not yet fully replied to.

This includes both RPCs awaiting their first reply byte (in the pending map) and streaming RPCs whose response is in flight but not yet complete.

pub fn last_rpc_at(&self) -> Option<Instant>

The Instant when the most recent RPC reply was received, or None if no reply has been received yet.

pub fn connected_since(&self) -> Instant

The Instant when the session was established (hello exchange complete and reader task started).

Trait Implementations

impl Drop for Session

fn drop(&mut self)

Executes the destructor for this type.