[HSS] Enable Change Streams (#1758)

* [HSS] Enable Change Streams
* Enable Events and Timers in HSS
* Integrate change streams in dbi
* mongodb should be configured with replica sets enabled to use feature
* Change streams are optional in HSS
* Timer will poll change stream for changes in the database
* As changes are detected, event is created to perform the correct
  action

* Changes made as suggested

configs/310014.yaml.in

@@ -19,6 +19,7 @@ parameter:
 #    no_sgwu: true
 #    no_pcrf: true
 #    no_hss: true
+#    use_mongodb_change_stream: true
 
 mme:
     freeDiameter:

configs/csfb.yaml.in

@@ -19,6 +19,7 @@ parameter:
 #    no_sgwu: true
 #    no_pcrf: true
 #    no_hss: true
+#    use_mongodb_change_stream: true
 
 mme:
     freeDiameter:

configs/non3gpp.yaml.in

@@ -19,6 +19,7 @@ parameter:
 #    no_sgwu: true
 #    no_pcrf: true
 #    no_hss: true
+#    use_mongodb_change_stream: true
 
 mme:
     freeDiameter:

configs/open5gs/hss.yaml.in

@@ -41,6 +41,7 @@ hss:
 #      prefer_ipv4: true
 #
 parameter:
+#    use_mongodb_change_stream: true
 
 #
 # max:

configs/sample.yaml.in

@@ -19,6 +19,7 @@ parameter:
 #    no_sgwu: true
 #    no_pcrf: true
 #    no_hss: true
+#    use_mongodb_change_stream: true
 
 mme:
     freeDiameter:

configs/slice.yaml.in

@@ -19,6 +19,7 @@ parameter:
 #    no_sgwu: true
 #    no_pcrf: true
 #    no_hss: true
+#    use_mongodb_change_stream: true
 
 mme:
     freeDiameter:

configs/srslte.yaml.in

@@ -19,6 +19,7 @@ parameter:
 #    no_sgwu: true
 #    no_pcrf: true
 #    no_hss: true
+#    use_mongodb_change_stream: true
 
 mme:
     freeDiameter:

configs/volte.yaml.in

@@ -19,6 +19,7 @@ parameter:
 #    no_sgwu: true
 #    no_pcrf: true
 #    no_hss: true
+#    use_mongodb_change_stream: true
 
 mme:
     freeDiameter:

configs/vonr.yaml.in

@@ -19,6 +19,7 @@ parameter:
 #    no_sgwu: true
 #    no_pcrf: true
 #    no_hss: true
+#    use_mongodb_change_stream: true
 
 mme:
     freeDiameter:

lib/app/ogs-context.c

@@ -350,6 +350,9 @@ int ogs_app_context_parse_config(void)
                 } else if (!strcmp(parameter_key, "no_pfcp_rr_select")) {
                     self.parameter.no_pfcp_rr_select =
                         ogs_yaml_iter_bool(&parameter_iter);
+                } else if (!strcmp(parameter_key, "use_mongodb_change_stream")) {
+                    self.use_mongodb_change_stream = 
+                        ogs_yaml_iter_bool(&parameter_iter);
                 } else
                     ogs_warn("unknown key `%s`", parameter_key);
             }

lib/app/ogs-context.h

@@ -35,6 +35,8 @@ typedef struct ogs_app_context_s {
     void *document;
 
     const char *db_uri;
+    int use_mongodb_change_stream;
+
     struct {
         const char *file;
         const char *level;

lib/dbi/meson.build

@@ -19,11 +19,14 @@ libdbi_sources = files('''
     ogs-dbi.h
 
     ogs-mongoc.h
+    timer.h
 
     ogs-mongoc.c
     subscription.c
     session.c
     ims.c
+    path.c
+    timer.c
 '''.split())
 
 libmongoc_dep = dependency('libmongoc-1.0')
@@ -35,10 +38,10 @@ libdbi = library('ogsdbi',
     version : libogslib_version,
     c_args : '-DOGS_DBI_COMPILATION',
     include_directories : [libdbi_inc, libinc],
-    dependencies : [libcrypt_dep, libmongoc_dep],
+    dependencies : [libcrypt_dep, libapp_dep, libmongoc_dep],
     install : true)
 
 libdbi_dep = declare_dependency(
     link_with : libdbi,
     include_directories : [libdbi_inc, libinc],
-    dependencies : [libcrypt_dep, libmongoc_dep])
+    dependencies : [libcrypt_dep, libapp_dep, libmongoc_dep])

lib/dbi/ogs-dbi.h

@@ -21,6 +21,7 @@
 #define OGS_DBI_H
 
 #include "crypt/ogs-crypt.h"
+#include "app/ogs-app.h"
 
 #define OGS_DBI_INSIDE
 
@@ -28,6 +29,8 @@
 #include "dbi/subscription.h"
 #include "dbi/session.h"
 #include "dbi/ims.h"
+#include "dbi/path.h"
+#include "dbi/timer.h"
 
 #undef OGS_DBI_INSIDE
 

lib/dbi/ogs-mongoc.c

@@ -182,5 +182,60 @@ void ogs_dbi_final()
         mongoc_collection_destroy(self.collection.subscriber);
     }
 
+    if (self.stream) {
+        mongoc_change_stream_destroy(self.stream);
+    }
+
     ogs_mongoc_final();
 }
+
+int ogs_dbi_collection_watch_init(void) {
+    bson_t empty = BSON_INITIALIZER;    
+    const bson_t *err_doc;
+    bson_error_t error;
+    bson_t *options = BCON_NEW("fullDocument", "updateLookup");
+   
+    ogs_mongoc()->stream = mongoc_collection_watch(self.collection.subscriber,
+        &empty, options);
+
+    if (mongoc_change_stream_error_document(ogs_mongoc()->stream, &error,
+            &err_doc)) {
+        if (!bson_empty (err_doc)) {
+            ogs_error("Change Stream Error.  Enable replica sets to "
+                "enable database updates to be sent to MME.");
+        } else {
+            ogs_error("Client Error: %s\n", error.message);
+        }
+        return OGS_ERROR;
+    } else {
+        ogs_info("Change Streams are Enabled.");
+    }
+
+    return OGS_OK;
+}
+
+int ogs_dbi_poll_change_stream(void) {
+    int rv;
+    
+    const bson_t *document;
+    const bson_t *err_document;
+    bson_error_t error;
+
+    while (mongoc_change_stream_next(ogs_mongoc()->stream, &document)) {
+        rv = ogs_dbi_process_change_stream(document);
+        if (rv != OGS_OK) return rv;
+    }
+
+    if (mongoc_change_stream_error_document(ogs_mongoc()->stream, &error, 
+            &err_document)) {
+        if (!bson_empty (err_document)) {
+            ogs_debug("Server Error: %s\n",
+            bson_as_relaxed_extended_json(err_document, NULL));
+        } else {
+            ogs_debug("Client Error: %s\n", error.message);
+        }
+        return OGS_ERROR;
+    }
+
+    return OGS_OK;
+}

lib/dbi/ogs-mongoc.h

@@ -37,6 +37,8 @@ typedef struct ogs_mongoc_s {
     void *client;
     void *database;
 
+    mongoc_change_stream_t *stream;
+
     char *masked_db_uri;
 
     struct {
@@ -51,6 +53,9 @@ ogs_mongoc_t *ogs_mongoc(void);
 int ogs_dbi_init(const char *db_uri);
 void ogs_dbi_final(void);
 
+int ogs_dbi_collection_watch_init(void);
+int ogs_dbi_poll_change_stream(void);
+
 #ifdef __cplusplus
 }
 #endif

lib/dbi/path.c

@@ -0,0 +1,40 @@
+/*
+ * Copyright (C) 2019,2020 by Sukchan Lee <acetcom@gmail.com>
+ *
+ * This file is part of Open5GS.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "ogs-dbi.h"
+
+int ogs_dbi_process_change_stream(const bson_t *document) {
+    int rv;
+
+    ogs_event_t *e = NULL;
+
+    e = ogs_event_new(OGS_EVENT_DBI_MESSAGE);
+    ogs_assert(e);
+    e->dbi.document = bson_copy(document);
+    rv = ogs_queue_push(ogs_app()->queue, e);
+    if (rv != OGS_OK) {
+        ogs_error("ogs_queue_push() failed:%d", (int)rv);
+        bson_destroy((bson_t*)e->dbi.document);
+        ogs_event_free(e);
+    } else {
+        ogs_pollset_notify(ogs_app()->pollset);
+    }
+
+    return OGS_OK;
+}

lib/dbi/path.h

@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2019,2020 by Sukchan Lee <acetcom@gmail.com>
+ *
+ * This file is part of Open5GS.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#ifndef OGS_DBI_PATH_H
+#define OGS_DBI_PATH_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int ogs_dbi_process_change_stream(const bson_t *document);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* OGS_DBI_PATH_H */

lib/dbi/timer.c

@@ -0,0 +1,41 @@
+/*
+ * Copyright (C) 2019 by Sukchan Lee <acetcom@gmail.com>
+ *
+ * This file is part of Open5GS.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "ogs-dbi.h"
+
+static void timer_send_event(int timer_id, void *data)
+{
+    int rv;
+    ogs_event_t *e = NULL;
+
+    e = ogs_event_new(OGS_EVENT_DBI_POLL_TIMER);
+    ogs_assert(e);
+    e->timer_id = timer_id;
+
+    rv = ogs_queue_push(ogs_app()->queue, e);
+    if (rv != OGS_OK) {
+        ogs_error("ogs_queue_push() failed:%d", (int)rv);
+        ogs_event_free(e);
+    }
+}
+
+void ogs_timer_dbi_poll_change_stream(void *data)
+{
+    timer_send_event(OGS_TIMER_DBI_POLL_CHANGE_STREAM, data);
+}

lib/dbi/timer.h

@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2019 by Sukchan Lee <acetcom@gmail.com>
+ *
+ * This file is part of Open5GS.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#ifndef OGS_DBI_TIMER_H
+#define OGS_DBI_TIMER_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+void ogs_timer_dbi_poll_change_stream(void *data);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* OGS_DBI_TIMER_H */

lib/proto/event.c

@@ -65,6 +65,11 @@ const char *ogs_event_get_name(ogs_event_t *e)
     case OGS_EVENT_SBI_TIMER:
         return OGS_EVENT_NAME_SBI_TIMER;
 
+    case OGS_EVENT_DBI_POLL_TIMER:
+        return "OGS_EVENT_DBI_POLL_TIMER";
+    case OGS_EVENT_DBI_MESSAGE:
+        return "OGS_EVENT_DBI_MESSAGE";
+
     default:
         break;
     }

lib/proto/event.h

@@ -39,6 +39,9 @@ typedef enum {
     OGS_EVENT_SBI_CLIENT,
     OGS_EVENT_SBI_TIMER,
 
+    OGS_EVENT_DBI_POLL_TIMER,
+    OGS_EVENT_DBI_MESSAGE,
+
     OGS_MAX_NUM_OF_PROTO_EVENT,
 
 } ogs_event_e;
@@ -47,6 +50,8 @@ typedef struct ogs_sbi_request_s ogs_sbi_request_t;
 typedef struct ogs_sbi_response_s ogs_sbi_response_t;
 typedef struct ogs_sbi_message_s ogs_sbi_message_t;
 
+typedef struct _bson_t bson_t;
+
 typedef struct ogs_event_s {
     int id;
     int timer_id;
@@ -59,6 +64,10 @@ typedef struct ogs_event_s {
 
         ogs_sbi_message_t *message;
     } sbi;
+
+    struct {
+        const bson_t *document;
+    } dbi;
 } ogs_event_t;
 
 void *ogs_event_size(int id, size_t size);

lib/proto/timer.c

@@ -31,6 +31,8 @@ const char *OGS_TIMER_NAME_SUBSCRIPTION_VALIDITY =
     "OGS_TIMER_SUBSCRIPTION_VALIDITY";
 const char *OGS_TIMER_NAME_SBI_CLIENT_WAIT =
     "OGS_TIMER_SBI_CLIENT_WAIT";
+const char *OGS_TIMER_NAME_DBI_POLL_CHANGE_STREAM =
+    "OGS_TIMER_DBI_POLL_CHANGE_STREAM";
 
 const char *ogs_timer_get_name(int timer_id)
 {
@@ -47,6 +49,8 @@ const char *ogs_timer_get_name(int timer_id)
         return OGS_TIMER_NAME_SUBSCRIPTION_VALIDITY;
     case OGS_TIMER_SBI_CLIENT_WAIT:
         return OGS_TIMER_NAME_SBI_CLIENT_WAIT;
+    case OGS_TIMER_DBI_POLL_CHANGE_STREAM:
+        return OGS_TIMER_NAME_DBI_POLL_CHANGE_STREAM;
     default: 
        break;
     }

lib/proto/timer.h

@@ -46,6 +46,8 @@ typedef enum {
     OGS_TIMER_SUBSCRIPTION_VALIDITY,
     OGS_TIMER_SBI_CLIENT_WAIT,
 
+    OGS_TIMER_DBI_POLL_CHANGE_STREAM,
+
     OGS_MAX_NUM_OF_PROTO_TIMER,
 
 } ogs_timer_e;

src/hss/hss-context.c

@@ -19,6 +19,9 @@
 
 #include "ogs-dbi.h"
 #include "hss-context.h"
+#include "hss-event.h"
+#include "hss-s6a-path.h"
+
 
 typedef struct hss_impi_s hss_impi_t;
 
@@ -680,7 +683,7 @@ void hss_cx_associate_identity(char *user_name, char *public_identity)
 {
     hss_impi_t *impi = NULL;
     hss_impu_t *impu = NULL;
-    
+
     ogs_assert(user_name);
     ogs_assert(public_identity);
 
@@ -1171,3 +1174,132 @@ char *hss_cx_download_user_data(
 
     return user_data;
 }
+
+int hss_db_poll_change_stream(void) {
+    int rv;
+
+    ogs_thread_mutex_lock(&self.db_lock);
+
+    rv = ogs_dbi_poll_change_stream();
+
+    ogs_thread_mutex_unlock(&self.db_lock);
+
+    return rv;
+}
+
+int hss_handle_change_event(const bson_t *document) {
+    bson_iter_t iter, child1_iter, child2_iter, child3_iter;
+
+    char *utf8 = NULL;
+    uint32_t length = 0;
+
+    bool send_clr_flag = false;
+    bool send_idr_flag = false;
+    uint32_t subdatamask = 0;
+
+    char *imsi_bcd;
+
+    char *as_json = bson_as_relaxed_extended_json(document, NULL);
+    ogs_debug("Got document: %s\n", as_json);
+    if (!bson_iter_init_find(&iter, document, "fullDocument")) {
+        ogs_error("No 'imsi' field in this document.");
+        return OGS_ERROR;
+    } else {
+        bson_iter_recurse(&iter, &child1_iter);
+        while (bson_iter_next(&child1_iter)) {
+            const char *key = bson_iter_key(&child1_iter);
+            if (!strcmp(key, "imsi") && 
+                    BSON_ITER_HOLDS_UTF8(&child1_iter)) {
+                utf8 = (char *)bson_iter_utf8(&child1_iter, &length);
+                imsi_bcd = ogs_strndup(utf8,
+                    ogs_min(length, OGS_MAX_IMSI_BCD_LEN) + 1);
+                ogs_assert(imsi_bcd);
+            }
+        }
+    }
+
+    if (!imsi_bcd) {
+        ogs_error("No 'imsi' field in this document.");
+        return OGS_ERROR;
+    }
+
+    if (bson_iter_init_find(&iter, document, "updateDescription")) {
+        bson_iter_recurse(&iter, &child1_iter);
+        while (bson_iter_next(&child1_iter)) {
+            const char *key = bson_iter_key(&child1_iter);
+            if (!strcmp(key, "updatedFields") && 
+                    BSON_ITER_HOLDS_DOCUMENT(&child1_iter)) {
+                bson_iter_recurse(&child1_iter, &child2_iter);
+                while (bson_iter_next(&child2_iter)) {
+                    const char *child2_key = bson_iter_key(&child2_iter);
+                    if (!strcmp(child2_key, 
+                            "request_cancel_location") && 
+                            BSON_ITER_HOLDS_BOOL(&child2_iter)) {
+                        send_clr_flag = (char *)bson_iter_bool(&child2_iter);
+                    } else if (!strncmp(child2_key, "msisdn",
+                            strlen("msisdn"))) {
+                        int msisdn_count = 0;
+                        bson_iter_recurse(&child2_iter, &child3_iter);
+                        while (bson_iter_next(&child3_iter)) {
+                            if (BSON_ITER_HOLDS_UTF8(&child3_iter)) {
+                                msisdn_count++;
+                            }
+                        }
+                        if (msisdn_count) {
+                            send_idr_flag = true;
+                            subdatamask = (subdatamask | 
+                                OGS_HSS_SUBDATA_MSISDN);
+                        } else {
+                            send_clr_flag = true;
+                        }
+                    } else if (!strncmp(child2_key, 
+                            "access_restriction_data",
+                            strlen("access_restriction_data"))) {
+                        send_idr_flag = true;
+                        subdatamask = (subdatamask | OGS_HSS_SUBDATA_ARD);
+                    } else if (!strncmp(child2_key, 
+                            "subscriber_status", 
+                            strlen("subscriber_status"))) {
+                        send_idr_flag = true;
+                        subdatamask = (subdatamask | 
+                            OGS_HSS_SUBDATA_SUB_STATUS);
+                    } else if (!strncmp(child2_key, 
+                            "network_access_mode",
+                            strlen("network_access_mode"))) {
+                        send_idr_flag = true;
+                        subdatamask = (subdatamask | OGS_HSS_SUBDATA_NAM);
+                    } else if (!strncmp(child2_key, "ambr", strlen("ambr"))) {
+                        send_idr_flag = true;
+                        subdatamask = (subdatamask | OGS_HSS_SUBDATA_UEAMBR);
+                    } else if (!strncmp(child2_key, 
+                            "subscribed_rau_tau_timer",
+                            strlen("subscribed_rau_tau_timer"))) {
+                        send_idr_flag = true;
+                        subdatamask = (subdatamask | 
+                            OGS_HSS_SUBDATA_RAU_TAU_TIMER);
+                    } else if (!strncmp(child2_key, "slice", strlen("slice"))) {
+                        send_idr_flag = true;
+                        subdatamask = (subdatamask | OGS_HSS_SUBDATA_SLICE);
+                    }
+                }
+            }
+        }
+    } else {
+        ogs_debug("No 'updateDescription' field in this document");
+    }
+
+    bson_free (as_json);
+
+    if (send_clr_flag) {
+        ogs_info("[%s] Cancel Location Requested", imsi_bcd);
+        hss_s6a_send_clr(imsi_bcd, NULL, NULL,
+            OGS_DIAM_S6A_CT_SUBSCRIPTION_WITHDRAWL);
+    } else if (send_idr_flag) {
+        ogs_info("[%s] Subscription-Data Changed", imsi_bcd);
+        hss_s6a_send_idr(imsi_bcd, 0, subdatamask);
+    }
+
+    ogs_free(imsi_bcd);
+
+    return OGS_OK;
+}

src/hss/hss-context.h

@@ -92,6 +92,9 @@ char *hss_cx_download_user_data(
         char *user_name, char *visited_network_identifier,
         ogs_ims_data_t *ims_data);
 
+int hss_db_poll_change_stream(void);
+int hss_handle_change_event(const bson_t *document);
+
 
 #ifdef __cplusplus
 }

src/hss/hss-event.c

@@ -0,0 +1,27 @@
+/*
+ * Copyright (C) 2019 by Sukchan Lee <acetcom@gmail.com>
+ *
+ * This file is part of Open5GS.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "hss-event.h"
+#include "ogs-app.h"
+
+void hss_event_term(void)
+{
+    ogs_queue_term(ogs_app()->queue);
+    ogs_pollset_notify(ogs_app()->pollset);
+}

src/hss/hss-event.h

@@ -0,0 +1,35 @@
+/*
+ * Copyright (C) 2019 by Sukchan Lee <acetcom@gmail.com>
+ *
+ * This file is part of Open5GS.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#ifndef HSS_EVENT_H
+#define HSS_EVENT_H
+
+#include "ogs-core.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+void hss_event_term(void);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* HSS_EVENT_H */

src/hss/hss-fd-path.h

@@ -39,4 +39,3 @@ void hss_swx_final(void);
 #endif
 
 #endif /* HSS_FD_PATH_H */
-

src/hss/hss-init.c

@@ -19,6 +19,11 @@
 
 #include "hss-context.h"
 #include "hss-fd-path.h"
+#include "hss-sm.h"
+
+
+static ogs_thread_t *thread;
+static void hss_main(void *data);
 
 static int initialized = 0;
 
@@ -41,6 +46,9 @@ int hss_initialize(void)
     rv = hss_fd_init();
     if (rv != OGS_OK) return OGS_ERROR;
 
+    thread = ogs_thread_create(hss_main, NULL);
+    if (!thread) return OGS_ERROR;
+
     initialized = 1;
 
     return OGS_OK;
@@ -50,6 +58,9 @@ void hss_terminate(void)
 {
     if (!initialized) return;
 
+    hss_event_term();
+    ogs_thread_destroy(thread);
+
     hss_fd_final();
 
     ogs_dbi_final();
@@ -57,3 +68,49 @@ void hss_terminate(void)
 
     return;
 }
+
+static void hss_main(void *data)
+{
+    ogs_fsm_t hss_sm;
+    int rv;
+
+    ogs_fsm_init(&hss_sm, hss_state_initial, hss_state_final, 0);
+
+    for ( ;; ) {
+        ogs_pollset_poll(ogs_app()->pollset,
+                ogs_timer_mgr_next(ogs_app()->timer_mgr));
+
+        /*
+         * After ogs_pollset_poll(), ogs_timer_mgr_expire() must be called.
+         *
+         * The reason is why ogs_timer_mgr_next() can get the corrent value
+         * when ogs_timer_stop() is called internally in ogs_timer_mgr_expire().
+         *
+         * You should not use event-queue before ogs_timer_mgr_expire().
+         * In this case, ogs_timer_mgr_expire() does not work
+         * because 'if rv == OGS_DONE' statement is exiting and
+         * not calling ogs_timer_mgr_expire().
+         */
+        ogs_timer_mgr_expire(ogs_app()->timer_mgr);
+
+        for ( ;; ) {
+            ogs_event_t *e = NULL;
+
+            rv = ogs_queue_trypop(ogs_app()->queue, (void**)&e);
+            ogs_assert(rv != OGS_ERROR);
+
+            if (rv == OGS_DONE)
+                goto done;
+
+            if (rv == OGS_RETRY)
+                break;
+
+            ogs_assert(e);
+            ogs_fsm_dispatch(&hss_sm, e);
+            ogs_event_free(e);
+        }
+    }
+done:
+
+    ogs_fsm_fini(&hss_sm, 0);
+}

src/hss/hss-s6a-path.c

@@ -719,7 +719,7 @@ static int hss_s6a_avp_add_subscription_data(
             ret = fd_msg_avp_add(apn_configuration_profile,
                     MSG_BRW_LAST_CHILD, apn_configuration);
             ogs_assert(ret == 0);
-            
+
         }
         ret = fd_msg_avp_add(avp, MSG_BRW_LAST_CHILD,
                 apn_configuration_profile);
@@ -825,7 +825,7 @@ static int hss_ogs_diam_s6a_ulr_cb( struct msg **msg, struct avp *avp,
     ogs_assert(mme_host);
     ogs_assert(mme_realm);
 
-    /* If UE is not purged at MME, determine if the MME sending the ULR  
+    /* If UE is not purged at MME, determine if the MME sending the ULR 
      * is different from the one that was last used.  if so, send CLR.
      */
     if (subscription_data.mme_host != NULL &&
@@ -1269,6 +1269,7 @@ void hss_s6a_send_clr(char *imsi_bcd, char *mme_host, char *mme_realm,
     ogs_assert(pthread_mutex_lock(&ogs_diam_logger_self()->stats_lock) == 0);
     ogs_diam_logger_self()->stats.nb_sent++;
     ogs_assert(pthread_mutex_unlock(&ogs_diam_logger_self()->stats_lock) == 0);
+
 }
 
 /* HSS received Cancel Location Answer from MME */
@@ -1286,7 +1287,7 @@ static void hss_s6a_cla_cb(void *data, struct msg **msg)
     ret = fd_msg_sess_get(fd_g_config->cnf_dict, *msg, &session, &new);
     ogs_expect_or_return(ret == 0);
     ogs_expect_or_return(new == 0);
-    
+
     ret = fd_sess_state_retrieve(hss_s6a_reg, session, &sess_data);
     ogs_expect_or_return(ret == 0);
     ogs_expect_or_return(sess_data);
@@ -1326,8 +1327,8 @@ int hss_s6a_send_idr(char *imsi_bcd, uint32_t idr_flags, uint32_t subdatamask)
 
     if (subscription_data.purge_flag) {
         ogs_error("    [%s] UE Purged at MME.  Cannot send IDR.", imsi_bcd);
-        return OGS_ERROR;        
-    }    
+        return OGS_ERROR;
+    }
 
     /* Create the random value to store with the session */
     sess_data = ogs_calloc(1, sizeof(*sess_data));
@@ -1405,7 +1406,7 @@ int hss_s6a_send_idr(char *imsi_bcd, uint32_t idr_flags, uint32_t subdatamask)
         ret = fd_msg_avp_add(req, MSG_BRW_LAST_CHILD, avp);
         ogs_assert(ret == 0);
     }
-    
+
     /* Set the Subscription Data */
     ret = fd_msg_avp_new(ogs_diam_s6a_subscription_data, 0, &avp);
     ogs_assert(ret == 0);
@@ -1415,11 +1416,11 @@ int hss_s6a_send_idr(char *imsi_bcd, uint32_t idr_flags, uint32_t subdatamask)
             if (ret != OGS_OK) {
                 ogs_error("    [%s] Could not build Subscription-Data.", 
                     imsi_bcd);
-                return OGS_ERROR;   
+                return OGS_ERROR;
             }
         }
     ret = fd_msg_avp_add(req, MSG_BRW_LAST_CHILD, avp);
-    ogs_assert(ret == 0);    
+    ogs_assert(ret == 0);
 
     /* Set Vendor-Specific-Application-Id AVP */
     ret = ogs_diam_message_vendor_specific_appid_set(
@@ -1464,7 +1465,7 @@ static void hss_s6a_ida_cb(void *data, struct msg **msg)
     ret = fd_msg_sess_get(fd_g_config->cnf_dict, *msg, &session, &new);
     ogs_expect_or_return(ret == 0);
     ogs_expect_or_return(new == 0);
-    
+
     ret = fd_sess_state_retrieve(hss_s6a_reg, session, &sess_data);
     ogs_expect_or_return(ret == 0);
     ogs_expect_or_return(sess_data);
@@ -1527,7 +1528,7 @@ int hss_s6a_init(void)
 void hss_s6a_final(void)
 {
     int ret;
-    
+
     ret = fd_sess_handler_destroy(&hss_s6a_reg, NULL);
     ogs_assert(ret == OGS_OK);
 

src/hss/hss-sm.c

@@ -0,0 +1,98 @@
+/*
+ * Copyright (C) 2019-2022 by Sukchan Lee <acetcom@gmail.com>
+ *
+ * This file is part of Open5GS.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#include "hss-sm.h"
+#include "hss-context.h"
+#include "hss-event.h"
+
+static ogs_timer_t *t_db_polling = NULL;
+
+void hss_state_initial(ogs_fsm_t *s, ogs_event_t *e)
+{
+    hss_sm_debug(e);
+
+    ogs_assert(s);
+
+    if (ogs_app()->use_mongodb_change_stream) {
+        ogs_dbi_collection_watch_init();
+
+        t_db_polling = ogs_timer_add(ogs_app()->timer_mgr,
+                ogs_timer_dbi_poll_change_stream, 0);
+        ogs_assert(t_db_polling);
+        ogs_timer_start(t_db_polling, DB_POLLING_TIME);
+
+        OGS_FSM_TRAN(s, &hss_state_operational);
+    }
+}
+
+void hss_state_final(ogs_fsm_t *s, ogs_event_t *e)
+{
+    hss_sm_debug(e);
+
+    if (t_db_polling)
+        ogs_timer_delete(t_db_polling);
+
+    ogs_assert(s);
+}
+
+void hss_state_operational(ogs_fsm_t *s, ogs_event_t *e)
+{
+    hss_sm_debug(e);
+
+    ogs_assert(s);
+
+    switch (e->id) {
+    case OGS_FSM_ENTRY_SIG:
+        break;
+
+    case OGS_FSM_EXIT_SIG:
+        if (t_db_polling) {
+            ogs_timer_stop(t_db_polling);
+        }
+        break;
+
+    case OGS_EVENT_DBI_POLL_TIMER:
+        ogs_assert(e);
+
+        switch(e->timer_id) {
+        case OGS_TIMER_DBI_POLL_CHANGE_STREAM:
+            hss_db_poll_change_stream();
+            ogs_timer_start(t_db_polling, DB_POLLING_TIME);
+            break;
+
+        default:
+            ogs_error("Unknown timer[%s:%d]",
+                    ogs_timer_get_name(e->timer_id), e->timer_id);
+        }
+        break;
+
+    case OGS_EVENT_DBI_MESSAGE:
+        ogs_assert(e);
+
+        ogs_assert(e->dbi.document);
+        hss_handle_change_event(e->dbi.document);
+
+        bson_destroy((bson_t*)e->dbi.document);
+        break;
+
+    default:
+        ogs_error("No handler for event %s", ogs_event_get_name(e));
+        break;
+    }
+}

src/hss/hss-sm.h

@@ -0,0 +1,44 @@
+/*
+ * Copyright (C) 2019-2022 by Sukchan Lee <acetcom@gmail.com>
+ *
+ * This file is part of Open5GS.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#ifndef HSS_SM_H
+#define HSS_SM_H
+
+#include "hss-event.h"
+#include "ogs-proto.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define DB_POLLING_TIME ogs_time_from_msec(100)
+
+void hss_state_initial(ogs_fsm_t *s, ogs_event_t *e);
+void hss_state_final(ogs_fsm_t *s, ogs_event_t *e);
+void hss_state_operational(ogs_fsm_t *s, ogs_event_t *e);
+void hss_state_exception(ogs_fsm_t *s, ogs_event_t *e);
+
+#define hss_sm_debug(__pe) \
+    ogs_debug("%s(): %s", __func__, ogs_event_get_name(__pe))
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* HSS_SM_H */

src/hss/meson.build

@@ -19,9 +19,13 @@ libhss_sources = files('''
     hss-context.h
     hss-fd-path.h
     hss-s6a-path.h
+    hss-event.h
+    hss-sm.h
 
     hss-init.c
     hss-context.c
+    hss-event.c
+    hss-sm.c
 
     hss-s6a-path.c
     hss-cx-path.c