Add YANG notification subscription with circular buffer

src/internal_commands.rs

@@ -20,6 +20,7 @@ use yang5::schema::SchemaNodeKind;
 use crate::YANG_CTX;
 use crate::error::CallbackError;
 use crate::grpc::proto;
+use crate::notifications::NotificationEntry;
 use crate::parser::ParsedArgs;
 use crate::session::{CommandMode, ConfigurationType, Session};
 use crate::token::{Commands, TokenKind};
@@ -488,6 +489,128 @@ pub fn cmd_validate(
     Ok(false)
 }
 
+// ===== "notification" =====
+
+pub fn cmd_notification_start(
+    _commands: &Commands,
+    session: &mut Session,
+    _args: ParsedArgs,
+) -> Result<bool, CallbackError> {
+    match session.notification_start() {
+        Ok(()) => println!("% subscribed to notifications"),
+        Err(error) => println!("% {}", error),
+    }
+
+    Ok(false)
+}
+
+pub fn cmd_notification_stop(
+    _commands: &Commands,
+    session: &mut Session,
+    _args: ParsedArgs,
+) -> Result<bool, CallbackError> {
+    match session.notification_stop() {
+        Ok(()) => println!("% unsubscribed from notifications"),
+        Err(error) => println!("% {}", error),
+    }
+
+    Ok(false)
+}
+
+// ===== "show notifications" =====
+
+// Formats a notification as a single human-readable line by parsing its
+// JSON payload against the YANG schema and flattening all leaf values.
+fn format_notification_line(entry: &NotificationEntry) -> String {
+    let yang_ctx = YANG_CTX.get().unwrap();
+    let timestamp = DateTime::from_timestamp(entry.timestamp, 0)
+        .unwrap_or_default()
+        .with_timezone(&Local)
+        .format("%Y-%m-%d %H:%M:%S");
+
+    // Fall back to the raw JSON payload if parsing fails.
+    let Ok(tree) = DataTree::parse_op_string(
+        yang_ctx,
+        entry.data_json.as_str(),
+        DataFormat::JSON,
+        DataParserFlags::empty(),
+        DataOperation::NotificationYang,
+    ) else {
+        return format!(
+            "{} {} {}",
+            timestamp, entry.module_path, entry.data_json
+        );
+    };
+
+    let mut line = timestamp.to_string();
+
+    // Notification name.
+    match tree
+        .traverse()
+        .find(|dnode| dnode.schema().kind() == SchemaNodeKind::Notification)
+    {
+        Some(dnode) => {
+            let snode = dnode.schema();
+            write!(line, " {}:{}", snode.module().name(), snode.name())
+                .unwrap();
+        }
+        None => write!(line, " {}", entry.module_path).unwrap(),
+    }
+
+    // Leaf values, including list keys of ancestor nodes (e.g. the BGP
+    // neighbor's remote-address).
+    for dnode in tree.traverse().filter(|dnode| {
+        matches!(
+            dnode.schema().kind(),
+            SchemaNodeKind::Leaf | SchemaNodeKind::LeafList
+        )
+    }) {
+        if let Some(value) = dnode.value_canonical() {
+            let snode = dnode.schema();
+            let name = snode.name();
+            if value.contains(' ') {
+                write!(line, " {}=\"{}\"", name, value).unwrap();
+            } else {
+                write!(line, " {}={}", name, value).unwrap();
+            }
+        }
+    }
+
+    line
+}
+
+pub fn cmd_show_notifications(
+    _commands: &Commands,
+    session: &mut Session,
+    mut args: ParsedArgs,
+) -> Result<bool, CallbackError> {
+    let json = get_opt_arg(&mut args, "format").is_some();
+    let buffer = session.notification_buffer();
+    let buffer = buffer.lock().unwrap_or_else(|error| error.into_inner());
+
+    if buffer.is_empty() {
+        writeln!(session.writer(), "% no notifications received")?;
+    }
+    for entry in buffer.entries() {
+        if json {
+            let timestamp = DateTime::from_timestamp(entry.timestamp, 0)
+                .unwrap_or_default()
+                .with_timezone(&Local)
+                .format("%Y-%m-%d %H:%M:%S");
+            writeln!(session.writer(), "{} {}", timestamp, entry.module_path)?;
+            writeln!(session.writer(), "{}", entry.data_json)?;
+            writeln!(session.writer(), "!")?;
+        } else {
+            writeln!(session.writer(), "{}", format_notification_line(entry))?;
+        }
+    }
+    if let Some(status) = buffer.status() {
+        writeln!(session.writer(), "% subscription terminated: {}", status)?;
+    }
+
+    Ok(false)
+}
+
 // ===== "show <candidate|running>" =====
 
 fn cmd_show_config_cmds(

src/internal_commands.xml

@@ -5,6 +5,10 @@
     <token name="exit" help="Exit the management session." cmd="cmd_exit_exec"/>
     <token name="end" help="Terminate configuration session." cmd="cmd_end"/>
     <token name="list" help="Print command list." cmd="cmd_list" pipe="true"/>
+    <token name="notification" help="Notification subscription management.">
+      <token name="start" help="Subscribe to YANG notifications." cmd="cmd_notification_start"/>
+      <token name="stop" help="Unsubscribe from YANG notifications." cmd="cmd_notification_stop"/>
+    </token>
     <token name="show" help="Show information about the system." pipe="true">
       <token name="running" argument="configuration" help="Show running configuration." cmd="cmd_show_config">
         <token name="format" help="Configuration format.">
@@ -33,6 +37,9 @@
       <token name="yang" help="YANG information.">
         <token name="modules" help="Show loaded YANG modules." cmd="cmd_show_yang_modules"/>
       </token>
+      <token name="notifications" help="Show buffered notifications." cmd="cmd_show_notifications">
+        <token name="json" argument="format" help="Raw JSON output format." cmd="cmd_show_notifications"/>
+      </token>
       <!-- IS-IS show commands -->
       <token name="isis" argument="protocol" help="IS-IS information">
         <token name="interface" help="Interface information" cmd="cmd_show_isis_interface">

src/main.rs

@@ -7,6 +7,7 @@
 mod error;
 mod grpc;
 mod internal_commands;
+mod notifications;
 mod parser;
 mod pipe;
 mod session;
@@ -41,13 +42,17 @@ pub struct Cli {
 // ===== impl Cli =====
 
 impl Cli {
-    fn new(use_pager: bool, grpc_client: GrpcClient) -> Cli {
+    fn new(
+        use_pager: bool,
+        grpc_client: GrpcClient,
+        grpc_addr: &'static str,
+    ) -> Cli {
         // Generate commands.
         let mut commands = Commands::new();
         commands.gen_cmds();
 
         // Create CLI session.
-        let session = Session::new(use_pager, grpc_client);
+        let session = Session::new(use_pager, grpc_client, grpc_addr);
 
         Cli { commands, session }
     }
@@ -266,7 +271,7 @@ fn main() {
     // Initialize CLI master structure.
     let use_pager = matches.values_of("command").is_none()
         && !matches.is_present("no-pager");
-    let mut cli = Cli::new(use_pager, grpc_client);
+    let mut cli = Cli::new(use_pager, grpc_client, grpc_addr);
 
     // Read configuration file.
     if let Some(path) = matches.value_of("file") {

src/notifications.rs

@@ -0,0 +1,174 @@
+//
+// Copyright (c) The Holo Core Contributors
+//
+// SPDX-License-Identifier: MIT
+//
+
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex};
+
+use crate::grpc::proto;
+use crate::grpc::proto::northbound_client::NorthboundClient;
+
+// Maximum number of notifications retained in the circular buffer.
+pub const NOTIFICATION_BUFFER_SIZE: usize = 1024;
+
+#[derive(Debug)]
+pub struct NotificationEntry {
+    // Timestamp in seconds since Epoch.
+    pub timestamp: i64,
+    // The YANG notification data path.
+    pub module_path: String,
+    // The notification data, JSON-encoded.
+    pub data_json: String,
+}
+
+#[derive(Debug, Default)]
+pub struct NotificationBuffer {
+    entries: VecDeque<NotificationEntry>,
+    // Reason the receiver stopped on its own, if it did.
+    status: Option<String>,
+}
+
+// ===== impl NotificationBuffer =====
+
+impl NotificationBuffer {
+    pub fn push(&mut self, entry: NotificationEntry) {
+        if self.entries.len() == NOTIFICATION_BUFFER_SIZE {
+            self.entries.pop_front();
+        }
+        self.entries.push_back(entry);
+    }
+
+    pub fn entries(&self) -> impl Iterator<Item = &NotificationEntry> {
+        self.entries.iter()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.entries.is_empty()
+    }
+
+    pub fn status(&self) -> Option<&str> {
+        self.status.as_deref()
+    }
+
+    pub fn set_status(&mut self, status: Option<String>) {
+        self.status = status;
+    }
+}
+
+#[derive(Debug)]
+pub struct Subscription {
+    shutdown_tx: tokio::sync::oneshot::Sender<()>,
+    handle: std::thread::JoinHandle<()>,
+}
+
+// ===== impl Subscription =====
+
+impl Subscription {
+    // Connects to holod, subscribes to all notifications and spawns a
+    // background thread that pushes received notifications into the given
+    // buffer.
+    pub fn start(
+        addr: &'static str,
+        buffer: Arc<Mutex<NotificationBuffer>>,
+    ) -> Result<Subscription, String> {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .map_err(|error| format!("failed to create runtime: {}", error))?;
+
+        // Connect and subscribe synchronously so failures are reported
+        // immediately, before any thread is spawned.
+        let (client, stream) = runtime.block_on(async {
+            let mut client = NorthboundClient::connect(addr)
+                .await
+                .map_err(|error| {
+                    format!("connection to holod failed: {}", error)
+                })?
+                .max_decoding_message_size(usize::MAX);
+            let stream = client
+                .subscribe(proto::SubscribeRequest {
+                    encoding: proto::Encoding::Json as i32,
+                    path: String::new(),
+                })
+                .await
+                .map_err(|error| {
+                    format!("subscribe request failed: {}", error)
+                })?
+                .into_inner();
+            Ok::<_, String>((client, stream))
+        })?;
+
+        let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
+        let handle = std::thread::spawn(move || {
+            // The client and the stream must be dropped before the runtime
+            // to avoid a deadlock (see the comment on `GrpcClient`), hence
+            // both are moved into the async block driven by `block_on`.
+            runtime.block_on(async move {
+                let _client = client;
+                let mut stream = stream;
+                loop {
+                    tokio::select! {
+                        _ = &mut shutdown_rx => break,
+                        message = stream.message() => {
+                            let mut buffer = buffer
+                                .lock()
+                                .unwrap_or_else(|error| error.into_inner());
+                            match message {
+                                Ok(Some(notification)) => {
+                                    buffer.push(notification.into());
+                                }
+                                Ok(None) => {
+                                    buffer.set_status(Some(
+                                        "stream closed by server".to_owned(),
+                                    ));
+                                    break;
+                                }
+                                Err(error) => {
+                                    buffer.set_status(Some(format!(
+                                        "stream error: {}",
+                                        error
+                                    )));
+                                    break;
+                                }
+                            }
+                        }
+                    }
+                }
+            });
+        });
+
+        Ok(Subscription {
+            shutdown_tx,
+            handle,
+        })
+    }
+
+    // Returns whether the receiver thread has terminated on its own.
+    pub fn is_finished(&self) -> bool {
+        self.handle.is_finished()
+    }
+
+    // Signals the receiver thread to stop and waits for it to finish.
+    pub fn stop(self) {
+        let _ = self.shutdown_tx.send(());
+        let _ = self.handle.join();
+    }
+}
+
+// ===== impl NotificationEntry =====
+
+impl From<proto::Notification> for NotificationEntry {
+    fn from(notification: proto::Notification) -> NotificationEntry {
+        let data_json = match notification.data.and_then(|data| data.data) {
+            Some(proto::data_tree::Data::DataString(string)) => string,
+            _ => "<no data>".to_owned(),
+        };
+        NotificationEntry {
+            timestamp: notification.timestamp,
+            module_path: notification.module_path,
+            data_json,
+        }
+    }
+}