custom event example

examples/appsink/main.go

@@ -129,5 +129,4 @@ func main() {
 		}
 		return mainLoop(pipeline)
 	})
-
 }

examples/appsrc/main.go

@@ -1,6 +1,13 @@
 package main
 
 import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"math"
+	"time"
+
+	"github.com/tinyzimmer/go-gst/examples"
 	"github.com/tinyzimmer/go-gst/gst"
 	"github.com/tinyzimmer/go-gst/gst/app"
 )
@@ -8,34 +15,116 @@ import (
 func createPipeline() (*gst.Pipeline, error) {
 	gst.Init(nil)
 
+	// Create a pipeline
 	pipeline, err := gst.NewPipeline("")
 	if err != nil {
 		return nil, err
 	}
 
-	// Should this actually be a *gst.Element that produces an Appsrc interface like the
-	// rust examples?
-	src, err := app.NewAppSrc()
+	// Create the elements
+	elems, err := gst.NewElementMany("appsrc", "autoaudiosink")
 	if err != nil {
 		return nil, err
 	}
 
-	elems, err := gst.NewElementMany("videoconvert", "autovideosink")
-	if err != nil {
-		return nil, err
-	}
-
-	// Place the app source at the top of the element list for linking
-	elems = append([]*gst.Element{src.Element}, elems...)
-
+	// Add the elements to the pipeline and link them
 	pipeline.AddMany(elems...)
 	gst.ElementLinkMany(elems...)
 
-	// TODO: need to implement video
+	// Get the app sourrce from the first element returned
+	src := app.SrcFromElement(elems[0])
+
+	// We are instructing downstream elements that we are producing raw signed 16-bit integers.
+	src.SetCaps(gst.NewCapsFromString(
+		"audio/x-raw, format=S16LE, layout=interleaved, channels=1, rate=44100",
+	))
+
+	// Add a callback for whene the sink requests a sample
+	i := 1
+	src.SetCallbacks(&app.SourceCallbacks{
+		NeedDataFunc: func(src *app.Source, _ uint) {
+			// Stop after 10 samples
+			if i == 10 {
+				src.EndStream()
+				return
+			}
+
+			fmt.Println("Producing sample", i)
+
+			sinWave := newSinWave(44100, 440.0, 1.0, time.Second)
+
+			// Allocate a new buffer with the sin wave
+			buffer := gst.NewBufferFromBytes(sinWave)
+
+			// Set the presentation timestamp on thee buffer
+			pts := time.Second * time.Duration(i)
+			buffer.SetPresentationTimestamp(pts)
+			buffer.SetDuration(time.Second)
+
+			// Push tehe buffer onto the src
+			src.PushBuffer(buffer)
+
+			i++
+		},
+	})
 
 	return pipeline, nil
 }
 
-func main() {
-
+func newSinWave(sampleRate int64, freq, vol float64, duration time.Duration) []byte {
+	numSamples := duration.Milliseconds() * (sampleRate / 1000.0)
+	buf := new(bytes.Buffer)
+	for i := int64(0); i < numSamples; i++ {
+		data := vol * math.Sin(2.0*math.Pi*freq*(1/float64(sampleRate)))
+		binary.Write(buf, binary.LittleEndian, data)
+	}
+	return buf.Bytes()
+}
+
+func handleMessage(msg *gst.Message) error {
+	defer msg.Unref() // Messages are a good candidate for trying out runtime finalizers
+
+	switch msg.Type() {
+	case gst.MessageEOS:
+		return app.ErrEOS
+	case gst.MessageError:
+		return msg.ParseError()
+	}
+
+	return nil
+}
+
+func mainLoop(pipeline *gst.Pipeline) error {
+
+	defer pipeline.Destroy() // Will stop and unref the pipeline when this function returns
+
+	// Start the pipeline
+	pipeline.SetState(gst.StatePlaying)
+
+	// Retrieve the bus from the pipeline
+	bus := pipeline.GetPipelineBus()
+
+	// Loop over messsages from the pipeline
+	for {
+		msg := bus.TimedPop(time.Duration(-1))
+		if msg == nil {
+			break
+		}
+		if err := handleMessage(msg); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func main() {
+	examples.Run(func() error {
+		var pipeline *gst.Pipeline
+		var err error
+		if pipeline, err = createPipeline(); err != nil {
+			return err
+		}
+		return mainLoop(pipeline)
+	})
 }

examples/common.go

@@ -21,3 +21,14 @@ func Run(f func() error) {
 
 	mainLoop.Run()
 }
+
+// RunLoop is used to wrap the given function in a main loop and print any error.
+// The main loop itself is passed to the function for more control over exiting.
+func RunLoop(f func(*gst.MainLoop) error) {
+	mainLoop := gst.NewMainLoop(gst.DefaultMainContext(), false)
+	defer mainLoop.Unref()
+
+	if err := f(mainLoop); err != nil {
+		fmt.Println("ERROR!", err)
+	}
+}

examples/custom_events/main.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/tinyzimmer/go-gst/examples"
+	"github.com/tinyzimmer/go-gst/gst"
+)
+
+type ExampleCustomEvent struct {
+	Count   int
+	SendEOS bool
+}
+
+func createPipeline() (*gst.Pipeline, error) {
+	gst.Init(nil)
+
+	// Create a new pipeline from a launch string
+	pipeline, err := gst.NewPipelineFromString(
+		"audiotestsrc name=src ! queue max-size-time=2000000000 ! fakesink name=sink sync=true",
+	)
+
+	// Retrieve the sink element
+	sinks, err := pipeline.GetSinkElements()
+	if err != nil {
+		return nil, err
+	} else if len(sinks) != 1 {
+		return nil, errors.New("Expected one sink back")
+	}
+	sink := sinks[0]
+
+	// Get the sink pad
+	sinkpad := sink.GetStaticPad("sink")
+
+	// Add a probe for out custom event
+	sinkpad.AddProbe(gst.PadProbeTypeEventDownstream, func(p *gst.Pad, info *gst.PadProbeInfo) gst.PadProbeReturn {
+		// Retrieve the event from the probe
+		ev := info.GetEvent()
+
+		// Extra check to make sure it is the right type.
+		if ev.Type() != gst.EventTypeCustomDownstream {
+			return gst.PadProbeUnhandled
+		}
+
+		// Unmarshal the event into our custom one
+		var customEvent ExampleCustomEvent
+		if err := ev.GetStructure().UnmarshalInto(&customEvent); err != nil {
+			fmt.Println("Could not parse the custom event!")
+			return gst.PadProbeUnhandled
+		}
+
+		// Log and act accordingly
+		fmt.Printf("Received custom event with count=%d send_eos=%v\n", customEvent.Count, customEvent.SendEOS)
+		if customEvent.SendEOS {
+			fmt.Println("Send EOS is true, sending eos")
+			if !pipeline.SendEvent(gst.NewEOSEvent()) {
+				fmt.Println("WARNING: Failed to send EOS to pipeline")
+			}
+		} else {
+			fmt.Println("Send EOS is false ignoring")
+		}
+		return gst.PadProbeOK
+	})
+
+	return pipeline, nil
+}
+
+func mainLoop(loop *gst.MainLoop, pipeline *gst.Pipeline) error {
+
+	// Start the pipeline
+	pipeline.SetState(gst.StatePlaying)
+
+	// Create a watch on the pipeline to kill the main loop when EOS is received
+	pipeline.GetPipelineBus().AddWatch(func(msg *gst.Message) bool {
+		switch msg.Type() {
+		case gst.MessageEOS:
+			pipeline.Destroy()
+			loop.Quit()
+		}
+		return true
+	})
+
+	// Loop and on the third iteration send the custom event.
+	ticker := time.NewTicker(time.Second * 2)
+	count := 0
+	for range ticker.C {
+		ev := ExampleCustomEvent{Count: count}
+		if count == 3 {
+			ev.SendEOS = true
+		}
+		st := gst.MarshalStructure(ev)
+		if !pipeline.SendEvent(gst.NewCustomEvent(gst.EventTypeCustomDownstream, st)) {
+			fmt.Println("Warning: failed to send custom event")
+		}
+		if count == 3 {
+			break
+		}
+		count++
+	}
+
+	return loop.RunError()
+}
+
+func main() {
+	examples.RunLoop(func(loop *gst.MainLoop) error {
+		var pipeline *gst.Pipeline
+		var err error
+		if pipeline, err = createPipeline(); err != nil {
+			return err
+		}
+		return mainLoop(loop, pipeline)
+	})
+}