stream replay and subscription update

apps/backend/backend_client.c

@@ -107,7 +107,6 @@ ce_event_cb(clicon_handle h,
 	    backend_client_rm(h, ce);
 #endif
 	}
-
     }
     return 0;
 }
@@ -845,14 +844,17 @@ from_client_create_subscription(clicon_handle        h,
     if ((x = xpath_first(xe, "//stopTime")) != NULL){
 	stoptime = xml_find_value(x, "body");
 	if (str2time(stoptime, &stop) < 0){
-	    clicon_err(OE_PROTO, errno, "str2time");
+	    if (netconf_bad_element(cbret, "application", "<bad-element>stopTime</bad-element>", "Expected timestamp") < 0)
 		goto done;
+	    goto ok;	
 	}
     }
     if ((x = xpath_first(xe, "//startTime")) != NULL){
 	starttime = xml_find_value(x, "body");
 	if (str2time(starttime, &start) < 0){
-	    clicon_err(OE_PROTO, errno, "str2time");
+	    if (netconf_bad_element(cbret, "application", "<bad-element>startTime</bad-element>", "Expected timestamp") < 0)
+		goto done;
+	    goto ok;	
 	    goto done;	
 	}	
     }

apps/backend/backend_main.c

@@ -729,8 +729,12 @@ main(int    argc,
 	return -1;
     }
 
-    /* Publish stream on pubsub channels XXX conditional? */
+    /* Publish stream on pubsub channels.
-    if (stream_publish_init() < 0)
+     * CLICON_STREAM_PUB should be set to URL to where streams are published
+     * and configure should be run with --enable-publish
+     */
+    if (clicon_option_exists(h, "CLICON_STREAM_PUB") &&
+	stream_publish_init() < 0)
 	goto done;
     if ((xmldb_plugin = clicon_xmldb_plugin(h)) == NULL){
 	clicon_log(LOG_ERR, "No xmldb plugin given (specify option CLICON_XMLDB_PLUGIN).\n"); 

example/example_backend.c

@@ -305,7 +305,8 @@ clixon_plugin_init(clicon_handle h)
 	goto done;
     /* assumes: CLIXON_PUBLISH_STREAMS, eg configure --enable-publish
      */
-    if (stream_publish(h, "EXAMPLE") < 0)
+    if (clicon_option_exists(h, "CLICON_STREAM_PUB") &&
+	stream_publish(h, "EXAMPLE") < 0)
 	goto done;
     if (example_stream_timer_setup(h) < 0)
 	goto done;

lib/clixon/clixon_handle.h

@@ -77,6 +77,7 @@ clicon_hash_t *clicon_data(clicon_handle h);
 /* Return internal stream hash-array given a handle.*/
 struct event_stream *clicon_stream(clicon_handle h);
 struct event_stream;
+int clicon_stream_set(clicon_handle h, struct event_stream *es);
 int clicon_stream_append(clicon_handle h, struct event_stream *es);
 
 #endif  /* _CLIXON_HANDLE_H_ */

lib/clixon/clixon_stream.h

@@ -67,7 +67,7 @@ struct stream_replay{
 /* See RFC8040 9.3, stream list, no replay support for now
  */
 struct event_stream{
-    struct event_stream *es_next;
+    qelem_t              es_q;   /* queue header */
     char                *es_name; /* name of notification event stream */
     char                *es_description;
     struct stream_subscription *es_subscription;
@@ -81,7 +81,7 @@ typedef struct event_stream event_stream_t;
  */
 event_stream_t *stream_find(clicon_handle h, const char *name);
 int stream_register(clicon_handle h, const char *name, const char *description, int replay_enabled);
-int stream_delete_all(event_stream_t *es);
+int stream_delete_all(clicon_handle h);
 int stream_get_xml(clicon_handle h, int access, cbuf *cb);
 int stream_timer_setup(int fd, void *arg);
 /* Subscriptions */

lib/src/clixon_handle.c

@@ -135,7 +135,7 @@ clicon_handle_exit(clicon_handle h)
 	hash_free(ha);
     if ((ha = clicon_data(h)) != NULL)
 	hash_free(ha);
-    stream_delete_all(clicon_stream(h));
+    stream_delete_all(h);
     free(ch);
     return 0;
 }
@@ -187,14 +187,22 @@ clicon_stream(clicon_handle h)
     return ch->ch_stream;
 }
 
+int
+clicon_stream_set(clicon_handle   h,
+		  event_stream_t *es)
+{
+    struct clicon_handle *ch = handle(h);
+
+    ch->ch_stream = es;
+    return 0;
+}
+
 int
 clicon_stream_append(clicon_handle h,
 		     event_stream_t *es)
 {
     struct clicon_handle *ch = handle(h);
-    event_stream_t      **ep;
     
-    for (ep = &ch->ch_stream; (*ep); ep=&(*ep)->es_next);
+    ADDQ(es, ch->ch_stream);
-    *ep = es;
     return 0;
 }

lib/src/clixon_stream.c

@@ -90,12 +90,17 @@ event_stream_t *
 stream_find(clicon_handle h,
 	    const char   *name)
 {
+    event_stream_t *es0;
     event_stream_t *es = NULL;
 
-    for (es=clicon_stream(h); es; es=es->es_next)
+    es0 = clicon_stream(h);
+    if ((es = es0) != NULL)
+	do {
 	    if (strcmp(name, es->es_name)==0)
-	    break;
 		return es;
+	    es = NEXTQ(struct event_stream *, es);
+	} while (es && es != es0);
+    return NULL;
 }
 
 /*! Add notification event stream
@@ -137,14 +142,16 @@ stream_register(clicon_handle h,
  * @param[in] es 
  */
 int
-stream_delete_all(event_stream_t *es)
+stream_delete_all(clicon_handle h)
 {
-    event_stream_t       *e_next;
     struct stream_replay *r;
     struct stream_subscription *ss;
+    event_stream_t       *es;
+    event_stream_t       *head = clicon_stream(h);
     
-    while (es){
+    while ((es = clicon_stream(h)) != NULL){
-	e_next = es->es_next;
+	DELQ(es, head, event_stream_t *);
+	clicon_stream_set(h, head);
 	if (es->es_name)
 	    free(es->es_name);
 	if (es->es_description)
@@ -158,7 +165,6 @@ stream_delete_all(event_stream_t *es)
 	    free(r);
 	}
 	free(es);
-	es = e_next;
     }
     return 0;
 }
@@ -180,7 +186,8 @@ stream_get_xml(clicon_handle h,
     char           *stream_path;
 
     cprintf(cb, "<streams>");
-    for (es=clicon_stream(h); es; es=es->es_next){
+    if ((es = clicon_stream(h)) != NULL){
+	do {
 	    cprintf(cb, "<stream>");
 	    cprintf(cb, "<name>%s</name>", es->es_name);
 	    if (es->es_description)
@@ -196,6 +203,8 @@ stream_get_xml(clicon_handle h,
 		cprintf(cb, "</access>");
 	    }
 	    cprintf(cb, "</stream>");
+	    es = NEXTQ(struct event_stream *, es);
+	} while (es && es != clicon_stream(h));
     }
     cprintf(cb, "</streams>");
     return 0;
@@ -224,7 +233,11 @@ stream_timer_setup(int   fd,
      */
     gettimeofday(&now, NULL);
     /* for all event streams, remove subscription if past stop time */
-    for (es=clicon_stream(h); es; es=es->es_next){
+
+
+
+    if ((es = clicon_stream(h)) != NULL){
+	do {
 	    if ((ss = es->es_subscription) != NULL)
 		do {
 		    if (timerisset(&ss->ss_stoptime) && timercmp(&ss->ss_stoptime, &now, <)){
@@ -236,6 +249,8 @@ stream_timer_setup(int   fd,
 		    else
 			ss = NEXTQ(struct stream_subscription *, ss);
 		} while (ss && ss != es->es_subscription);
+	    es = NEXTQ(struct event_stream *, es);
+	} while (es && es != clicon_stream(h));
     }
     /* Initiate new timer */
     timeradd(&now, &t1, &t);
@@ -374,10 +389,14 @@ stream_ss_delete_all(clicon_handle     h,
     event_stream_t              *es;
     struct stream_subscription  *ss;
 
-    for (es=clicon_stream(h); es; es=es->es_next)
+    if ((es = clicon_stream(h)) != NULL){
+	do {
 	    if ((ss = stream_ss_find(es, fn, arg)) != NULL)
 		if (stream_ss_rm(es, ss) < 0)
 		    goto done;
+	    es = NEXTQ(struct event_stream *, es);
+	} while (es && es != clicon_stream(h));
+    }	    
     retval = 0;
  done:
     return retval;

test/test_stream.sh

@@ -10,10 +10,9 @@ APPNAME=example
 # include err() and new() functions and creates $dir
 . ./lib.sh
 cfg=$dir/conf.xml
-fyang=$dir/restconf.yang
+fyang=$dir/stream.yang
 xml=$dir/xml.xml
 
-
 #  <CLICON_YANG_MODULE_MAIN>example</CLICON_YANG_MODULE_MAIN>
 cat <<EOF > $cfg
 <config>
@@ -34,7 +33,6 @@ cat <<EOF > $cfg
   <CLICON_STREAM_DISCOVERY_RFC8040>true</CLICON_STREAM_DISCOVERY_RFC8040>
   <CLICON_STREAM_PATH>streams</CLICON_STREAM_PATH>
   <CLICON_STREAM_URL>https://localhost</CLICON_STREAM_URL>
-  <CLICON_STREAM_PUB>http://localhost/pub</CLICON_STREAM_PUB>
 </config>
 EOF
 
@@ -98,8 +96,6 @@ sudo pkill -u www-data clixon_restconf
 new "start restconf daemon"
 sudo start-stop-daemon -S -q -o -b -x /www-data/clixon_restconf -d /www-data -c www-data -- -f $cfg  -y $fyang # -D 1
 
-sleep 1
-
 new "netconf event stream discovery RFC5277 Sec 3.2.5"
 expecteof "$clixon_netconf -qf $cfg -y $fyang" 0 '<rpc><get><filter type="xpath" select="netconf/streams" xmlns="urn:ietf:params:xml:ns:netmod:notification"/></get></rpc>]]>]]>' '<rpc-reply><data><netconf><streams><stream><name>EXAMPLE</name><description>Example event stream</description><replay-support>false</replay-support></stream></streams></netconf></data></rpc-reply>]]>]]>'
 
@@ -153,8 +149,13 @@ new "netconf NONEXIST subscription"
 expectwait "$clixon_netconf -qf $cfg -y $fyang" '<rpc><create-subscription><stream>NONEXIST</stream></create-subscription></rpc>]]>]]>' '^<rpc-reply><rpc-error><error-tag>invalid-value</error-tag><error-type>application</error-type><error-severity>error</error-severity><error-message>No such stream</error-message></rpc-error></rpc-reply>]]>]]>$' 5
 fi
 
-#new "netconf EXAMPLE subscription with replay"
+new "netconf EXAMPLE subscription with wrong date"
-#expectwait "$clixon_netconf -qf $cfg -y $fyang" '<rpc><create-subscription><stream>EXAMPLE</stream><startTime>2018-10-21T19:22:16</startTime><stopTime>2018-10-21T19:25:00</stopTime></create-subscription></rpc>]]>]]>' '^<rpc-reply><ok/></rpc-reply>]]>]]><notification xmlns="urn:ietf:params:xml:ns:netconf:notification:1.0"><eventTime>20' 5
+expectwait "$clixon_netconf -qf $cfg -y $fyang" '<rpc><create-subscription><stream>EXAMPLE</stream><startTime>kallekaka</startTime></create-subscription></rpc>]]>]]>' '^<rpc-reply><rpc-error><error-tag>bad-element</error-tag><error-type>application</error-type><error-info><bad-element>startTime</bad-element></error-info><error-severity>error</error-severity><error-message>Expected timestamp</error-message></rpc-error></rpc-reply>]]>]]>$' 0
+
+new "netconf EXAMPLE subscription with replay"
+NOW=$(date +"%Y-%m-%dT%H:%M:%S")
+sleep 10
+expectwait "$clixon_netconf -qf $cfg -y $fyang" "<rpc><create-subscription><stream>EXAMPLE</stream><startTime>$NOW</startTime></create-subscription></rpc>]]>]]>" '^<rpc-reply><ok/></rpc-reply>]]>]]><notification xmlns="urn:ietf:params:xml:ns:netconf:notification:1.0"><eventTime>20' 10
 
 new "Kill restconf daemon"
 sudo pkill -u www-data clixon_restconf

yang/clixon-config@2018-10-21.yang

@@ -411,11 +411,12 @@ module clixon-config {
 	}
 	leaf CLICON_STREAM_PUB {
 	    type string;
-	    default "http://localhost/pub";
     	    description "For stream publish using eg nchan, the base address
-	          to publish to. 
+	          to publish to. Example value: http://localhost/pub
-                  Example: http://localhost/pub/NETCONF. Note this may
+                  Example: stream NETCONF would then be pushed to
-		  be local URL behind reverse-proxy";
+                  http://localhost/pub/NETCONF. 
+                  Note this may be a local/provate URL behind reverse-proxy.
+                  If not given, do NOT enable stream publishing using NCHAN.";
 	}
     }
 }