clixon/lib/src/clixon_stream.c

/*
 *
  ***** BEGIN LICENSE BLOCK *****
 
  Copyright (C) 2009-2018 Olof Hagsand and Benny Holmgren

  This file is part of CLIXON.

  Licensed under the Apache License, Version 2.0 (the "License");
  you may not use this file except in compliance with the License.
  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

  Unless required by applicable law or agreed to in writing, software
  distributed under the License is distributed on an "AS IS" BASIS,
  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  See the License for the specific language governing permissions and
  limitations under the License.

  Alternatively, the contents of this file may be used under the terms of
  the GNU General Public License Version 3 or later (the "GPL"),
  in which case the provisions of the GPL are applicable instead
  of those above. If you wish to allow use of your version of this file only
  under the terms of the GPL, and not to allow others to
  use your version of this file under the terms of Apache License version 2, 
  indicate your decision by deleting the provisions above and replace them with
  the  notice and other provisions required by the GPL. If you do not delete
  the provisions above, a recipient may use your version of this file under
  the terms of any one of the Apache License version 2 or the GPL.

  ***** END LICENSE BLOCK *****

 * Event notification streams according to RFC5277
 * See (old) subscription code in clixon_backend_handle.c and backend_client.c
 */

#ifdef HAVE_CONFIG_H
#include "clixon_config.h" /* generated by config & autoconf */
#endif

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <inttypes.h>
#include <syslog.h>
#include <sys/time.h>

/* cligen */
#include <cligen/cligen.h>

/* clicon */
#include "clixon_queue.h"
#include "clixon_err.h"
#include "clixon_log.h"
#include "clixon_string.h"
#include "clixon_hash.h"
#include "clixon_handle.h"
#include "clixon_yang.h"
#include "clixon_xml.h"
#include "clixon_options.h"
#include "clixon_xpath_ctx.h"
#include "clixon_xpath.h"
#include "clixon_stream.h"

/*! Find an event notification stream given name
 * @param[in]  h    Clicon handle
 * @param[in]  name Name of stream
 * @retval     es   Event notification stream structure
 * @retval     NULL Not found
 */
event_stream_t *
stream_find(clicon_handle h,
	    const char   *name)
{
    event_stream_t *es = NULL;

    for (es=clicon_stream(h); es; es=es->es_next)
	if (strcmp(name, es->es_name)==0)
	    break;
    return es;    
}

/*! Add notification event stream
 * 
 */
int
stream_register(clicon_handle h,
		const char   *name, 
		const char   *description) 
{
    int             retval = -1;
    event_stream_t *es;

    if ((es = stream_find(h, name)) != NULL)
	goto ok;
    if ((es = malloc(sizeof(event_stream_t))) == NULL){
	clicon_err(OE_XML, errno, "malloc");
	goto done;
    }
    memset(es, 0, sizeof(event_stream_t));
    if ((es->es_name = strdup(name)) == NULL){
	clicon_err(OE_XML, errno, "strdup");
	goto done;
    }
    if ((es->es_description = strdup(description)) == NULL){
	clicon_err(OE_XML, errno, "strdup");
	goto done;
    }
    clicon_stream_append(h, es);
 ok:
    retval = 0;
 done:
    return retval;
}

/*! Delete complete notification event stream list (not just single stream)
 * @param[in] es 
 */
int
stream_delete_all(event_stream_t *es)
{
    event_stream_t *e_next;

    while (es){
	e_next = es->es_next;
	if (es->es_name)
	    free(es->es_name);
	if (es->es_description)
	    free(es->es_description);
	free(es);
	es = e_next;
    }
    return 0;
}

/*! Return stream definition
 * @param[in]  h      Clicon handle
 * @param[in]  access If set, include access/location
 * @param[out] cb     Output buffer containing XML on exit
 * @retval     0      OK
 * @retval     -1     Error
 */ 
int
stream_get_xml(clicon_handle h,
	       int           access,
	       cbuf         *cb)
{
    event_stream_t *es = NULL;
    char           *url_prefix;
    char           *stream_path;

    cprintf(cb, "<streams>");
    for (es=clicon_stream(h); es; es=es->es_next){
	cprintf(cb, "<stream>");
	cprintf(cb, "<name>%s</name>", es->es_name);
	if (es->es_description)
	    cprintf(cb, "<description>%s</description>", es->es_description);
	cprintf(cb, "<replay-support>false</replay-support>");
	if (access){
	    cprintf(cb, "<access>");
	    cprintf(cb, "<encoding>xml</encoding>");
	    url_prefix = clicon_option_str(h, "CLICON_STREAM_URL");
	    stream_path = clicon_option_str(h, "CLICON_STREAM_PATH");
	    cprintf(cb, "<location>%s/%s/%s</location>", 
		    url_prefix, stream_path, es->es_name);
	    cprintf(cb, "</access>");
	}
	cprintf(cb, "</stream>");
    }
    cprintf(cb, "</streams>");
    return 0;
}

#ifdef NYI
/*! Delete single notification event stream 
 * XXX notused
 */
int
stream_del()
{
    return 0;
}
#endif


/*! Add an event notification callback to a stream given a callback function
 * @param[in]  h      Clicon handle
 * @param[in]  stream Name of stream
 * @param[in]  xpath  Filter selector - xpath
 * @param[in]  fn     Callback when event occurs
 * @param[in]  arg    Argument to use with callback. Also handle when deleting
 * @retval     0      OK
 * @retval     -1     Error, ie no such stream 
 */
int
stream_cb_add(clicon_handle     h,
	      char             *stream,
	      char             *xpath,
	      stream_fn_t       fn,
	      void             *arg)
{
    int                         retval = -1;
    event_stream_t             *es;
    struct stream_subscription *ss;

    clicon_debug(1, "%s", __FUNCTION__);
    if ((es = stream_find(h, stream)) == NULL){
	clicon_err(OE_CFG, ENOENT, "Stream %s not found", stream);
	goto done;
    }
    if ((ss = malloc(sizeof(*ss))) == NULL){
	clicon_err(OE_CFG, errno, "malloc");
	goto done;
    }
    memset(ss, 0, sizeof(*ss));
    if ((ss->ss_stream = strdup(stream)) == NULL){
	clicon_err(OE_CFG, errno, "strdup");
	goto done;
    }
    if (xpath && (ss->ss_xpath = strdup(xpath)) == NULL){
	clicon_err(OE_CFG, errno, "strdup");
	goto done;
    }
    ss->ss_fn     = fn;
    ss->ss_arg    = arg;
    ss->ss_next   = es->es_subscription;
    es->es_subscription = ss;
    retval = 0;
  done:
    return retval;
}

/*! Delete event notification callback to a stream given a callback and arg
 * @param[in]  h      Clicon handle
 * @param[in]  stream Name of stream or NULL for all streams
 * @param[in]  fn     Callback when event occurs
 * @param[in]  arg    Argument to use with callback. Also handle when deleting
 * @retval     0      OK
 * @retval     -1     Error
 */
int
stream_cb_delete(clicon_handle     h,
		 char             *stream,
		 stream_fn_t       fn,
		 void             *arg)
{
    int                          retval = -1;
    event_stream_t              *es;
    struct stream_subscription **ss_prev;
    struct stream_subscription  *ss;
    struct stream_subscription  *ss_next;
    
    for (es=clicon_stream(h); es; es=es->es_next){
	if (stream && strcmp(stream, es->es_name)!=0)
	    continue;
	ss_prev = &es->es_subscription;
	for (ss = *ss_prev; ss; ss = ss_next){
	    ss_next = ss->ss_next;
	    if (fn == ss->ss_fn && arg == ss->ss_arg){
		*ss_prev = ss->ss_next;
		if (ss->ss_stream)
		    free(ss->ss_stream);
		if (ss->ss_xpath)
		    free(ss->ss_xpath);
		free(ss);
		continue;
	//	    break; if more > 1
	    }
	    ss_prev = &ss->ss_next;
	}
    }
    retval = 0;
    return retval;
}

/*! Stream notify event and distribute to all registered callbacks
 * @param[in]  h       Clicon handle
 * @param[in]  stream  Name of event stream. CLICON is predefined as LOG stream
 * @param[in]  event   Notification as xml tree
 * @retval  0  OK
 * @retval -1  Error with clicon_err called
 * @see stream_notify
 */
int
stream_notify_xml(clicon_handle h, 
		  char         *stream, 
		  cxobj        *xevent)
{
    int                         retval = -1;
    event_stream_t             *es;
    struct stream_subscription *ss;
    
    clicon_debug(1, "%s", __FUNCTION__);
    if ((es = stream_find(h, stream)) == NULL)
	goto ok;
    /* Go thru all global (handle) subscriptions and find matches */
    for (ss = es->es_subscription; ss; ss = ss->ss_next){
	if (ss->ss_xpath == NULL ||
	    strlen(ss->ss_xpath)==0 ||
	    xpath_first(xevent, "%s", ss->ss_xpath) != NULL)
	    if ((*ss->ss_fn)(h, xevent, ss->ss_arg) < 0)
		goto done;
    }
 ok:
    retval = 0;
  done:
    return retval;
}

/*! Stream notify event and distribute to all registered callbacks
 * @param[in]  h       Clicon handle
 * @param[in]  stream  Name of event stream. CLICON is predefined as LOG stream
 * @param[in]  event   Notification as format string according to printf(3)
 * @retval  0  OK
 * @retval -1  Error with clicon_err called
 * @code
 *  if (stream_notify(h, "NETCONF", "<event><event-class>fault</event-class><reportingEntity><card>Ethernet0</card></reportingEntity><severity>major</severity></event>") < 0)
 *    err;
 * @endcode
 * @see stream_notify_xml
 */
int
stream_notify(clicon_handle h, 
	      char         *stream, 
	      const char   *event, ...)
{
    int        retval = -1;
    va_list    args;
    int        len;
    cxobj     *xev = NULL;
    yang_spec *yspec = NULL;
    char      *str = NULL;
    cbuf      *cb = NULL;
    char       timestr[27];
    struct timeval tv;

    clicon_debug(1, "%s", __FUNCTION__);
    va_start(args, event);
    len = vsnprintf(NULL, 0, event, args) + 1;
    va_end(args);
    if ((str = malloc(len)) == NULL){
	clicon_err(OE_UNIX, errno, "malloc");
	goto done;
    }
    memset(str, 0, len);
    va_start(args, event);
    len = vsnprintf(str, len, event, args) + 1;
    va_end(args);
    if ((yspec = clicon_dbspec_yang(h)) == NULL){
	clicon_err(OE_YANG, 0, "No yang spec");
	goto done;
    }
    if ((cb = cbuf_new()) == NULL){
	clicon_err(OE_UNIX, errno, "cbuf_new");
	goto done;
    }

    gettimeofday(&tv, NULL);
    if (time2str(tv, timestr, sizeof(timestr)) < 0){
	clicon_err(OE_UNIX, errno, "time2str");
	goto done;
    }
    cprintf(cb, "<notification xmlns=\"urn:ietf:params:xml:ns:netconf:notification:1.0\"><eventTime>%s</eventTime>%s</notification>", timestr, str);
    if (xml_parse_string(cbuf_get(cb), yspec, &xev) < 0)
	goto done;
    if (xml_rootchild(xev, 0, &xev) < 0)
	goto done;
    if (stream_notify_xml(h, stream, xev) < 0)
	goto done;
    retval = 0;
  done:
    if (cb)
	cbuf_free(cb);
    if (xev)
	xml_free(xev);
    if (str)
	free(str);
    return retval;
}

#ifdef CLIXON_PUBLISH_STREAMS

#include <curl/curl.h>

/*
 * Types (curl)
 */
struct curlbuf{
    size_t b_len;
    char  *b_buf;
};

/*
 * For the asynchronous case. I think we must handle the case where of many of these
 * come in before we can handle them in the upper-level polling routine.
 * realloc. Therefore, we append new data to the userdata buffer.
 */
static size_t
curl_get_cb(void *ptr, 
	    size_t size, 
	    size_t nmemb, 
	    void *userdata)
{
    struct curlbuf *buf = (struct curlbuf *)userdata;
    int len;

    len = size*nmemb;
    if ((buf->b_buf = realloc(buf->b_buf, buf->b_len+len+1)) == NULL)
	return 0;
    memcpy(buf->b_buf+buf->b_len, ptr, len);
    buf->b_len += len;
    buf->b_buf[buf->b_len] = '\0';
    fprintf(stderr, "%s: %s\n", __FUNCTION__, buf->b_buf);
    return len;
}

/*! Send a curl POST request
 * @retval  -1   fatal error
 * @retval   0   expect set but did not expected return or other non-fatal error
 * @retval   1   ok
 * Note: curl_easy_perform blocks
 * Note: New handle is created every time, the handle can be re-used for better TCP performance
 * @see same function (url_post) in grideye_curl.c
 */
static int
url_post(char *url, 
	 char *postfields, 
	 char **getdata)
{
    int            retval = -1;
    CURL          *curl = NULL;
    char          *err = NULL;
    struct curlbuf cb = {0, };
    CURLcode       errcode;

    /* Try it with  curl -X PUT -d '*/
    clicon_debug(1, "%s:  curl -X POST -d '%s' %s",
	__FUNCTION__, postfields, url);
    /* Set up curl for doing the communication with the controller */
    if ((curl = curl_easy_init()) == NULL) {
	clicon_debug(1, "curl_easy_init");
	goto done;
    }
    if ((err = malloc(CURL_ERROR_SIZE)) == NULL) {
	clicon_debug(1, "%s: malloc", __FUNCTION__);
	goto done;
    }
    curl_easy_setopt(curl, CURLOPT_URL, url);
    curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_get_cb);
    curl_easy_setopt(curl, CURLOPT_WRITEDATA, &cb);
    curl_easy_setopt(curl, CURLOPT_ERRORBUFFER, err);
    curl_easy_setopt(curl, CURLOPT_POST, 1);
    curl_easy_setopt(curl, CURLOPT_POSTFIELDS, postfields);
    curl_easy_setopt(curl, CURLOPT_POSTFIELDSIZE, strlen(postfields));

    if (debug)
	curl_easy_setopt(curl, CURLOPT_VERBOSE, 1);   
    if ((errcode = curl_easy_perform(curl)) != CURLE_OK){
	clicon_debug(1, "%s: curl: %s(%d)", __FUNCTION__, err, errcode);
	retval = 0;
	goto done; 
    }
    if (getdata && cb.b_buf){
	*getdata = cb.b_buf;
	cb.b_buf = NULL;
    }
    retval = 1;
  done:
    if (err)
	free(err);
    if (cb.b_buf)
	free(cb.b_buf);
    if (curl)
	curl_easy_cleanup(curl);   /* cleanup */ 
    return retval;
}

/*! Stream callback for example stream notification 
 * Push via curl_post to publish stream event
 * @param[in]  h     Clicon handle
 * @param[in]  event Event as XML
 * @param[in]  arg   Extra argument provided in stream_cb_add
 * @see stream_cb_add
 */
static int 
stream_publish_cb(clicon_handle h, 
		  cxobj        *event,
		  void         *arg)
{
    int   retval = -1;
    cbuf *u = NULL; /* stream pub (push) url */
    cbuf *d = NULL; /* (XML) data to push */
    char *pub_prefix;
    char *result = NULL;
    char *stream = (char*)arg;

    clicon_debug(1, "%s", __FUNCTION__); 

    /* Create pub url */
    if ((u = cbuf_new()) == NULL){
	clicon_err(OE_XML, errno, "cbuf_new");
	goto done;
    }
    if ((pub_prefix = clicon_option_str(h, "CLICON_STREAM_PUB")) == NULL){
	clicon_err(OE_CFG, ENOENT, "CLICON_STREAM_PUB not defined");
	goto done;
    }

    cprintf(u, "%s/%s", pub_prefix, stream);
    /* Create XML data as string */
    if ((d = cbuf_new()) == NULL){
	clicon_err(OE_XML, errno, "cbuf_new");
	goto done;
    }
    if (clicon_xml2cbuf(d, event, 0, 0) < 0)
	goto done;
    if (url_post(cbuf_get(u),     /* url+stream */
		 cbuf_get(d),     /* postfields */
		 &result) < 0)    /* result as xml */
	goto done;
    if (result)
	clicon_debug(1, "%s: %s", __FUNCTION__, result);
    retval = 0;
 done:
    if (u)
	cbuf_free(u);
    if (d)
	cbuf_free(d);
    if (result)
	free(result);
    return retval;
}
#endif /* CLIXON_PUBLISH_STREAMS */

/*! Publish all streams on a pubsub channel, eg using SSE
 */
int
stream_publish(clicon_handle h,
	       char         *stream)
{
#ifdef CLIXON_PUBLISH_STREAMS
    int retval = -1;

    if (stream_cb_add(h, stream, NULL, stream_publish_cb, (void*)stream) < 0)
	goto done;
    retval = 0;
 done:
    return retval;
#else
   clicon_log(LOG_WARNING, "%s called but CLIXON_PUBLISH_STREAMS not enabled (enable with configure --enable-publish)", __FUNCTION__);
   clicon_log_init("xpath", LOG_WARNING, CLICON_LOG_STDERR); 
   return 0;
#endif
}

int 
stream_publish_init()
{
#ifdef CLIXON_PUBLISH_STREAMS
    int retval = -1;

    if (curl_global_init(CURL_GLOBAL_ALL) != 0){
	clicon_err(OE_PLUGIN, errno, "curl_global_init");
	goto done;
    }    
    retval = 0;
 done:
    return retval;
#else
    return 0;
#endif
}

int 
stream_publish_exit()
{
#ifdef CLIXON_PUBLISH_STREAMS
    curl_global_cleanup();
#endif 
    return 0;
}