Release v1.3.1 - Multicast/Broadcast Services Transport Function

meson.build

@@ -10,7 +10,7 @@
 # Meson module fs and its functions like fs.hash_file require atleast meson 0.59.0 
 
 project('rt-mbs-transport-function', 'c', 'cpp',
-        version : '1.3.0',
+        version : '1.3.1',
         license : '5G-MAG Public',
         meson_version : '>= 1.4.0',
         default_options : [

src/mbstf/Curl.cc

@@ -11,6 +11,8 @@
  * https://drive.google.com/file/d/1cinCiA778IErENZ3JN52VFW-1ffHpx7Z/view
  */
 
+#include <unistd.h>
+
 #include <chrono>
 #include <iostream>
 #include <map>
@@ -230,6 +232,16 @@ int Curl::getResponseCode() const
     return m_statusCode;
 }
 
+void Curl::abortFetch()
+{
+    if (!m_curl) return;
+    curl_socket_t sockfd;
+    CURLcode res = curl_easy_getinfo(m_curl, CURLINFO_ACTIVESOCKET, &sockfd);
+    if (res == CURLE_OK) {
+        close(sockfd);
+    }
+}
+
 bool Curl::extractProtocolAndStatusCode(std::string_view &header_line)
 {
     header_line.remove_prefix(5); // Skip "HTTP/"

src/mbstf/Curl.hh

@@ -45,6 +45,8 @@ public:
     unsigned long getAge() const;
     int getResponseCode() const;
 
+    void abortFetch();
+
     Curl &setUserAgent(const std::string &user_agent);
 
 private:

src/mbstf/DistributionSession.cc

@@ -388,8 +388,7 @@ bool DistributionSession::processEvent(Open5GSEvent &event)
                                 ogs_assert(true == Open5GSSBIServer::sendResponse(stream, *response));
                             } else {
                                 std::ostringstream err;
-                                err << "Distribution Session [" << ptr_resource1 << "], method [" << method
-                                    << "] is not allowed for a Distribution Session";
+                                err << "Distribution Sessions method [" << method << "] is not allowed for a Distribution Session";
                                 ogs_error("%s", err.str().c_str());
                                 ogs_assert(true == NfServer::sendError(stream, OGS_SBI_HTTP_STATUS_MEHTOD_NOT_ALLOWED,
                                                                           2, message, app_meta, api, std::nullopt, err.str()));

src/mbstf/ObjectCarouselPackager.cc

@@ -124,7 +124,7 @@ ObjectCarouselPackager::ObjectCarouselPackager(ObjectStore &object_store, Object
     ,m_maxStreams(0)
 {
     if (tunnel_address) {
-        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::address::from_string(tunnel_address.value()), tunnel_port);
+        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::make_address(tunnel_address.value()), tunnel_port);
     }
     startWorker();
     startScheduler();
@@ -146,7 +146,7 @@ ObjectCarouselPackager::ObjectCarouselPackager(ObjectStore &object_store, Object
     ,m_maxStreams(0)
 {
     if (tunnel_address) {
-        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::address::from_string(tunnel_address.value()), tunnel_port);
+        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::make_address(tunnel_address.value()), tunnel_port);
     }
     startWorker();
     startScheduler();
@@ -168,7 +168,7 @@ ObjectCarouselPackager::ObjectCarouselPackager(ObjectStore &object_store, Object
     ,m_maxStreams(0)
 {
     if (tunnel_address) {
-        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::address::from_string(tunnel_address.value()), tunnel_port);
+        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::make_address(tunnel_address.value()), tunnel_port);
     }
     startWorker();
     startScheduler();

src/mbstf/ObjectListPackager.cc

@@ -91,7 +91,7 @@ ObjectListPackager::ObjectListPackager(ObjectStore &object_store, ObjectControll
 {
     sortListByPolicy();
     if (tunnel_address) {
-        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::address::from_string(tunnel_address.value()), tunnel_port);
+        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::make_address(tunnel_address.value()), tunnel_port);
     }
     startWorker();
 }
@@ -106,7 +106,7 @@ ObjectListPackager::ObjectListPackager(ObjectStore &object_store, ObjectControll
 {
     sortListByPolicy();
     if (tunnel_address) {
-        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::address::from_string(tunnel_address.value()), tunnel_port);
+        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::make_address(tunnel_address.value()), tunnel_port);
     }
     startWorker();
 }
@@ -120,7 +120,7 @@ ObjectListPackager::ObjectListPackager(ObjectStore &object_store, ObjectControll
     ,m_tunnelEndpoint()
 {
     if (tunnel_address) {
-        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::address::from_string(tunnel_address.value()), tunnel_port);
+        m_tunnelEndpoint = boost::asio::ip::udp::endpoint(boost::asio::ip::make_address(tunnel_address.value()), tunnel_port);
     }
     startWorker();
 }

src/mbstf/ObjectManifestHandler.cc

@@ -176,6 +176,7 @@ bool ObjectManifestHandler::update(const std::shared_ptr<ObjectStore::Object> &n
 
     const auto &new_objects = new_manifest.getObjects();
     const auto &old_objects = m_objectManifest.getObjects();
+    ObjectManifest::ObjectsType to_del;
     for (auto old_it = old_objects.begin(); old_it != old_objects.end(); old_it++) {
         if (!*old_it || !old_it->value()) continue;
         auto new_it = std::find_if(new_objects.begin(), new_objects.end(), [&old_it](const auto &new_obj) -> bool {
@@ -190,14 +191,18 @@ bool ObjectManifestHandler::update(const std::shared_ptr<ObjectStore::Object> &n
                 object_store.removeObject(metadata->objectId());
             }
             m_objectMetadataCache.erase(old_it->value().get());
-            m_objectManifest.removeObjects(*old_it);
+            to_del.push_back(*old_it);
         } else {
             /* object same or update */
             (*old_it->value()) = std::move(*new_it->value());
             new_manifest.removeObjects(*new_it);
         }
     }
 
+    for (const auto &obj : to_del) {
+        m_objectManifest.removeObjects(obj);
+    }
+
     auto now = datetime_type::clock::now();
     for (const auto &obj : new_objects) {
         if (!obj || !obj.value()) continue;

src/mbstf/ObjectPackager.hh

@@ -21,7 +21,7 @@
 
 #include <netinet/in.h>
 
-#include <boost/asio/io_service.hpp>
+#include <boost/asio/io_context.hpp>
 
 #include "common.hh"
 #include "Event.hh"
@@ -174,7 +174,7 @@ protected:
 
     std::shared_ptr<std::recursive_mutex> m_transmitterMutex;
     std::shared_ptr<LibFlute::Transmitter> m_transmitter;
-    boost::asio::io_service m_io;
+    boost::asio::io_context m_io;
     uint32_t m_queuedToi;
     bool m_queued;
     std::atomic_bool m_deactivating;

src/mbstf/PullObjectIngester.cc

@@ -86,7 +86,10 @@ std::string PullObjectIngester::PullIngestFailedEvent::reprString() const {
 
 /***** PullObjectIngester methods *****/
 
-PullObjectIngester::~PullObjectIngester() {abort();}
+PullObjectIngester::~PullObjectIngester() {
+    if (m_curl) m_curl->abortFetch();
+    abort();
+}
 
 bool PullObjectIngester::fetch(const std::string &object_id, const std::optional<time_type> &download_deadline)
 {

src/mbstf/meson.build

@@ -34,7 +34,7 @@ fiveg_api_release = get_option('fiveg_api_release')
 
 fs = import('fs')
 
-boost_dep = dependency('boost', required: true)
+boost_dep = dependency('boost', version: '>=1.66.0', required: true)
 uuid_dep = dependency('uuid', required: true)
 zlib_dep = dependency('zlib', required: true)
 libmpdpp_dep = dependency('mpd++', fallback: ['libmpdpp'])