Persist observe: Support tracking observe information over server restarts

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 */