vrxtx: rm process_sender_message from API

MartinPulec · MartinPulec · commit 9c3fc1018e99 · 2026-03-26T11:44:54.000+01:00
Since the vrxtx modules are no longer subclass of video_rxtx, we can
split the handling of the messages by video_rxtx itself (query format)
and vrxtx module (RTP specific stuff).

As a consequence, the message workaround from previous commit can be
dismissed. Single message dispatch is processed in rtp_video_rxtx
constructor (also moved from video_rxtx).
diff --git a/src/control_socket.cpp b/src/control_socket.cpp
@@ -518,10 +518,12 @@ static int process_msg(struct control_state *s, fd_t client_fd, char *message, s
                         msg_audio->tx_port = msg->tx_port + 2;
                 }
 
-                enum module_class path_sender[] = { MODULE_CLASS_SENDER, MODULE_CLASS_NONE };
+                enum module_class path_sender_video[] = { MODULE_CLASS_SENDER,
+                                                          MODULE_CLASS_DATA,
+                                                          MODULE_CLASS_NONE };
                 enum module_class path_sender_audio[] = { MODULE_CLASS_AUDIO, MODULE_CLASS_SENDER, MODULE_CLASS_NONE };
                 memcpy(path_audio, path, sizeof(path_audio));
-                append_message_path(path, sizeof(path), path_sender);
+                append_message_path(path, sizeof(path), path_sender_video);
                 append_message_path(path_audio, sizeof(path_audio), path_sender_audio);
 
                 resp =
@@ -546,10 +548,12 @@ static int process_msg(struct control_state *s, fd_t client_fd, char *message, s
                 struct msg_sender *msg_audio = (struct msg_sender *) malloc(sizeof(struct msg_sender));
                 memcpy(msg_audio, msg, sizeof(struct msg_sender));
 
-                enum module_class path_sender[] = { MODULE_CLASS_SENDER, MODULE_CLASS_NONE };
+                enum module_class path_sender_video[] = { MODULE_CLASS_SENDER,
+                                                          MODULE_CLASS_DATA,
+                                                          MODULE_CLASS_NONE };
                 enum module_class path_sender_audio[] = { MODULE_CLASS_AUDIO, MODULE_CLASS_SENDER, MODULE_CLASS_NONE };
                 memcpy(path_audio, path, sizeof(path_audio));
-                append_message_path(path, sizeof(path), path_sender);
+                append_message_path(path, sizeof(path), path_sender_video);
                 append_message_path(path_audio, sizeof(path_audio), path_sender_audio);
 
                 resp =
diff --git a/src/video_rxtx.cpp b/src/video_rxtx.cpp
@@ -117,13 +117,6 @@ video_rxtx::video_rxtx(const char                *protocol_name,
         pthread_mutex_init(&m_lock, nullptr);
 }
 
-static void flush_messages(struct module *mod) {
-        struct message *m = nullptr;
-        while ((m = check_message(mod)) != nullptr) {
-                free_message(m, nullptr);
-        }
-}
-
 video_rxtx::~video_rxtx() noexcept
 {
         join();
@@ -134,7 +127,6 @@ video_rxtx::~video_rxtx() noexcept
         m_impl_funcs->done(m_impl_state);
         compress_done(m_compression);
         module_done(&m_receiver_mod);
-        flush_messages(&m_sender_mod);
         module_done(&m_sender_mod);
 
         pthread_mutex_destroy(&m_lock);
@@ -202,11 +194,12 @@ void video_rxtx::check_sender_messages() {
                                 r = new_response(RESPONSE_OK,
                                                  oss.str().c_str());
                         }
-                } else { // delegate to implementations
-                        if (m_impl_funcs->process_sender_message != nullptr) {
-                                r = m_impl_funcs->process_sender_message(
-                                    m_impl_state, msg);
-                        }
+                } else {
+                        char buf[200];
+                        snprintf_ch(buf, "Unexpected sender message type %d",
+                                    msg->type);
+                        MSG(WARNING, "%s\n", buf);
+                        r = new_response(RESPONSE_BAD_REQUEST, buf);
                 }
 
                 free_message(msg_external, r);
@@ -264,10 +257,6 @@ video_rxtx::create(string const &proto, const struct vrxtx_params *params,
                 return nullptr;
         }
 
-        // The idea of doing that is to display help on '-f ldgm:help' even if UG would exit
-        // immediately. The encoder is actually created by a message.
-        ret->check_sender_messages();
-
         int rc = pthread_create(&ret->m_thread_id, nullptr, video_rxtx::sender_thread,
                            (void *) ret);
         assert(rc == 0);
diff --git a/src/video_rxtx.hpp b/src/video_rxtx.hpp
@@ -110,8 +110,6 @@ struct video_rxtx_info {
                                       const struct audio_desc *desc,
                                       int audio_rx_port, int audio_tx_port,
                                       bool ipv6);
-        struct response *(*process_sender_message)(void *state,
-                                                   struct msg_sender *);
         void *(*receiver_routine)(void *state);
 };
 
diff --git a/src/video_rxtx/h264_rtp.cpp b/src/video_rxtx/h264_rtp.cpp
@@ -128,6 +128,7 @@ h264_rtp_video_rxtx::configure_rtsp_server_video()
 void
 h264_rtp_video_rxtx::send_frame(shared_ptr<video_frame> tx_frame) noexcept
 {
+        rtp_process_sender_messages();
         // requestt compress reconfiguration if receivng raw data
         if (!is_codec_opaque(tx_frame->color_spec)) {
                 if (!m_sent_compress_change) {
@@ -276,7 +277,6 @@ static const struct video_rxtx_info h264_video_rxtx_info = {
         .send_frame             = send_frame,
         .join_sender            = join,
         .set_sender_audio_spec  = set_audio_spec,
-        .process_sender_message = rtp_process_sender_message,
         .receiver_routine       = nullptr,
 };
 
diff --git a/src/video_rxtx/h264_sdp.cpp b/src/video_rxtx/h264_sdp.cpp
@@ -123,6 +123,7 @@ void h264_sdp_video_rxtx::sdp_add_video(codec_t codec)
 void
 h264_sdp_video_rxtx::send_frame(shared_ptr<video_frame> tx_frame) noexcept
 {
+        rtp_process_sender_messages();
         if (!is_codec_opaque(tx_frame->color_spec)) {
 		if (m_sent_compress_change) {
 			return;
@@ -223,7 +224,6 @@ static const struct video_rxtx_info h264_sdp_video_rxtx_info = {
         .send_frame             = send_frame,
         .join_sender            = nullptr,
         .set_sender_audio_spec  = set_audio_spec,
-        .process_sender_message = rtp_process_sender_message,
         .receiver_routine       = nullptr,
 };
 
diff --git a/src/video_rxtx/ihdtv.cpp b/src/video_rxtx/ihdtv.cpp
@@ -217,7 +217,6 @@ static const struct video_rxtx_info ihdtv_video_rxtx_info = {
         .send_frame             = send_frame,
         .join_sender            = nullptr,
         .set_sender_audio_spec  = nullptr,
-        .process_sender_message = nullptr,
         .receiver_routine       = ihdtv_video_rxtx::receiver_thread,
 };
 
diff --git a/src/video_rxtx/loopback.cpp b/src/video_rxtx/loopback.cpp
@@ -176,7 +176,6 @@ static const struct video_rxtx_info loopback_video_rxtx_info = {
         .send_frame             = send_frame,
         .join_sender            = nullptr,
         .set_sender_audio_spec  = nullptr,
-        .process_sender_message = nullptr,
         .receiver_routine       = loopback_video_rxtx::receiver_thread,
 };
 
diff --git a/src/video_rxtx/omt.cpp b/src/video_rxtx/omt.cpp
@@ -241,7 +241,6 @@ constexpr video_rxtx_info omt_video_rxtx_info = {
         omt_rxtx_send_frame,
         nullptr,
         nullptr,
-        nullptr,
         omt_rxtx_recv_worker
 };
 
diff --git a/src/video_rxtx/rtp.cpp b/src/video_rxtx/rtp.cpp
@@ -186,6 +186,19 @@ rtp_video_rxtx::process_sender_message(struct msg_sender *msg)
         return new_response(RESPONSE_OK, nullptr);
 }
 
+void
+rtp_video_rxtx::rtp_process_sender_messages()
+{
+        struct message *msg_external = nullptr;
+        while((msg_external = check_message(&m_rtp_sender_mod))) {
+                struct response *r = nullptr;
+                auto *msg = (struct msg_sender *) msg_external;
+                process_sender_message(msg);
+
+                free_message(msg_external, r);
+        }
+}
+
 rtp_video_rxtx::rtp_video_rxtx(const struct vrxtx_params *params,
                                const struct common_opts  *common) :
         m_force_ip_version(common->force_ip_version),
@@ -208,11 +221,21 @@ rtp_video_rxtx::rtp_video_rxtx(const struct vrxtx_params *params,
                                        EXIT_FAIL_NETWORK);
         }
 
-        m_tx = tx_init(params->sender_mod, common->mtu, TX_MEDIA_VIDEO,
+        module_init_default(&m_rtp_sender_mod);
+        m_rtp_sender_mod.cls = MODULE_CLASS_DATA;
+        module_register(&m_rtp_sender_mod, params->sender_mod);
+
+        m_tx = tx_init(&m_rtp_sender_mod, common->mtu, TX_MEDIA_VIDEO,
                        params->fec, common->encryption, params->bitrate_limit);
         if (m_tx == nullptr) {
                 throw ug_runtime_error("Unable to initialize transmitter", EXIT_FAIL_TRANSMIT);
         }
+
+        // The idea of doing that is to display help on '-f ldgm:help' even if UG would exit
+        // immediately. The encoder is actually created by a message.
+        // Also for `-x sdp:help` the message will get discarded and the warning that message quie
+        rtp_process_sender_messages();
+
 }
 
 rtp_video_rxtx::~rtp_video_rxtx()
@@ -230,6 +253,7 @@ rtp_video_rxtx::~rtp_video_rxtx()
         }
 
         delete m_fec_state;
+        module_done(&m_rtp_sender_mod);
 }
 
 void rtp_video_rxtx::display_buf_increase_warning(int size)
@@ -313,7 +337,3 @@ void rtp_video_rxtx::destroy_rtp_device(struct rtp *network_device)
         rtp_done(network_device);
 }
 
-struct response *rtp_process_sender_message(void *state, struct msg_sender *msg) {
-        auto *s = static_cast<rtp_video_rxtx*>(state);
-        return s->process_sender_message(msg);
-}
diff --git a/src/video_rxtx/rtp.hpp b/src/video_rxtx/rtp.hpp
@@ -69,7 +69,7 @@ class rtp_video_rxtx {
                         const char *mcast_if, int ttl);
         static void destroy_rtp_device(struct rtp * network_devices);
         static void display_buf_increase_warning(int size);
-        struct response *process_sender_message(struct msg_sender *msg);
+        void rtp_process_sender_messages();
 
 protected:
         struct rtp *m_network_device; // ULTRAGRID_RTP
@@ -86,9 +86,13 @@ class rtp_video_rxtx {
         int              m_rxtx_mode;
         bool             m_used = false; ///< at least one frame was sent
 private:
+        struct response *process_sender_message(struct msg_sender *msg);
+        /// This is child of vrxtx sender module to process specific messages.
+        /// The receiver module is used directly (vrxtx doesn't process any
+        /// message and also the send/receive handling is not entirely
+        /// symetric).
+        struct module m_rtp_sender_mod;
 };
 
-struct response *rtp_process_sender_message(void *state, struct msg_sender *);
-
 #endif // VIDEO_RXTX_RTP_H_
 
diff --git a/src/video_rxtx/ultragrid_rtp.cpp b/src/video_rxtx/ultragrid_rtp.cpp
@@ -112,6 +112,7 @@ void *ultragrid_rtp_video_rxtx::receiver_thread(void *arg) {
 void
 ultragrid_rtp_video_rxtx::send_frame(shared_ptr<video_frame> tx_frame) noexcept
 {
+        rtp_process_sender_messages();
         m_used = true;
         if (m_fec_state) {
                 tx_frame = m_fec_state->encode(tx_frame);
@@ -457,7 +458,6 @@ static const struct video_rxtx_info ultragrid_rtp_video_rxtx_info = {
         .send_frame             = send_frame,
         .join_sender            = join,
         .set_sender_audio_spec  = nullptr,
-        .process_sender_message = rtp_process_sender_message,
         .receiver_routine       = ultragrid_rtp_video_rxtx::receiver_thread,
 };
 
