gstreamer + golang pipeline

Purpose
The purpose of this implementation is to provide a means to connect to an IP camera over RTSP and ingest the video+audio streams using gstreamer with golang and the go-gst bindings.

Once the stream is ingested we need to break-out in to children pipelines to perform various tasks such as creating HLS segments and persisting to disk for future playback and transcoding H.265 to H.264 so that the stream is compatible with the WebRTC requirement of H.264 support.

Functionality
The functionality demonstrated in the above diagram is broken out to:

Ingest Pipeline
Responsible for connecting to the IP camera over the RTSP protocol using the rtspsrc element. Once connected the pipeline needs to prepare the stream for consumption by lower pipelines as below.

Segment Pipeline
Take the input stream and output HLS segment files which will be persisted to disk over NFS. It is important to note that we need a hook so that when the files are written we can send a notification message over the message broker announcing the segment file size, name, etc.

Transcode Pipeline
This child pipeline, when activated, is responsible for transcoding the input H.265 stream to H.264 to prepare it to be consumed over WebRTC.

Requirements
Solution must be implemented using golang.

Solution must use the go-gst library for accessing gstreamer bindings.

Code must be documented.

Pipeline must be construct by code. Do NOT expect to take a pipeline string and parse it as we will need low level access to each element!

---

Similar Example
// This example demonstrates the use of GStreamer's pad probe APIs.
//
// Probes are callbacks that can be installed by the application and will notify
// the application about the states of the dataflow. Those are mostly used for
// changing pipelines dynamically at runtime or for inspecting/modifying buffers or events
//
// |-[probe]
// /
// {audiotestsrc} - {fakesink}
package main

import (
"errors"
"fmt"
"math"

"github.com/go-gst/go-glib/glib"
"github.com/go-gst/go-gst/examples"
"github.com/go-gst/go-gst/gst"
)

func padProbes(mainLoop *glib.MainLoop) error {
gst.Init(nil)

// Parse the pipeline we want to probe from a static in-line string.
// Here we give our audiotestsrc a name, so we can retrieve that element
// from the resulting pipeline.
pipeline, err := gst.NewPipelineFromString(
"audiotestsrc name=src ! audio/x-raw,format=S16LE,channels=1 ! fakesink",
)

if err != nil {
return err
}

// Get the audiotestsrc element from the pipeline that GStreamer
// created for us while parsing the launch syntax above.
//
// TODO: There are some discrepancies still between methods that check the nil
// value and return an error, versus those that will instead just return nil.
// Need to settle on one way or the other.
src, err := pipeline.GetElementByName("src")
if err != nil {
return err
}
// Get the audiotestsrc's src-pad.
srcPad := src.GetStaticPad("src")
if srcPad == nil {
return errors.New("src pad on src element was nil")
}

// Add a probe handler on the audiotestsrc's src-pad.
// This handler gets called for every buffer that passes the pad we probe.
srcPad.AddProbe(gst.PadProbeTypeBuffer, func(self *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn {
// Interpret the data sent over the pad as a buffer. We know to expect this because of
// the probe mask defined above.
buffer := info.GetBuffer()

// At this point, buffer is only a reference to an existing memory region somewhere.
// When we want to access its content, we have to map it while requesting the required
// mode of access (read, read/write).
// This type of abstraction is necessary, because the buffer in question might not be
// on the machine's main memory itself, but rather in the GPU's memory.
// So mapping the buffer makes the underlying memory region accessible to us.
// See: https://gstreamer.freedesktop.org/documentation/plugin-development/advanced/allocation.html
mapInfo := buffer.Map(gst.MapRead)
defer buffer.Unmap()

// We know what format the data in the memory region has, since we requested
// it by setting the fakesink's caps. So what we do here is interpret the
// memory region we mapped as an array of signed 16 bit integers.
samples := mapInfo.AsInt16LESlice()
if len(samples) == 0 {
return gst.PadProbeOK
}

// For each buffer (= chunk of samples) calculate the root mean square.
var square float64
for _, i := range samples {
square += float64(i * i)
}
rms := math.Sqrt(square / float64(len(samples)))
fmt.Println("rms:", rms)

return gst.PadProbeOK
})

// Start the pipeline
pipeline.SetState(gst.StatePlaying)

// Block on messages coming in from the bus instead of using the main loop
for {
msg := pipeline.GetPipelineBus().TimedPop(gst.ClockTimeNone)
if msg == nil {
break
}
if err := handleMessage(msg); err != nil {
return err
}
}

return nil
}

func handleMessage(msg *gst.Message) error {
defer msg.Unref()
switch msg.Type() {
case gst.MessageEOS:
return errors.New("end-of-stream")
case gst.MessageError:
return msg.ParseError()
}
return nil
}

func main() {
examples.RunLoop(padProbes)
}