* Major rewrite of event streams

* If you used old event callbacks API, you need to switch to the streams API
    * See clixon_stream.[ch]
  * Old streams API which needs to be removed include:
    * clicon_log_register_callback()
    * subscription_add() --> stream_register()
    * backend_notify() and backend_notify_xml() - use stream_notify() instead
* Example uses "NETCONF" stream instead of "ROUTING"
* Added timeout option -t for clixon_netconf - quit after max time.
This commit is contained in:
Olof hagsand 2018-09-30 14:51:30 +02:00
parent d7fbe75c9e
commit 98f3cd0e32
31 changed files with 597 additions and 635 deletions

View file

@ -91,7 +91,6 @@ struct backend_handle {
/* ------ end of common handle ------ */
struct client_entry *bh_ce_list; /* The client list */
int bh_ce_nr; /* Number of clients, just increment */
struct handle_subscription *bh_subscription; /* Event subscription list */
cxobj *bh_nacm; /* NACM external struct */
};
@ -121,155 +120,6 @@ backend_handle_exit(clicon_handle h)
return 0;
}
/*! Notify event and distribute to all registered clients
*
* @param[in] h Clicon handle
* @param[in] stream Name of event stream. CLICON is predefined as LOG stream
* @param[in] level Event level (not used yet)
* @param[in] event Actual message as text format
*
* Stream is a string used to qualify the event-stream. Distribute the
* event to all clients registered to this backend.
* XXX: event-log NYI.
* @see also subscription_add()
* @see also backend_notify_xml()
*/
int
backend_notify(clicon_handle h,
char *stream,
int level,
char *event)
{
struct client_entry *ce;
struct client_entry *ce_next;
struct client_subscription *su;
struct handle_subscription *hs;
int retval = -1;
clicon_debug(2, "%s %s", __FUNCTION__, stream);
/* First thru all clients(sessions), and all subscriptions and find matches */
for (ce = backend_client_list(h); ce; ce = ce_next){
ce_next = ce->ce_next;
for (su = ce->ce_subscription; su; su = su->su_next)
if (strcmp(su->su_stream, stream) == 0){
if (strlen(su->su_filter)==0 || fnmatch(su->su_filter, event, 0) == 0){
if (send_msg_notify(ce->ce_s, level, event) < 0){
if (errno == ECONNRESET || errno == EPIPE){
clicon_log(LOG_WARNING, "client %d reset", ce->ce_nr);
#if 0
/* We should remove here but removal is not possible
from a client since backend_client is not linked.
Maybe we should add it to the plugin, but it feels
"safe" that you cant remove a client.
Instead, the client is (hopefully) removed elsewhere?
*/
backend_client_rm(h, ce);
#endif
break;
}
goto done;
}
}
}
}
/* Then go thru all global (handle) subscriptions and find matches */
hs = NULL;
while ((hs = subscription_each(h, hs)) != NULL){
if (hs->hs_format != FORMAT_TEXT)
continue;
if (strcmp(hs->hs_stream, stream))
continue;
if (hs->hs_filter==NULL ||
strlen(hs->hs_filter)==0 ||
fnmatch(hs->hs_filter, event, 0) == 0)
if ((*hs->hs_fn)(h, event, hs->hs_arg) < 0)
goto done;
}
retval = 0;
done:
return retval;
}
/*! Notify event and distribute to all registered clients
*
* @param[in] h Clicon handle
* @param[in] stream Name of event stream. CLICON is predefined as LOG stream
* @param[in] level Event level (not used yet)
* @param[in] x Actual message as xml tree
*
* Stream is a string used to qualify the event-stream. Distribute the
* event to all clients registered to this backend.
* XXX: event-log NYI.
* @see also subscription_add()
* @see also backend_notify()
*/
int
backend_notify_xml(clicon_handle h,
char *stream,
int level,
cxobj *x)
{
struct client_entry *ce;
struct client_entry *ce_next;
struct client_subscription *su;
int retval = -1;
cbuf *cb = NULL;
struct handle_subscription *hs;
clicon_debug(1, "%s %s", __FUNCTION__, stream);
/* Now go thru all clients(sessions), and all subscriptions and find matches */
for (ce = backend_client_list(h); ce; ce = ce_next){
ce_next = ce->ce_next;
for (su = ce->ce_subscription; su; su = su->su_next)
if (strcmp(su->su_stream, stream) == 0){
if (strlen(su->su_filter)==0 || xpath_first(x, "%s", su->su_filter) != NULL){
if (cb==NULL){
if ((cb = cbuf_new()) == NULL){
clicon_err(OE_PLUGIN, errno, "cbuf_new");
goto done;
}
if (clicon_xml2cbuf(cb, x, 0, 0) < 0)
goto done;
}
if (send_msg_notify(ce->ce_s, level, cbuf_get(cb)) < 0){
if (errno == ECONNRESET || errno == EPIPE){
clicon_log(LOG_WARNING, "client %d reset", ce->ce_nr);
#if 0
/* We should remove here but removal is not possible
from a client since backend_client is not linked.
Maybe we should add it to the plugin, but it feels
"safe" that you cant remove a client.
Instead, the client is (hopefully) removed elsewhere?
*/
backend_client_rm(h, ce);
#endif
break;
}
goto done;
}
}
}
}
/* Then go thru all global (handle) subscriptions and find matches */
hs = NULL;
while ((hs = subscription_each(h, hs)) != NULL){
if (hs->hs_format != FORMAT_XML)
continue;
if (strcmp(hs->hs_stream, stream))
continue;
if (strlen(hs->hs_filter)==0 || xpath_first(x, "%s", hs->hs_filter) != NULL){
if ((*hs->hs_fn)(h, x, hs->hs_arg) < 0)
goto done;
}
}
retval = 0;
done:
if (cb)
cbuf_free(cb);
return retval;
}
/*! Add new client, typically frontend such as cli, netconf, restconf
* @param[in] h Clicon handle
* @param[in] addr Address of client
@ -332,111 +182,6 @@ backend_client_delete(clicon_handle h,
return 0;
}
/*! Add subscription given stream name, callback and argument
* @param[in] h Clicon handle
* @param[in] stream Name of event stream
* @param[in] format Expected format of event, eg text or xml
* @param[in] filter Filter to match event, depends on format, eg xpath for xml
* @param[in] fn Callback when event occurs
* @param[in] arg Argument to use with callback. Also handle when deleting
* Note that arg is not a real handle.
* @see subscription_delete
* @see subscription_each
*/
struct handle_subscription *
subscription_add(clicon_handle h,
char *stream,
enum format_enum format,
char *filter,
subscription_fn_t fn,
void *arg)
{
struct backend_handle *bh = handle(h);
struct handle_subscription *hs = NULL;
if ((hs = malloc(sizeof(*hs))) == NULL){
clicon_err(OE_PLUGIN, errno, "malloc");
goto done;
}
memset(hs, 0, sizeof(*hs));
hs->hs_stream = strdup(stream);
hs->hs_format = format;
hs->hs_filter = filter?strdup(filter):NULL;
hs->hs_next = bh->bh_subscription;
hs->hs_fn = fn;
hs->hs_arg = arg;
bh->bh_subscription = hs;
done:
return hs;
}
/*! Delete subscription given stream name, callback and argument
* @param[in] h Clicon handle
* @param[in] stream Name of event stream
* @param[in] fn Callback when event occurs
* @param[in] arg Argument to use with callback and handle
* Note that arg is not a real handle.
* @see subscription_add
* @see subscription_each
*/
int
subscription_delete(clicon_handle h,
char *stream,
subscription_fn_t fn,
void *arg)
{
struct backend_handle *bh = handle(h);
struct handle_subscription *hs;
struct handle_subscription **hs_prev;
hs_prev = &bh->bh_subscription; /* this points to stack and is not real backpointer */
for (hs = *hs_prev; hs; hs = hs->hs_next){
/* XXX arg == hs->hs_arg */
if (strcmp(hs->hs_stream, stream)==0 && hs->hs_fn == fn){
*hs_prev = hs->hs_next;
free(hs->hs_stream);
if (hs->hs_filter)
free(hs->hs_filter);
if (hs->hs_arg)
free(hs->hs_arg);
free(hs);
break;
}
hs_prev = &hs->hs_next;
}
return 0;
}
/*! Iterator over subscriptions
*
* NOTE: Never manipulate the child-list during operation or using the
* same object recursively, the function uses an internal field to remember the
* index used. It works as long as the same object is not iterated concurrently.
*
* @param[in] h clicon handle
* @param[in] hprev iterator, initialize with NULL
* @code
* clicon_handle h;
* struct handle_subscription *hs = NULL;
* while ((hs = subscription_each(h, hs)) != NULL) {
* ...
* }
* @endcode
*/
struct handle_subscription *
subscription_each(clicon_handle h,
struct handle_subscription *hprev)
{
struct backend_handle *bh = handle(h);
struct handle_subscription *hs = NULL;
if (hprev)
hs = hprev->hs_next;
else
hs = bh->bh_subscription;
return hs;
}
int
backend_nacm_list_set(clicon_handle h,
cxobj *xnacm)