util.hpp

//
// © NTT DOCOMO BUSINESS, Inc. All Rights Reserved.
//
 
#ifndef SKYWAY_GLOBAL_UTIL_HPP_
#define SKYWAY_GLOBAL_UTIL_HPP_
 
#include <atomic>
#include <boost/archive/iterators/binary_from_base64.hpp>
#include <boost/archive/iterators/transform_width.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <condition_variable>
#include <functional>
#include <json.hpp>
#include <mutex>
#include <sstream>
#include <thread>
#include <vector>
 
#include "skyway/global/config.hpp"
#include "skyway/model/domain.hpp"
 
namespace skyway {
namespace global {
namespace util {
 
enum class SleepIntervalMs : int {
    kLocal   = config::kLocalSleepIntervalMs,
    kNetwork = config::kNetworkSleepIntervalMs,
};
 
bool WaitUntilWithTimeoutMs(std::function<bool()> release_condition,
                            int interval_ms,
                            int timeout_ms);
 
bool WaitUntilWithTimeoutMs(std::function<bool()> release_condition,
                            SleepIntervalMs interval,
                            int timeout_ms);
 
bool WaitUntilWithTimeoutMs(std::atomic<bool>& release_condition_boolean,
                            int interval_ms,
                            int timeout_ms);
 
bool WaitUntilWithTimeoutMs(std::atomic<bool>& release_condition_boolean,
                            SleepIntervalMs interval,
                            int timeout_ms);
 
class ScopeExit {
public:
    explicit ScopeExit(std::function<void()> on_exit) : on_exit_(on_exit) {}
 
    ~ScopeExit() { on_exit_(); }
 
    ScopeExit(const ScopeExit&) = delete;
    ScopeExit& operator=(const ScopeExit&) = delete;
 
private:
    std::function<void()> on_exit_;
};
 
inline bool SpinLockWithTimeoutMs(std::function<bool()> release_condition, int timeout_ms) {
    std::condition_variable cv;
    std::mutex mtx;
    std::unique_ptr<std::thread> observer_thread = nullptr;
    bool is_succeeded                            = false;
    {
        std::unique_lock<std::mutex> lock(mtx);
        // The flag used for Spurious Wakeup
        auto is_notified     = std::make_shared<bool>(false);
        auto result_returned = std::make_shared<std::atomic<bool>>(false);
        observer_thread      = std::make_unique<std::thread>([&, result_returned, is_notified] {
            while (!release_condition()) {
                // Block this thread until release_condition will be true
                if (result_returned->load()) {
                    return;
                }
            }
            std::lock_guard<std::mutex> lk(mtx);
            *is_notified = true;
            cv.notify_one();
        });
        is_succeeded =
            cv.wait_for(lock, std::chrono::milliseconds(timeout_ms), [&] { return *is_notified; });
        result_returned->store(true);
    }
    observer_thread->join();
 
    return is_succeeded;
}
 
inline bool SpinLockWithTimeoutMs(std::atomic<bool>& release_condition, int timeout_ms) {
    std::function<bool()> f = [&] { return release_condition.load(); };
    return SpinLockWithTimeoutMs(f, timeout_ms);
}
 
inline bool SpinLockWithTimeout(std::function<bool()> release_condition,
                                int timeout_sec = config::kDefaultTimeoutSec) {
    return SpinLockWithTimeoutMs(release_condition, timeout_sec * 1000);
}
 
inline bool SpinLockWithTimeout(std::atomic<bool>& release_condition,
                                int timeout_sec = config::kDefaultTimeoutSec) {
    return SpinLockWithTimeoutMs(release_condition, timeout_sec * 1000);
}
 
inline std::string Uuid() {
    auto id            = boost::uuids::random_generator()();
    std::string result = boost::lexical_cast<std::string>(id);
    return result;
}
 
// REF:
// https://stackoverflow.com/questions/14265581/parse-split-a-string-in-c-using-string-delimiter-standard-c
 
inline std::vector<std::string> Split(const std::string& s, const std::string& delimiter) {
    size_t pos_start = 0, pos_end, delim_len = delimiter.length();
    std::string token;
    std::vector<std::string> res;
 
    while ((pos_end = s.find(delimiter, pos_start)) != std::string::npos) {
        token     = s.substr(pos_start, pos_end - pos_start);
        pos_start = pos_end + delim_len;
        res.push_back(token);
    }
 
    res.push_back(s.substr(pos_start));
    return res;
}
 
inline std::string DecodeBase64Token(const std::string& token) {
    using InputItr  = std::istreambuf_iterator<char>;
    using OutputItr = std::ostream_iterator<char>;
    using DecodeItr = boost::archive::iterators::
        transform_width<boost::archive::iterators::binary_from_base64<InputItr>, 8, 6, char>;
 
    std::stringstream input_stream, output_stream;
    input_stream << token;
    copy(DecodeItr(InputItr(input_stream)), DecodeItr(InputItr()), OutputItr(output_stream));
    std::string res;
    output_stream >> res;
    return res;
}
 
inline void SetEncodingIndexToIdIfNeeded(std::vector<model::Encoding>& encodings) {
    for (int i = 0; i < encodings.size(); i++) {
        if (!encodings[i].id) {
            encodings[i].id = std::to_string(i);
        }
    }
}
 
inline void SortEncodingsForMediaSoup(std::vector<model::Encoding>& encodings) {
    if (encodings.empty()) return;
    auto encoding = encodings[0];
    if (encoding.max_bitrate) {
        std::sort(encodings.begin(),
                  encodings.end(),
                  [](const model::Encoding& lhs, const model::Encoding& rhs) {
                      if (lhs.max_bitrate == std::nullopt || rhs.max_bitrate == std::nullopt) {
                          return true;
                      }
                      return *lhs.max_bitrate < *rhs.max_bitrate;
                  });
    } else if (encoding.scale_resolution_down_by) {
        std::sort(encodings.begin(),
                  encodings.end(),
                  [](const model::Encoding& lhs, const model::Encoding& rhs) {
                      if (lhs.scale_resolution_down_by == std::nullopt ||
                          rhs.scale_resolution_down_by == std::nullopt) {
                          return true;
                      }
                      return *lhs.scale_resolution_down_by > *rhs.scale_resolution_down_by;
                  });
    } else if (encoding.max_framerate) {
        std::sort(encodings.begin(),
                  encodings.end(),
                  [](const model::Encoding& lhs, const model::Encoding& rhs) {
                      if (lhs.max_framerate == std::nullopt || rhs.max_framerate == std::nullopt) {
                          return true;
                      }
                      return *lhs.max_framerate < *rhs.max_framerate;
                  });
    }
}
 
inline const std::chrono::milliseconds CurrentUnixTimestampMs() {
    return std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::system_clock::now().time_since_epoch());
}
 
}  // namespace util
}  // namespace global
}  // namespace skyway
 
#endif /* SKYWAY_GLOBAL_UTIL_HPP_ */