AppSink向AppSrc推送数据：单/双管道选择及数据处理疑问

Great question! Let’s break this down clearly—GStreamer’s app elements can feel a bit counterintuitive when wiring them together, but you’ve got the right pieces with AppSink and AppSrc.

First off: you absolutely can implement this in a single pipeline—it just depends on the complexity of your data processing:

• If your processing is lightweight and synchronous, a single pipeline is more efficient. You avoid the overhead of passing data between separate pipeline contexts (like cross-thread or inter-process communication).
• If your processing is slow, asynchronous, or needs to interact with other independent GStreamer workflows, two pipelines will give you more flexibility. This prevents long-running processing from blocking the entire pipeline’s data flow.

Short answer: You don’t need to. AppSink is designed to handle this for you. When you use gst_app_sink_pull_sample() or set up a new-sample callback, AppSink already wraps the incoming data into a GstSample (which contains a GstBuffer with your media data).

That said, if your processing requires custom buffer handling (like reallocating memory, converting formats, or modifying raw data), you can absolutely create new buffers manually. Here’s a quick C example of how that might look:

// Pull a sample from AppSink
GstSample *in_sample = gst_app_sink_pull_sample(GST_APP_SINK(app_sink));
GstBuffer *in_buf = gst_sample_get_buffer(in_sample);

// Create a new buffer (copy data from the original, or allocate fresh)
GstBuffer *out_buf = gst_buffer_copy(in_buf);
// --- Your data processing logic goes here ---
// Example: Modify buffer metadata or raw pixel/audio data

// Wrap the processed buffer into a new sample
GstSample *out_sample = gst_sample_new(out_buf, gst_sample_get_caps(in_sample), NULL, NULL);

// Push the sample to AppSrc
gst_app_src_push_sample(GST_APP_SRC(app_src), out_sample);

// Clean up references
gst_sample_unref(in_sample);
gst_sample_unref(out_sample);

• Thread Safety: If you’re using a single pipeline, avoid long-running blocking operations in the AppSink callback. The callback runs in the pipeline’s stream thread, so blocking it will slow down the entire pipeline. For heavy processing, offload the work to a separate thread (use a queue to pass data between the callback and worker thread).
• Caps Negotiation: Make sure the caps (media format, resolution, etc.) coming out of AppSink match what AppSrc expects. You can set explicit caps on AppSrc with gst_app_src_set_caps(), or use a capsfilter element to enforce compatibility. Mismatched caps will break your pipeline.
• Flow Control: AppSrc has built-in flow control. If you’re using push mode (push_sample/push_buffer), don’t flood it with data faster than the downstream elements can process—this causes buffer overflow. For better control, enable emit-signals on AppSrc and use the need-data signal to only push data when downstream is ready.

Here’s a simplified snippet to illustrate the workflow with GStreamer’s Python bindings:

import gi
gi.require_version('Gst', '1.0')
from gi.repository import Gst, GLib

Gst.init(None)

# Create pipeline and elements
pipeline = Gst.Pipeline.new("app-data-pipeline")
src = Gst.ElementFactory.make("videotestsrc", "test-src")
app_sink = Gst.ElementFactory.make("appsink", "data-sink")
app_sink.set_property("emit-signals", True)
app_src = Gst.ElementFactory.make("appsrc", "data-src")
app_src.set_property("caps", Gst.Caps.from_string("video/x-raw,format=I420,width=640,height=480"))
video_sink = Gst.ElementFactory.make("autovideosink", "video-output")

# Add elements to pipeline and link
pipeline.add(src, app_sink, app_src, video_sink)
src.link(app_sink)
app_src.link(video_sink)

# Define AppSink callback to process and push data
def handle_new_sample(sink):
    in_sample = sink.pull_sample()
    buf = in_sample.get_buffer()
    
    # --- Your data processing here ---
    # For example: Adjust brightness, crop frames, etc.
    
    # Push processed data to AppSrc
    out_sample = Gst.Sample.new(buf, in_sample.get_caps(), None, None)
    app_src.push_sample(out_sample)
    
    # Clean up
    in_sample.unref()
    out_sample.unref()
    return Gst.FlowReturn.OK

app_sink.connect("new-sample", handle_new_sample)

# Start pipeline and run main loop
pipeline.set_state(Gst.State.PLAYING)
loop = GLib.MainLoop()
loop.run()

内容的提问来源于stack exchange，提问作者Rhythm Chopra