]> git.sesse.net Git - mlt/blob - src/framework/mlt_consumer.c
Remove the tractor service locking.
[mlt] / src / framework / mlt_consumer.c
1 /**
2  * \file mlt_consumer.c
3  * \brief abstraction for all consumer services
4  * \see mlt_consumer_s
5  *
6  * Copyright (C) 2003-2009 Ushodaya Enterprises Limited
7  * \author Charles Yates <charles.yates@pandora.be>
8  *
9  * This library is free software; you can redistribute it and/or
10  * modify it under the terms of the GNU Lesser General Public
11  * License as published by the Free Software Foundation; either
12  * version 2.1 of the License, or (at your option) any later version.
13  *
14  * This library is distributed in the hope that it will be useful,
15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
17  * Lesser General Public License for more details.
18  *
19  * You should have received a copy of the GNU Lesser General Public
20  * License along with this library; if not, write to the Free Software
21  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
22  */
23
24 #include "mlt_consumer.h"
25 #include "mlt_factory.h"
26 #include "mlt_producer.h"
27 #include "mlt_frame.h"
28 #include "mlt_profile.h"
29 #include "mlt_log.h"
30
31 #include <stdio.h>
32 #include <string.h>
33 #include <stdlib.h>
34 #include <sys/time.h>
35
36 /** Define this if you want an automatic deinterlace (if necessary) when the
37  * consumer's producer is not running at normal speed.
38  */
39 #undef DEINTERLACE_ON_NOT_NORMAL_SPEED
40
41 /** This is not the ideal place for this, but it is needed by VDPAU as well.
42  */
43 pthread_mutex_t mlt_sdl_mutex = PTHREAD_MUTEX_INITIALIZER;
44
45 static void mlt_consumer_frame_render( mlt_listener listener, mlt_properties owner, mlt_service this, void **args );
46 static void mlt_consumer_frame_show( mlt_listener listener, mlt_properties owner, mlt_service this, void **args );
47 static void mlt_consumer_property_changed( mlt_properties owner, mlt_consumer this, char *name );
48 static void apply_profile_properties( mlt_consumer this, mlt_profile profile, mlt_properties properties );
49 static void on_consumer_frame_show( mlt_properties owner, mlt_consumer this, mlt_frame frame );
50
51 /** Initialize a consumer service.
52  *
53  * \public \memberof mlt_consumer_s
54  * \param this the consumer to initialize
55  * \param child a pointer to the object for the subclass
56  * \param profile the \p mlt_profile_s to use (optional but recommended,
57  * uses the environment variable MLT if this is NULL)
58  * \return true if there was an error
59  */
60
61 int mlt_consumer_init( mlt_consumer this, void *child, mlt_profile profile )
62 {
63         int error = 0;
64         memset( this, 0, sizeof( struct mlt_consumer_s ) );
65         this->child = child;
66         error = mlt_service_init( &this->parent, this );
67         if ( error == 0 )
68         {
69                 // Get the properties from the service
70                 mlt_properties properties = MLT_SERVICE_PROPERTIES( &this->parent );
71
72                 // Apply profile to properties
73                 if ( profile == NULL )
74                 {
75                         // Normally the application creates the profile and controls its lifetime
76                         // This is the fallback exception handling
77                         profile = mlt_profile_init( NULL );
78                         mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
79                         mlt_properties_set_data( properties, "_profile", profile, 0, (mlt_destructor)mlt_profile_close, NULL );
80                 }
81                 apply_profile_properties( this, profile, properties );
82
83                 // Default rescaler for all consumers
84                 mlt_properties_set( properties, "rescale", "bilinear" );
85
86                 // Default read ahead buffer size
87                 mlt_properties_set_int( properties, "buffer", 25 );
88
89                 // Default audio frequency and channels
90                 mlt_properties_set_int( properties, "frequency", 48000 );
91                 mlt_properties_set_int( properties, "channels", 2 );
92
93                 // Default of all consumers is real time
94                 mlt_properties_set_int( properties, "real_time", 1 );
95
96                 // Default to environment test card
97                 mlt_properties_set( properties, "test_card", mlt_environment( "MLT_TEST_CARD" ) );
98
99                 // Hmm - default all consumers to yuv422 :-/
100                 this->format = mlt_image_yuv422;
101
102                 mlt_events_register( properties, "consumer-frame-show", ( mlt_transmitter )mlt_consumer_frame_show );
103                 mlt_events_register( properties, "consumer-frame-render", ( mlt_transmitter )mlt_consumer_frame_render );
104                 mlt_events_register( properties, "consumer-stopped", NULL );
105                 mlt_events_listen( properties, this, "consumer-frame-show", ( mlt_listener )on_consumer_frame_show );
106
107                 // Register a property-changed listener to handle the profile property -
108                 // subsequent properties can override the profile
109                 this->event_listener = mlt_events_listen( properties, this, "property-changed", ( mlt_listener )mlt_consumer_property_changed );
110
111                 // Create the push mutex and condition
112                 pthread_mutex_init( &this->put_mutex, NULL );
113                 pthread_cond_init( &this->put_cond, NULL );
114
115         }
116         return error;
117 }
118
119 /** Convert the profile into properties on the consumer.
120  *
121  * \private \memberof mlt_consumer_s
122  * \param this a consumer
123  * \param profile a profile
124  * \param properties a properties list (typically, the consumer's)
125  */
126
127 static void apply_profile_properties( mlt_consumer this, mlt_profile profile, mlt_properties properties )
128 {
129         mlt_event_block( this->event_listener );
130         mlt_properties_set_double( properties, "fps", mlt_profile_fps( profile ) );
131         mlt_properties_set_int( properties, "frame_rate_num", profile->frame_rate_num );
132         mlt_properties_set_int( properties, "frame_rate_den", profile->frame_rate_den );
133         mlt_properties_set_int( properties, "width", profile->width );
134         mlt_properties_set_int( properties, "height", profile->height );
135         mlt_properties_set_int( properties, "progressive", profile->progressive );
136         mlt_properties_set_double( properties, "aspect_ratio", mlt_profile_sar( profile )  );
137         mlt_properties_set_int( properties, "sample_aspect_num", profile->sample_aspect_num );
138         mlt_properties_set_int( properties, "sample_aspect_den", profile->sample_aspect_den );
139         mlt_properties_set_double( properties, "display_ratio", mlt_profile_dar( profile )  );
140         mlt_properties_set_int( properties, "display_aspect_num", profile->display_aspect_num );
141         mlt_properties_set_int( properties, "display_aspect_num", profile->display_aspect_num );
142         mlt_properties_set_int( properties, "colorspace", profile->colorspace );
143         mlt_event_unblock( this->event_listener );
144 }
145
146 /** The property-changed event listener
147  *
148  * \private \memberof mlt_consumer_s
149  * \param owner the events object
150  * \param this the consumer
151  * \param name the name of the property that changed
152  */
153
154 static void mlt_consumer_property_changed( mlt_properties owner, mlt_consumer this, char *name )
155 {
156         if ( !strcmp( name, "profile" ) )
157         {
158                 // Get the properies
159                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
160
161                 // Get the current profile
162                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
163
164                 // Load the new profile
165                 mlt_profile new_profile = mlt_profile_init( mlt_properties_get( properties, name ) );
166
167                 if ( new_profile )
168                 {
169                         // Copy the profile
170                         if ( profile != NULL )
171                         {
172                                 free( profile->description );
173                                 memcpy( profile, new_profile, sizeof( struct mlt_profile_s ) );
174                                 profile->description = strdup( new_profile->description );
175                                 mlt_profile_close( new_profile );
176                         }
177                         else
178                         {
179                                 profile = new_profile;
180                         }
181
182                         // Apply to properties
183                         apply_profile_properties( this, profile, properties );
184                 }
185         }
186         else if ( !strcmp( name, "frame_rate_num" ) )
187         {
188                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
189                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
190                 if ( profile )
191                 {
192                         profile->frame_rate_num = mlt_properties_get_int( properties, "frame_rate_num" );
193                         mlt_properties_set_double( properties, "fps", mlt_profile_fps( profile ) );
194                 }
195         }
196         else if ( !strcmp( name, "frame_rate_den" ) )
197         {
198                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
199                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
200                 if ( profile )
201                 {
202                         profile->frame_rate_den = mlt_properties_get_int( properties, "frame_rate_den" );
203                         mlt_properties_set_double( properties, "fps", mlt_profile_fps( profile ) );
204                 }
205         }
206         else if ( !strcmp( name, "width" ) )
207         {
208                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
209                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
210                 if ( profile )
211                         profile->width = mlt_properties_get_int( properties, "width" );
212         }
213         else if ( !strcmp( name, "height" ) )
214         {
215                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
216                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
217                 if ( profile )
218                         profile->height = mlt_properties_get_int( properties, "height" );
219         }
220         else if ( !strcmp( name, "progressive" ) )
221         {
222                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
223                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
224                 if ( profile )
225                         profile->progressive = mlt_properties_get_int( properties, "progressive" );
226         }
227         else if ( !strcmp( name, "sample_aspect_num" ) )
228         {
229                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
230                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
231                 profile->sample_aspect_num = mlt_properties_get_int( properties, "sample_aspect_num" );
232                 if ( profile )
233                         mlt_properties_set_double( properties, "aspect_ratio", mlt_profile_sar( profile )  );
234         }
235         else if ( !strcmp( name, "sample_aspect_den" ) )
236         {
237                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
238                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
239                 profile->sample_aspect_den = mlt_properties_get_int( properties, "sample_aspect_den" );
240                 if ( profile )
241                         mlt_properties_set_double( properties, "aspect_ratio", mlt_profile_sar( profile )  );
242         }
243         else if ( !strcmp( name, "display_aspect_num" ) )
244         {
245                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
246                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
247                 if ( profile )
248                 {
249                         profile->display_aspect_num = mlt_properties_get_int( properties, "display_aspect_num" );
250                         mlt_properties_set_double( properties, "display_ratio", mlt_profile_dar( profile )  );
251                 }
252         }
253         else if ( !strcmp( name, "display_aspect_den" ) )
254         {
255                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
256                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
257                 if ( profile )
258                 {
259                         profile->display_aspect_den = mlt_properties_get_int( properties, "display_aspect_den" );
260                         mlt_properties_set_double( properties, "display_ratio", mlt_profile_dar( profile )  );
261                 }
262         }
263         else if ( !strcmp( name, "colorspace" ) )
264         {
265                 mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
266                 mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
267                 if ( profile )
268                         profile->colorspace = mlt_properties_get_int( properties, "colorspace" );
269         }
270 }
271
272 /** The transmitter for the consumer-frame-show event
273  *
274  * Invokes the listener.
275  *
276  * \private \memberof mlt_consumer_s
277  * \param listener a function pointer that will be invoked
278  * \param owner the events object that will be passed to \p listener
279  * \param this  a service that will be passed to \p listener
280  * \param args an array of pointers - the first entry is passed as a string to \p listener
281  */
282
283 static void mlt_consumer_frame_show( mlt_listener listener, mlt_properties owner, mlt_service this, void **args )
284 {
285         if ( listener != NULL )
286                 listener( owner, this, ( mlt_frame )args[ 0 ] );
287 }
288
289 /** The transmitter for the consumer-frame-render event
290  *
291  * Invokes the listener.
292  *
293  * \private \memberof mlt_consumer_s
294  * \param listener a function pointer that will be invoked
295  * \param owner the events object that will be passed to \p listener
296  * \param this  a service that will be passed to \p listener
297  * \param args an array of pointers - the first entry is passed as a string to \p listener
298  */
299
300 static void mlt_consumer_frame_render( mlt_listener listener, mlt_properties owner, mlt_service this, void **args )
301 {
302         if ( listener != NULL )
303                 listener( owner, this, ( mlt_frame )args[ 0 ] );
304 }
305
306 /** A listener on the consumer-frame-show event
307  *
308  * Saves the position of the frame shown.
309  *
310  * \private \memberof mlt_consumer_s
311  * \param owner the events object
312  * \param consumer the consumer on which this event occurred
313  * \param frame the frame that was shown
314  */
315
316 static void on_consumer_frame_show( mlt_properties owner, mlt_consumer consumer, mlt_frame frame )
317 {
318         if ( frame )
319                 consumer->position = mlt_frame_get_position( frame );
320 }
321
322 /** Create a new consumer.
323  *
324  * \public \memberof mlt_consumer_s
325  * \param profile a profile (optional, but recommended)
326  * \return a new consumer
327  */
328
329 mlt_consumer mlt_consumer_new( mlt_profile profile )
330 {
331         // Create the memory for the structure
332         mlt_consumer this = malloc( sizeof( struct mlt_consumer_s ) );
333
334         // Initialise it
335         if ( this != NULL )
336                 mlt_consumer_init( this, NULL, profile );
337
338         // Return it
339         return this;
340 }
341
342 /** Get the parent service object.
343  *
344  * \public \memberof mlt_consumer_s
345  * \param this a consumer
346  * \return the parent service class
347  * \see MLT_CONSUMER_SERVICE
348  */
349
350 mlt_service mlt_consumer_service( mlt_consumer this )
351 {
352         return this != NULL ? &this->parent : NULL;
353 }
354
355 /** Get the consumer properties.
356  *
357  * \public \memberof mlt_consumer_s
358  * \param this a consumer
359  * \return the consumer's properties list
360  * \see MLT_CONSUMER_PROPERTIES
361  */
362
363 mlt_properties mlt_consumer_properties( mlt_consumer this )
364 {
365         return this != NULL ? MLT_SERVICE_PROPERTIES( &this->parent ) : NULL;
366 }
367
368 /** Connect the consumer to the producer.
369  *
370  * \public \memberof mlt_consumer_s
371  * \param this a consumer
372  * \param producer a producer
373  * \return > 0 warning, == 0 success, < 0 serious error,
374  *         1 = this service does not accept input,
375  *         2 = the producer is invalid,
376  *         3 = the producer is already registered with this consumer
377  */
378
379 int mlt_consumer_connect( mlt_consumer this, mlt_service producer )
380 {
381         return mlt_service_connect_producer( &this->parent, producer, 0 );
382 }
383
384 /** Start the consumer.
385  *
386  * \public \memberof mlt_consumer_s
387  * \param this a consumer
388  * \return true if there was an error
389  */
390
391 int mlt_consumer_start( mlt_consumer this )
392 {
393         // Stop listening to the property-changed event
394         mlt_event_block( this->event_listener );
395
396         // Get the properies
397         mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
398
399         // Determine if there's a test card producer
400         char *test_card = mlt_properties_get( properties, "test_card" );
401
402         // Just to make sure nothing is hanging around...
403         mlt_frame_close( this->put );
404         this->put = NULL;
405         this->put_active = 1;
406
407         // Deal with it now.
408         if ( test_card != NULL )
409         {
410                 if ( mlt_properties_get_data( properties, "test_card_producer", NULL ) == NULL )
411                 {
412                         // Create a test card producer
413                         mlt_profile profile = mlt_service_profile( MLT_CONSUMER_SERVICE( this ) );
414                         mlt_producer producer = mlt_factory_producer( profile, NULL, test_card );
415
416                         // Do we have a producer
417                         if ( producer != NULL )
418                         {
419                                 // Test card should loop I guess...
420                                 mlt_properties_set( MLT_PRODUCER_PROPERTIES( producer ), "eof", "loop" );
421                                 //mlt_producer_set_speed( producer, 0 );
422                                 //mlt_producer_set_in_and_out( producer, 0, 0 );
423
424                                 // Set the test card on the consumer
425                                 mlt_properties_set_data( properties, "test_card_producer", producer, 0, ( mlt_destructor )mlt_producer_close, NULL );
426                         }
427                 }
428         }
429         else
430         {
431                 // Allow the hash table to speed things up
432                 mlt_properties_set_data( properties, "test_card_producer", NULL, 0, NULL, NULL );
433         }
434
435         // Set the frame duration in microseconds for the frame-dropping heuristic
436         int frame_duration = 1000000 / mlt_properties_get_int( properties, "frame_rate_num" ) *
437                         mlt_properties_get_int( properties, "frame_rate_den" );
438         mlt_properties_set_int( properties, "frame_duration", frame_duration );
439
440         // Check and run an ante command
441         if ( mlt_properties_get( properties, "ante" ) )
442                 if ( system( mlt_properties_get( properties, "ante" ) ) == -1 )
443                         mlt_log( MLT_CONSUMER_SERVICE( this ), MLT_LOG_ERROR, "system(%s) failed!\n", mlt_properties_get( properties, "ante" ) );
444
445         // Set the real_time preference
446         this->real_time = mlt_properties_get_int( properties, "real_time" );
447
448         // For worker threads implementation, buffer must be at least # threads
449         if ( abs( this->real_time ) > 1 && mlt_properties_get_int( properties, "buffer" ) <= abs( this->real_time ) )
450                 mlt_properties_set_int( properties, "buffer", abs( this->real_time ) + 1 );
451
452         // Start the service
453         if ( this->start != NULL )
454                 return this->start( this );
455
456         return 0;
457 }
458
459 /** An alternative method to feed frames into the consumer.
460  *
461  * Only valid if the consumer itself is not connected.
462  *
463  * \public \memberof mlt_consumer_s
464  * \param this a consumer
465  * \param frame a frame
466  * \return true (ignore this for now)
467  */
468
469 int mlt_consumer_put_frame( mlt_consumer this, mlt_frame frame )
470 {
471         int error = 1;
472
473         // Get the service assoicated to the consumer
474         mlt_service service = MLT_CONSUMER_SERVICE( this );
475
476         if ( mlt_service_producer( service ) == NULL )
477         {
478                 struct timeval now;
479                 struct timespec tm;
480                 pthread_mutex_lock( &this->put_mutex );
481                 while ( this->put_active && this->put != NULL )
482                 {
483                         gettimeofday( &now, NULL );
484                         tm.tv_sec = now.tv_sec + 1;
485                         tm.tv_nsec = now.tv_usec * 1000;
486                         pthread_cond_timedwait( &this->put_cond, &this->put_mutex, &tm );
487                 }
488                 if ( this->put_active && this->put == NULL )
489                         this->put = frame;
490                 else
491                         mlt_frame_close( frame );
492                 pthread_cond_broadcast( &this->put_cond );
493                 pthread_mutex_unlock( &this->put_mutex );
494         }
495         else
496         {
497                 mlt_frame_close( frame );
498         }
499
500         return error;
501 }
502
503 /** Protected method for consumer to get frames from connected service
504  *
505  * \public \memberof mlt_consumer_s
506  * \param this a consumer
507  * \return a frame
508  */
509
510 mlt_frame mlt_consumer_get_frame( mlt_consumer this )
511 {
512         // Frame to return
513         mlt_frame frame = NULL;
514
515         // Get the service assoicated to the consumer
516         mlt_service service = MLT_CONSUMER_SERVICE( this );
517
518         // Get the consumer properties
519         mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
520
521         // Get the frame
522         if ( mlt_service_producer( service ) == NULL && mlt_properties_get_int( properties, "put_mode" ) )
523         {
524                 struct timeval now;
525                 struct timespec tm;
526                 pthread_mutex_lock( &this->put_mutex );
527                 while ( this->put_active && this->put == NULL )
528                 {
529                         gettimeofday( &now, NULL );
530                         tm.tv_sec = now.tv_sec + 1;
531                         tm.tv_nsec = now.tv_usec * 1000;
532                         pthread_cond_timedwait( &this->put_cond, &this->put_mutex, &tm );
533                 }
534                 frame = this->put;
535                 this->put = NULL;
536                 pthread_cond_broadcast( &this->put_cond );
537                 pthread_mutex_unlock( &this->put_mutex );
538                 if ( frame != NULL )
539                         mlt_service_apply_filters( service, frame, 0 );
540         }
541         else if ( mlt_service_producer( service ) != NULL )
542         {
543                 mlt_service_get_frame( service, &frame, 0 );
544         }
545         else
546         {
547                 frame = mlt_frame_init( service );
548         }
549
550         if ( frame != NULL )
551         {
552                 // Get the frame properties
553                 mlt_properties frame_properties = MLT_FRAME_PROPERTIES( frame );
554
555                 // Get the test card producer
556                 mlt_producer test_card = mlt_properties_get_data( properties, "test_card_producer", NULL );
557
558                 // Attach the test frame producer to it.
559                 if ( test_card != NULL )
560                         mlt_properties_set_data( frame_properties, "test_card_producer", test_card, 0, NULL, NULL );
561
562                 // Attach the rescale property
563                 mlt_properties_set( frame_properties, "rescale.interp", mlt_properties_get( properties, "rescale" ) );
564
565                 // Aspect ratio and other jiggery pokery
566                 mlt_properties_set_double( frame_properties, "consumer_aspect_ratio", mlt_properties_get_double( properties, "aspect_ratio" ) );
567                 mlt_properties_set_int( frame_properties, "consumer_deinterlace", mlt_properties_get_int( properties, "progressive" ) | mlt_properties_get_int( properties, "deinterlace" ) );
568                 mlt_properties_set( frame_properties, "deinterlace_method", mlt_properties_get( properties, "deinterlace_method" ) );
569         }
570
571         // Return the frame
572         return frame;
573 }
574
575 /** Compute the time difference between now and a time value.
576  *
577  * \private \memberof mlt_consumer_s
578  * \param time1 a time value to be compared against now
579  * \return the difference in microseconds
580  */
581
582 static inline long time_difference( struct timeval *time1 )
583 {
584         struct timeval time2;
585         time2.tv_sec = time1->tv_sec;
586         time2.tv_usec = time1->tv_usec;
587         gettimeofday( time1, NULL );
588         return time1->tv_sec * 1000000 + time1->tv_usec - time2.tv_sec * 1000000 - time2.tv_usec;
589 }
590
591 /** The thread procedure for asynchronously pulling frames through the service
592  * network connected to a consumer.
593  *
594  * \private \memberof mlt_consumer_s
595  * \param arg a consumer
596  */
597
598 static void *consumer_read_ahead_thread( void *arg )
599 {
600         // The argument is the consumer
601         mlt_consumer this = arg;
602
603         // Get the properties of the consumer
604         mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
605
606         // Get the width and height
607         int width = mlt_properties_get_int( properties, "width" );
608         int height = mlt_properties_get_int( properties, "height" );
609
610         // See if video is turned off
611         int video_off = mlt_properties_get_int( properties, "video_off" );
612         int preview_off = mlt_properties_get_int( properties, "preview_off" );
613         int preview_format = mlt_properties_get_int( properties, "preview_format" );
614
615         // Get the audio settings
616         mlt_audio_format afmt = mlt_audio_s16;
617         int counter = 0;
618         double fps = mlt_properties_get_double( properties, "fps" );
619         int channels = mlt_properties_get_int( properties, "channels" );
620         int frequency = mlt_properties_get_int( properties, "frequency" );
621         int samples = 0;
622         void *audio = NULL;
623
624         // See if audio is turned off
625         int audio_off = mlt_properties_get_int( properties, "audio_off" );
626
627         // Get the maximum size of the buffer
628         int buffer = mlt_properties_get_int( properties, "buffer" ) + 1;
629
630         // General frame variable
631         mlt_frame frame = NULL;
632         uint8_t *image = NULL;
633
634         // Time structures
635         struct timeval ante;
636
637         // Average time for get_frame and get_image
638         int count = 1;
639         int skipped = 0;
640         int64_t time_wait = 0;
641         int64_t time_frame = 0;
642         int64_t time_process = 0;
643         int skip_next = 0;
644
645         if ( preview_off && preview_format != 0 )
646                 this->format = preview_format;
647
648         // Get the first frame
649         frame = mlt_consumer_get_frame( this );
650
651         // Get the image of the first frame
652         if ( !video_off )
653         {
654                 mlt_events_fire( MLT_CONSUMER_PROPERTIES( this ), "consumer-frame-render", frame, NULL );
655                 mlt_frame_get_image( frame, &image, &this->format, &width, &height, 0 );
656         }
657
658         if ( !audio_off )
659         {
660                 samples = mlt_sample_calculator( fps, frequency, counter++ );
661                 mlt_frame_get_audio( frame, &audio, &afmt, &frequency, &channels, &samples );
662         }
663
664         // Mark as rendered
665         mlt_properties_set_int( MLT_FRAME_PROPERTIES( frame ), "rendered", 1 );
666
667         // Get the starting time (can ignore the times above)
668         gettimeofday( &ante, NULL );
669
670         // Continue to read ahead
671         while ( this->ahead )
672         {
673                 // Fetch width/height again
674                 width = mlt_properties_get_int( properties, "width" );
675                 height = mlt_properties_get_int( properties, "height" );
676
677                 // Put the current frame into the queue
678                 pthread_mutex_lock( &this->frame_queue_mutex );
679                 while( this->ahead && mlt_deque_count( this->frame_queue ) >= buffer )
680                         pthread_cond_wait( &this->frame_queue_cond, &this->frame_queue_mutex );
681                 mlt_deque_push_back( this->frame_queue, frame );
682                 pthread_cond_broadcast( &this->frame_queue_cond );
683                 pthread_mutex_unlock( &this->frame_queue_mutex );
684
685                 time_wait += time_difference( &ante );
686
687                 // Get the next frame
688                 frame = mlt_consumer_get_frame( this );
689                 time_frame += time_difference( &ante );
690
691                 // If there's no frame, we're probably stopped...
692                 if ( frame == NULL )
693                         continue;
694
695                 // Increment the count
696                 count ++;
697
698                 // All non normal playback frames should be shown
699                 if ( mlt_properties_get_int( MLT_FRAME_PROPERTIES( frame ), "_speed" ) != 1 )
700                 {
701 #ifdef DEINTERLACE_ON_NOT_NORMAL_SPEED
702                         mlt_properties_set_int( MLT_FRAME_PROPERTIES( frame ), "consumer_deinterlace", 1 );
703 #endif
704                         skipped = 0;
705                         time_frame = 0;
706                         time_process = 0;
707                         time_wait = 0;
708                         count = 1;
709                         skip_next = 0;
710                 }
711
712                 // Get the image
713                 if ( !skip_next || this->real_time == -1 )
714                 {
715                         // Get the image, mark as rendered and time it
716                         if ( !video_off )
717                         {
718                                 mlt_events_fire( MLT_CONSUMER_PROPERTIES( this ), "consumer-frame-render", frame, NULL );
719                                 mlt_frame_get_image( frame, &image, &this->format, &width, &height, 0 );
720                         }
721                         mlt_properties_set_int( MLT_FRAME_PROPERTIES( frame ), "rendered", 1 );
722                 }
723                 else
724                 {
725                         // Increment the number of sequentially skipped frames
726                         skipped ++;
727                         skip_next = 0;
728
729                         // If we've reached an unacceptable level, reset everything
730                         if ( skipped > 5 )
731                         {
732                                 skipped = 0;
733                                 time_frame = 0;
734                                 time_process = 0;
735                                 time_wait = 0;
736                                 count = 1;
737                         }
738                 }
739
740                 // Always process audio
741                 if ( !audio_off )
742                 {
743                         samples = mlt_sample_calculator( fps, frequency, counter++ );
744                         mlt_frame_get_audio( frame, &audio, &afmt, &frequency, &channels, &samples );
745                 }
746
747                 // Increment the time take for this frame
748                 time_process += time_difference( &ante );
749
750                 // Determine if the next frame should be skipped
751                 if ( mlt_deque_count( this->frame_queue ) <= 5 )
752                 {
753                         int frame_duration = mlt_properties_get_int( properties, "frame_duration" );
754                         if ( ( ( time_wait + time_frame + time_process ) / count ) > frame_duration )
755                                 skip_next = 1;
756                 }
757         }
758
759         // Remove the last frame
760         mlt_frame_close( frame );
761
762         return NULL;
763 }
764
765 static int compare_frame_position( mlt_frame a, mlt_frame b )
766 {
767         return mlt_frame_get_position( a ) - mlt_frame_get_position( b );
768 }
769
770 /** The worker thread procedure for parallel processing frames.
771  *
772  * \private \memberof mlt_consumer_s
773  * \param arg a consumer
774  */
775
776 static void *consumer_worker_thread( void *arg )
777 {
778         // The argument is the consumer
779         mlt_consumer this = arg;
780
781         // Get the properties of the consumer
782         mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
783
784         // Get the width and height
785         int width = mlt_properties_get_int( properties, "width" );
786         int height = mlt_properties_get_int( properties, "height" );
787
788         // See if video is turned off
789         int video_off = mlt_properties_get_int( properties, "video_off" );
790         int preview_off = mlt_properties_get_int( properties, "preview_off" );
791         int preview_format = mlt_properties_get_int( properties, "preview_format" );
792
793         // Get the audio settings
794         mlt_audio_format afmt = mlt_audio_s16;
795         int counter = 0;
796         double fps = mlt_properties_get_double( properties, "fps" );
797         int channels = mlt_properties_get_int( properties, "channels" );
798         int frequency = mlt_properties_get_int( properties, "frequency" );
799         int samples = 0;
800         void *audio = NULL;
801
802         // See if audio is turned off
803         int audio_off = mlt_properties_get_int( properties, "audio_off" );
804
805         // General frame variable
806         mlt_frame frame = NULL;
807         uint8_t *image = NULL;
808
809         if ( preview_off && preview_format != 0 )
810                 this->format = preview_format;
811
812         // Get the first frame from the work queue
813         pthread_mutex_lock( &this->frame_queue_mutex );
814         while( this->ahead && mlt_deque_count( this->frame_queue ) <= 0 )
815                 pthread_cond_wait( &this->frame_queue_cond, &this->frame_queue_mutex );
816         frame = mlt_deque_pop_front( this->frame_queue );
817         pthread_cond_signal( &this->frame_queue_cond );
818         pthread_mutex_unlock( &this->frame_queue_mutex );
819
820         // Get the image of the first frame
821         if ( !video_off )
822         {
823                 mlt_events_fire( MLT_CONSUMER_PROPERTIES( this ), "consumer-frame-render", frame, NULL );
824                 mlt_frame_get_image( frame, &image, &this->format, &width, &height, 0 );
825         }
826
827         if ( !audio_off )
828         {
829                 samples = mlt_sample_calculator( fps, frequency, counter++ );
830                 mlt_frame_get_audio( frame, &audio, &afmt, &frequency, &channels, &samples );
831         }
832
833         // Mark as rendered
834         mlt_properties_set_int( MLT_FRAME_PROPERTIES( frame ), "rendered", 1 );
835
836         // Continue to read ahead
837         while ( this->ahead )
838         {
839                 // Fetch width/height again
840                 width = mlt_properties_get_int( properties, "width" );
841                 height = mlt_properties_get_int( properties, "height" );
842
843                 // Put the processed frame into the done queue
844                 if ( frame )
845                 {
846                         pthread_mutex_lock( &this->done_queue_mutex );
847                         mlt_deque_insert( this->done_queue, frame, ( mlt_deque_compare ) compare_frame_position );
848                         pthread_cond_signal( &this->done_queue_cond );
849                         pthread_mutex_unlock( &this->done_queue_mutex );
850                 }
851
852                 // Get the next frame from the work queue
853                 pthread_mutex_lock( &this->frame_queue_mutex );
854                 while( this->ahead && mlt_deque_count( this->frame_queue ) <= 0 )
855                         pthread_cond_wait( &this->frame_queue_cond, &this->frame_queue_mutex );
856                 frame = mlt_deque_pop_front( this->frame_queue );
857                 pthread_cond_signal( &this->frame_queue_cond );
858                 pthread_mutex_unlock( &this->frame_queue_mutex );
859
860                 // If there's no frame, we're probably stopped...
861                 if ( frame == NULL )
862                         continue;
863
864                 // All non normal playback frames should be shown
865                 if ( mlt_properties_get_int( MLT_FRAME_PROPERTIES( frame ), "_speed" ) != 1 )
866                 {
867 #ifdef DEINTERLACE_ON_NOT_NORMAL_SPEED
868                         mlt_properties_set_int( MLT_FRAME_PROPERTIES( frame ), "consumer_deinterlace", 1 );
869 #endif
870                 }
871
872                 // Get the image
873                 if ( !video_off )
874                 {
875                         mlt_events_fire( MLT_CONSUMER_PROPERTIES( this ), "consumer-frame-render", frame, NULL );
876                         mlt_frame_get_image( frame, &image, &this->format, &width, &height, 0 );
877                 }
878                 mlt_properties_set_int( MLT_FRAME_PROPERTIES( frame ), "rendered", 1 );
879
880                 // Always process audio
881                 if ( !audio_off )
882                 {
883                         samples = mlt_sample_calculator( fps, frequency, counter++ );
884                         mlt_frame_get_audio( frame, &audio, &afmt, &frequency, &channels, &samples );
885                 }
886         }
887
888         // Remove the last frame
889         mlt_frame_close( frame );
890
891         return NULL;
892 }
893
894 /** Start the read/render thread.
895  *
896  * \private \memberof mlt_consumer_s
897  * \param this a consumer
898  */
899
900 static void consumer_read_ahead_start( mlt_consumer this )
901 {
902         // We're running now
903         this->ahead = 1;
904
905         // Create the frame queue
906         this->frame_queue = mlt_deque_init( );
907
908         // Create the frame_queue mutex
909         pthread_mutex_init( &this->frame_queue_mutex, NULL );
910
911         // Create the condition
912         pthread_cond_init( &this->frame_queue_cond, NULL );
913
914         // Create the read ahead
915         if ( mlt_properties_get( MLT_CONSUMER_PROPERTIES( this ), "priority" ) )
916         {
917                 struct sched_param priority;
918                 priority.sched_priority = mlt_properties_get_int( MLT_CONSUMER_PROPERTIES( this ), "priority" );
919                 pthread_attr_t thread_attributes;
920                 pthread_attr_init( &thread_attributes );
921                 pthread_attr_setschedpolicy( &thread_attributes, SCHED_OTHER );
922                 pthread_attr_setschedparam( &thread_attributes, &priority );
923                 pthread_attr_setinheritsched( &thread_attributes, PTHREAD_EXPLICIT_SCHED );
924                 pthread_attr_setscope( &thread_attributes, PTHREAD_SCOPE_SYSTEM );
925                 if ( pthread_create( &this->ahead_thread, &thread_attributes, consumer_read_ahead_thread, this ) < 0 )
926                         pthread_create( &this->ahead_thread, NULL, consumer_read_ahead_thread, this );
927                 pthread_attr_destroy( &thread_attributes );
928         }
929         else
930         {
931                 pthread_create( &this->ahead_thread, NULL, consumer_read_ahead_thread, this );
932         }
933 }
934
935 /** Start the worker threads.
936  *
937  * \private \memberof mlt_consumer_s
938  * \param this a consumer
939  */
940
941 static void consumer_work_start( mlt_consumer this )
942 {
943         int n = abs( this->real_time );
944         pthread_t thread;
945
946         // We're running now
947         this->ahead = 1;
948
949         // Create the queues
950         this->frame_queue = mlt_deque_init();
951         this->done_queue = mlt_deque_init();
952         this->worker_threads = mlt_deque_init();
953
954         // Create the mutexes
955         pthread_mutex_init( &this->frame_queue_mutex, NULL );
956         pthread_mutex_init( &this->done_queue_mutex, NULL );
957
958         // Create the conditions
959         pthread_cond_init( &this->frame_queue_cond, NULL );
960         pthread_cond_init( &this->done_queue_cond, NULL );
961
962         // Create the read ahead
963         if ( mlt_properties_get( MLT_CONSUMER_PROPERTIES( this ), "priority" ) )
964         {
965
966                 struct sched_param priority;
967                 pthread_attr_t thread_attributes;
968
969                 priority.sched_priority = mlt_properties_get_int( MLT_CONSUMER_PROPERTIES( this ), "priority" );
970                 pthread_attr_init( &thread_attributes );
971                 pthread_attr_setschedpolicy( &thread_attributes, SCHED_OTHER );
972                 pthread_attr_setschedparam( &thread_attributes, &priority );
973                 pthread_attr_setinheritsched( &thread_attributes, PTHREAD_EXPLICIT_SCHED );
974                 pthread_attr_setscope( &thread_attributes, PTHREAD_SCOPE_SYSTEM );
975
976                 while ( n-- )
977                 {
978                         if ( pthread_create( &thread, &thread_attributes, consumer_read_ahead_thread, this ) < 0 )
979                                 if ( pthread_create( &thread, NULL, consumer_read_ahead_thread, this ) == 0 )
980                                         mlt_deque_push_back( this->worker_threads, (void*) thread );
981                 }
982                 pthread_attr_destroy( &thread_attributes );
983         }
984
985         else
986         {
987                 while ( n-- )
988                 {
989                         if ( pthread_create( &thread, NULL, consumer_worker_thread, this ) == 0 )
990                                 mlt_deque_push_back( this->worker_threads, (void*) thread );
991                 }
992         }
993 }
994
995 /** Stop the read/render thread.
996  *
997  * \private \memberof mlt_consumer_s
998  * \param this a consumer
999  */
1000
1001 static void consumer_read_ahead_stop( mlt_consumer this )
1002 {
1003         // Make sure we're running
1004         if ( this->ahead )
1005         {
1006                 // Inform thread to stop
1007                 this->ahead = 0;
1008
1009                 // Broadcast to the condition in case it's waiting
1010                 pthread_mutex_lock( &this->frame_queue_mutex );
1011                 pthread_cond_broadcast( &this->frame_queue_cond );
1012                 pthread_mutex_unlock( &this->frame_queue_mutex );
1013
1014                 // Broadcast to the put condition in case it's waiting
1015                 pthread_mutex_lock( &this->put_mutex );
1016                 pthread_cond_broadcast( &this->put_cond );
1017                 pthread_mutex_unlock( &this->put_mutex );
1018
1019                 // Join the thread
1020                 pthread_join( this->ahead_thread, NULL );
1021
1022                 // Destroy the frame queue mutex
1023                 pthread_mutex_destroy( &this->frame_queue_mutex );
1024
1025                 // Destroy the condition
1026                 pthread_cond_destroy( &this->frame_queue_cond );
1027
1028                 // Wipe the queue
1029                 while ( mlt_deque_count( this->frame_queue ) )
1030                         mlt_frame_close( mlt_deque_pop_back( this->frame_queue ) );
1031
1032                 // Close the queue
1033                 mlt_deque_close( this->frame_queue );
1034         }
1035 }
1036
1037 /** Stop the worker threads.
1038  *
1039  * \private \memberof mlt_consumer_s
1040  * \param this a consumer
1041  */
1042
1043 static void consumer_work_stop( mlt_consumer this )
1044 {
1045         // Make sure we're running
1046         if ( this->ahead )
1047         {
1048                 // Inform thread to stop
1049                 this->ahead = 0;
1050
1051                 // Broadcast to the frame_queue condition in case it's waiting
1052                 pthread_mutex_lock( &this->frame_queue_mutex );
1053                 pthread_cond_broadcast( &this->frame_queue_cond );
1054                 pthread_mutex_unlock( &this->frame_queue_mutex );
1055
1056                 // Broadcast to the done_queue condition in case it's waiting
1057                 pthread_mutex_lock( &this->done_queue_mutex );
1058                 pthread_cond_broadcast( &this->done_queue_cond );
1059                 pthread_mutex_unlock( &this->done_queue_mutex );
1060
1061                 // Broadcast to the put condition in case it's waiting
1062                 pthread_mutex_lock( &this->put_mutex );
1063                 pthread_cond_broadcast( &this->put_cond );
1064                 pthread_mutex_unlock( &this->put_mutex );
1065
1066                 // Join the threads
1067                 pthread_t thread;
1068                 while ( ( thread = (pthread_t) mlt_deque_pop_front( this->worker_threads ) ) )
1069                         pthread_join( thread, NULL );
1070
1071                 // Destroy the mutexes
1072                 pthread_mutex_destroy( &this->frame_queue_mutex );
1073                 pthread_mutex_destroy( &this->done_queue_mutex );
1074
1075                 // Destroy the conditions
1076                 pthread_cond_destroy( &this->frame_queue_cond );
1077                 pthread_cond_destroy( &this->done_queue_cond );
1078
1079                 // Wipe the queues
1080                 while ( mlt_deque_count( this->frame_queue ) )
1081                         mlt_frame_close( mlt_deque_pop_back( this->frame_queue ) );
1082                 while ( mlt_deque_count( this->done_queue ) )
1083                         mlt_frame_close( mlt_deque_pop_back( this->done_queue ) );
1084
1085                 // Close the queues
1086                 mlt_deque_close( this->frame_queue );
1087                 mlt_deque_close( this->done_queue );
1088                 mlt_deque_close( this->worker_threads );
1089         }
1090 }
1091
1092 /** Flush the read/render thread's buffer.
1093  *
1094  * \public \memberof mlt_consumer_s
1095  * \param this a consumer
1096  */
1097
1098 void mlt_consumer_purge( mlt_consumer this )
1099 {
1100         if ( this->ahead )
1101         {
1102                 pthread_mutex_lock( &this->frame_queue_mutex );
1103                 while ( mlt_deque_count( this->frame_queue ) )
1104                         mlt_frame_close( mlt_deque_pop_back( this->frame_queue ) );
1105                 pthread_cond_broadcast( &this->frame_queue_cond );
1106                 pthread_mutex_unlock( &this->frame_queue_mutex );
1107
1108                 if ( this->done_queue )
1109                 {
1110                         pthread_mutex_lock( &this->done_queue_mutex );
1111                         while ( mlt_deque_count( this->done_queue ) )
1112                                 mlt_frame_close( mlt_deque_pop_back( this->done_queue ) );
1113                         pthread_cond_broadcast( &this->done_queue_cond );
1114                         pthread_mutex_unlock( &this->done_queue_mutex );
1115                 }
1116         }
1117 }
1118
1119 /** Get the next frame from the producer connected to a consumer.
1120  *
1121  * Typically, one uses this instead of \p mlt_consumer_get_frame to make
1122  * the asynchronous/real-time behavior configurable at runtime.
1123  * You should close the frame returned from this when you are done with it.
1124  *
1125  * \public \memberof mlt_consumer_s
1126  * \param this a consumer
1127  * \return a frame
1128  */
1129
1130 mlt_frame mlt_consumer_rt_frame( mlt_consumer this )
1131 {
1132         // Frame to return
1133         mlt_frame frame = NULL;
1134
1135         // Get the properties
1136         mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
1137
1138         // Check if the user has requested real time or not
1139         if ( this->real_time == 1 || this->real_time == -1 )
1140         {
1141                 int size = 1;
1142
1143                 // Is the read ahead running?
1144                 if ( this->ahead == 0 )
1145                 {
1146                         int buffer = mlt_properties_get_int( properties, "buffer" );
1147                         int prefill = mlt_properties_get_int( properties, "prefill" );
1148                         consumer_read_ahead_start( this );
1149                         if ( buffer > 1 )
1150                                 size = prefill > 0 && prefill < buffer ? prefill : buffer;
1151                 }
1152
1153                 // Get frame from queue
1154                 pthread_mutex_lock( &this->frame_queue_mutex );
1155                 while( this->ahead && mlt_deque_count( this->frame_queue ) < size )
1156                         pthread_cond_wait( &this->frame_queue_cond, &this->frame_queue_mutex );
1157                 frame = mlt_deque_pop_front( this->frame_queue );
1158                 pthread_cond_broadcast( &this->frame_queue_cond );
1159                 pthread_mutex_unlock( &this->frame_queue_mutex );
1160         }
1161         else if ( this->real_time > 1 || this->real_time < -1 )
1162         {
1163                 // Use multiple worker threads and a work queue
1164                 int size = abs( this->real_time );
1165                 int buffer = mlt_properties_get_int( properties, "buffer" );
1166                 buffer = buffer < size ? size : buffer;
1167
1168                 // Start worker threads if not already
1169                 if ( ! this->ahead )
1170                 {
1171                         int prefill = mlt_properties_get_int( properties, "prefill" );
1172                         prefill = prefill > 0 && prefill < buffer ? prefill : buffer;
1173
1174                         consumer_work_start( this );
1175
1176                         // Fill the work queue
1177                         int i = buffer;
1178                         while ( i-- )
1179                         {
1180                                 frame = mlt_consumer_get_frame( this );
1181                                 if ( frame )
1182                                 {
1183                                         pthread_mutex_lock( &this->frame_queue_mutex );
1184                                         mlt_deque_push_back( this->frame_queue, frame );
1185                                         pthread_cond_signal( &this->frame_queue_cond );
1186                                         pthread_mutex_unlock( &this->frame_queue_mutex );
1187                                 }
1188                         }
1189
1190                         // Wait for prefill
1191                         pthread_mutex_lock( &this->done_queue_mutex );
1192                         while( this->ahead && mlt_deque_count( this->done_queue ) < prefill )
1193                                 pthread_cond_wait( &this->done_queue_cond, &this->done_queue_mutex );
1194                         pthread_mutex_unlock( &this->done_queue_mutex );
1195                 }
1196
1197                 // Try to get frame from the done queue
1198                 mlt_log_verbose( MLT_CONSUMER_SERVICE(this), "size %d done count %d work count %d\n",
1199                         size, mlt_deque_count( this->done_queue ), mlt_deque_count( this->frame_queue ) );
1200                 if ( this->real_time > 0 && mlt_deque_count( this->done_queue ) <= 0 )
1201                 {
1202                         // Non-realtime and no ready frames
1203                         frame = mlt_consumer_get_frame( this );
1204                 }
1205                 else
1206                 {
1207                         // Wait for work queue space
1208                         pthread_mutex_lock( &this->frame_queue_mutex );
1209                         while( this->ahead && mlt_deque_count( this->frame_queue ) >= buffer )
1210                                 pthread_cond_wait( &this->frame_queue_cond, &this->frame_queue_mutex );
1211                         pthread_mutex_unlock( &this->frame_queue_mutex );
1212
1213                         // Feed the work queue
1214                         frame = mlt_consumer_get_frame( this );
1215                         if ( frame )
1216                         {
1217                                 pthread_mutex_lock( &this->frame_queue_mutex );
1218                                 mlt_deque_push_back( this->frame_queue, frame );
1219                                 pthread_cond_signal( &this->frame_queue_cond );
1220                                 pthread_mutex_unlock( &this->frame_queue_mutex );
1221                         }
1222
1223                         // Get the frame from the done queue
1224                         pthread_mutex_lock( &this->done_queue_mutex );
1225                         while( this->ahead && mlt_deque_count( this->done_queue ) <= 0 )
1226                                 pthread_cond_wait( &this->done_queue_cond, &this->done_queue_mutex );
1227                         frame = mlt_deque_pop_front( this->done_queue );
1228                         pthread_mutex_unlock( &this->done_queue_mutex );
1229                 }
1230         }
1231         else
1232         {
1233                 // Get the frame in non real time
1234                 frame = mlt_consumer_get_frame( this );
1235
1236                 // This isn't true, but from the consumers perspective it is
1237                 if ( frame != NULL )
1238                         mlt_properties_set_int( MLT_FRAME_PROPERTIES( frame ), "rendered", 1 );
1239         }
1240
1241         return frame;
1242 }
1243
1244 /** Callback for the implementation to indicate a stopped condition.
1245  *
1246  * \public \memberof mlt_consumer_s
1247  * \param this a consumer
1248  */
1249
1250 void mlt_consumer_stopped( mlt_consumer this )
1251 {
1252         mlt_properties_set_int( MLT_CONSUMER_PROPERTIES( this ), "running", 0 );
1253         mlt_events_fire( MLT_CONSUMER_PROPERTIES( this ), "consumer-stopped", NULL );
1254         mlt_event_unblock( this->event_listener );
1255 }
1256
1257 /** Stop the consumer.
1258  *
1259  * \public \memberof mlt_consumer_s
1260  * \param this a consumer
1261  * \return true if there was an error
1262  */
1263
1264 int mlt_consumer_stop( mlt_consumer this )
1265 {
1266         // Get the properies
1267         mlt_properties properties = MLT_CONSUMER_PROPERTIES( this );
1268
1269         // Just in case...
1270         mlt_log( MLT_CONSUMER_SERVICE( this ), MLT_LOG_DEBUG, "stopping put waiting\n" );
1271         pthread_mutex_lock( &this->put_mutex );
1272         this->put_active = 0;
1273         pthread_cond_broadcast( &this->put_cond );
1274         pthread_mutex_unlock( &this->put_mutex );
1275
1276         // Stop the consumer
1277         mlt_log( MLT_CONSUMER_SERVICE( this ), MLT_LOG_DEBUG, "stopping consumer\n" );
1278         if ( this->stop != NULL )
1279                 this->stop( this );
1280
1281         // Check if the user has requested real time or not and stop if necessary
1282         mlt_log( MLT_CONSUMER_SERVICE( this ), MLT_LOG_DEBUG, "stopping read_ahead\n" );
1283         if ( abs( this->real_time ) == 1 )
1284                 consumer_read_ahead_stop( this );
1285         else if ( abs( this->real_time ) > 1 )
1286                 consumer_work_stop( this );
1287
1288         // Kill the test card
1289         mlt_properties_set_data( properties, "test_card_producer", NULL, 0, NULL, NULL );
1290
1291         // Check and run a post command
1292         if ( mlt_properties_get( properties, "post" ) )
1293                 if (system( mlt_properties_get( properties, "post" ) ) == -1 )
1294                         mlt_log( MLT_CONSUMER_SERVICE( this ), MLT_LOG_ERROR, "system(%s) failed!\n", mlt_properties_get( properties, "post" ) );
1295
1296         mlt_log( MLT_CONSUMER_SERVICE( this ), MLT_LOG_DEBUG, "stopped\n" );
1297
1298         return 0;
1299 }
1300
1301 /** Determine if the consumer is stopped.
1302  *
1303  * \public \memberof mlt_consumer_s
1304  * \param this a consumer
1305  * \return true if the consumer is stopped
1306  */
1307
1308 int mlt_consumer_is_stopped( mlt_consumer this )
1309 {
1310         // Check if the consumer is stopped
1311         if ( this->is_stopped != NULL )
1312                 return this->is_stopped( this );
1313
1314         return 0;
1315 }
1316
1317 /** Close and destroy the consumer.
1318  *
1319  * \public \memberof mlt_consumer_s
1320  * \param this a consumer
1321  */
1322
1323 void mlt_consumer_close( mlt_consumer this )
1324 {
1325         if ( this != NULL && mlt_properties_dec_ref( MLT_CONSUMER_PROPERTIES( this ) ) <= 0 )
1326         {
1327                 // Get the childs close function
1328                 void ( *consumer_close )( ) = this->close;
1329
1330                 if ( consumer_close )
1331                 {
1332                         // Just in case...
1333                         //mlt_consumer_stop( this );
1334
1335                         this->close = NULL;
1336                         consumer_close( this );
1337                 }
1338                 else
1339                 {
1340                         // Make sure it only gets called once
1341                         this->parent.close = NULL;
1342
1343                         // Destroy the push mutex and condition
1344                         pthread_mutex_destroy( &this->put_mutex );
1345                         pthread_cond_destroy( &this->put_cond );
1346
1347                         mlt_service_close( &this->parent );
1348                 }
1349         }
1350 }
1351
1352 /** Get the position of the last frame shown.
1353  *
1354  * \public \memberof mlt_consumer_s
1355  * \param consumer a consumer
1356  * \return the position
1357  */
1358
1359 mlt_position mlt_consumer_position( mlt_consumer consumer )
1360 {
1361         return consumer->position;
1362 }
1363