mgmtd: add notify selectors to filter datastore notifications

- Additionally push the selectors down to the backends

Signed-off-by: Christian Hopps <chopps@labn.net>

lib/mgmt_be_client.c

@@ -322,8 +322,6 @@ static int __send_notification(struct mgmt_be_client *client, const char *xpath,
 	LY_ERR err;
 	int ret = 0;
 
-	assert(tree);
-
 	debug_be_client("%s: sending %sYANG %snotification: %s", __func__,
 			op == NOTIFY_OP_DS_DELETE    ? "delete "
 			: op == NOTIFY_OP_DS_REPLACE ? "replace "
@@ -1154,6 +1152,22 @@ static void be_client_handle_notify(struct mgmt_be_client *client, void *msgbuf,
 	lyd_free_all(dnode);
 }
 
+/*
+ * Process a notify select msg
+ */
+static void be_client_handle_notify_select(struct mgmt_be_client *client, void *msgbuf,
+					   size_t msg_len)
+{
+	struct mgmt_msg_notify_select *msg = msgbuf;
+	const char **selectors = NULL;
+
+	debug_be_client("Received notify-select for client %s", client->name);
+
+	if (msg_len >= sizeof(*msg))
+		selectors = mgmt_msg_native_strings_decode(msg, msg_len, msg->selectors);
+	nb_notif_set_filters(selectors, msg->replace);
+}
+
 /*
  * Handle a native encoded message
  *
@@ -1175,6 +1189,9 @@ static void be_client_handle_native_msg(struct mgmt_be_client *client,
 	case MGMT_MSG_CODE_NOTIFY:
 		be_client_handle_notify(client, msg, msg_len);
 		break;
+	case MGMT_MSG_CODE_NOTIFY_SELECT:
+		be_client_handle_notify_select(client, msg, msg_len);
+		break;
 	default:
 		log_err_be_client("unknown native message txn-id %" PRIu64
 				  " req-id %" PRIu64 " code %u to client %s",

lib/mgmt_msg_native.c

@@ -14,7 +14,8 @@ DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_ERROR, "native error msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_GET_TREE, "native get tree msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_TREE_DATA, "native tree data msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_GET_DATA, "native get data msg");
-DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_NOTIFY, "native get data msg");
+DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_NOTIFY, "native notify msg");
+DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_NOTIFY_SELECT, "native notify select msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_EDIT, "native edit msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_EDIT_REPLY, "native edit reply msg");
 DEFINE_MTYPE(MSG_NATIVE, MSG_NATIVE_RPC, "native RPC msg");

lib/mgmt_msg_native.h

@@ -159,6 +159,7 @@ DECLARE_MTYPE(MSG_NATIVE_GET_TREE);
 DECLARE_MTYPE(MSG_NATIVE_TREE_DATA);
 DECLARE_MTYPE(MSG_NATIVE_GET_DATA);
 DECLARE_MTYPE(MSG_NATIVE_NOTIFY);
+DECLARE_MTYPE(MSG_NATIVE_NOTIFY_SELECT);
 DECLARE_MTYPE(MSG_NATIVE_EDIT);
 DECLARE_MTYPE(MSG_NATIVE_EDIT_REPLY);
 DECLARE_MTYPE(MSG_NATIVE_RPC);

lib/northbound.h

@@ -1815,6 +1815,15 @@ extern struct lyd_node *nb_op_updatef(struct lyd_node *tree, const char *path, c
 extern struct lyd_node *nb_op_vupdatef(struct lyd_node *tree, const char *path, const char *val_fmt,
 				       va_list ap);
 
+/**
+ * nb_notif_set_filters() - add or replace notification filters
+ * @selectors: darr array of selector (filter) xpath strings, can be NULL if
+ *	       @replace is true. nb_notif_set_filters takes ownership of this
+ *	       array and the contained darr strings.
+ * @replace: true to replace existing set otherwise append.
+ */
+extern void nb_notif_set_filters(const char **selectors, bool replace);
+
 extern void nb_notif_init(struct event_loop *loop);
 extern void nb_notif_terminate(void);
 

lib/northbound_notif.c

@@ -64,6 +64,8 @@ struct event_loop *nb_notif_master;
 struct event *nb_notif_timer;
 void *nb_notif_walk;
 
+const char **nb_notif_filters;
+
 /*
  * We maintain a queue of change lists one entry per query and notification send
  * action
@@ -635,6 +637,20 @@ static void nb_notif_set_walk_timer(void)
 			     &nb_notif_timer);
 }
 
+void nb_notif_set_filters(const char **selectors, bool replace)
+{
+	const char **csp;
+
+	if (replace) {
+		darr_free_free(nb_notif_filters);
+		nb_notif_filters = selectors;
+		return;
+	}
+	darr_foreach_p (selectors, csp)
+		*darr_append(nb_notif_filters) = *csp;
+	darr_free(selectors);
+}
+
 void nb_notif_init(struct event_loop *tm)
 {
 	nb_notif_master = tm;
@@ -659,4 +675,6 @@ void nb_notif_terminate(void)
 
 	while ((group = op_changes_group_next()))
 		op_changes_group_free(group);
+
+	darr_free_free(nb_notif_filters);
 }

mgmtd/mgmt_be_adapter.c

@@ -320,7 +320,7 @@ static void mgmt_be_xpath_map_init(void)
 
 	__dbg("Total Cfg XPath Maps: %u", darr_len(be_cfg_xpath_map));
 	__dbg("Total Oper XPath Maps: %u", darr_len(be_oper_xpath_map));
-	__dbg("Total Noitf XPath Maps: %u", darr_len(be_notif_xpath_map));
+	__dbg("Total Notif XPath Maps: %u", darr_len(be_notif_xpath_map));
 	__dbg("Total RPC XPath Maps: %u", darr_len(be_rpc_xpath_map));
 }
 
@@ -651,13 +651,17 @@ int mgmt_be_send_native(enum mgmt_be_client_id id, void *msg)
 	return mgmt_msg_native_send_msg(adapter->conn, msg, false);
 }
 
+/*
+ * Send notification to back-ends that subscribed for them.
+ */
 static void mgmt_be_adapter_send_notify(struct mgmt_msg_notify_data *msg,
 					size_t msglen)
 {
 	struct mgmt_be_client_adapter *adapter;
 	struct mgmt_be_xpath_map *map;
-	struct nb_node *nb_node;
+	struct nb_node *nb_node = NULL;
 	const char *notif;
+	bool is_root;
 	uint id, len;
 
 	if (!darr_len(be_notif_xpath_map))
@@ -669,28 +673,34 @@ static void mgmt_be_adapter_send_notify(struct mgmt_msg_notify_data *msg,
 		return;
 	}
 
-	nb_node = nb_node_find(notif);
+	is_root = !strcmp(notif, "/");
-	if (!nb_node) {
+	if (!is_root) {
-		__log_err("No schema found for notification: %s", notif);
+		nb_node = nb_node_find(notif);
-		return;
+		if (!nb_node) {
+			__log_err("No schema found for notification: %s", notif);
+			return;
+		}
 	}
 
 	darr_foreach_p (be_notif_xpath_map, map) {
-		len = strlen(map->xpath_prefix);
+		if (!is_root) {
-		if (strncmp(map->xpath_prefix, nb_node->xpath, len) &&
+			len = strlen(map->xpath_prefix);
-		    strncmp(map->xpath_prefix, notif, len))
+			if (strncmp(map->xpath_prefix, nb_node->xpath, len) &&
-			continue;
+			    strncmp(map->xpath_prefix, notif, len))
-
+				continue;
+		}
 		FOREACH_BE_CLIENT_BITS (id, map->clients) {
 			adapter = mgmt_be_get_adapter_by_id(id);
 			if (!adapter)
 				continue;
+
 			msg_conn_send_msg(adapter->conn, MGMT_MSG_VERSION_NATIVE,
 					  msg, msglen, NULL, false);
 		}
 	}
 }
 
+
 /*
  * Handle a native encoded message
  */
@@ -735,6 +745,9 @@ static void be_adapter_handle_native_msg(struct mgmt_be_client_adapter *adapter,
 		mgmt_txn_notify_rpc_reply(adapter, rpc_msg, msg_len);
 		break;
 	case MGMT_MSG_CODE_NOTIFY:
+		/*
+		 * Handle notify message from a back-end client
+		 */
 		notify_msg = (typeof(notify_msg))msg;
 		__dbg("Got NOTIFY from '%s'", adapter->name);
 		mgmt_be_adapter_send_notify(notify_msg, msg_len);

mgmtd/mgmt_fe_adapter.c

@@ -9,6 +9,7 @@
 
 #include <zebra.h>
 #include "darr.h"
+#include "frrstr.h"
 #include "sockopt.h"
 #include "network.h"
 #include "libfrr.h"
@@ -31,6 +32,7 @@
 #define FOREACH_ADAPTER_IN_LIST(adapter)                                       \
 	frr_each_safe (mgmt_fe_adapters, &mgmt_fe_adapters, (adapter))
 
+
 enum mgmt_session_event {
 	MGMTD_FE_SESSION_CFG_TXN_CLNUP = 1,
 	MGMTD_FE_SESSION_SHOW_TXN_CLNUP,
@@ -55,6 +57,22 @@ DECLARE_LIST(mgmt_fe_sessions, struct mgmt_fe_session_ctx, list_linkage);
 #define FOREACH_SESSION_IN_LIST(adapter, session)                              \
 	frr_each_safe (mgmt_fe_sessions, &(adapter)->fe_sessions, (session))
 
+/*
+ * A tree for storing unique notify-select strings.
+ */
+PREDECL_RBTREE_UNIQ(ns_string);
+struct ns_string {
+	struct ns_string_item link;
+	struct list *sessions;
+	char s[];
+};
+static uint32_t ns_string_compare(const struct ns_string *ns1, const struct ns_string *ns2);
+DECLARE_RBTREE_UNIQ(ns_string, struct ns_string, link, ns_string_compare);
+
+/* ---------------- */
+/* Global variables */
+/* ---------------- */
+
 static struct event_loop *mgmt_loop;
 static struct msg_server mgmt_fe_server = {.fd = -1};
 
@@ -63,6 +81,72 @@ static struct mgmt_fe_adapters_head mgmt_fe_adapters;
 static struct hash *mgmt_fe_sessions;
 static uint64_t mgmt_fe_next_session_id;
 
+static struct ns_string_head mgmt_fe_ns_strings;
+
+/* ------------------------------ */
+/* Notify select string functions */
+/* ------------------------------ */
+
+static uint32_t ns_string_compare(const struct ns_string *ns1, const struct ns_string *ns2)
+{
+	return strcmp(ns1->s, ns2->s);
+}
+
+static void mgmt_fe_free_ns_string(struct ns_string *ns)
+{
+	list_delete(&ns->sessions);
+	XFREE(MTYPE_MGMTD_XPATH, ns);
+}
+
+static void mgmt_fe_free_ns_strings(struct ns_string_head *head)
+{
+	struct ns_string *ns;
+
+	while ((ns = ns_string_pop(head)))
+		mgmt_fe_free_ns_string(ns);
+	ns_string_fini(head);
+}
+
+static void mgmt_fe_ns_string_remove_session(struct ns_string_head *head,
+					     struct mgmt_fe_session_ctx *session)
+{
+	struct ns_string *ns;
+
+	frr_each_safe (ns_string, head, ns) {
+		listnode_delete(ns->sessions, session);
+		if (list_isempty(ns->sessions)) {
+			ns_string_del(head, ns);
+			mgmt_fe_free_ns_string(ns);
+		}
+	}
+}
+
+static void mgmt_fe_add_ns_string(struct ns_string_head *head, const char *path, size_t plen,
+				  struct mgmt_fe_session_ctx *session)
+{
+	struct ns_string *e, *ns;
+
+	ns = XCALLOC(MTYPE_MGMTD_XPATH, sizeof(*ns) + plen + 1);
+	strlcpy(ns->s, path, plen + 1);
+	e = ns_string_add(head, ns);
+	if (!e)
+		ns->sessions = list_new();
+	if (!listnode_lookup(ns->sessions, session))
+		listnode_add(ns->sessions, session);
+}
+
+char **mgmt_fe_get_all_selectors(void)
+{
+	char **selectors = NULL;
+	struct ns_string *ns;
+
+	frr_each (ns_string, &mgmt_fe_ns_strings, ns)
+		*darr_append(selectors) = darr_strdup(ns->s);
+
+	return selectors;
+}
+
+
 /* Forward declarations */
 static void
 mgmt_fe_session_register_event(struct mgmt_fe_session_ctx *session,
@@ -190,6 +274,7 @@ static void mgmt_fe_cleanup_session(struct mgmt_fe_session_ctx **sessionp)
 		assert(session->adapter->refcount > 1);
 		mgmt_fe_adapter_unlock(&session->adapter);
 	}
+	mgmt_fe_ns_string_remove_session(&mgmt_fe_ns_strings, session);
 	darr_free_free(session->notify_xpaths);
 	hash_release(mgmt_fe_sessions, session);
 	XFREE(MTYPE_MGMTD_FE_SESSION, session);
@@ -1542,32 +1627,90 @@ static void fe_adapter_handle_edit(struct mgmt_fe_session_ctx *session,
  * @__msg: the message data.
  * @msg_len: the length of the message data.
  */
-static void fe_adapter_handle_notify_select(struct mgmt_fe_session_ctx *session,
+static void fe_adapter_handle_notify_select(struct mgmt_fe_session_ctx *session, void *__msg,
-					    void *__msg, size_t msg_len)
+					    size_t msg_len)
 {
 	struct mgmt_msg_notify_select *msg = __msg;
 	uint64_t req_id = msg->req_id;
 	const char **selectors = NULL;
 	const char **new;
+	const char **sp;
+	char *selstr = NULL;
+	uint64_t clients = 0;
+	uint ret;
 
 	if (msg_len >= sizeof(*msg)) {
-		selectors = mgmt_msg_native_strings_decode(msg, msg_len,
+		selectors = mgmt_msg_native_strings_decode(msg, msg_len, msg->selectors);
-							   msg->selectors);
 		if (!selectors) {
-			fe_adapter_send_error(session, req_id, false, -EINVAL,
+			fe_adapter_send_error(session, req_id, false, -EINVAL, "Invalid message");
-					      "Invalid message");
 			return;
 		}
 	}
+	if (DEBUG_MODE_CHECK(&mgmt_debug_fe, DEBUG_MODE_ALL)) {
+		selstr = frrstr_join(selectors, darr_len(selectors), ", ");
+		if (!selstr)
+			selstr = XSTRDUP(MTYPE_TMP, "");
+	}
+
 	if (msg->replace) {
+		mgmt_fe_ns_string_remove_session(&mgmt_fe_ns_strings, session);
+		// [ ] Keep a local tree to optimize sending selectors to BE?
+		// [*] Or just KISS and fanout the original message to BEs?
+		// mgmt_remove_add_notify_selectors(session->notify_xpaths, selectors);
 		darr_free_free(session->notify_xpaths);
 		session->notify_xpaths = selectors;
 	} else if (selectors) {
-		new = darr_append_nz(session->notify_xpaths,
+		// [ ] Keep a local tree to optimize sending selectors to BE?
-				     darr_len(selectors));
+		// [*] Or just KISS and fanout the original message to BEs?
+		// mgmt_remove_add_notify_selectors(session->notify_xpaths, selectors);
+		new = darr_append_nz(session->notify_xpaths, darr_len(selectors));
 		memcpy(new, selectors, darr_len(selectors) * sizeof(*selectors));
-		darr_free(selectors);
+	} else {
+		__log_err("Invalid msg from session-id: %Lu: no selectors present in non-replace msg",
+			  session->session_id);
+		darr_free_free(selectors);
+		selectors = NULL;
+		goto done;
 	}
+
+
+	if (session->notify_xpaths && DEBUG_MODE_CHECK(&mgmt_debug_fe, DEBUG_MODE_ALL)) {
+		const char **sel = session->notify_xpaths;
+		char *s = frrstr_join(sel, darr_len(sel), ", ");
+		__dbg("New NOTIF %d selectors '%s' (replace: %d) txn-id: %Lu for session-id: %Lu",
+		      darr_len(sel), s, msg->replace, session->cfg_txn_id, session->session_id);
+		XFREE(MTYPE_TMP, s);
+	}
+
+	/* Add the new selectors to the global tree */
+	darr_foreach_p (selectors, sp)
+		mgmt_fe_add_ns_string(&mgmt_fe_ns_strings, *sp, darr_strlen(*sp), session);
+
+	/* Check if any backends are interested in the new selectors. */
+	if (msg->replace) {
+		/* If we are replacing we'll send all the selectors again with replace flag */
+		clients = mgmt_be_interested_clients("/", MGMT_BE_XPATH_SUBSCR_TYPE_OPER);
+	} else {
+		darr_foreach_p (selectors, sp)
+			clients |= mgmt_be_interested_clients(*sp, MGMT_BE_XPATH_SUBSCR_TYPE_OPER);
+	}
+	if (!clients) {
+		__dbg("No backends provide oper for notify selectors: '%s' txn-id %Lu session-id: %Lu",
+		      selstr, session->txn_id, session->session_id);
+		goto done;
+	}
+
+	/* We don't use a transaction for this, just send the message */
+	ret = mgmt_txn_send_notify_selectors(req_id, clients, msg->replace ? NULL : selectors);
+	if (ret) {
+		fe_adapter_send_error(session, req_id, false, -EINPROGRESS,
+				      "Failed to create a NOTIFY_SELECT transaction");
+	}
+done:
+	if (session->notify_xpaths != selectors)
+		darr_free(selectors);
+	if (selstr)
+		XFREE(MTYPE_TMP, selstr);
 }
 
 /**
@@ -1758,10 +1901,11 @@ void mgmt_fe_adapter_send_notify(struct mgmt_msg_notify_data *msg, size_t msglen
 {
 	struct mgmt_fe_client_adapter *adapter;
 	struct mgmt_fe_session_ctx *session;
-	struct nb_node *nb_node;
+	struct nb_node *nb_node = NULL;
-	const char **xpath_prefix;
+	struct listnode *node;
+	struct ns_string *ns;
 	const char *notif;
-	bool sendit;
+	bool is_root;
 	uint len;
 
 	assert(msg->refer_id == 0);
@@ -1772,36 +1916,48 @@ void mgmt_fe_adapter_send_notify(struct mgmt_msg_notify_data *msg, size_t msglen
 		return;
 	}
 
-	/*
+	is_root = !strcmp(notif, "/");
-	 * We need the nb_node to obtain a path which does not include any
+	if (!is_root) {
-	 * specific list entry selectors
+		/*
-	 */
+		 * We need the nb_node to obtain a path which does not include any
-	nb_node = nb_node_find(notif);
+		 * specific list entry selectors
-	if (!nb_node) {
+		 */
-		__log_err("No schema found for notification: %s", notif);
+		nb_node = nb_node_find(notif);
-		return;
+		if (!nb_node) {
+			__log_err("No schema found for notification: %s", notif);
+			return;
+		}
 	}
 
-	FOREACH_ADAPTER_IN_LIST (adapter) {
+	frr_each (ns_string, &mgmt_fe_ns_strings, ns) {
-		FOREACH_SESSION_IN_LIST (adapter, session) {
+		if (!is_root) {
-			/* If no selectors then always send */
+			len = strlen(ns->s);
-			sendit = !session->notify_xpaths;
+			if (strncmp(ns->s, notif, len) && strncmp(ns->s, nb_node->xpath, len))
-			darr_foreach_p (session->notify_xpaths, xpath_prefix) {
+				continue;
-				len = strlen(*xpath_prefix);
+		}
-				if (!strncmp(*xpath_prefix, notif, len) ||
+		for (ALL_LIST_ELEMENTS_RO(ns->sessions, node, session)) {
-				    !strncmp(*xpath_prefix, nb_node->xpath,
+			msg->refer_id = session->session_id;
-					     len)) {
+			(void)fe_adapter_send_native_msg(session->adapter, msg, msglen, false);
-					sendit = true;
+		}
-					break;
+	}
-				}
+
-			}
+	/*
-			if (sendit) {
+	 * Send all YANG defined notifications to all sesisons with *no*
+	 * selectors as well (i.e., original NETCONF/RESTCONF notification
+	 * scheme).
+	 */
+	if (!is_root && CHECK_FLAG(nb_node->snode->nodetype, LYS_NOTIF)) {
+		FOREACH_ADAPTER_IN_LIST (adapter) {
+			FOREACH_SESSION_IN_LIST (adapter, session) {
+				if (session->notify_xpaths)
+					continue;
 				msg->refer_id = session->session_id;
 				(void)fe_adapter_send_native_msg(adapter, msg,
 								 msglen, false);
 			}
 		}
 	}
+
 	msg->refer_id = 0;
 }
 
@@ -1810,9 +1966,10 @@ void mgmt_fe_adapter_lock(struct mgmt_fe_client_adapter *adapter)
 	adapter->refcount++;
 }
 
-extern void mgmt_fe_adapter_unlock(struct mgmt_fe_client_adapter **adapter)
+void mgmt_fe_adapter_unlock(struct mgmt_fe_client_adapter **adapter)
 {
 	struct mgmt_fe_client_adapter *a = *adapter;
+
 	assert(a && a->refcount);
 
 	if (!--a->refcount) {
@@ -1840,6 +1997,8 @@ void mgmt_fe_adapter_init(struct event_loop *tm)
 		hash_create(mgmt_fe_session_hash_key, mgmt_fe_session_hash_cmp,
 			    "MGMT Frontend Sessions");
 
+	ns_string_init(&mgmt_fe_ns_strings);
+
 	snprintf(server_path, sizeof(server_path), MGMTD_FE_SOCK_NAME);
 
 	if (msg_server_init(&mgmt_fe_server, server_path, tm,
@@ -1869,10 +2028,13 @@ void mgmt_fe_adapter_destroy(void)
 
 	msg_server_cleanup(&mgmt_fe_server);
 
+
 	/* Deleting the adapters will delete all the sessions */
 	FOREACH_ADAPTER_IN_LIST (adapter)
 		mgmt_fe_adapter_delete(adapter);
 
+	mgmt_fe_free_ns_strings(&mgmt_fe_ns_strings);
+
 	hash_clean_and_free(&mgmt_fe_sessions, mgmt_fe_abort_if_session);
 }
 
@@ -1885,8 +2047,7 @@ struct msg_conn *mgmt_fe_create_adapter(int conn_fd, union sockunion *from)
 
 	adapter = mgmt_fe_find_adapter_by_fd(conn_fd);
 	if (!adapter) {
-		adapter = XCALLOC(MTYPE_MGMTD_FE_ADPATER,
+		adapter = XCALLOC(MTYPE_MGMTD_FE_ADPATER, sizeof(struct mgmt_fe_client_adapter));
-				sizeof(struct mgmt_fe_client_adapter));
 		snprintf(adapter->name, sizeof(adapter->name), "Unknown-FD-%d",
 			 conn_fd);
 

mgmtd/mgmt_fe_adapter.h

@@ -225,6 +225,13 @@ extern int mgmt_fe_adapter_txn_error(uint64_t txn_id, uint64_t req_id,
 				     const char *errstr);
 
 
+/**
+ * mgmt_fe_get_all_selectors() - Get all selectors for all frontend adapters.
+ *
+ * Returns: A darr array of all selectors for all frontend adapters.
+ */
+extern char **mgmt_fe_get_all_selectors(void);
+
 /* Fetch frontend client session set-config stats */
 extern struct mgmt_setcfg_stats *
 mgmt_fe_get_session_setcfg_stats(uint64_t session_id);

mgmtd/mgmt_txn.c

@@ -237,6 +237,7 @@ struct mgmt_txn_ctx {
 	struct event *clnup;
 
 	/* List of backend adapters involved in this transaction */
+	/* XXX reap this */
 	struct mgmt_txn_badapters_head be_adapters;
 
 	int refcount;
@@ -2651,6 +2652,52 @@ int mgmt_txn_send_rpc(uint64_t txn_id, uint64_t req_id, uint64_t clients,
 	return 0;
 }
 
+int mgmt_txn_send_notify_selectors(uint64_t req_id, uint64_t clients, const char **selectors)
+{
+	struct mgmt_msg_notify_select *msg;
+	char **all_selectors = NULL;
+	uint64_t id;
+	int ret;
+	uint i;
+
+	msg = mgmt_msg_native_alloc_msg(struct mgmt_msg_notify_select, 0,
+					MTYPE_MSG_NATIVE_NOTIFY_SELECT);
+	msg->refer_id = MGMTD_TXN_ID_NONE;
+	msg->req_id = req_id;
+	msg->code = MGMT_MSG_CODE_NOTIFY_SELECT;
+	msg->replace = selectors == NULL;
+
+	if (selectors == NULL) {
+		/* Get selectors for all sessions */
+		all_selectors = mgmt_fe_get_all_selectors();
+		selectors = (const char **)all_selectors;
+	}
+
+	darr_foreach_i (selectors, i)
+		mgmt_msg_native_add_str(msg, selectors[i]);
+
+	assert(clients);
+	FOREACH_BE_CLIENT_BITS (id, clients) {
+		/* make sure the backend is running/connected */
+		if (!mgmt_be_get_adapter_by_id(id))
+			continue;
+		ret = mgmt_be_send_native(id, msg);
+		if (ret) {
+			__log_err("Could not send notify-select message to backend client %s",
+				  mgmt_be_client_id2name(id));
+			continue;
+		}
+
+		__dbg("Sent notify-select req to backend client %s", mgmt_be_client_id2name(id));
+	}
+	mgmt_msg_native_free_msg(msg);
+
+	if (all_selectors)
+		darr_free_free(all_selectors);
+
+	return 0;
+}
+
 /*
  * Error reply from the backend client.
  */

mgmtd/mgmt_txn.h

@@ -297,6 +297,16 @@ extern int mgmt_txn_send_rpc(uint64_t txn_id, uint64_t req_id, uint64_t clients,
 			     LYD_FORMAT result_type, const char *xpath,
 			     const char *data, size_t data_len);
 
+/**
+ * mgmt_txn_send_notify_selectors() - Send NOTIFY SELECT request.
+ * @req_id: FE client request identifier.
+ * @clients: Bitmask of clients to send RPC to.
+ * @selectors: Array of selectors or NULL to resend all selectors to BE clients.
+ *
+ * Returns 0 on success.
+ */
+extern int mgmt_txn_send_notify_selectors(uint64_t req_id, uint64_t clients, const char **selectors);
+
 /*
  * Notifiy backend adapter on connection.
  */