static void foreach_consumer_stop( mlt_consumer consumer )
{
    mlt_properties properties = MLT_CONSUMER_PROPERTIES( consumer );
    mlt_consumer nested = NULL;
    char key[30];
    int index = 0;
    struct timespec tm = { 0, 1000 * 1000 };

    do {
        snprintf( key, sizeof(key), "%d.consumer", index++ );
        nested = mlt_properties_get_data( properties, key, NULL );
        if ( nested )
        {
            // Let consumer with terminate_on_pause stop on their own
            if ( mlt_properties_get_int( MLT_CONSUMER_PROPERTIES(nested), "terminate_on_pause" ) )
            {
                // Send additional dummy frame to unlatch nested consumer's threads
                mlt_consumer_put_frame( nested, mlt_frame_init( MLT_CONSUMER_SERVICE(consumer) ) );
                // wait for stop
                while ( !mlt_consumer_is_stopped( nested ) )
                    nanosleep( &tm, NULL );
            }
            else
            {
                mlt_consumer_stop( nested );
            }
        }
    } while ( nested );
}

static int stop( mlt_consumer consumer )
{
    // Check that we're running
    if ( !mlt_properties_get_int( MLT_CONSUMER_PROPERTIES(consumer), "joined" ) )
    {
        mlt_properties properties = MLT_CONSUMER_PROPERTIES( consumer );
        pthread_t *thread = mlt_properties_get_data( properties, "thread", NULL );

        // Stop the thread
        mlt_properties_set_int( properties, "running", 0 );

        // Wait for termination
        if ( thread )
        {
            foreach_consumer_refresh( consumer );
            pthread_join( *thread, NULL );
        }
        mlt_properties_set_int( properties, "joined", 1 );

        // Stop nested consumers
        foreach_consumer_stop( consumer );
    }

    return 0;
}

static void consumer_read_ahead_start( mlt_consumer self )
{
	// We're running now
	self->ahead = 1;

	// Create the frame queue
	self->queue = mlt_deque_init( );

	// Create the queue mutex
	pthread_mutex_init( &self->queue_mutex, NULL );

	// Create the condition
	pthread_cond_init( &self->queue_cond, NULL );

	// Create the read ahead
	if ( mlt_properties_get( MLT_CONSUMER_PROPERTIES( self ), "priority" ) )
	{
		struct sched_param priority;
		priority.sched_priority = mlt_properties_get_int( MLT_CONSUMER_PROPERTIES( self ), "priority" );
		pthread_attr_t thread_attributes;
		pthread_attr_init( &thread_attributes );
		pthread_attr_setschedpolicy( &thread_attributes, SCHED_OTHER );
		pthread_attr_setschedparam( &thread_attributes, &priority );
		pthread_attr_setinheritsched( &thread_attributes, PTHREAD_EXPLICIT_SCHED );
		pthread_attr_setscope( &thread_attributes, PTHREAD_SCOPE_SYSTEM );
		if ( pthread_create( &self->ahead_thread, &thread_attributes, consumer_read_ahead_thread, self ) < 0 )
			pthread_create( &self->ahead_thread, NULL, consumer_read_ahead_thread, self );
		pthread_attr_destroy( &thread_attributes );
	}
	else
	{
		pthread_create( &self->ahead_thread, NULL, consumer_read_ahead_thread, self );
	}
	self->started = 1;
}

static void on_jack_stopped( mlt_properties owner, mlt_consumer consumer, mlt_position *position )
{
    mlt_producer producer = mlt_properties_get_data( MLT_CONSUMER_PROPERTIES(consumer), "transport_producer", NULL );
    if ( producer )
    {
        mlt_producer_set_speed( producer, 0 );
        mlt_consumer_purge( consumer );
        mlt_producer_seek( producer, *position );
        mlt_properties_set_int( MLT_CONSUMER_PROPERTIES( consumer ), "refresh", 1 );
    }
}

static void foreach_consumer_refresh( mlt_consumer consumer )
{
    mlt_properties properties = MLT_CONSUMER_PROPERTIES( consumer );
    mlt_consumer nested = NULL;
    char key[30];
    int index = 0;

    do {
        snprintf( key, sizeof(key), "%d.consumer", index++ );
        nested = mlt_properties_get_data( properties, key, NULL );
        if ( nested ) mlt_properties_set_int( MLT_CONSUMER_PROPERTIES(nested), "refresh", 1 );
    } while ( nested );
}

void mlt_consumer_close( mlt_consumer self )
{
	if ( self != NULL && mlt_properties_dec_ref( MLT_CONSUMER_PROPERTIES( self ) ) <= 0 )
	{
		// Get the childs close function
		void ( *consumer_close )( ) = self->close;

		if ( consumer_close )
		{
			// Just in case...
			//mlt_consumer_stop( self );

			self->close = NULL;
			consumer_close( self );
		}
		else
		{
			// Make sure it only gets called once
			self->parent.close = NULL;

			// Destroy the push mutex and condition
			pthread_mutex_destroy( &self->put_mutex );
			pthread_cond_destroy( &self->put_cond );

			mlt_service_close( &self->parent );
		}
	}
}

static int start( mlt_consumer consumer )
{
    // Check that we're not already running
    if ( is_stopped( consumer ) )
    {
        mlt_properties properties = MLT_CONSUMER_PROPERTIES( consumer );
        pthread_t *thread = calloc( 1, sizeof( pthread_t ) );

        // Assign the thread to properties with automatic dealloc
        mlt_properties_set_data( properties, "thread", thread, sizeof( pthread_t ), free, NULL );

        // Set the running state
        mlt_properties_set_int( properties, "running", 1 );
        mlt_properties_set_int( properties, "joined", 0 );

        // Construct and start nested consumers
        if ( !mlt_properties_get_data( properties, "0.consumer", NULL ) )
            foreach_consumer_init( consumer );
        foreach_consumer_start( consumer );

        // Create the thread
        pthread_create( thread, NULL, consumer_thread, consumer );
    }
    return 0;
}

void consumer_frame_show_cb( mlt_consumer sdl, mlt_consumer parent, mlt_frame frame )
{
	consumer_sdl self = parent->child;
	self->last_speed = mlt_properties_get_double( MLT_FRAME_PROPERTIES( frame ), "_speed" );
	self->last_position = mlt_frame_get_position( frame );
	mlt_events_fire( MLT_CONSUMER_PROPERTIES( parent ), "consumer-frame-show", frame, NULL );
}

int consumer_start( mlt_consumer parent )
{
	consumer_sdl self = parent->child;

	if ( !self->running )
	{
		consumer_stop( parent );

		mlt_properties properties = MLT_CONSUMER_PROPERTIES( parent );
		char *audio_driver = mlt_properties_get( properties, "audio_driver" );
		char *audio_device = mlt_properties_get( properties, "audio_device" );

		if ( audio_driver && strcmp( audio_driver, "" ) )
			setenv( "SDL_AUDIODRIVER", audio_driver, 1 );

		if ( audio_device && strcmp( audio_device, "" ) )
			setenv( "AUDIODEV", audio_device, 1 );

		pthread_mutex_lock( &mlt_sdl_mutex );
		int ret = SDL_Init( SDL_INIT_AUDIO | SDL_INIT_NOPARACHUTE );
		pthread_mutex_unlock( &mlt_sdl_mutex );
		if ( ret < 0 )
		{
			mlt_log_error( MLT_CONSUMER_SERVICE(parent), "Failed to initialize SDL: %s\n", SDL_GetError() );
			return -1;
		}

		self->running = 1;
		self->joined = 0;
		pthread_create( &self->thread, NULL, consumer_thread, self );
	}

	return 0;
}

static int start( mlt_consumer consumer )
{
	// Get the properties
	mlt_properties properties = MLT_CONSUMER_PROPERTIES( consumer );
	DeckLinkConsumer* decklink = (DeckLinkConsumer*) consumer->child;
	int result = decklink->start( mlt_properties_get_int( properties, "preroll" ) ) ? 0 : 1;

	// Check that we're not already running
	if ( !result && !mlt_properties_get_int( properties, "running" ) )
	{
		// Allocate a thread
		pthread_t *pthread = (pthread_t*) calloc( 1, sizeof( pthread_t ) );

		// Assign the thread to properties
		mlt_properties_set_data( properties, "pthread", pthread, sizeof( pthread_t ), free, NULL );

		// Set the running state
		mlt_properties_set_int( properties, "running", 1 );
		mlt_properties_set_int( properties, "joined", 0 );

		// Create the thread
		pthread_create( pthread, NULL, run, consumer->child );
	}
	return result;
}

static int consumer_stop( mlt_consumer parent )
{
	// Get the actual object
	consumer_sdl self = parent->child;

	if ( self->joined == 0 )
	{
		mlt_properties properties = MLT_CONSUMER_PROPERTIES( parent );
		int app_locked = mlt_properties_get_int( properties, "app_locked" );
		void ( *lock )( void ) = mlt_properties_get_data( properties, "app_lock", NULL );
		void ( *unlock )( void ) = mlt_properties_get_data( properties, "app_unlock", NULL );

		if ( app_locked && unlock ) unlock( );

		// Kill the thread and clean up
		self->running = 0;

		pthread_mutex_lock( &self->refresh_mutex );
		pthread_cond_broadcast( &self->refresh_cond );
		pthread_mutex_unlock( &self->refresh_mutex );
#ifndef WIN32
		if ( self->thread )
#endif
			pthread_join( self->thread, NULL );
		self->joined = 1;

		if ( app_locked && lock ) lock( );
		
		pthread_mutex_lock( &mlt_sdl_mutex );
		SDL_Quit( );
		pthread_mutex_unlock( &mlt_sdl_mutex );
	}

	return 0;
}

int mlt_consumer_init( mlt_consumer self, void *child, mlt_profile profile )
{
	int error = 0;
	memset( self, 0, sizeof( struct mlt_consumer_s ) );
	self->child = child;
	error = mlt_service_init( &self->parent, self );
	if ( error == 0 )
	{
		// Get the properties from the service
		mlt_properties properties = MLT_SERVICE_PROPERTIES( &self->parent );

		// Apply profile to properties
		if ( profile == NULL )
		{
			// Normally the application creates the profile and controls its lifetime
			// This is the fallback exception handling
			profile = mlt_profile_init( NULL );
			mlt_properties properties = MLT_CONSUMER_PROPERTIES( self );
			mlt_properties_set_data( properties, "_profile", profile, 0, (mlt_destructor)mlt_profile_close, NULL );
		}
		apply_profile_properties( self, profile, properties );

		// Default rescaler for all consumers
		mlt_properties_set( properties, "rescale", "bilinear" );

		// Default read ahead buffer size
		mlt_properties_set_int( properties, "buffer", 25 );
		mlt_properties_set_int( properties, "drop_max", 5 );

		// Default audio frequency and channels
		mlt_properties_set_int( properties, "frequency", 48000 );
		mlt_properties_set_int( properties, "channels", 2 );

		// Default of all consumers is real time
		mlt_properties_set_int( properties, "real_time", 1 );

		// Default to environment test card
		mlt_properties_set( properties, "test_card", mlt_environment( "MLT_TEST_CARD" ) );

		// Hmm - default all consumers to yuv422 :-/
		self->format = mlt_image_yuv422;
		mlt_properties_set( properties, "mlt_image_format", mlt_image_format_name( self->format ) );
		mlt_properties_set( properties, "mlt_audio_format", mlt_audio_format_name( mlt_audio_s16 ) );

		mlt_events_register( properties, "consumer-frame-show", ( mlt_transmitter )mlt_consumer_frame_show );
		mlt_events_register( properties, "consumer-frame-render", ( mlt_transmitter )mlt_consumer_frame_render );
		mlt_events_register( properties, "consumer-stopped", NULL );
		mlt_events_listen( properties, self, "consumer-frame-show", ( mlt_listener )on_consumer_frame_show );

		// Register a property-changed listener to handle the profile property -
		// subsequent properties can override the profile
		self->event_listener = mlt_events_listen( properties, self, "property-changed", ( mlt_listener )mlt_consumer_property_changed );

		// Create the push mutex and condition
		pthread_mutex_init( &self->put_mutex, NULL );
		pthread_cond_init( &self->put_cond, NULL );

	}
	return error;
}

mlt_consumer consumer_multi_init( mlt_profile profile, mlt_service_type type, const char *id, char *arg )
{
    mlt_consumer consumer = mlt_consumer_new( profile );

    if ( consumer )
    {
        mlt_properties properties = MLT_CONSUMER_PROPERTIES(consumer);

        // Set defaults
        mlt_properties_set( properties, "resource", arg );
        mlt_properties_set_int( properties, "real_time", -1 );
        mlt_properties_set_int( properties, "terminate_on_pause", 1 );

        // Init state
        mlt_properties_set_int( properties, "joined", 1 );

        // Assign callbacks
        consumer->close = consumer_close;
        consumer->start = start;
        consumer->stop = stop;
        consumer->is_stopped = is_stopped;
    }

    return consumer;
}

static void *consumer_thread( void *arg )
{
	// Map the argument to the object
	mlt_consumer consumer = arg;

	// Get the properties
	mlt_properties properties = MLT_CONSUMER_PROPERTIES( consumer );

	// Convenience functionality
	int terminate_on_pause = mlt_properties_get_int( properties, "terminate_on_pause" );
	int terminated = 0;

	// Frame and size
	mlt_frame frame = NULL;

	// Loop while running
	while( !terminated && mlt_properties_get_int( properties, "_running" ) )
	{
		// Get the frame
		frame = mlt_consumer_rt_frame( consumer );

		// Check for termination
		if ( terminate_on_pause && frame != NULL )
			terminated = mlt_properties_get_double( MLT_FRAME_PROPERTIES( frame ), "_speed" ) == 0.0;

		// Check that we have a frame to work with
		if ( frame )
		{
			avsync_stats* stats = mlt_properties_get_data( properties, "_stats", NULL );
			double fps = mlt_properties_get_double( properties, "fps" );
			mlt_position pos = mlt_frame_get_position( frame );

			 if( !strcmp( mlt_properties_get( properties, "report" ), "frame" ) )
			 {
				 stats->report_frames = 1;
			 }
			 else
			 {
				 stats->report_frames = 0;
			 }

			detect_flash( frame, pos, fps, stats );
			detect_blip( frame, pos, fps, stats );
			calculate_sync( stats );
			report_results( stats, pos );

			// Close the frame
			mlt_events_fire( properties, "consumer-frame-show", frame, NULL );
			mlt_frame_close( frame );
		}
	}

	// Indicate that the consumer is stopped
	mlt_properties_set_int( properties, "_running", 0 );
	mlt_consumer_stopped( consumer );

	return NULL;
}

int consumer_stop( mlt_consumer parent )
{
	// Get the actual object
	consumer_sdl self = parent->child;

	if ( self->joined == 0 )
	{
		// Kill the thread and clean up
		self->joined = 1;
		self->running = 0;

#ifndef _WIN32
		if ( self->thread )
#endif
			pthread_join( self->thread, NULL );

		// cleanup SDL
		if ( self->sdl_texture )
			SDL_DestroyTexture( self->sdl_texture );
		self->sdl_texture = NULL;
		if ( self->sdl_renderer )
			SDL_DestroyRenderer( self->sdl_renderer );
		self->sdl_renderer = NULL;
		if ( self->sdl_window )
			SDL_DestroyWindow( self->sdl_window );
		self->sdl_window = NULL;

		if ( !mlt_properties_get_int( MLT_CONSUMER_PROPERTIES( parent ), "audio_off" ) )
		{
			pthread_mutex_lock( &self->audio_mutex );
			pthread_cond_broadcast( &self->audio_cond );
			pthread_mutex_unlock( &self->audio_mutex );
			SDL_QuitSubSystem( SDL_INIT_AUDIO );
		}

		if ( mlt_properties_get_int( MLT_CONSUMER_PROPERTIES( parent ), "sdl_started" ) == 0 )
		{
			pthread_mutex_lock( &mlt_sdl_mutex );
			SDL_Quit( );
			pthread_mutex_unlock( &mlt_sdl_mutex );
		}
	}

	return 0;
}

static mlt_consumer generate_consumer( mlt_consumer consumer, mlt_properties props, int index )
{
    mlt_profile profile = NULL;
    if ( mlt_properties_get( props, "mlt_profile" ) )
        profile = mlt_profile_init( mlt_properties_get( props, "mlt_profile" ) );
    if ( !profile )
        profile = mlt_profile_clone( mlt_service_profile( MLT_CONSUMER_SERVICE(consumer) ) );
    mlt_consumer nested = create_consumer( profile, mlt_properties_get( props, "mlt_service" ),
                                           mlt_properties_get( props, "target" ) );

    if ( nested )
    {
        mlt_properties properties = MLT_CONSUMER_PROPERTIES(consumer);
        mlt_properties nested_props = MLT_CONSUMER_PROPERTIES(nested);
        char key[30];

        snprintf( key, sizeof(key), "%d.consumer", index );
        mlt_properties_set_data( properties, key, nested, 0, (mlt_destructor) mlt_consumer_close, NULL );
        snprintf( key, sizeof(key), "%d.profile", index );
        mlt_properties_set_data( properties, key, profile, 0, (mlt_destructor) mlt_profile_close, NULL );

        mlt_properties_set_int( nested_props, "put_mode", 1 );
        mlt_properties_pass_list( nested_props, properties, "terminate_on_pause" );
        mlt_properties_set( props, "consumer", NULL );
        // set mlt_profile before other properties to facilitate presets
        mlt_properties_pass_list( nested_props, props, "mlt_profile" );
        mlt_properties_inherit( nested_props, props );

        attach_normalisers( profile, MLT_CONSUMER_SERVICE(nested) );

        // Relay the first available consumer-frame-show event
        mlt_event event = mlt_properties_get_data( properties, "frame-show-event", NULL );
        if ( !event )
        {
            event = mlt_events_listen( nested_props, properties, "consumer-frame-show", (mlt_listener) on_frame_show );
            mlt_properties_set_data( properties, "frame-show-event", event, 0, /*mlt_event_close*/ NULL, NULL );
        }
    }
    else
    {
        mlt_profile_close( profile );
    }
    return nested;
}

static void on_jack_started( mlt_properties owner, mlt_consumer consumer, mlt_position *position )
{
    mlt_producer producer = mlt_properties_get_data( MLT_CONSUMER_PROPERTIES(consumer), "transport_producer", NULL );
    if ( producer )
    {
        if ( mlt_producer_get_speed( producer ) != 0 )
        {
            mlt_properties jack = mlt_properties_get_data( MLT_CONSUMER_PROPERTIES( consumer ), "jack_filter", NULL );
            mlt_events_fire( jack, "jack-stop", NULL );
        }
        else
        {
            mlt_producer_set_speed( producer, 1 );
            mlt_consumer_purge( consumer );
            mlt_producer_seek( producer, *position );
            mlt_properties_set_int( MLT_CONSUMER_PROPERTIES( consumer ), "refresh", 1 );
        }
    }
}

static void *consumer_thread( void *arg )
{
    mlt_consumer consumer = arg;
    mlt_properties properties = MLT_CONSUMER_PROPERTIES( consumer );
    mlt_frame frame = NULL;

    // Determine whether to stop at end-of-media
    int terminate_on_pause = mlt_properties_get_int( properties, "terminate_on_pause" );
    int terminated = 0;

    // Loop while running
    while ( !terminated && !is_stopped( consumer ) )
    {
        // Get the next frame
        frame = mlt_consumer_rt_frame( consumer );

        // Check for termination
        if ( terminate_on_pause && frame )
            terminated = mlt_properties_get_double( MLT_FRAME_PROPERTIES( frame ), "_speed" ) == 0.0;

        // Check that we have a frame to work with
        if ( frame && !terminated && !is_stopped( consumer ) )
        {
            if ( mlt_properties_get_int( MLT_FRAME_PROPERTIES(frame), "rendered" ) )
            {
                if ( mlt_properties_get_int( MLT_FRAME_PROPERTIES(frame), "_speed" ) == 0 )
                    foreach_consumer_refresh( consumer );
                foreach_consumer_put( consumer, frame );
            }
            else
            {
                int dropped = mlt_properties_get_int( properties, "_dropped" );
                mlt_log_info( MLT_CONSUMER_SERVICE(consumer), "dropped frame %d\n", ++dropped );
                mlt_properties_set_int( properties, "_dropped", dropped );
            }
            mlt_frame_close( frame );
        }
        else
        {
            if ( frame && terminated )
            {
                // Send this termination frame to nested consumers for their cancellation
                foreach_consumer_put( consumer, frame );
            }
            if ( frame )
                mlt_frame_close( frame );
            terminated = 1;
        }
    }

    // Indicate that the consumer is stopped
    mlt_consumer_stopped( consumer );

    return NULL;
}

static void foreach_consumer_start( mlt_consumer consumer )
{
    mlt_properties properties = MLT_CONSUMER_PROPERTIES( consumer );
    mlt_consumer nested = NULL;
    char key[30];
    int index = 0;

    do {
        snprintf( key, sizeof(key), "%d.consumer", index++ );
        nested = mlt_properties_get_data( properties, key, NULL );
        if ( nested )
        {
            mlt_properties nested_props = MLT_CONSUMER_PROPERTIES(nested);
            mlt_properties_set_position( nested_props, "_multi_position", 0 );
            mlt_properties_set_data( nested_props, "_multi_audio", NULL, 0, NULL, NULL );
            mlt_properties_set_int( nested_props, "_multi_samples", 0 );
            mlt_consumer_start( nested );
        }
    } while ( nested );
}

void ServiceManager::setupConsumerListener(mlt_frame frame)
{
    // If there is a consumer property, listen to the consumer-stopping event to cancel rendering.
    if (!event) {
        mlt_consumer consumer = static_cast<mlt_consumer>(mlt_properties_get_data(MLT_FRAME_PROPERTIES(frame), "consumer", 0));
        if (consumer) {
            event = MLT_CONSUMER_PROPERTIES(consumer);
            mlt_events_listen(event, this, "consumer-stopping",
                reinterpret_cast<mlt_listener>(MLTWebVfx::consumerStoppingListener));
        }
    }
}