Skip to content

Gst.Event

Gst.Event is a control message that travels through the pipeline pads. Buffers carry media data; events carry pipeline control information such as EOS, CAPS, SEGMENT, FLUSH_START, or application-defined commands.

The main idea is direction:

Event direction Flow Common use
Downstream source to sink caps, segment, eos, custom app command
Upstream sink to source seek, navigation, latency query result path

In a Python plugin, a simple way to listen for downstream events is to inherit from GstBase.BaseTransform and override do_sink_event(). The plugin checks the event type, reads the attached GstStructure, handles the fields it knows, then calls the parent implementation so normal GStreamer events still continue through the element.

Minimal Plugin

This element listens for a custom downstream event named app-control.

plugin 
 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
import gi

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

from gi.repository import Gst, GstBase, GObject

Gst.init(None)


class EventListener(GstBase.BaseTransform):

    __gstmetadata__ = (
        "EventListener",
        "Transform",
        "Listen to custom downstream events",
        "example",
    )

    __gsttemplates__ = (
        Gst.PadTemplate.new(
            "sink",
            Gst.PadDirection.SINK,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.new_any(),
        ),
        Gst.PadTemplate.new(
            "src",
            Gst.PadDirection.SRC,
            Gst.PadPresence.ALWAYS,
            Gst.Caps.new_any(),
        ),
    )

    def do_sink_event(self, event):

        if event.type == Gst.EventType.CUSTOM_DOWNSTREAM:
            structure = event.get_structure()

            if structure and structure.get_name() == "app-control":
                print(
                    "event:",
                    structure.get_value("command"),
                    structure.get_value("value"),
                )

        return GstBase.BaseTransform.do_sink_event(self, event)

    def do_transform_ip(self, buffer):
        return Gst.FlowReturn.OK


GObject.type_register(EventListener)

__gstelementfactory__ = (
    "eventlistener",
    Gst.Rank.NONE,
    EventListener,
)

Minimal Application

The application creates a normal pipeline, finds the plugin by name, builds a custom downstream event, and sends it to the plugin sink pad.

usage 
 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
import gi

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

from gi.repository import Gst

Gst.init(None)

pipeline = Gst.parse_launch(
    "videotestsrc is-live=true ! eventlistener name=listener ! fakesink"
)

listener = pipeline.get_by_name("listener")
sinkpad = listener.get_static_pad("sink")

pipeline.set_state(Gst.State.PLAYING)

Gst.util_usleep(500_000)

structure = Gst.Structure.new_empty("app-control")
structure.set_value("command", "set-threshold")
structure.set_value("value", 42)

event = Gst.Event.new_custom(
    Gst.EventType.CUSTOM_DOWNSTREAM,
    structure,
)

sinkpad.send_event(event)

Gst.util_usleep(500_000)

pipeline.set_state(Gst.State.NULL)

For a quick test, put the plugin file under python/eventlistener.py, then run the application with GST_PLUGIN_PATH pointing at the directory that contains that python/ folder:

1
GST_PLUGIN_PATH=$PWD python3 send_event.py

Expected plugin output:

1
event: set-threshold 42