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" "[login to view URL]" "[login to view URL]" "[login to view URL]" ) func padProbes(mainLoop *[login to view URL]) error { [login to view URL](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 := [login to view URL]( "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 := [login to view URL]("src") if err != nil { return err } // Get the audiotestsrc's src-pad. srcPad := [login to view URL]("src") if srcPad == nil { return [login to view URL]("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. [login to view URL]([login to view URL], func(self *[login to view URL], info *[login to view URL]) [login to view URL] { // Interpret the data sent over the pad as a buffer. We know to expect this because of // the probe mask defined above. buffer := [login to view URL]() // 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: [login to view URL] mapInfo := [login to view URL]([login to view URL]) defer [login to view URL]() // 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 := [login to view URL]() if len(samples) == 0 { return [login to view URL] } // For each buffer (= chunk of samples) calculate the root mean square. var square float64 for _, i := range samples { square += float64(i * i) } rms := [login to view URL](square / float64(len(samples))) [login to view URL]("rms:", rms) return [login to view URL] }) // Start the pipeline [login to view URL]([login to view URL]) // Block on messages coming in from the bus instead of using the main loop for { msg := [login to view URL]().TimedPop([login to view URL]) if msg == nil { break } if err := handleMessage(msg); err != nil { return err } } return nil } func handleMessage(msg *[login to view URL]) error { defer [login to view URL]() switch [login to view URL]() { case [login to view URL]: return [login to view URL]("end-of-stream") case [login to view URL]: return [login to view URL]() } return nil } func main() { [login to view URL](padProbes) }