mediagst.c (11266B)
1 /* 2 This file is part of libextractor. 3 Copyright (C) 2026 Vidyut Samanta and Christian Grothoff 4 5 libextractor is free software; you can redistribute it and/or modify 6 it under the terms of the GNU General Public License as published 7 by the Free Software Foundation; either version 3, or (at your 8 option) any later version. 9 10 libextractor is distributed in the hope that it will be useful, but 11 WITHOUT ANY WARRANTY; without even the implied warranty of 12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 13 General Public License for more details. 14 15 You should have received a copy of the GNU General Public License 16 along with libextractor; see the file COPYING. If not, write to the 17 Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, 18 Boston, MA 02110-1301, USA. 19 */ 20 /** 21 * @file plugins/mediagst.c 22 * @brief plumbing between an extraction context and a GStreamer pipeline 23 * @author Christian Grothoff 24 */ 25 #include "platform.h" 26 #include "mediagst.h" 27 28 #include <gst/app/gstappsrc.h> 29 30 31 /** 32 * Implementation of GstAppSrc's "need-data" callback. Reads data from 33 * the extraction context and passes it to GStreamer. 34 * 35 * @param appsrc the source that ran out of data 36 * @param size number of bytes requested 37 * @param io our execution context 38 */ 39 static void 40 feed_data (GstElement *appsrc, 41 guint size, 42 struct EXTRACTOR_GstIO *io) 43 { 44 GstBuffer *buffer; 45 GstMemory *mem; 46 GstMapInfo mi; 47 guint accumulated; 48 ssize_t data_len; 49 uint8_t *le_data; 50 51 g_mutex_lock (&io->lock); 52 if ( (0 != io->length) && 53 (io->offset >= io->length) ) 54 { 55 g_mutex_unlock (&io->lock); 56 gst_app_src_end_of_stream (GST_APP_SRC (appsrc)); 57 return; 58 } 59 if ( (0 != io->length) && 60 (io->offset + size > io->length) ) 61 size = io->length - io->offset; 62 if (0 == size) 63 { 64 g_mutex_unlock (&io->lock); 65 gst_app_src_end_of_stream (GST_APP_SRC (appsrc)); 66 return; 67 } 68 mem = gst_allocator_alloc (NULL, size, NULL); 69 if (! gst_memory_map (mem, &mi, GST_MAP_WRITE)) 70 { 71 g_mutex_unlock (&io->lock); 72 gst_memory_unref (mem); 73 gst_app_src_end_of_stream (GST_APP_SRC (appsrc)); 74 return; 75 } 76 accumulated = 0; 77 data_len = 1; 78 while ( (accumulated < size) && 79 (data_len > 0) ) 80 { 81 data_len = io->ec->read (io->ec->cls, 82 (void **) &le_data, 83 size - accumulated); 84 if (data_len > 0) 85 { 86 memcpy (&mi.data[accumulated], 87 le_data, 88 data_len); 89 accumulated += data_len; 90 } 91 } 92 gst_memory_unmap (mem, &mi); 93 if (0 == accumulated) 94 { 95 g_mutex_unlock (&io->lock); 96 gst_memory_unref (mem); 97 gst_app_src_end_of_stream (GST_APP_SRC (appsrc)); 98 return; 99 } 100 gst_memory_resize (mem, 0, accumulated); 101 buffer = gst_buffer_new (); 102 gst_buffer_append_memory (buffer, mem); 103 /* the offsets matter: they are what makes random access work */ 104 GST_BUFFER_OFFSET (buffer) = io->offset; 105 GST_BUFFER_OFFSET_END (buffer) = io->offset + accumulated; 106 io->offset += accumulated; 107 g_mutex_unlock (&io->lock); 108 gst_app_src_push_buffer (GST_APP_SRC (appsrc), buffer); 109 } 110 111 112 /** 113 * Implementation of GstAppSrc's "seek-data" callback. 114 * 115 * @param appsrc the source that wants to seek 116 * @param position new desired absolute position in the file 117 * @param io our execution context 118 * @return TRUE if seeking succeeded 119 */ 120 static gboolean 121 seek_data (GstElement *appsrc, 122 guint64 position, 123 struct EXTRACTOR_GstIO *io) 124 { 125 int64_t ret; 126 127 gboolean ok; 128 129 (void) appsrc; 130 g_mutex_lock (&io->lock); 131 ret = io->ec->seek (io->ec->cls, 132 position, 133 SEEK_SET); 134 if (0 > ret) 135 { 136 g_mutex_unlock (&io->lock); 137 return FALSE; 138 } 139 io->offset = (uint64_t) ret; 140 ok = (io->offset == position); 141 g_mutex_unlock (&io->lock); 142 return ok; 143 } 144 145 146 /** 147 * Configure @a source to read from the extraction context of @a io. 148 * 149 * @param source the appsrc to configure 150 * @param io our execution context 151 */ 152 static void 153 configure_appsrc (GstElement *source, 154 struct EXTRACTOR_GstIO *io) 155 { 156 if (0 != io->length) 157 g_object_set (source, 158 "size", (gint64) io->length, 159 "stream-type", GST_APP_STREAM_TYPE_RANDOM_ACCESS, 160 NULL); 161 else 162 g_object_set (source, 163 "size", (gint64) - 1, 164 "stream-type", GST_APP_STREAM_TYPE_STREAM, 165 NULL); 166 g_object_set (source, 167 "format", GST_FORMAT_BYTES, 168 NULL); 169 g_signal_connect (source, 170 "need-data", 171 G_CALLBACK (&feed_data), 172 io); 173 g_signal_connect (source, 174 "seek-data", 175 G_CALLBACK (&seek_data), 176 io); 177 } 178 179 180 /** 181 * Implementation of playbin's "source-setup" signal: configure the 182 * appsrc it created for our "appsrc://" URI. 183 * 184 * @param playbin the pipeline 185 * @param source the appsrc to configure 186 * @param io our execution context 187 */ 188 static void 189 source_setup (GstElement *playbin, 190 GstElement *source, 191 struct EXTRACTOR_GstIO *io) 192 { 193 (void) playbin; 194 configure_appsrc (source, 195 io); 196 } 197 198 199 /** 200 * Initialize @a io for reading @a ec from the beginning. 201 * 202 * @param[out] io I/O state to initialize 203 * @param ec extraction context to read from 204 */ 205 static void 206 io_init (struct EXTRACTOR_GstIO *io, 207 struct EXTRACTOR_ExtractContext *ec) 208 { 209 uint64_t size; 210 211 io->ec = ec; 212 io->offset = 0; 213 /* a zero-filled GMutex is a valid, unlocked mutex that needs no 214 g_mutex_clear(), which saves every caller a cleanup call */ 215 memset (&io->lock, 0, sizeof (io->lock)); 216 io->last_ping = g_get_monotonic_time (); 217 size = ec->get_size (ec->cls); 218 io->length = (UINT64_MAX == size) ? 0 : size; 219 /* an earlier pipeline may have left the context anywhere */ 220 (void) ec->seek (ec->cls, 221 0, 222 SEEK_SET); 223 } 224 225 226 GstElement * 227 EXTRACTOR_gst_appsrc_ (struct EXTRACTOR_GstIO *io, 228 struct EXTRACTOR_ExtractContext *ec) 229 { 230 GstElement *source; 231 232 if (! gst_is_initialized ()) 233 gst_init (NULL, NULL); 234 io_init (io, 235 ec); 236 if (NULL == (source = gst_element_factory_make ("appsrc", 237 NULL))) 238 return NULL; 239 configure_appsrc (source, 240 io); 241 return source; 242 } 243 244 245 GstElement * 246 EXTRACTOR_gst_playbin_ (struct EXTRACTOR_GstIO *io, 247 struct EXTRACTOR_ExtractContext *ec, 248 unsigned int flags, 249 GstElement *video_sink, 250 GstElement *audio_sink) 251 { 252 GstElement *pipeline; 253 254 if (! gst_is_initialized ()) 255 gst_init (NULL, NULL); 256 io_init (io, 257 ec); 258 pipeline = gst_element_factory_make ("playbin", NULL); 259 if ( (NULL == pipeline) || 260 (NULL == video_sink) || 261 (NULL == audio_sink) ) 262 { 263 /* gst-plugins-base is not installed, or not completely */ 264 if (NULL != video_sink) 265 gst_object_unref (video_sink); 266 if (NULL != audio_sink) 267 gst_object_unref (audio_sink); 268 if (NULL != pipeline) 269 gst_object_unref (pipeline); 270 return NULL; 271 } 272 g_object_set (pipeline, 273 "uri", "appsrc://", 274 "video-sink", video_sink, 275 "audio-sink", audio_sink, 276 "flags", flags, 277 NULL); 278 g_signal_connect (pipeline, 279 "source-setup", 280 G_CALLBACK (&source_setup), 281 io); 282 return pipeline; 283 } 284 285 286 void 287 EXTRACTOR_gst_keepalive_ (struct EXTRACTOR_GstIO *io) 288 { 289 gint64 now = g_get_monotonic_time (); 290 291 if (now - io->last_ping < 100 * G_TIME_SPAN_MILLISECOND) 292 return; 293 io->last_ping = now; 294 g_mutex_lock (&io->lock); 295 /* a no-op seek makes the plugin side of the IPC tell the host that 296 we are still busy (MESSAGE_ALIVE) */ 297 (void) io->ec->seek (io->ec->cls, 298 0, 299 SEEK_CUR); 300 g_mutex_unlock (&io->lock); 301 } 302 303 304 int 305 EXTRACTOR_gst_wait_async_alive_ (GstElement *pipeline, 306 GstClockTime timeout, 307 struct EXTRACTOR_GstIO *io) 308 { 309 GstBus *bus; 310 GstMessage *msg; 311 GstClockTime end; 312 int ret = 0; 313 314 if (0 == timeout) 315 return 0; 316 if (NULL == (bus = gst_element_get_bus (pipeline))) 317 return 0; 318 end = (GstClockTime) g_get_monotonic_time () * GST_USECOND + timeout; 319 while (1) 320 { 321 GstClockTime now = (GstClockTime) g_get_monotonic_time () * GST_USECOND; 322 GstClockTime slice; 323 324 if (now >= end) 325 break; /* timed out */ 326 slice = end - now; 327 if (slice > 100 * GST_MSECOND) 328 slice = 100 * GST_MSECOND; 329 msg = gst_bus_timed_pop_filtered (bus, 330 slice, 331 GST_MESSAGE_ASYNC_DONE 332 | GST_MESSAGE_ERROR 333 | GST_MESSAGE_EOS); 334 if (NULL == msg) 335 { 336 EXTRACTOR_gst_keepalive_ (io); 337 continue; 338 } 339 if (GST_MESSAGE_ASYNC_DONE != GST_MESSAGE_TYPE (msg)) 340 { 341 /* an error, or the file ended before we ever saw a frame */ 342 gst_message_unref (msg); 343 break; 344 } 345 if (GST_OBJECT (pipeline) == GST_MESSAGE_SRC (msg)) 346 { 347 gst_message_unref (msg); 348 ret = 1; 349 break; 350 } 351 /* ASYNC_DONE from some element inside the pipeline; keep waiting */ 352 gst_message_unref (msg); 353 } 354 gst_object_unref (bus); 355 return ret; 356 } 357 358 359 int 360 EXTRACTOR_gst_wait_async_ (GstElement *pipeline, 361 GstClockTime timeout) 362 { 363 GstBus *bus; 364 GstMessage *msg; 365 int ret = 0; 366 367 if (0 == timeout) 368 return 0; 369 if (NULL == (bus = gst_element_get_bus (pipeline))) 370 return 0; 371 while (1) 372 { 373 msg = gst_bus_timed_pop_filtered (bus, 374 timeout, 375 GST_MESSAGE_ASYNC_DONE 376 | GST_MESSAGE_ERROR 377 | GST_MESSAGE_EOS); 378 if (NULL == msg) 379 break; /* timed out */ 380 if (GST_MESSAGE_ASYNC_DONE != GST_MESSAGE_TYPE (msg)) 381 { 382 /* an error, or the file ended before we ever saw a frame */ 383 gst_message_unref (msg); 384 break; 385 } 386 if (GST_OBJECT (pipeline) == GST_MESSAGE_SRC (msg)) 387 { 388 gst_message_unref (msg); 389 ret = 1; 390 break; 391 } 392 /* ASYNC_DONE from some element inside the pipeline; keep waiting */ 393 gst_message_unref (msg); 394 } 395 gst_object_unref (bus); 396 return ret; 397 } 398 399 400 gint64 401 EXTRACTOR_gst_sample_position_ (GstElement *pipeline, 402 unsigned int offset, 403 gint64 *duration) 404 { 405 gint64 target; 406 gint64 dur = 0; 407 408 target = ((gint64) offset) * GST_SECOND; 409 if (! gst_element_query_duration (pipeline, 410 GST_FORMAT_TIME, 411 &dur)) 412 dur = 0; 413 if (dur < 0) 414 dur = 0; 415 if ( (0 != dur) && 416 (target > dur / 3) ) 417 target = dur / 3; 418 *duration = dur; 419 return target; 420 } 421 422 423 /* end of mediagst.c */