Persist: Support tracking observe information over server restarts

Add in optional callbacks for when dynamic resources are setup and deleted.

Add in optional callbacks for when observe subscriptions start and stop.

Add in optional callbacks for when the observe counter is updated (can be
ratelimited) for a resource as well as track when a resource is deleted.

Add is support for new functions

Main functions with libcoap tracking support are
 coap_persist_startup()
 coap_persist_stop()

coap-server has -t option to track observe information.

UDP sessions supported, as well as OSCORE over UDP.

Documentation updated

libcoap tracking support (as opposed to application tracking) is configurable.

CMakeLists.txt

@@ -71,6 +71,10 @@ option(
   ENABLE_OSCORE
   "compile with support for OSCORE"
   ON)
+option(
+  WITH_OBSERVE_PERSIST
+  "compile with observe persist support for server restarts"
+  ON)
 option(
   WITH_EPOLL
   "compile with epoll support"
@@ -238,6 +242,13 @@ else()
   message(STATUS "compiling without WebSockets support")
 endif()
 
+if(${WITH_OBSERVE_PERSIST})
+  set(COAP_WITH_OBSERVE_PERSIST "1")
+  message(STATUS "compiling with observe persistence support")
+else()
+  message(STATUS "compiling without observe persistence support")
+endif()
+
 if(${WITH_EPOLL}
    AND HAVE_EPOLL_H
    AND HAVE_TIMERFD_H)

cmake_coap_config.h.in

@@ -32,6 +32,9 @@
 /* Define if the library has server support */
 #cmakedefine COAP_SERVER_SUPPORT @COAP_SERVER_SUPPORT@
 
+/* Define if the library is to have observe persistence */
+#cmakedefine COAP_WITH_OBSERVE_PERSIST @COAP_WITH_OBSERVE_PERSIST@
+
 /* Define if the system has epoll support */
 #cmakedefine COAP_EPOLL_SUPPORT @COAP_EPOLL_SUPPORT@
 

configure.ac

@@ -787,6 +787,17 @@ AC_ARG_ENABLE([async],
 AS_IF([test "x$build_async" != "xyes"],
       [AC_DEFINE(WITHOUT_ASYNC, [1], [Define to build without support for separate responses.])])
 
+# configure options
+# __observe_persist__
+AC_ARG_ENABLE([async],
+              [AS_HELP_STRING([--enable-observe-persist],
+                              [Enable building with support for persisting observes over a server restart [default=yes]])],
+              [build_observe_persist="$enableval"],
+              [build_observe_persist="yes"])
+
+AS_IF([test "x$build_observe_persist" = "xyes"],
+      [AC_DEFINE(COAP_WITH_OBSERVE_PERSIST, [1], [Define to build support for persisting observes.])])
+
 # configure options
 # __add_default_names__
 AC_ARG_ENABLE([add-default-names],
@@ -1043,6 +1054,7 @@ man/coap_observe.txt
 man/coap_oscore.txt
 man/coap_pdu_access.txt
 man/coap_pdu_setup.txt
+man/coap_persist.txt
 man/coap_recovery.txt
 man/coap_resource.txt
 man/coap_session.txt
@@ -1120,6 +1132,11 @@ if test "x$build_add_default_names" = "xyes"; then
 else
     AC_MSG_RESULT([      add default names        : "no"])
 fi
+if test "x$build_observe_persist" = "xyes"; then
+    AC_MSG_RESULT([      build Observe Persist    : "yes"])
+else
+    AC_MSG_RESULT([      build Observe Persist    : "no"])
+fi
 if test "x$have_epoll" = "xyes"; then
     AC_MSG_RESULT([      build using epoll        : "$with_epoll"])
 fi

examples/coap-client.c

@@ -235,12 +235,12 @@ track_check_token(coap_bin_const_t *token) {
 }
 
 static void
-track_flush_token(coap_bin_const_t *token) {
+track_flush_token(coap_bin_const_t *token, int force) {
   size_t i;
 
   for (i = 0; i < tracked_tokens_count; i++) {
     if (coap_binary_equal(token, tracked_tokens[i].token)) {
-      if (!tracked_tokens[i].observe || !obs_started) {
+      if (force || !tracked_tokens[i].observe || !obs_started) {
         /* Only remove if not Observing */
         coap_delete_binary(tracked_tokens[i].token);
         if (tracked_tokens_count-i > 1) {
@@ -445,7 +445,7 @@ message_handler(coap_session_t *session COAP_UNUSED,
       } else {
         doing_getting_block = 0;
         if (!is_mcast)
-          track_flush_token(&token);
+          track_flush_token(&token, 0);
       }
       return COAP_RESPONSE_OK;
     }
@@ -462,11 +462,12 @@ message_handler(coap_session_t *session COAP_UNUSED,
         }
       }
       fprintf(stderr, "\n");
+      track_flush_token(&token, 1);
     }
 
   }
   if (!is_mcast)
-    track_flush_token(&token);
+    track_flush_token(&token, 0);
 
   /* our job is done, we can exit at any time */
   ready = doing_observe ? coap_check_option(received,

examples/coap-server.c

@@ -77,13 +77,17 @@ static int doing_oscore = 0;
 /* set to 1 to request clean server shutdown */
 static int quit = 0;
 
+/* set to 1 if persist information is to be kept on server shutdown */
+static int keep_persist = 0;
+
 /* changeable clock base (see handle_put_time()) */
 static time_t clock_offset;
 static time_t my_clock_base = 0;
 
 coap_resource_t *time_resource = NULL;
 
 static int resource_flags = COAP_RESOURCE_FLAGS_NOTIFY_CON;
+static int track_observes = 0;
 
 /*
  * For PKI, if one or more of cert_file, key_file and ca_file is in PKCS11 URI
@@ -184,6 +188,19 @@ handle_sigint(int signum COAP_UNUSED) {
   quit = 1;
 }
 
+#ifndef _WIN32
+/*
+ * SIGUSR2 handler: set quit to 1 for graceful termination
+ * Disable sending out 4.04 for any active observations.
+ * Note: coap_*() functions should not be called at sig interrupt.
+ */
+static void
+handle_sigusr2(int signum COAP_UNUSED) {
+  quit = 1;
+  keep_persist = 1;
+}
+#endif /* ! _WIN32 */
+
 /*
  * This will return a correctly formed transient_value_t *, or NULL.
  * If an error, the passed in coap_binary_t * will get deleted.
@@ -1446,6 +1463,7 @@ hnd_put_post(coap_resource_t *resource,
   }
   /* Need to de-reference as value may be in use elsewhere */
   release_resource_data(session, resource_entry->value);
+  resource_entry->value = NULL;
   transient_value = alloc_resource_data(data_so_far);
   if (!transient_value) {
     coap_pdu_set_code(response, COAP_RESPONSE_CODE_INTERNAL_ERROR);
@@ -1538,6 +1556,7 @@ hnd_put_post_unknown(coap_resource_t *resource COAP_UNUSED,
   coap_register_request_handler(r, COAP_REQUEST_PUT, hnd_put_post);
   coap_register_request_handler(r, COAP_REQUEST_POST, hnd_put_post);
   coap_register_request_handler(r, COAP_REQUEST_DELETE, hnd_delete);
+  coap_register_request_handler(r, COAP_REQUEST_FETCH, hnd_get);
   /* We possibly want to Observe the GETs */
   coap_resource_set_get_observable(r, 1);
   coap_register_request_handler(r, COAP_REQUEST_GET, hnd_get);
@@ -2159,6 +2178,9 @@ usage( const char *program, const char *version) {
      "\t       \t\tonly '/', '/async' and '/.well-known/core' are enabled\n"
      "\t       \t\tfor multicast requests support, otherwise all\n"
      "\t       \t\tresources are enabled\n"
+     "\t-t     \t\tTrack resource's observe values so observe\n"
+     "\t       \t\tsubscriptions can be maintained over a server restart.\n"
+     "\t       \t\tNote: Use 'kill SIGUSR2 <pid>' for controlled shutdown\n"
      "\t-v num \t\tVerbosity level (default 4, maximum is 8) for general\n"
      "\t       \t\tCoAP logging\n"
      "\t-w [port][,secure_port]\n"
@@ -2728,7 +2750,7 @@ main(int argc, char **argv) {
 
   clock_offset = time(NULL);
 
-  while ((opt = getopt(argc, argv, "c:d:eg:G:h:i:j:J:k:l:mnp:rs:u:v:w:A:C:E:L:M:NP:R:S:T:U:V:X:")) != -1) {
+  while ((opt = getopt(argc, argv, "c:d:eg:G:h:i:j:J:k:l:mnp:rs:tu:v:w:A:C:E:L:M:NP:R:S:T:U:V:X:")) != -1) {
     switch (opt) {
     case 'A' :
       strncpy(addr_str, optarg, NI_MAXHOST-1);
@@ -2846,6 +2868,9 @@ main(int argc, char **argv) {
         goto failed;
       }
       break;
+    case 't':
+      track_observes = 1;
+      break;
     case 'u':
 #if SERVER_CAN_PROXY
       user_length = cmdline_read_user(optarg, &user, MAX_USER);
@@ -2891,6 +2916,8 @@ main(int argc, char **argv) {
   sa.sa_flags = 0;
   sigaction (SIGINT, &sa, NULL);
   sigaction (SIGTERM, &sa, NULL);
+  sa.sa_handler = handle_sigusr2;
+  sigaction (SIGUSR2, &sa, NULL);
   /* So we do not exit on a SIGPIPE */
   sa.sa_handler = SIG_IGN;
   sigaction (SIGPIPE, &sa, NULL);
@@ -2924,6 +2951,20 @@ main(int argc, char **argv) {
   if (group)
     coap_join_mcast_group_intf(ctx, group, group_if);
 
+  if (track_observes) {
+    /*
+     * Read in and set up appropriate persist information.
+     * Note that this should be done after ctx is properly set up.
+     */
+    if (!coap_persist_startup(ctx,
+                              "/tmp/coap_dyn_resource_save_file",
+                              "/tmp/coap_observe_save_file",
+                              "/tmp/coap_obs_cnt_save_file", 10)) {
+      fprintf(stderr, "Unable to set up persist logic\n");
+      goto finish;
+    }
+  }
+
   coap_fd = coap_context_get_coap_fd(ctx);
   if (coap_fd != -1) {
     /* if coap_fd is -1, then epoll is not supported within libcoap */
@@ -3014,6 +3055,9 @@ main(int argc, char **argv) {
 
 finish:
   /* Clean up local usage */
+  if (keep_persist)
+    coap_persist_stop(ctx);
+
   coap_free(ca_mem);
   coap_free(cert_mem);
   coap_free(key_mem);

include/coap3/coap_net_internal.h

@@ -19,6 +19,7 @@
 #define COAP_NET_INTERNAL_H_
 
 #include "coap_internal.h"
+#include "coap_subscribe.h"
 
 /**
  * @ingroup internal_api
@@ -95,11 +96,40 @@ struct coap_context_t {
 #endif /* HAVE_OSCORE */
 
 #if COAP_CLIENT_SUPPORT
-  coap_response_handler_t response_handler;
+  coap_response_handler_t response_handler; /**< Called when a response is
+                                                 received */
 #endif /* COAP_CLIENT_SUPPORT */
-  coap_nack_handler_t nack_handler;
-  coap_ping_handler_t ping_handler;
-  coap_pong_handler_t pong_handler;
+  coap_nack_handler_t nack_handler; /**< Called when a response issue has
+                                         occurred */
+  coap_ping_handler_t ping_handler; /**< Called when a CoAP ping is received */
+  coap_pong_handler_t pong_handler; /**< Called when a ping response
+                                         is received */
+
+#if COAP_SERVER_SUPPORT
+  coap_observe_added_t observe_added; /**< Called when there is a new observe
+                                           subscription request */
+  coap_observe_deleted_t observe_deleted; /**< Called when there is a observe
+                                           subscription de-register request */
+  void *observe_user_data; /**< App provided data for use in observe_added or
+                                observe_deleted */
+  uint32_t observe_save_freq; /**< How frequently to update observe value */
+  coap_track_observe_value_t track_observe_value; /**< Callback to save observe
+                                                       value when updated */
+  coap_dyn_resource_added_t dyn_resource_added; /**< Callback to save dynamic
+                                                     resource when created */
+  coap_resource_deleted_t resource_deleted; /**< Invoked when resource
+                                                 is deleted */
+#if COAP_WITH_OBSERVE_PERSIST
+  coap_bin_const_t *dyn_resource_save_file; /** Where dynamic resource requests
+                                                that create resources are
+                                                tracked */
+  coap_bin_const_t *obs_cnt_save_file; /** Where resource observe counters are
+                                            tracked */
+  coap_bin_const_t *observe_save_file; /** Where observes are tracked */
+  coap_pdu_t *unknown_pdu;        /** PDU used for unknown resource request */
+  coap_session_t *unknown_session; /** Session used for unknown resource request */
+#endif /* COAP_WITH_OBSERVE_PERSIST */
+#endif /* COAP_SERVER_SUPPORT */
 
   /**
    * Callback function that is used to signal events to the
@@ -153,6 +183,8 @@ struct coap_context_t {
 #endif /* ! COAP_EPOLL_SUPPORT */
 #if COAP_SERVER_SUPPORT
   uint8_t observe_pending;         /**< Observe response pending */
+  uint8_t observe_no_clear;        /**< Observe 4.04 not to be sent on deleting
+                                        resource */
   uint8_t mcast_per_resource;      /**< Mcast controlled on a per resource
                                         basis */
 #endif /* COAP_SERVER_SUPPORT */