Implement pubsub for channel events.

app/channels/channel.go

@@ -11,6 +11,7 @@ type Channel struct {
 	metadataLock    sync.RWMutex
 	metadata        map[string]string
 	metadataChannel chan map[string]string
+	Id              string
 	Name            string
 	Description     string
 	MimeType        string

app/channels/channel_manager.go

@@ -3,15 +3,19 @@ package channels
 import (
 	"errors"
 	"sync"
+
+	"github.com/cskr/pubsub"
 )
 
 type ChannelManager struct {
 	channels     map[string]*Channel
 	channelsLock sync.RWMutex
+	pubsub       *pubsub.PubSub
 }
 
 func NewChannelManager() *ChannelManager {
-	return &ChannelManager{
+	mgr := &ChannelManager{
+		pubsub:   pubsub.New(1),
 		channels: map[string]*Channel{},
 	}
 }
@@ -38,6 +42,7 @@ func (manager *ChannelManager) Close(uuid string) (err error) {
 		return
 	}
 
+	pubsub.PubSub.Pub("close", manager.channels[uuid])
 	delete(manager.channels, uuid)
 
 	return
@@ -52,8 +57,14 @@ func (manager *ChannelManager) Open(uuid string) (channel *Channel, err error) {
 		return
 	}
 
-	channel = new(Channel)
+	channel = &Channel{Id: uuid}
 	manager.channels[uuid] = channel
 
+	pubsub.PubSub.Pub("open", channel)
+
 	return
 }
+
+func (manager *ChannelManager) Shutdown() {
+	manager.pubsub.Shutdown()
+}

app/streams/metadata.go

@@ -1,10 +1,10 @@
 package streams
 
 import (
-	"errors"
 	"fmt"
-	"math"
+	"sort"
 	"strings"
+	"unicode"
 )
 
 func quote(text string) string {
@@ -16,7 +16,7 @@ func quote(text string) string {
 
 func unquote(text string) string {
 	if strings.HasPrefix(text, "'") && strings.HasSuffix(text, "'") {
-		text = text[1 : len(text)-2]
+		text = text[1 : len(text)-1]
 		text = strings.Replace(text, "\\'", "'", -1)
 		text = strings.Replace(text, "\\\\", "\\", -1)
 	}
@@ -25,14 +25,21 @@ func unquote(text string) string {
 
 type Metadata map[string]string
 
-func DecodeMetadataFromBytes(b []byte) {
-	// TODO
+func (meta Metadata) sortedKeys() []string {
+	keys := make([]string, len(meta))
+	i := 0
+	for key, _ := range meta {
+		keys[i] = key
+		i++
+	}
+	sort.Strings(keys)
+	return keys
 }
 
-func decodeMetadataItem(text string, metadata *map[string]string) (err error) {
+func decodeMetadataItem(text string, metadata *Metadata) (err error) {
 	parts := strings.SplitN(text, "=", 2)
 	if len(parts) < 2 {
-		err = errors.New("expected key=value but only got key")
+		err = fmt.Errorf("expected key=value but only got key (in %q)", text)
 		return
 	}
 
@@ -45,34 +52,55 @@ func decodeMetadataItem(text string, metadata *map[string]string) (err error) {
 func DecodeMetadata(source string) (meta Metadata, err error) {
 	// name='value'; name='value';name='value';
 	meta = make(Metadata)
-	// TODO
+
+	escape := false
+	quoted := false
+	fieldsFunc := func(c rune) (retval bool) {
+		retval = false
+		if escape {
+			escape = false
+		} else {
+			switch {
+			case unicode.IsSpace(c) && !quoted:
+				retval = true
+				return
+			case c == '\'':
+				quoted = !quoted
+			case c == '\\' && quoted:
+				escape = true
+			case c == ';' && !quoted:
+				retval = true
+				return
+			}
+		}
+		return
+	}
+
+	for _, field := range strings.FieldsFunc(source, fieldsFunc) {
+		if len(field) == 0 {
+			continue
+		}
+		if err = decodeMetadataItem(field, &meta); err != nil {
+			return
+		}
+	}
 
 	return
 }
 
 func (meta Metadata) String() string {
-	return string(meta.Bytes())
-}
-
-func (meta Metadata) Bytes() (buf []byte) {
 	mstr := ""
 
 	if meta != nil {
-		for key, value := range meta {
+		for _, key := range meta.sortedKeys() {
+			value := meta[key]
 			mstr += fmt.Sprintf("%s=%s;", key, quote(value))
 		}
 	}
 
-	if len(mstr) > 16*256-1 {
-		mstr = mstr[0 : 16*256]
+	return mstr
 }
 
-	lengthDiv := int(math.Ceil(float64(len(mstr)) / 16))
-	lengthByte := byte(lengthDiv)
-
-	buf = make([]byte, lengthDiv*16+1)
-	buf[0] = lengthByte
-	copy(buf[1:], []byte(mstr))
-
-	return
+func (meta Metadata) Bytes() (buf []byte) {
+	return []byte(meta.String())
 }

app/streams/metadata_extractor.go

@@ -43,6 +43,20 @@ func (me *MetadataExtractor) close() {
 }
 
 func (mi *MetadataExtractor) Read(data []byte) (n int, err error) {
+	bytesToRead := mi.MetadataInterval - mi.blockOffset
+
+	if bytesToRead <= 0 {
+		// time to prepare for metadata
+		lenBuf := make([]byte, 1)
+		n, err = mi.Reader.Read(lenBuf)
+		if n < 1 {
+			return
+		}
+		length := int(lenBuf[0]) * 16
+		mi.metadataToRead = length
+		mi.metadataBuf = make([]byte, length)
+	}
+
 	if mi.metadataBuf != nil && mi.metadataToRead > 0 {
 		n, err = mi.Reader.Read(mi.metadataBuf[len(mi.metadataBuf)-mi.metadataToRead:])
 		if err != nil {
@@ -51,17 +65,16 @@ func (mi *MetadataExtractor) Read(data []byte) (n int, err error) {
 		mi.metadataToRead -= n
 		if mi.metadataToRead <= 0 {
 			var meta map[string]string
-			meta, err = decodeMetadata(string(mi.metadataBuf))
+			meta, err = DecodeMetadata(string(mi.metadataBuf))
 			if err != nil {
 				return
 			}
-			me.pubsub.Pub(meta)
+			mi.pubsub.Pub(meta)
 			mi.metadataBuf = nil
 		}
 		return
 	}
 
-	bytesToRead := mi.MetadataInterval - mi.blockOffset
 	if bytesToRead > len(data) {
 		bytesToRead = len(data)
 	}
@@ -72,11 +85,5 @@ func (mi *MetadataExtractor) Read(data []byte) (n int, err error) {
 		}
 		mi.blockOffset += n
 	}
-	if mi.blockOffset == mi.MetadataInterval {
-		mi.generateMetadataBuf() // will be read in on next Read call
-		mi.blockOffset = 0
-	} else if mi.blockOffset > mi.MetadataInterval {
-		panic("block offset higher than metadata interval, logical error")
-	}
 	return
 }

app/streams/metadata_extractor_test.go

@@ -0,0 +1,32 @@
+package streams
+
+var (
+	exampleMetadataStr = `StreamTitle='Test 123';`
+	exampleMetadata    = map[string]string{
+		"StreamTitle": "Test 123",
+	}
+
+	exampleData = []byte{
+		0, 1, 2, 3, 4, 5, 6, 7,
+		0, 1, 2, 3, 4, 5, 6, 7,
+	}
+
+	exampleCompleteData = append(
+		append(
+			append(
+				append(
+					append(
+						exampleData[0:4], // content
+						2,                // 2*16 = 32 bytes in length
+					),
+					[]byte(exampleMetadataStr)..., // actual metadata
+				),
+				make([]byte, 9)..., // padding
+			),
+			exampleData[4:8]...,
+		),
+		0, // 0*16 = 0 bytes, no change in length
+	)
+)
+
+func Test_MetadataExtractor()

app/streams/metadata_injector.go

@@ -8,7 +8,7 @@ type MetadataInjector struct {
 	io.Reader
 	MetadataInterval int
 	blockOffset      int
-	Metadata         map[string]string
+	Metadata         Metadata
 	metadataBuf      []byte
 }
 
@@ -48,7 +48,12 @@ func (mi *MetadataInjector) Read(data []byte) (n int, err error) {
 		mi.blockOffset += n
 	}
 	if mi.blockOffset == mi.MetadataInterval {
-		mi.generateMetadataBuf() // will be read in on next Read call
+		// the metadata generated here will be read on the next Read call
+		metadataBytes := mi.Metadata.Bytes()
+		lenByte := byte((len(metadataBytes) + 15) / 16)
+		mi.metadataBuf = make([]byte, int(lenByte)*16+1)
+		mi.metadataBuf[0] = lenByte
+		copy(mi.metadataBuf[1:], metadataBytes)
 		mi.blockOffset = 0
 	} else if mi.blockOffset > mi.MetadataInterval {
 		panic("block offset higher than metadata interval, logical error")

app/streams/metadata_test.go

@@ -0,0 +1,51 @@
+package streams
+
+import (
+	"fmt"
+	"testing"
+
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+const (
+	originalStringA = "Aäöü´ß - \\_-_-_"
+	quotedStringA   = "'Aäöü´ß - \\\\_-_-_'"
+	originalStringB = "hfn9'07';137gr\tbqp9\"ui"
+	quotedStringB   = "'hfn9\\'07\\';137gr\tbqp9\"ui'"
+)
+
+func Test_Metadata(t *testing.T) {
+	Convey("Metadata", t, func() {
+		Convey("decoding", func() {
+			meta, err := DecodeMetadata("")
+			So(err, ShouldBeNil)
+			So(meta, ShouldHaveLength, 0)
+
+			meta, err = DecodeMetadata("StreamTitle=" + quotedStringA + ";")
+			So(err, ShouldBeNil)
+			So(meta, ShouldHaveLength, 1)
+			So(meta, ShouldContainKey, "StreamTitle")
+			So(meta["StreamTitle"], ShouldEqual, originalStringA)
+		})
+
+		Convey("encoding", func() {
+			meta := Metadata{}
+			So(meta.String(), ShouldHaveLength, 0)
+
+			meta = Metadata{
+				"StreamTitle": originalStringA,
+			}
+			So(meta.String(), ShouldEqual,
+				fmt.Sprintf("StreamTitle=%s;",
+					quotedStringA))
+
+			meta = Metadata{
+				"StreamTitle": originalStringA,
+				"test":        originalStringB,
+			}
+			So(meta.String(), ShouldEqual,
+				fmt.Sprintf("StreamTitle=%s;test=%s;",
+					quotedStringA, quotedStringB))
+		})
+	})
+}