Add ST2110 Rx/Tx 20/22/30 connections with basic UT

Signed-off-by: Tomasz Szumski <tomasz.szumski@intel.com>

media-proxy/include/mesh/st2110.h

@@ -0,0 +1,63 @@
+#ifndef ST2110_H
+#define ST2110_H
+
+#include <thread>
+#include <bsd/string.h>
+#include <arpa/inet.h>
+#include <mtl/st_pipeline_api.h>
+#include <mtl/st30_pipeline_api.h>
+
+#include "conn.h"
+#include "mesh_dp.h"
+#include "logger.h"
+
+namespace mesh
+{
+
+namespace connection
+{
+
+#define ST_APP_PAYLOAD_TYPE_ST30 (111)
+#define ST_APP_PAYLOAD_TYPE_ST20 (112)
+#define ST_APP_PAYLOAD_TYPE_ST22 (114)
+
+/**
+ * ST2110
+ *
+ * Base abstract class of ST2110. ST2110Rx/ST2110Tx
+ * inherit this class.
+ */
+class ST2110 : public Connection
+{
+  public:
+    static st_frame_fmt mesh_video_format_to_st_format(int fmt);
+    static st30_fmt mesh_audio_format_to_st_format(int fmt);
+    static st30_sampling mesh_audio_sampling_to_st_sampling(int sampling);
+    static st30_ptime mesh_audio_ptime_to_st_ptime(int ptime);
+    static void *get_frame_data_ptr(st_frame *src);
+    static void *get_frame_data_ptr(st30_frame *src);
+
+    static void get_mtl_dev_params(mtl_init_params &st_param, const std::string &dev_port,
+                                   mtl_log_level log_level,
+                                   const char local_ip_addr[MESH_IP_ADDRESS_SIZE]);
+    static mtl_handle get_mtl_handle(const std::string &dev_port, mtl_log_level log_level,
+                                     const char local_ip_addr[MESH_IP_ADDRESS_SIZE]);
+
+    ST2110() : _st(0){};
+    virtual ~ST2110(){};
+
+  protected:
+    static int frame_available_cb(void *ptr);
+
+    mtl_handle _st;
+    std::atomic<bool> _stop;
+    std::condition_variable_any _cv;
+
+  private:
+};
+
+} // namespace connection
+
+} // namespace mesh
+
+#endif // ST2110_H

media-proxy/include/mesh/st2110rx.h

@@ -0,0 +1,148 @@
+#ifndef ST2110RX_H
+#define ST2110RX_H
+
+#include "st2110.h"
+
+namespace mesh
+{
+
+namespace connection
+{
+
+/**
+ * ST2110Rx
+ *
+ * Base abstract class of ST2110Rx. ST2110_20Rx/ST2110_22Rx/ST2110_30Rx
+ * inherit this class.
+ */
+template <typename FRAME, typename HANDLE, typename OPS> class ST2110Rx : public ST2110
+{
+  public:
+    ST2110Rx()
+    {
+        _kind = Kind::receiver;
+
+        _handle = nullptr;
+        _ops = {0};
+        _transfer_size = 0;
+    }
+    ~ST2110Rx() {}
+
+  protected:
+    HANDLE _handle;
+    OPS _ops;
+    size_t _transfer_size;
+    std::jthread _frame_thread_handle;
+    context::Context _ctx;
+
+    std::function<FRAME *(HANDLE)> _get_frame_fn;
+    std::function<int(HANDLE, FRAME *)> _put_frame_fn;
+    std::function<HANDLE(mtl_handle, OPS *)> _create_session_fn;
+    std::function<int(HANDLE)> _close_session_fn;
+
+    Result on_establish(context::Context &ctx) override
+    {
+        _ctx = context::WithCancel(ctx);
+        _stop = false;
+
+        _handle = _create_session_fn(_st, &_ops);
+        if (!_handle) {
+            log::error("Failed to create session");
+            set_state(ctx, State::closed);
+            return set_result(Result::error_general_failure);
+        }
+
+        /* Start MTL session thread. */
+        try {
+            _frame_thread_handle = std::jthread(&ST2110Rx::frame_thread, this);
+        } catch (const std::system_error &e) {
+            log::error("Failed to create thread");
+            set_state(ctx, State::closed);
+            return set_result(Result::error_out_of_memory);
+        }
+
+        set_state(ctx, State::active);
+        return set_result(Result::success);
+    }
+
+    Result on_shutdown(context::Context &ctx) override
+    {
+        _ctx.cancel();
+
+        _frame_thread_handle.join();
+
+        if (_handle) {
+            _close_session_fn(_handle);
+            _handle = nullptr;
+        }
+        set_state(ctx, State::closed);
+        return set_result(Result::success);
+    };
+
+    virtual void on_delete(context::Context &ctx) override {}
+
+  private:
+    void frame_thread()
+    {
+        while (!_ctx.cancelled()) {
+            // Get full buffer from MTL
+            FRAME *frame_ptr = _get_frame_fn(_handle);
+            if (!frame_ptr) { /* no frame */
+                std::mutex mx;
+                std::unique_lock lk(mx);
+                _cv.wait(lk, _ctx.stop_token(), [this] { return _stop.load(); });
+                _stop = false;
+                if (_ctx.cancelled()) {
+                    return;
+                }
+                continue;
+            }
+            // Forward buffer to emulated receiver
+            transmit(_ctx, get_frame_data_ptr(frame_ptr), _transfer_size);
+            // Return used buffer to MTL
+            _put_frame_fn(_handle, frame_ptr);
+        }
+    }
+};
+
+class ST2110_20Rx : public ST2110Rx<st_frame, st20p_rx_handle, st20p_rx_ops>
+{
+  public:
+    ST2110_20Rx();
+    ~ST2110_20Rx();
+
+    Result configure(context::Context &ctx, const std::string &dev_port,
+                     const MeshConfig_ST2110 &cfg_st2110, const MeshConfig_Video &cfg_video);
+
+  private:
+};
+
+class ST2110_22Rx : public ST2110Rx<st_frame, st22p_rx_handle, st22p_rx_ops>
+{
+  public:
+    ST2110_22Rx();
+    ~ST2110_22Rx();
+
+    Result configure(context::Context &ctx, const std::string &dev_port,
+                     const MeshConfig_ST2110 &cfg_st2110, const MeshConfig_Video &cfg_video);
+
+  private:
+};
+
+class ST2110_30Rx : public ST2110Rx<st30_frame, st30p_rx_handle, st30p_rx_ops>
+{
+  public:
+    ST2110_30Rx();
+    ~ST2110_30Rx();
+
+    Result configure(context::Context &ctx, const std::string &dev_port,
+                     const MeshConfig_ST2110 &cfg_st2110, const MeshConfig_Audio &cfg_audio);
+
+  private:
+};
+
+} // namespace connection
+
+} // namespace mesh
+
+#endif // ST2110RX_H

media-proxy/include/mesh/st2110tx.h

@@ -0,0 +1,146 @@
+#ifndef ST2110TX_H
+#define ST2110TX_H
+
+#include "st2110.h"
+#include <algorithm>
+
+namespace mesh
+{
+
+namespace connection
+{
+
+/**
+ * ST2110
+ *
+ * Base abstract class of ST2110. ST2110_20Tx/ST2110_22Tx/ST2110_30Tx
+ * inherit this class.
+ */
+template <typename FRAME, typename HANDLE, typename OPS> class ST2110Tx : public ST2110
+{
+  public:
+    ST2110Tx()
+    {
+        _kind = Kind::transmitter;
+
+        _handle = nullptr;
+        _ops = {0};
+        _transfer_size = 0;
+    };
+    ~ST2110Tx(){};
+
+  protected:
+    HANDLE _handle;
+    OPS _ops;
+    uint32_t _transfer_size;
+    context::Context _ctx;
+
+    std::function<FRAME *(HANDLE)> _get_frame_fn;
+    std::function<int(HANDLE, FRAME *)> _put_frame_fn;
+    std::function<HANDLE(mtl_handle, OPS *)> _create_session_fn;
+    std::function<int(HANDLE)> _close_session_fn;
+
+    Result on_establish(context::Context &ctx) override
+    {
+        _ctx = context::WithCancel(ctx);
+        _stop = false;
+
+        _handle = _create_session_fn(_st, &_ops);
+        if (!_handle) {
+            log::error("Failed to create session");
+            set_state(ctx, State::closed);
+            return set_result(Result::error_general_failure);
+        }
+
+        set_state(ctx, State::active);
+        return set_result(Result::success);
+    };
+
+    Result on_shutdown(context::Context &ctx) override
+    {
+        _ctx.cancel();
+
+        if (_handle) {
+            _close_session_fn(_handle);
+            _handle = nullptr;
+        }
+        set_state(ctx, State::closed);
+        return set_result(Result::success);
+    };
+
+    Result on_receive(context::Context &ctx, void *ptr, uint32_t sz, uint32_t &sent) override
+    {
+        sent = std::min(_transfer_size, sz);
+        // TODO: add error/warning if sent is different than _transfer_size
+
+        FRAME *frame = NULL;
+        do {
+            // Get empty buffer from MTL
+            frame = _get_frame_fn(_handle);
+            if (!frame) {
+                std::mutex mx;
+                std::unique_lock lk(mx);
+                _cv.wait(lk, _ctx.stop_token(), [this] { return _stop.load(); });
+                _stop = false;
+                if (_ctx.cancelled()) {
+                    break;
+                }
+            }
+        } while (!frame);
+
+        if (frame) {
+            // Copy data from emulated transmitter to MTL empty buffer
+            mtl_memcpy(get_frame_data_ptr(frame), ptr, sent);
+            // Return full buffer to MTL
+            _put_frame_fn(_handle, frame);
+        } else {
+            sent = 0;
+            return set_result(Result::error_shutdown);
+        }
+        return set_result(Result::success);
+    };
+
+  private:
+};
+
+class ST2110_20Tx : public ST2110Tx<st_frame, st20p_tx_handle, st20p_tx_ops>
+{
+  public:
+    ST2110_20Tx();
+    ~ST2110_20Tx();
+
+    Result configure(context::Context &ctx, const std::string &dev_port,
+                     const MeshConfig_ST2110 &cfg_st2110, const MeshConfig_Video &cfg_video);
+
+  private:
+};
+
+class ST2110_22Tx : public ST2110Tx<st_frame, st22p_tx_handle, st22p_tx_ops>
+{
+  public:
+    ST2110_22Tx();
+    ~ST2110_22Tx();
+
+    Result configure(context::Context &ctx, const std::string &dev_port,
+                     const MeshConfig_ST2110 &cfg_st2110, const MeshConfig_Video &cfg_video);
+
+  private:
+};
+
+class ST2110_30Tx : public ST2110Tx<st30_frame, st30p_tx_handle, st30p_tx_ops>
+{
+  public:
+    ST2110_30Tx();
+    ~ST2110_30Tx();
+
+    Result configure(context::Context &ctx, const std::string &dev_port,
+                     const MeshConfig_ST2110 &cfg_st2110, const MeshConfig_Audio &cfg_audio);
+
+  private:
+};
+
+} // namespace connection
+
+} // namespace mesh
+
+#endif // ST2110TX_H