add ric_subscription class

srsenb/hdr/enb.h

@@ -34,6 +34,7 @@
 #include "srsenb/hdr/stack/mac/sched_interface.h"
 #include "srsgnb/hdr/stack/gnb_stack_nr.h"
 #include "srsgnb/hdr/stack/ric/ric_client.h"
+#include "srsgnb/hdr/stack/ric/ric_subscription.h"
 #include "srsran/common/bcd_helpers.h"
 #include "srsran/common/buffer_pool.h"
 #include "srsran/common/interfaces_common.h"

srsenb/src/enb.cc

@@ -274,6 +274,11 @@ void enb::tti_clock()
   if (!started) {
     return;
   }
+
+  if (ric) {
+    ric->tic();
+  }
+
   if (eutra_stack) {
     eutra_stack->tti_clock();
   }

srsgnb/hdr/stack/ric/ric_client.h

@@ -48,10 +48,13 @@ public:
   bool init(ric_args_t args);
   void stop();
   void run_thread();
+  void tic();
 
   // Send messages to RIC
   bool send_sctp(srsran::unique_byte_buffer_t& buf);
   bool send_e2_msg(e2_msg_type_t msg_type);
+  bool queue_send_e2ap_pdu(e2_ap_pdu_c send_pdu);
+  bool send_e2ap_pdu(e2_ap_pdu_c send_pdu);
   bool send_reset_response();
 
   // Handle messages received from RIC
@@ -66,6 +69,8 @@ public:
   bool handle_reset_response(reset_resp_s& reset_response);
   bool handle_reset_request(reset_request_s& reset_request);
 
+  class ric_subscription;
+
 private:
   e2ap                      e2ap_;
   srsran::unique_socket     ric_socket;
@@ -76,6 +81,8 @@ private:
   struct sockaddr_in        ric_addr = {}; // RIC address
   bool                      running  = false;
   srsenb::e2_interface_metrics* gnb_metrics = nullptr;
+
+  std::vector<std::unique_ptr<ric_subscription> > active_subscriptions;
 };
 } // namespace srsenb
 

srsgnb/hdr/stack/ric/ric_subscription.h

@@ -0,0 +1,51 @@
+/**
+ *
+ * \section COPYRIGHT
+ *
+ * Copyright 2013-2022 Software Radio Systems Limited
+ *
+ * By using this file, you agree to the terms and conditions set
+ * forth in the LICENSE file which can be found at the top level of
+ * the distribution.
+ *
+ *
+ */
+
+#ifndef SRSRAN_RIC_SUBSCRIPTION_H
+#define SRSRAN_RIC_SUBSCRIPTION_H
+
+#include "srsgnb/hdr/stack/ric/e2ap.h"
+#include "srsgnb/hdr/stack/ric/ric_client.h"
+#include "srsran/common/task_scheduler.h"
+#include "srsran/common/threads.h"
+#include "srsran/srsran.h"
+
+namespace srsenb {
+
+class ric_client::ric_subscription
+{
+public:
+  ric_subscription(ric_client* ric_client, ricsubscription_request_s ric_subscription_request);
+  virtual ~ric_subscription() = default;
+
+  uint32_t get_ric_requestor_id() { return ric_requestor_id; };
+  uint32_t get_ric_instance_id() { return ric_instance_id; };
+
+  void start_ric_indication_reporting();
+  void stop_ric_indication_reporting();
+
+private:
+  void send_ric_indication();
+
+  ric_client* parent = nullptr;
+
+  uint32_t ric_requestor_id;
+  uint32_t ric_instance_id;
+  uint16_t ra_nfunction_id;
+
+  uint32_t             reporting_period; // ms
+  srsran::unique_timer reporting_timer;  // for RIC indication reporting
+};
+
+} // namespace srsenb
+#endif // SRSRAN_RIC_SUBSCRIPTION_H

srsgnb/src/stack/ric/CMakeLists.txt

@@ -1,4 +1,4 @@
-set(SOURCES ric_client.cc e2ap.cc e2sm_kpm.cc)
+set(SOURCES ric_client.cc ric_subscription.cc e2ap.cc e2sm_kpm.cc)
 add_library(srsgnb_ric STATIC ${SOURCES})
 target_link_libraries(srsgnb_ric srsran_asn1 ric_e2)
 

srsgnb/src/stack/ric/ric_client.cc

@@ -11,6 +11,7 @@
  */
 
 #include "srsgnb/hdr/stack/ric/ric_client.h"
+#include "srsgnb/hdr/stack/ric/ric_subscription.h"
 #include "srsran/asn1/e2ap.h"
 #include "stdint.h"
 
@@ -66,6 +67,12 @@ void ric_client::stop()
   wait_thread_finish();
 }
 
+void ric_client::tic()
+{
+  // get tick every 1ms to advance timers
+  task_sched.tic();
+}
+
 void ric_client::run_thread()
 {
   using namespace asn1::e2ap;
@@ -75,7 +82,6 @@ void ric_client::run_thread()
       send_e2_msg(E2_SETUP_REQUEST);
       printf("e2 setup request sent\n");
     }
-    sleep(1);
     task_sched.run_next_task();
   }
 }
@@ -114,18 +120,6 @@ bool ric_client::send_e2_msg(e2_msg_type_t msg_type)
       send_pdu     = e2ap_.generate_setup_request();
       message_name = "E2 SETUP REQUEST";
       break;
-    case e2_msg_type_t::E2_SUB_RESPONSE:
-      send_pdu     = e2ap_.generate_subscription_response();
-      message_name = "E2 SUBSCRIPTION RESPONSE";
-      break;
-    case e2_msg_type_t::E2_SUB_DEL_RESPONSE:
-      send_pdu     = e2ap_.generate_subscription_delete_response();
-      message_name = "E2 SUBSCRIPTION DELETE RESPONSE";
-      break;
-    case e2_msg_type_t::E2_INDICATION:
-      send_pdu     = e2ap_.generate_indication();
-      message_name = "E2 INDICATION";
-      break;
     case e2_msg_type_t::E2_RESET:
       send_pdu     = e2ap_.generate_reset_request();
       message_name = "E2 RESET REQUEST";
@@ -152,6 +146,34 @@ bool ric_client::send_e2_msg(e2_msg_type_t msg_type)
   return true;
 }
 
+bool ric_client::queue_send_e2ap_pdu(e2_ap_pdu_c e2ap_pdu)
+{
+  auto send_e2ap_pdu_task = [this, e2ap_pdu]() { send_e2ap_pdu(e2ap_pdu); };
+  ric_rece_task_queue.push(send_e2ap_pdu_task);
+  return true;
+}
+
+bool ric_client::send_e2ap_pdu(e2_ap_pdu_c send_pdu)
+{
+  srsran::unique_byte_buffer_t buf = srsran::make_byte_buffer();
+  if (buf == nullptr) {
+    // logger.error("Fatal Error: Couldn't allocate buffer for %s.", procedure_name);
+    return false;
+  }
+  asn1::bit_ref bref(buf->msg, buf->get_tailroom());
+  if (send_pdu.pack(bref) != asn1::SRSASN_SUCCESS) {
+    logger.error("Failed to pack TX E2 PDU");
+    return false;
+  }
+  buf->N_bytes = bref.distance_bytes();
+  printf("try to send %d bytes to addr %s \n", buf->N_bytes, inet_ntoa(ric_addr.sin_addr));
+  if (!send_sctp(buf)) {
+    logger.error("failed to send");
+    return false;
+  }
+  return true;
+}
+
 bool ric_client::handle_e2_rx_msg(srsran::unique_byte_buffer_t pdu,
                                   const sockaddr_in&           from,
                                   const sctp_sndrcvinfo&       sri,
@@ -257,27 +279,39 @@ bool ric_client::handle_e2_unsuccessful_outcome(asn1::e2ap::unsuccessful_outcome
 
 bool ric_client::handle_ric_subscription_request(ricsubscription_request_s ric_subscription_request)
 {
-  auto send_sub_resp = [this]() { send_e2_msg(E2_SUB_RESPONSE); };
+  logger.info("Received RIC Subscription Request from RIC ID: %i (instance id %i) to RAN Function ID: %i",
-  ric_rece_task_queue.push(send_sub_resp);
-
-  logger.info("Received subscription request from RIC ID: %i (instance id %i) to RAN Function ID: %i \n",
               ric_subscription_request->ri_crequest_id->ric_requestor_id,
               ric_subscription_request->ri_crequest_id->ric_instance_id,
               ric_subscription_request->ra_nfunction_id->value);
 
-  e2ap_.process_subscription_request(ric_subscription_request);
+  std::unique_ptr<ric_client::ric_subscription> new_ric_subs =
-
+      std::make_unique<ric_client::ric_subscription>(this, ric_subscription_request);
-  auto send_ind = [this]() { for(uint16_t i = 0; i < 5; i++) {send_e2_msg(E2_INDICATION);
+  new_ric_subs->start_ric_indication_reporting();
-      sleep(1);}};
+  active_subscriptions.push_back(std::move(new_ric_subs));
-  ric_rece_task_queue.push(send_ind);
-
   return true;
 }
 
 bool ric_client::handle_ric_subscription_delete_request(ricsubscription_delete_request_s ricsubscription_delete_request)
 {
-  auto send_resp = [this]() { send_e2_msg(E2_SUB_DEL_RESPONSE); };
+  logger.info("Received RIC Subscription Delete request from RIC ID: %i (instance id %i) to RAN Function ID: %i",
-  ric_rece_task_queue.push(send_resp);
+              ricsubscription_delete_request->ri_crequest_id->ric_requestor_id,
+              ricsubscription_delete_request->ri_crequest_id->ric_instance_id,
+              ricsubscription_delete_request->ra_nfunction_id->value);
+
+  bool ric_subs_found = false;
+  for (auto it = active_subscriptions.begin(); it != active_subscriptions.end(); it++) {
+    if ((**it).get_ric_requestor_id() == ricsubscription_delete_request->ri_crequest_id->ric_requestor_id and
+        (**it).get_ric_instance_id() == ricsubscription_delete_request->ri_crequest_id->ric_instance_id) {
+      ric_subs_found = true;
+      (**it).stop_ric_indication_reporting();
+      active_subscriptions.erase(it);
+      break;
+    }
+  }
+
+  if (not ric_subs_found) {
+    // TODO: send failure
+  }
 
   return true;
 }

srsgnb/src/stack/ric/ric_subscription.cc

@@ -0,0 +1,66 @@
+/**
+ *
+ * \section COPYRIGHT
+ *
+ * Copyright 2013-2022 Software Radio Systems Limited
+ *
+ * By using this file, you agree to the terms and conditions set
+ * forth in the LICENSE file which can be found at the top level of
+ * the distribution.
+ *
+ *
+ */
+
+#include "srsgnb/hdr/stack/ric/ric_subscription.h"
+
+namespace srsenb {
+
+ric_client::ric_subscription::ric_subscription(ric_client*               ric_client,
+                                               ricsubscription_request_s ric_subscription_request) :
+  parent(ric_client),
+  ric_requestor_id(ric_subscription_request->ri_crequest_id->ric_requestor_id),
+  ric_instance_id(ric_subscription_request->ri_crequest_id->ric_instance_id),
+  ra_nfunction_id(ric_subscription_request->ra_nfunction_id->value)
+{
+  reporting_period = 1000;
+  reporting_timer  = parent->task_sched.get_unique_timer();
+
+  // TODO: process request to know what to report
+  parent->e2ap_.process_subscription_request(ric_subscription_request);
+}
+
+void ric_client::ric_subscription::start_ric_indication_reporting()
+{
+  parent->logger.debug("Send RIC Subscription Response to RIC Requestor ID: %i\n", ric_requestor_id);
+  e2_ap_pdu_c send_pdu = parent->e2ap_.generate_subscription_response();
+  parent->queue_send_e2ap_pdu(send_pdu);
+
+  printf("Start sending RIC indication msgs every %i ms\n", reporting_period);
+  parent->logger.debug("Start sending RIC indication msgs every %i ms", reporting_period);
+  reporting_timer.set(reporting_period, [this](uint32_t tid) { send_ric_indication(); });
+  reporting_timer.run();
+}
+
+void ric_client::ric_subscription::stop_ric_indication_reporting()
+{
+  if (reporting_timer.is_running()) {
+    parent->logger.debug("Stop sending RIC indication msgs");
+    reporting_timer.stop();
+  }
+
+  parent->logger.debug("Send RIC Subscription Delete Response to RIC Requestor ID: %i\n", ric_requestor_id);
+  e2_ap_pdu_c send_pdu = parent->e2ap_.generate_subscription_delete_response();
+  parent->queue_send_e2ap_pdu(send_pdu);
+}
+
+void ric_client::ric_subscription::send_ric_indication()
+{
+  printf("Sending RIC indication msg to RIC Requestor ID: %i\n", ric_requestor_id);
+  e2_ap_pdu_c send_pdu = parent->e2ap_.generate_indication();
+  parent->queue_send_e2ap_pdu(send_pdu);
+
+  // reschedule sending RIC indication
+  reporting_timer.run();
+}
+
+} // namespace srsenb