Added messages wrapper and bus message pop

go.mod

@@ -2,4 +2,7 @@ module github.com/vinijabes/gostreamer
 
 go 1.13
 
-require github.com/notedit/gstreamer-go v0.3.1 // indirect
+require (
+ github.com/labstack/gommon v0.3.0 // indirect
+ github.com/notedit/gstreamer-go v0.3.1 // indirect
+)

go.sum

@@ -1,2 +1,21 @@
+github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/labstack/gommon v0.3.0 h1:JEeO0bvc78PKdyHxloTKiF8BD5iGrH8T6MSeGvSgob0=
+github.com/labstack/gommon v0.3.0/go.mod h1:MULnywXg0yavhxWKc+lOruYdAhDwPK9wf0OL7NoOu+k=
+github.com/mattn/go-colorable v0.1.2/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
+github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
+github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
 github.com/notedit/gstreamer-go v0.3.1 h1:gkHgNY4Qswv5hZxq5GbZ1lPGUX26BoMAqWVaoIX7ohE=
 github.com/notedit/gstreamer-go v0.3.1/go.mod h1:CGCLZR47aRzzEmup8uQcplfOdLqEtJjFihiu+A2aIZw=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
+github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
+golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

include/element.h

@@ -3,4 +3,6 @@
 
 #include "pch.h"
 
+GstFlowReturn gostreamer_element_push_buffer(GstElement *element, void *buffer,int len);
+
 #endif

include/message.h

@@ -0,0 +1,8 @@
+#ifndef MESSAGE_H
+#define MESSAGE_H
+
+#include "pch.h"
+
+GstMessageType gostreamer_get_message_type(GstMessage *message);
+
+#endif

include/object.h

@@ -7,6 +7,7 @@ void gostreamer_object_set_string(GstObject* object, const char* first_property_
 void gostreamer_object_set_int(GstObject* object, const char* first_property_name, gint arg);
 void gostreamer_object_set_uint(GstObject* object, const char* first_property_name, guint arg);
 void gostreamer_object_set_bool(GstObject* object, const char* first_property_name, gboolean arg);
+void gostreamer_object_set_caps(GstObject* object, const char* first_property_name, const GstCaps *arg);
 
 gchar* gostreamer_object_get_string(GstObject* object, const char* first_property_name);
 gint gostreamer_object_get_int(GstObject* object, const char* first_property_name);

pkg/gstreamer/bus.go

@@ -16,6 +16,8 @@ type Bus interface {
  Object
 
  HavePending() bool
+ Pop() (Message, error)
+
  GetBusPointer() *C.GstBus
 }
 
@@ -24,7 +26,8 @@ type bus struct {
 }
 
 var (
- ErrFailedToCreateBus = errors.New("failed to create bus")
+ ErrFailedToCreateBus = errors.New("failed to create bus")
+ ErrFailedToCreateMessage = errors.New("failed to create message")
 )
 
 func NewBus() (Bus, error) {
@@ -51,6 +54,11 @@ func (b *bus) HavePending() bool {
  return C.gst_bus_have_pending(b.GetBusPointer()) != 0
 }
 
+func (b *bus) Pop() (Message, error) {
+ cmessage := C.gst_bus_pop(b.GetBusPointer())
+ return newMessageFromPointer(cmessage)
+}
+
 func (b *bus) GetBusPointer() *C.GstBus {
  return (*C.GstBus)(unsafe.Pointer(b.GstObject))
 }

pkg/gstreamer/caps.go

@@ -22,47 +22,36 @@ type caps struct {
 
 func NewCapsAny() (Caps, error) {
  ccaps := C.gst_caps_new_any()
- if ccaps == nil {
- return nil, fmt.Errorf("failed to create Caps")
- }
-
- caps := &caps{}
- caps.GstCaps = ccaps
-
- runtime.SetFinalizer(caps, func(c Caps) {
- c.Unref()
- })
-
- return caps, nil
+ return newCapsFromPointer(ccaps)
 }
 
 func NewCapsEmpty() (Caps, error) {
  ccaps := C.gst_caps_new_empty()
- if ccaps == nil {
- return nil, fmt.Errorf("failed to create Caps")
- }
-
- caps := &caps{}
- caps.GstCaps = ccaps
-
- runtime.SetFinalizer(caps, func(c Caps) {
- c.Unref()
- })
-
- return caps, nil
+ return newCapsFromPointer(ccaps)
 }
 
 func NewCapsEmptySimple(mediaType string) (Caps, error) {
  cmediaType := C.CString(mediaType)
  defer C.free(unsafe.Pointer(cmediaType))
 
  ccaps := C.gst_caps_new_empty_simple(cmediaType)
- if ccaps == nil {
+ return newCapsFromPointer(ccaps)
+}
+
+func NewCapsFromString(caps string) (Caps, error) {
+ c := (*C.gchar)(unsafe.Pointer(C.CString(caps)))
+ defer C.g_free(C.gpointer(unsafe.Pointer(c)))
+ ccaps := C.gst_caps_from_string(c)
+ return newCapsFromPointer(ccaps)
+}
+
+func newCapsFromPointer(pointer *C.GstCaps) (Caps, error) {
+ if pointer == nil {
  return nil, fmt.Errorf("failed to create Caps")
  }
 
  caps := &caps{}
- caps.GstCaps = ccaps
+ caps.GstCaps = pointer
 
  runtime.SetFinalizer(caps, func(c Caps) {
  c.Unref()

pkg/gstreamer/element.c

@@ -0,0 +1,8 @@
+#include "element.h"
+
+GstFlowReturn gostreamer_element_push_buffer(GstElement *element, void *buffer, int len) {
+ gpointer p = g_memdup(buffer, len);
+ GstBuffer *data = gst_buffer_new_wrapped(p, len);
+
+ return gst_app_src_push_buffer(GST_APP_SRC(element), data);
+}

pkg/gstreamer/element.go

@@ -7,6 +7,7 @@ package gstreamer
 */
 import "C"
 import (
+ "errors"
  "fmt"
  "runtime"
  "unsafe"
@@ -26,6 +27,8 @@ type Element interface {
 
  GetBus() (Bus, error)
 
+ Push(buffer []byte) error
+
  GetElementPointer() *C.GstElement
 }
 
@@ -165,6 +168,19 @@ func (e *element) GetElementPointer() *C.GstElement {
  return (*C.GstElement)(unsafe.Pointer(e.GstObject))
 }
 
+func (e *element) Push(buffer []byte) (err error) {
+ b := C.CBytes(buffer)
+ defer C.free(unsafe.Pointer(b))
+ gstReturn := C.gostreamer_element_push_buffer(e.GetElementPointer(), b, C.int(len(buffer)))
+
+ if gstReturn != C.GST_FLOW_OK {
+ err = errors.New("could not push buffer on appsrc element")
+ return
+ }
+
+ return
+}
+
 func (ef *elementFactory) Create(name string) Element {
  elementNameStrUnsafe := C.CString(name)
  defer C.free(unsafe.Pointer(elementNameStrUnsafe))

pkg/gstreamer/gstreamer.go

@@ -9,7 +9,6 @@ import "C"
 import (
  "log"
  "sync"
- "unsafe"
 )
 
 //export goPrint
@@ -60,24 +59,10 @@ type Structure struct {
 // bus *C.GstBus
 // }
 
-//Message ...
-type Message struct {
- message *C.GstMessage
-}
-type MessageType int
-
-//MessageType constants
-const (
- MessageUnknown MessageType = C.GST_MESSAGE_UNKNOWN
- MessageEOS MessageType = C.GST_MESSAGE_EOS
- MessageError MessageType = C.GST_MESSAGE_ERROR
- MessageWarning MessageType = C.GST_MESSAGE_WARNING
- MessageInfo MessageType = C.GST_MESSAGE_INFO
- MessageTag MessageType = C.GST_MESSAGE_TAG
- MessageBuffering MessageType = C.GST_MESSAGE_BUFFERING
- MessageStateChanged MessageType = C.GST_MESSAGE_STATE_CHANGED
- MessageAny MessageType = C.GST_MESSAGE_ANY
-)
+// //Message ...
+// type Message struct {
+// message *C.GstMessage
+// }
 
 var pipelines = make(map[int]*Pipeline)
 var elements = make(map[int]*Element)
@@ -146,23 +131,23 @@ func btoi(b bool) int {
 // return element, nil
 // }
 
-//GetType ...
-func (m *Message) GetType() MessageType {
- c := C.toGstMessageType(unsafe.Pointer(m.message))
- return MessageType(c)
-}
+// //GetType ...
+// func (m *Message) GetType() MessageType {
+//  c := C.toGstMessageType(unsafe.Pointer(m.message))
+//  return MessageType(c)
+// }
 
-//GetTimestamp ...
-func (m *Message) GetTimestamp() uint64 {
- c := C.messageTimestamp(unsafe.Pointer(m.message))
- return uint64(c)
-}
+// //GetTimestamp ...
+// func (m *Message) GetTimestamp() uint64 {
+//  c := C.messageTimestamp(unsafe.Pointer(m.message))
+//  return uint64(c)
+// }
 
-//GetTypeName ...
-func (m *Message) GetTypeName() string {
- c := C.messageTypeName(unsafe.Pointer(m.message))
- return C.GoString(c)
-}
+// //GetTypeName ...
+// func (m *Message) GetTypeName() string {
+//  c := C.messageTypeName(unsafe.Pointer(m.message))
+//  return C.GoString(c)
+// }
 
 // //Start ...
 // func (p *Pipeline) Start() {

pkg/gstreamer/message.c

@@ -0,0 +1,5 @@
+#include "message.h"
+
+GstMessageType gostreamer_get_message_type(GstMessage *message){
+ return GST_MESSAGE_TYPE(message);
+}

pkg/gstreamer/message.go

@@ -0,0 +1,73 @@
+package gstreamer
+
+/*
+#cgo CFLAGS: -I ../../include
+#cgo pkg-config: gstreamer-1.0
+#include "../../include/message.h"
+*/
+import "C"
+import (
+ "runtime"
+)
+
+type Message interface {
+ GetType() MessageType
+ GetName() string
+
+ Unref()
+ GetMessagePointer() *C.GstMessage
+}
+
+type message struct {
+ GstMessage *C.GstMessage
+}
+
+type MessageType int
+
+//MessageType constants
+const (
+ MessageUnknown MessageType = C.GST_MESSAGE_UNKNOWN
+ MessageEOS MessageType = C.GST_MESSAGE_EOS
+ MessageError MessageType = C.GST_MESSAGE_ERROR
+ MessageWarning MessageType = C.GST_MESSAGE_WARNING
+ MessageInfo MessageType = C.GST_MESSAGE_INFO
+ MessageTag MessageType = C.GST_MESSAGE_TAG
+ MessageBuffering MessageType = C.GST_MESSAGE_BUFFERING
+ MessageStateChanged MessageType = C.GST_MESSAGE_STATE_CHANGED
+ MessageAny MessageType = C.GST_MESSAGE_ANY
+)
+
+func newMessageFromPointer(pointer *C.GstMessage) (Message, error) {
+ if pointer == nil {
+ return nil, ErrFailedToCreateBus
+ }
+
+ message := &message{}
+ message.GstMessage = pointer
+
+ runtime.SetFinalizer(message, func(m Message) {
+ m.Unref()
+ })
+
+ return message, nil
+}
+
+func (m *message) GetType() MessageType {
+ return MessageType(C.gostreamer_get_message_type(m.GetMessagePointer()))
+}
+
+func (m *message) GetName() string {
+ messageType := m.GetType()
+ cname := C.gst_message_type_get_name(C.GstMessageType(messageType))
+ name := C.GoString(cname)
+
+ return name
+}
+
+func (m *message) Unref() {
+ C.gst_message_unref(m.GetMessagePointer())
+}
+
+func (m *message) GetMessagePointer() *C.GstMessage {
+ return m.GstMessage
+}

pkg/gstreamer/object.c

@@ -17,6 +17,10 @@ void gostreamer_object_set_bool(GstObject* object, const char* first_property_na
  g_object_set(object, first_property_name, arg, NULL);
 }
 
+void gostreamer_object_set_caps(GstObject* object, const char* first_property_name,const GstCaps *arg){
+ g_object_set(object, first_property_name, arg, NULL);
+}
+
 gchar* gostreamer_object_get_string(GstObject* object, const char* first_property_name) {
  gchar *data;
  g_object_get (object, first_property_name, &data, NULL);

pkg/gstreamer/object.go

@@ -31,6 +31,10 @@ type object struct {
 }
 
 func (o *object) GetParent() Object {
+ if o.GstObject == nil {
+ return nil
+ }
+
  parent := C.gst_object_get_parent(o.GstObject)
  if parent != nil {
  newObj := &object{
@@ -89,6 +93,9 @@ func (o *object) Set(name string, value interface{}) {
  cvalue = 0
  }
  C.gostreamer_object_set_bool(o.GstObject, cname, C.gboolean(cvalue))
+ case Caps:
+ caps := value.(Caps)
+ C.gostreamer_object_set_caps(o.GstObject, cname, caps.GetCapsPointer())
  }
 }