自学教程：C++ GST_APP_SINK函数代码示例

bool  ofGstUtils::getIsMovieDone(){	if(isAppSink){		return gst_app_sink_is_eos(GST_APP_SINK(gstSink));	}else{		return bIsMovieDone;	}}

static gboolean create_pipeline(SpiceGstDecoder *decoder){    gchar *desc;    gboolean auto_enabled;    guint opt;    GstAppSinkCallbacks appsink_cbs = { NULL };    GError *err = NULL;    GstBus *bus;    auto_enabled = (g_getenv("SPICE_GSTVIDEO_AUTO") != NULL);    if (auto_enabled || !VALID_VIDEO_CODEC_TYPE(decoder->base.codec_type)) {        SPICE_DEBUG("Trying %s for codec type %d %s",                    gst_opts[0].dec_name, decoder->base.codec_type,                    (auto_enabled) ? "(SPICE_GSTVIDEO_AUTO is set)" : "");        opt = 0;    } else {        opt = decoder->base.codec_type;    }    /* - We schedule the frame display ourselves so set sync=false on appsink     *   so the pipeline decodes them as fast as possible. This will also     *   minimize the risk of frames getting lost when we rebuild the     *   pipeline.     * - Set max-bytes=0 on appsrc so it does not drop frames that may be     *   needed by those that follow.     */    desc = g_strdup_printf("appsrc name=src is-live=true format=time max-bytes=0 block=true "                           "%s ! %s ! videoconvert ! appsink name=sink "                           "caps=video/x-raw,format=BGRx sync=false drop=false",                           gst_opts[opt].dec_caps, gst_opts[opt].dec_name);    SPICE_DEBUG("GStreamer pipeline: %s", desc);    decoder->pipeline = gst_parse_launch_full(desc, NULL, GST_PARSE_FLAG_FATAL_ERRORS, &err);    g_free(desc);    if (!decoder->pipeline) {        spice_warning("GStreamer error: %s", err->message);        g_clear_error(&err);        return FALSE;    }    decoder->appsrc = GST_APP_SRC(gst_bin_get_by_name(GST_BIN(decoder->pipeline), "src"));    decoder->appsink = GST_APP_SINK(gst_bin_get_by_name(GST_BIN(decoder->pipeline), "sink"));    appsink_cbs.new_sample = new_sample;    gst_app_sink_set_callbacks(decoder->appsink, &appsink_cbs, decoder, NULL);    bus = gst_pipeline_get_bus(GST_PIPELINE(decoder->pipeline));    gst_bus_add_watch(bus, handle_pipeline_message, decoder);    gst_object_unref(bus);    decoder->clock = gst_pipeline_get_clock(GST_PIPELINE(decoder->pipeline));    if (gst_element_set_state(decoder->pipeline, GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE) {        SPICE_DEBUG("GStreamer error: Unable to set the pipeline to the playing state.");        free_pipeline(decoder);        return FALSE;    }    return TRUE;}

GstFlowReturn GStreamerWrapper::onNewBufferFromAudioSource( GstAppSink* appsink, void* listener ){	GstBuffer* gstAudioSinkBuffer = gst_app_sink_pull_buffer( GST_APP_SINK( appsink ) );	( ( GStreamerWrapper * )listener )->newAudioSinkBufferCallback( gstAudioSinkBuffer );	gst_buffer_unref( gstAudioSinkBuffer );	return GST_FLOW_OK;}

GstFlowReturn GStreamerWrapper::onNewPrerollFromVideoSource( GstAppSink* appsink, void* listener ){	GstBuffer* gstVideoSinkBuffer = gst_app_sink_pull_preroll( GST_APP_SINK( appsink ) );	( ( GStreamerWrapper *)listener )->newVideoSinkPrerollCallback( gstVideoSinkBuffer );	gst_buffer_unref( gstVideoSinkBuffer );	return GST_FLOW_OK;}

static GstFlowReturngst_app_sink_render_common (GstBaseSink * psink, GstMiniObject * data,    gboolean is_list){  GstAppSink *appsink = GST_APP_SINK (psink);  gboolean emit;  g_mutex_lock (appsink->priv->mutex);  if (appsink->priv->flushing)    goto flushing;  GST_DEBUG_OBJECT (appsink, "pushing render buffer%s %p on queue (%d)",      is_list ? " list" : "", data, appsink->priv->queue->length);  while (appsink->priv->max_buffers > 0 &&      appsink->priv->queue->length >= appsink->priv->max_buffers) {    if (appsink->priv->drop) {      GstMiniObject *obj;      /* we need to drop the oldest buffer/list and try again */      obj = g_queue_pop_head (appsink->priv->queue);      GST_DEBUG_OBJECT (appsink, "dropping old buffer/list %p", obj);      gst_mini_object_unref (obj);    } else {      GST_DEBUG_OBJECT (appsink, "waiting for free space, length %d >= %d",          appsink->priv->queue->length, appsink->priv->max_buffers);      /* wait for a buffer to be removed or flush */      g_cond_wait (appsink->priv->cond, appsink->priv->mutex);      if (appsink->priv->flushing)        goto flushing;    }  }  /* we need to ref the buffer when pushing it in the queue */  g_queue_push_tail (appsink->priv->queue, gst_mini_object_ref (data));  g_cond_signal (appsink->priv->cond);  emit = appsink->priv->emit_signals;  g_mutex_unlock (appsink->priv->mutex);  if (is_list) {    if (appsink->priv->callbacks.new_buffer_list)      appsink->priv->callbacks.new_buffer_list (appsink,          appsink->priv->user_data);  } else {    if (appsink->priv->callbacks.new_buffer)      appsink->priv->callbacks.new_buffer (appsink, appsink->priv->user_data);    else if (emit)      g_signal_emit (appsink, gst_app_sink_signals[SIGNAL_NEW_BUFFER], 0);  }  return GST_FLOW_OK;flushing:  {    GST_DEBUG_OBJECT (appsink, "we are flushing");    g_mutex_unlock (appsink->priv->mutex);    return GST_FLOW_WRONG_STATE;  }}

static GstFlowReturn on_new_preroll_from_source (GstAppSink * elt, void * data){#if GST_VERSION_MAJOR==0	GstBuffer *buffer;#else	GstSample *buffer;#endif	buffer = gst_app_sink_pull_preroll(GST_APP_SINK (elt));	return ((ofGstUtils*)data)->preroll_cb(buffer);}

void GstAppSinkPipeline::Initialize(std::string pipelineString){		GstPipelineWrapper::InitializePipelineWithString(pipelineString);		// setup appsink	appsink = GstPipelineWrapper::GetElementByName(APPSINK_NAME);	GstAppSinkCallbacks appsinkCallbacks;	appsinkCallbacks.new_preroll	= &GstAppSinkPipeline::NewPrerollCallback;	appsinkCallbacks.new_sample		= &GstAppSinkPipeline::NewSampleCallback;	appsinkCallbacks.eos			= &GstAppSinkPipeline::EndOfStreamCallback; 	//	std::cout << pipelineString << std::endl;		gst_app_sink_set_drop			(GST_APP_SINK(appsink), true);	gst_app_sink_set_max_buffers	(GST_APP_SINK(appsink), 1);	//gst_app_sink_set_emit_signals	(GST_APP_SINK(appsink), true);	gst_app_sink_set_callbacks		(GST_APP_SINK(appsink), &appsinkCallbacks, this, (GDestroyNotify)GstAppSinkPipeline::DestroyCallback);		}

GstElement *MediaPlayer::createVideoSink(){    GstElement * sink = createElement("appsink", "videosink");    if ( !sink )        return 0;    // Set the caps - so far we only want our image to be RGB    GstCaps *sinkCaps = gst_caps_new_simple( "video/x-raw", "format", G_TYPE_STRING, "BGRA", NULL );    gst_app_sink_set_caps( GST_APP_SINK( sink ), sinkCaps );    gst_caps_unref(sinkCaps);    // Set up the callbacks    GstAppSinkCallbacks callbacks = { 0, 0, 0, 0, 0 };    callbacks.new_sample = cb_new_sample;    gst_app_sink_set_callbacks( GST_APP_SINK(sink), &callbacks, this, NULL );    return sink;}

//// start the pipeline, grab a buffer, and pause again//bool CvCapture_GStreamer::grabFrame(){    if(!pipeline)        return false;    if(gst_app_sink_is_eos(GST_APP_SINK(sink)))        return false;    if(buffer)        gst_buffer_unref(buffer);    handleMessage();    buffer = gst_app_sink_pull_buffer(GST_APP_SINK(sink));    if(!buffer)        return false;    return true;}

static voidgst_app_sink_finalize (GObject * obj){  GstAppSink *appsink = GST_APP_SINK (obj);  g_mutex_free (appsink->priv->mutex);  g_cond_free (appsink->priv->cond);  g_queue_free (appsink->priv->queue);  G_OBJECT_CLASS (parent_class)->finalize (obj);}

GstFlowReturn MediaImpl::gstNewSampleCallback(GstElement*, MediaImpl *p){  GstSample *sample;  sample = gst_app_sink_pull_sample(GST_APP_SINK(p->_videoSink));  //g_signal_emit_by_name (p->_videoSink, "pull-sample", &sample);  p->get_queue_input_buf()->put(sample);  if (p->get_queue_output_buf()->size() > 1) {    sample = p->get_queue_output_buf()->get();    gst_sample_unref(sample);  }  return GST_FLOW_OK;}

static gbooleangst_app_sink_start (GstBaseSink * psink){  GstAppSink *appsink = GST_APP_SINK (psink);  g_mutex_lock (appsink->priv->mutex);  GST_DEBUG_OBJECT (appsink, "starting");  appsink->priv->started = TRUE;  g_mutex_unlock (appsink->priv->mutex);  return TRUE;}

/*! * /brief OpenIMAJCapGStreamer::removeFilter * /param filter filter to remove * remove the specified filter from the appsink template caps */void OpenIMAJCapGStreamer::removeFilter(const char *filter){    if(!caps)        return;        if (! gst_caps_is_writable(caps))        caps = gst_caps_make_writable (caps);        GstStructure *s = gst_caps_get_structure(caps, 0);    gst_structure_remove_field(s, filter);        gst_app_sink_set_caps(GST_APP_SINK(sink), caps);}

static gbooleangst_app_sink_unlock_stop (GstBaseSink * bsink){  GstAppSink *appsink = GST_APP_SINK (bsink);  g_mutex_lock (appsink->priv->mutex);  GST_DEBUG_OBJECT (appsink, "unlock stop");  appsink->priv->flushing = FALSE;  g_cond_signal (appsink->priv->cond);  g_mutex_unlock (appsink->priv->mutex);  return TRUE;}

static gbooleangst_app_sink_stop (GstBaseSink * psink){  GstAppSink *appsink = GST_APP_SINK (psink);  g_mutex_lock (appsink->priv->mutex);  GST_DEBUG_OBJECT (appsink, "stopping");  appsink->priv->flushing = TRUE;  appsink->priv->started = FALSE;  gst_app_sink_flush_unlocked (appsink);  g_mutex_unlock (appsink->priv->mutex);  return TRUE;}

void ofGstUtils::update(){	gstHandleMessage();	if (bLoaded == true){		if(!bFrameByFrame){			ofGstDataLock(&gstData);				bHavePixelsChanged = gstData.bHavePixelsChanged;				if (bHavePixelsChanged){					gstData.bHavePixelsChanged=false;					bIsMovieDone = false;					memcpy(pixels,gstData.pixels,width*height*bpp);				}			ofGstDataUnlock(&gstData);		}else{			GstBuffer *buffer;			//get the buffer from appsink			if(bPaused) buffer = gst_app_sink_pull_preroll (GST_APP_SINK (gstSink));			else buffer = gst_app_sink_pull_buffer (GST_APP_SINK (gstSink));			if(buffer){				guint size = GST_BUFFER_SIZE (buffer);				if(pixels){					memcpy (pixels, GST_BUFFER_DATA (buffer), size);					bHavePixelsChanged=true;				}				/// we don't need the appsink buffer anymore				gst_buffer_unref (buffer);			}		}	}	bIsFrameNew = bHavePixelsChanged;	bHavePixelsChanged = false;}

void CGstDecoder::OnDecodedBuffer(GstElement *appsink, void *data){  CGstDecoder *decoder = (CGstDecoder *)data;  GstBuffer *buffer = gst_app_sink_pull_buffer(GST_APP_SINK(appsink));  if (buffer)  {    if (decoder->m_callback)      decoder->m_callback->OnDecodedBuffer(buffer);    else      gst_buffer_unref(buffer);  }  else    printf("GStreamer: OnDecodedBuffer - Null Buffer/n");}

static GstCaps *gst_app_sink_getcaps (GstBaseSink * psink){  GstCaps *caps;  GstAppSink *appsink = GST_APP_SINK (psink);  GST_OBJECT_LOCK (appsink);  if ((caps = appsink->priv->caps))    gst_caps_ref (caps);  GST_DEBUG_OBJECT (appsink, "got caps %" GST_PTR_FORMAT, caps);  GST_OBJECT_UNLOCK (appsink);  return caps;}

void ofGstVideoUtils::update(){	if (isLoaded()){		if(!isFrameByFrame()){			mutex.lock();				bHavePixelsChanged = bBackPixelsChanged;				if (bHavePixelsChanged){					bBackPixelsChanged=false;					pixels.swap(backPixels);					if(prevBuffer) gst_buffer_unref (prevBuffer);					prevBuffer = buffer;				}			mutex.unlock();		}else{			GstBuffer *buffer;			//get the buffer from appsink			if(isPaused()) buffer = gst_app_sink_pull_preroll (GST_APP_SINK (getSink()));			else buffer = gst_app_sink_pull_buffer (GST_APP_SINK (getSink()));			if(buffer){				if(pixels.isAllocated()){					if(prevBuffer) gst_buffer_unref (prevBuffer);					//memcpy (pixels.getPixels(), GST_BUFFER_DATA (buffer), size);					pixels.setFromExternalPixels(GST_BUFFER_DATA (buffer),pixels.getWidth(),pixels.getHeight(),pixels.getNumChannels());					prevBuffer = buffer;					bHavePixelsChanged=true;				}			}		}	}else{		ofLog(OF_LOG_WARNING,"ofGstVideoUtils not loaded");	}	bIsFrameNew = bHavePixelsChanged;	bHavePixelsChanged = false;}

/*! * /brief CvCapture_GStreamer::removeFilter * /param filter filter to remove * remove the specified filter from the appsink template caps */void CvCapture_GStreamer::removeFilter(const char *filter){    if(!caps)        return;#if GST_VERSION_MAJOR > 0    if (! gst_caps_is_writable(caps))        caps = gst_caps_make_writable (caps);#endif    GstStructure *s = gst_caps_get_structure(caps, 0);    gst_structure_remove_field(s, filter);    gst_app_sink_set_caps(GST_APP_SINK(sink), caps);}

static GstFlowReturnon_new_sample_from_source (GstAppSink * elt, gpointer user_data){  ProgramData *data = (ProgramData *) user_data;  GstSample *sample;  GstBuffer *buffer;  GstElement *source;  sample = gst_app_sink_pull_sample (GST_APP_SINK (elt));  buffer = gst_sample_get_buffer (sample);  source = gst_bin_get_by_name (GST_BIN (data->sink), "testsource");  gst_app_src_push_buffer (GST_APP_SRC (source), gst_buffer_ref (buffer));  gst_sample_unref (sample);  g_object_unref (source);  return GST_FLOW_OK;}

GstFlowReturn MediaImpl::gstNewSampleCallback(GstElement*, MediaImpl *p){    // Make it thread-safe.    p->lockMutex();    // Get next frame.    GstSample *sample = gst_app_sink_pull_sample(GST_APP_SINK(p->_appsink0));    // Unref last frame.    p->_freeCurrentSample();    // Set current frame.    p->_currentFrameSample = sample;    // For live sources, video dimensions have not been set, because    // gstPadAddedCallback is never called. Fix dimensions from first sample /    // caps we receive.    if (p->_isSharedMemorySource && ( p->_padHandlerData.width == -1 ||                                      p->_padHandlerData.height == -1)) {        GstCaps *caps = gst_sample_get_caps(sample);        GstStructure *structure;        structure = gst_caps_get_structure(caps, 0);        gst_structure_get_int(structure, "width",  &p->_padHandlerData.width);        gst_structure_get_int(structure, "height", &p->_padHandlerData.height);        // g_print("Size is %u x %u/n", _padHandlerData.width, _padHandlerData.height);    }    // Try to retrieve data bits of frame.    GstMapInfo& map = p->_mapInfo;    GstBuffer *buffer = gst_sample_get_buffer( sample );    if (gst_buffer_map(buffer, &map, GST_MAP_READ))    {        p->_currentFrameBuffer = buffer;        // For debugging:        //gst_util_dump_mem(map.data, map.size)        // Retrieve data from map info.        p->_data = map.data;        // Bits have changed.        p->_bitsChanged = true;    }    p->unlockMutex();    return GST_FLOW_OK;}

bool ofGstUtils::loadMovie(string name){	bpp 				= 24;	bLoaded      		= false;	bPaused 			= true;	speed 				= 1.0f;	bHavePixelsChanged 	= false;	if( name.find( "://",0 ) == string::npos){		name 			= "file://"+ofToDataPath(name,true);		bIsStream		= false;	}else{		bIsStream		= true;	}	ofLog(OF_LOG_VERBOSE,"loading "+name);	gstData.loop		= g_main_loop_new (NULL, FALSE);	gstPipeline = gst_element_factory_make("playbin","player");	g_object_set(G_OBJECT(gstPipeline), "uri", name.c_str(), (void*)NULL);	// create the oF appsink for video rgb without sync to clock	gstSink = gst_element_factory_make("appsink", NULL);	GstCaps *caps = gst_caps_new_simple("video/x-raw-rgb",										"bpp", G_TYPE_INT, 24,										//"depth", G_TYPE_INT, 24,										/*"endianness",G_TYPE_INT,4321,										"red_mask",G_TYPE_INT,0xff0000,										"green_mask",G_TYPE_INT,0x00ff00,										"blue_mask",G_TYPE_INT,0x0000ff,*/										NULL);	gst_app_sink_set_caps(GST_APP_SINK(gstSink), caps);	gst_caps_unref(caps);	gst_base_sink_set_sync(GST_BASE_SINK(gstSink), false);	g_object_set (G_OBJECT(gstPipeline),"video-sink",gstSink,(void*)NULL);	GstElement *audioSink = gst_element_factory_make("gconfaudiosink", NULL);	g_object_set (G_OBJECT(gstPipeline),"audio-sink",audioSink,(void*)NULL);	return startPipeline();;}

bool ofGstUtils::startPipeline() {    gstData.pipeline=gstPipeline;    // pause the pipeline    if(gst_element_set_state(GST_ELEMENT(gstPipeline), GST_STATE_PAUSED) ==            GST_STATE_CHANGE_FAILURE) {//		ofLog(OF_LOG_ERROR, "GStreamer: unable to set pipeline to paused/n");        return false;    }    bool ret = false;    if(!bIsStream) {        ofGstDataLock(&gstData);        ret = allocate();        ofGstDataUnlock(&gstData);    } else {        ret = true;    }    if(gstSink) {        // set the appsink to emit signals to get eos and errors        g_object_set (G_OBJECT (gstSink), "emit-signals", FALSE, "sync", !bFrameByFrame, (void*)NULL);        /*g_signal_connect (gstSink, "new-buffer", G_CALLBACK (on_new_buffer_from_source), &gstData);        g_signal_connect (gstSink, "new-preroll", G_CALLBACK (on_new_preroll_from_source), &gstData);*/        if(!bFrameByFrame) {            //		printf("SET CALLBACKS/n");            GstAppSinkCallbacks gstCallbacks;            gstCallbacks.eos = &on_eos_from_source;            gstCallbacks.new_preroll = &on_new_preroll_from_source;            gstCallbacks.new_buffer = &on_new_buffer_from_source;            gst_app_sink_set_callbacks(GST_APP_SINK(gstSink), &gstCallbacks, &gstData, NULL);        }    }    setSpeed(1.0);    return ret;}

bool ofGstUtils::startPipeline(){	bPaused 			= true;	speed 				= 1.0f;	// pause the pipeline	if(gst_element_set_state(GST_ELEMENT(gstPipeline), GST_STATE_PAUSED) ==	GST_STATE_CHANGE_FAILURE) {		ofLog(OF_LOG_ERROR, "GStreamer: unable to set pipeline to paused/n");		return false;	}	// wait for paused state to query the duration	if(!isStream){		GstState state = GST_STATE_PAUSED;		if(gst_element_get_state(gstPipeline,&state,NULL,2*GST_SECOND)==GST_STATE_CHANGE_FAILURE){			return false;		}	}	bLoaded = true;	if(isAppSink){		// set the appsink to not emit signals, we are using callbacks instead		// and frameByFrame to get buffers by polling instead of callback		g_object_set (G_OBJECT (gstSink), "emit-signals", FALSE, "sync", !bFrameByFrame, (void*)NULL);		if(!bFrameByFrame){			GstAppSinkCallbacks gstCallbacks;			gstCallbacks.eos = &on_eos_from_source;			gstCallbacks.new_preroll = &on_new_preroll_from_source;			gstCallbacks.new_buffer = &on_new_buffer_from_source;			gst_app_sink_set_callbacks(GST_APP_SINK(gstSink), &gstCallbacks, this, NULL);		}	}	setSpeed(1.0);	ofAddListener(ofEvents.update,this,&ofGstUtils::update);	return true;}