Skip to content

GStreamer Pad Probe

A GStreamer pad probe is a callback attached to a pad. It lets your application observe or control data as it passes through a pipeline.

Pads are the connection points between elements:

1
source element src pad -> sink pad next element

When you add a probe to a pad, GStreamer calls your function when matching events, buffers, queries, or blocking conditions pass through that pad.

Common uses

Pad probes are useful when you need to:

  • inspect buffers without changing the pipeline structure
  • read metadata, timestamps, flags, or encoded bytes
  • drop selected buffers
  • modify buffers before they reach the next element
  • debug caps, events, or pipeline flow
  • block a pad briefly while relinking pipeline elements

Simple Python Example

This example prints the PTS and size of every buffer passing through an identity element:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
#!/usr/bin/env python3

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

Gst.init(None)


def buffer_probe(pad, info, user_data):
    buffer = info.get_buffer()
    if buffer is None:
        return Gst.PadProbeReturn.OK

    print(f"buffer pts={buffer.pts}, size={buffer.get_size()}")
    return Gst.PadProbeReturn.OK


pipeline = Gst.parse_launch(
    "videotestsrc is-live=true ! "
    "identity name=tap ! "
    "videoconvert ! "
    "autovideosink"
)

tap = pipeline.get_by_name("tap")
pad = tap.get_static_pad("sink")
pad.add_probe(Gst.PadProbeType.BUFFER, buffer_probe, None)

loop = GLib.MainLoop()
bus = pipeline.get_bus()
bus.add_signal_watch()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


bus.connect("message", on_message, loop)
pipeline.set_state(Gst.State.PLAYING)

try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    pipeline.set_state(Gst.State.NULL)

Return Values

The probe callback return value controls what happens next:

  • Gst.PadProbeReturn.OK: let the buffer/event continue
  • Gst.PadProbeReturn.DROP: drop this buffer/event
  • Gst.PadProbeReturn.REMOVE: remove this probe
  • Gst.PadProbeReturn.HANDLED: mark the item as handled

Most inspection probes should return OK.

Performance Notes

A pad probe runs on the streaming thread. Keep the callback short.

Avoid slow work inside the probe, such as disk I/O, network calls, heavy JSON parsing, or long locks. If the probe takes too long, it can slow the whole pipeline, increase latency, or cause dropped frames. For expensive processing, copy the small data you need and send it to another thread or queue.

Probe Types

pad.add_probe() receives a Gst.PadProbeType mask. Combine flags with | when a callback should handle more than one kind of item.

  • Gst.PadProbeType.BUFFER: called for each Gst.Buffer. Use this for frame or packet inspection, timestamps, metadata, encoded payload parsing, or selectively dropping buffers.
  • Gst.PadProbeType.BUFFER_LIST: called for Gst.BufferList batches. Use this when an element pushes grouped buffers and you need to inspect or drop the whole batch.
  • Gst.PadProbeType.EVENT_DOWNSTREAM: called for downstream events such as caps, segment, tag, EOS, or flush-stop. Use this to debug negotiation and stream-control events moving from source to sink.
  • Gst.PadProbeType.EVENT_UPSTREAM: called for upstream events such as seek, latency, navigation, or QoS. Use this to observe control requests moving from sink to source.
  • Gst.PadProbeType.EVENT_BOTH: shorthand for upstream and downstream events. Use this when the same callback should inspect all pad events.
  • Gst.PadProbeType.QUERY_DOWNSTREAM: called for downstream queries. Use this to inspect queries sent toward sinks, such as allocation or scheduling.
  • Gst.PadProbeType.QUERY_UPSTREAM: called for upstream queries. Use this to inspect queries sent toward sources, such as position, duration, latency, or caps queries.
  • Gst.PadProbeType.QUERY_BOTH: shorthand for upstream and downstream queries. Use this when the same callback should inspect all pad queries.
  • Gst.PadProbeType.BLOCK: blocks the pad before matching data continues. Use this for short, controlled pipeline changes such as relinking or replacing an element.
  • Gst.PadProbeType.IDLE: called when the pad is idle and keeps it blocked until the probe is removed. Use this when you need a stable point for dynamic pipeline edits.

Block and Idle

flowchart LR
    A["videotestsrc<br/>is-live=true<br/>pattern=smpte"] --> B["videoconvert"]
    B --> C["capsfilter<br/>video/x-raw,format=RGB"]
    C --> D["queue<br/>upstream_queue"]

    D --> Psrc pad probes<br/>BLOCK | BUFFER once<br/>IDLE every 5s

    P --> E["identity<br/>normal_filter"]
    P -. switch by IDLE probe .-> F["videobalance<br/>effect_filter<br/>saturation=0.0<br/>brightness=-0.15<br/>contrast=1.8"]

    E --> G["queue<br/>downstream_queue"]
    F -. active after switch .-> G

    G --> H["videoconvert"]
    H --> I["autovideosink"]

prob_block_idle_switch.py starts with the normal path linked as:

1
upstream_queue -> normal_filter -> downstream_queue

The switch point is the src pad of upstream_queue. After two seconds the example installs a BLOCK | BUFFER probe on that pad. It fires on the next buffer, prints the buffer PTS, and returns REMOVE, so this first probe only demonstrates where a buffer-blocking probe pauses data flow.

Every five seconds the example installs an IDLE probe on the same pad. An idle probe runs when the pad has no data passing through it and keeps the pad blocked while the callback runs, which gives the code a stable point to change links. The callback chooses the current filter and the next filter from the using_effect flag, unlinks the old path, sets the old filter to NULL, links upstream_queue through the new filter to downstream_queue, syncs the new filter state with the pipeline, flips using_effect, and returns REMOVE to unblock the pad.

The result is a live pipeline that alternates between the identity path and the videobalance grayscale effect without stopping the whole pipeline.

Demo 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
#!/usr/bin/env python3

import gi

gi.require_version("Gst", "1.0")

from gi.repository import GLib, Gst

Gst.init(None)


class ProbeSwitchExample:
    def __init__(self):
        self.pipeline = Gst.Pipeline.new("probe-switch-example")
        self.using_effect = False

        self.source = Gst.ElementFactory.make("videotestsrc", "source")
        self.pre_switch_converter = Gst.ElementFactory.make(
            "videoconvert",
            "pre_switch_converter",
        )
        self.pre_switch_caps = Gst.ElementFactory.make("capsfilter", "pre_switch_caps")
        self.upstream_queue = Gst.ElementFactory.make("queue", "upstream_queue")
        self.normal_filter = Gst.ElementFactory.make("identity", "normal_filter")
        self.effect_filter = Gst.ElementFactory.make("videobalance", "effect_filter")
        self.downstream_queue = Gst.ElementFactory.make("queue", "downstream_queue")
        self.converter = Gst.ElementFactory.make("videoconvert", "converter")
        self.sink = Gst.ElementFactory.make("autovideosink", "sink")

        self.elements = (
            self.source,
            self.pre_switch_converter,
            self.pre_switch_caps,
            self.upstream_queue,
            self.normal_filter,
            self.effect_filter,
            self.downstream_queue,
            self.converter,
            self.sink,
        )
        if any(element is None for element in self.elements):
            raise RuntimeError("failed to create one or more GStreamer elements")

        self.source.set_property("is-live", True)
        self.source.set_property("pattern", "smpte")
        self.pre_switch_caps.set_property(
            "caps",
            Gst.Caps.from_string("video/x-raw,format=RGB"),
        )
        self.effect_filter.set_property("brightness", -0.15)
        self.effect_filter.set_property("contrast", 1.8)
        self.effect_filter.set_property("saturation", 0.0)

        for element in self.elements:
            self.pipeline.add(element)

        if not self.source.link(self.pre_switch_converter):
            raise RuntimeError("failed to link source to pre-switch converter")
        if not self.pre_switch_converter.link(self.pre_switch_caps):
            raise RuntimeError("failed to link pre-switch converter to capsfilter")
        if not self.pre_switch_caps.link(self.upstream_queue):
            raise RuntimeError("failed to link capsfilter to upstream queue")
        if not self.upstream_queue.link(self.normal_filter):
            raise RuntimeError("failed to link upstream queue to normal filter")
        if not self.normal_filter.link(self.downstream_queue):
            raise RuntimeError("failed to link normal filter to downstream queue")
        if not self.downstream_queue.link(self.converter):
            raise RuntimeError("failed to link downstream queue to converter")
        if not self.converter.link(self.sink):
            raise RuntimeError("failed to link converter to sink")

        self.switch_pad = self.upstream_queue.get_static_pad("src")
        if self.switch_pad is None:
            raise RuntimeError("failed to get upstream queue src pad")

    def current_filter(self):
        return self.effect_filter if self.using_effect else self.normal_filter

    def next_filter(self):
        return self.normal_filter if self.using_effect else self.effect_filter

    def install_block_probe(self):
        print("install BLOCK | BUFFER probe")
        self.switch_pad.add_probe(
            Gst.PadProbeType.BLOCK | Gst.PadProbeType.BUFFER,
            self.block_buffer_probe,
            None,
        )
        return GLib.SOURCE_REMOVE

    def block_buffer_probe(self, pad, info, user_data):
        # BLOCK holds this pad at the next buffer. REMOVE unblocks it immediately.
        buffer = info.get_buffer()
        pts = buffer.pts if buffer is not None else Gst.CLOCK_TIME_NONE
        print(f"BLOCK | BUFFER probe fired, stream paused at buffer pts={pts}")
        return Gst.PadProbeReturn.REMOVE

    def install_idle_probe(self):
        print("install IDLE probe for pipeline switch")
        self.switch_pad.add_probe(Gst.PadProbeType.IDLE, self.idle_switch_probe, None)
        return GLib.SOURCE_CONTINUE

    def idle_switch_probe(self, pad, info, user_data):
        # IDLE runs while the pad is blocked, giving a stable point to relink.
        old_filter = self.current_filter()
        new_filter = self.next_filter()
        new_name = "grayscale effect" if not self.using_effect else "normal identity"

        print(f"IDLE probe fired, switching to {new_name}")

        old_filter.unlink(self.downstream_queue)
        self.upstream_queue.unlink(old_filter)
        old_filter.set_state(Gst.State.NULL)

        if not self.upstream_queue.link(new_filter):
            raise RuntimeError(f"failed to link upstream queue to {new_filter.name}")
        if not new_filter.link(self.downstream_queue):
            raise RuntimeError(f"failed to link {new_filter.name} to downstream queue")

        new_filter.sync_state_with_parent()
        self.using_effect = not self.using_effect
        return Gst.PadProbeReturn.REMOVE


example = ProbeSwitchExample()
loop = GLib.MainLoop()
bus = example.pipeline.get_bus()
bus.add_signal_watch()


def on_message(bus, message, loop):
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        print("ERROR:", err)
        print("DEBUG:", debug)
        loop.quit()
    elif message.type == Gst.MessageType.EOS:
        loop.quit()


bus.connect("message", on_message, loop)

GLib.timeout_add_seconds(2, example.install_block_probe)
GLib.timeout_add_seconds(5, example.install_idle_probe)

example.pipeline.set_state(Gst.State.PLAYING)

try:
    loop.run()
except KeyboardInterrupt:
    pass
finally:
    example.pipeline.set_state(Gst.State.NULL)