commit 6811a7400504184e126533848180e2aafb8e7aad
Author: Oliver Lowe <o@olowe.co>
Date: Tue, 30 Apr 2024 16:59:37 +1000
initial commit
Diffstat:
13 files changed, 686 insertions(+), 0 deletions(-)
diff --git a/LICENSE b/LICENSE
@@ -0,0 +1,13 @@
+Copyright (c) 2024 The Untangled Authors. All rights reserved.
+
+Permission to use, copy, modify, and distribute this software for any
+purpose with or without fee is hereby granted, provided that the above
+copyright notice and this permission notice appear in all copies.
+
+THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
diff --git a/README b/README
@@ -0,0 +1 @@
+This repository contains packages for developing media streaming systems in Go.
diff --git a/cmcd/cmcd.go b/cmcd/cmcd.go
@@ -0,0 +1,212 @@
+/*
+Package cmcd provides types and functions for exchanging
+Common Media Client Data (CMCD) as specified in CTA-5004.
+
+The typical use case for servers is to read client playback
+information from a HTTP GET request, then relay the information to a
+database for later analysis.
+For instance, clients sending CMCD information as a query parameter
+can be read with ParseInfo.
+
+ func (srv *Server) ServeSegment(w http.ResponseWriter, req *http.Request) {
+ v := req.URL.Query()
+ var info cmcd.Info
+ if v.Has("CMCD") {
+ info, err := cmcd.ParseInfo(v.Get("CMCD"))
+ if err != nil {
+ log.Println("parse cmcd info: %v: ignoring", err)
+ }
+ relayToMetricStore(&info)
+ }
+ // serve response...
+ }
+*/
+package cmcd
+
+import (
+ "fmt"
+ "net/http"
+ "strconv"
+ "strings"
+ "time"
+)
+
+const (
+ HeaderRequest = "CMCD-Request"
+ HeaderObject = "CMCD-Object"
+ HeaderStatus = "CMCD-Status"
+ HeaderSession = "CMCD-Session"
+)
+
+type Info struct {
+ Request
+ Object
+ Status
+ Session
+ // Holds custom attributes as either a string, integer or
+ // boolean.
+ Custom map[string]any
+}
+
+func (info Info) Encode() string {
+ ss := make([]string, 4)
+ ss[0] = info.Request.Encode()
+ ss[1] = info.Object.Encode()
+ ss[2] = info.Status.Encode()
+ ss[3] = info.Session.Encode()
+ if info.Custom != nil && len(info.Custom) > 0 {
+ for k, v := range info.Custom {
+ switch v.(type) {
+ case string:
+ ss = append(ss, fmt.Sprintf("%s=%q", k, v))
+ case int:
+ ss = append(ss, fmt.Sprintf("%s=%d", k, v))
+ case bool:
+ ss = append(ss, k)
+ default:
+ ss = append(ss, fmt.Sprintf("%s=%q", k, v))
+ }
+ }
+ }
+
+ var noEmpty []string
+ for _, s := range ss {
+ if s == "" {
+ continue
+ }
+ noEmpty = append(noEmpty, s)
+ }
+ return strings.Join(noEmpty, ",")
+}
+
+// ParseInfo returns the Info encoded in the string s.
+// Typical usage is to parse a URL containing the "CMCD" query parameter,
+// then pass the corresponding value to ParseInfo.
+// See ExampleParseInfo.
+func ParseInfo(s string) (Info, error) {
+ return parseInfo(lex(s))
+}
+
+func ExtractInfo(header http.Header) (Info, error) {
+ var fields []string
+ fields = append(fields, header.Get(HeaderRequest))
+ fields = append(fields, header.Get(HeaderObject))
+ fields = append(fields, header.Get(HeaderStatus))
+ fields = append(fields, header.Get(HeaderSession))
+ tokens := lex(strings.Join(fields, ","))
+ return parseInfo(tokens)
+}
+
+// Request represents data relating to the client's...
+type Request struct {
+ // Playback duration of the requested content. When encoded,
+ // values are rounded to the nearest millisecond.
+ BufLength time.Duration
+ // Time limit, in milliseconds, to receive a response to the
+ // request before the client may experience playback problems.
+ Deadline time.Duration
+ // Kilobits per second between client and server, as measured by the client.
+ Throughput int
+ // Relative path of the next request.
+ Next string
+ // Byte range of the next request.
+ NextRange Range
+ // If true, a response is needed urgently as playback may be
+ // starting, seeking, or the client has an empty playback buffer.
+ Startup bool
+}
+
+type Range [2]int
+
+func (r Range) String() string {
+ if r[1] < 0 {
+ return fmt.Sprintf("%d-", r[0])
+ }
+ return fmt.Sprintf("%d-%d", r[0], r[1])
+}
+
+type Object struct {
+ // Encoded bitrate, in kilobits per second.
+ Bitrate int
+ // Playback duration. When encoded, values are rounded to the
+ // nearest millisecond.
+ Duration time.Duration
+ // Media type, such as audio or video.
+ Type ObjectType
+ // The client's highest allowed bitrate, in kilobits per second.
+ TopBitrate int
+}
+
+type ObjectType string
+
+const (
+ ObjTypeText ObjectType = "m"
+ ObjTypeAudio ObjectType = "a"
+ ObjTypeVideo ObjectType = "v"
+ ObjTypeAV ObjectType = "av"
+ ObjTypeI ObjectType = "i"
+ ObjTypeCaption ObjectType = "c"
+ ObjTypeTimedText ObjectType = "tt"
+ ObjTypeKey ObjectType = "k"
+ ObjTypeOther ObjectType = "o"
+)
+
+func parseRange(s string) (Range, error) {
+ offset, end, found := strings.Cut(s, "-")
+ if !found {
+ return Range{}, fmt.Errorf("parse next range request: missing range separator %q", "-")
+ }
+ off, err := strconv.Atoi(offset)
+ if err != nil {
+ return Range{}, fmt.Errorf("offset: %w", err)
+ }
+ e, err := strconv.Atoi(end)
+ if err != nil {
+ return Range{}, fmt.Errorf("end: %w", err)
+ }
+ return Range{off, e}, nil
+}
+
+type Status struct {
+ Starved bool
+ MaxThroughput int
+}
+
+type Session struct {
+ // A GUID uniquely identifying the session, no longer than 64
+ // characters.
+ ID string
+ // Type of the stream. If false, all playback segments are
+ // available. Otherwise, the stream is considered live, and
+ // segments become available over time.
+ Live bool
+ // A unique identifier of the client's requested content, no
+ // longer than 64 characters.
+ ContentID string
+ // The playback rate of the content.
+ PlayRate PlayRate
+ // The format of the stream, such as HLS or MPEG-DASH.
+ Format StreamFormat
+ version int
+}
+
+type PlayRate uint8
+
+const (
+ Stopped PlayRate = iota
+ RealTime
+ DoubleTime
+)
+
+type StreamFormat byte
+
+const (
+ FormatDASH StreamFormat = 'd'
+ FormatHLS StreamFormat = 'h'
+ FormatSmooth StreamFormat = 's'
+ FormatOther StreamFormat = 'o'
+)
+
+func (c StreamFormat) String() string {
+ return fmt.Sprintf("%c", c)
+}
diff --git a/cmcd/cmcd_test.go b/cmcd/cmcd_test.go
@@ -0,0 +1,21 @@
+package cmcd
+
+import (
+ "fmt"
+ "net/url"
+)
+
+func ExampleParseInfo() {
+ ex := "http://test.example.com/?CMCD=br%3D3200%2Cbs%2Cd%3D4004%2Cmtp%3D25400"
+ u, err := url.Parse(ex)
+ param := u.Query().Get("CMCD")
+ fmt.Println(param)
+ info, err := ParseInfo(param)
+ if err != nil {
+ // handle...
+ }
+ fmt.Println(info.Bitrate)
+ // Output:
+ // br=3200,bs,d=4004,mtp=25400
+ // 3200
+}
diff --git a/cmcd/encode.go b/cmcd/encode.go
@@ -0,0 +1,86 @@
+package cmcd
+
+import (
+ "fmt"
+ "net/url"
+ "strings"
+)
+
+func (r Request) Encode() string {
+ var attrs []string
+ if r.BufLength > 0 {
+ // TODO(otl): round to nearest 100ms
+ a := fmt.Sprintf("bl=%d", r.BufLength.Milliseconds())
+ attrs = append(attrs, a)
+ }
+ if r.Deadline > 0 {
+ // TODO(otl): round to nearest 100ms
+ a := fmt.Sprintf("dl=%d", r.Deadline.Milliseconds())
+ attrs = append(attrs, a)
+ }
+ if r.Throughput > 0 {
+ attrs = append(attrs, fmt.Sprintf("mtp=%d", r.Throughput))
+ }
+ if r.Next != "" {
+ a := fmt.Sprintf("nor=%q", url.QueryEscape(r.Next))
+ attrs = append(attrs, a)
+ }
+ if r.NextRange != [2]int{0, 0} {
+ a := fmt.Sprintf("nrr=%q", r.NextRange)
+ attrs = append(attrs, a)
+ }
+ if r.Startup == true {
+ attrs = append(attrs, "su")
+ }
+ return strings.Join(attrs, ",")
+}
+
+func (o Object) Encode() string {
+ var attrs []string
+ if o.Bitrate > 0 {
+ attrs = append(attrs, fmt.Sprintf("br=%d", o.Bitrate))
+ }
+ if o.Duration > 0 {
+ // TODO(otl): round to nearest 100ms
+ a := fmt.Sprintf("d=%d", o.Duration.Milliseconds())
+ attrs = append(attrs, a)
+ }
+ if o.Type != "" {
+ // not a quoted string as these are reserved keywords.
+ attrs = append(attrs, fmt.Sprintf("ot=%s", o.Type))
+ }
+ if o.TopBitrate > 0 {
+ attrs = append(attrs, fmt.Sprintf("tb=%d", o.TopBitrate))
+ }
+ return strings.Join(attrs, ",")
+}
+
+func (s Status) Encode() string {
+ var attrs []string
+ if s.Starved {
+ attrs = append(attrs, "bs")
+ }
+ if s.MaxThroughput > 0 {
+ attrs = append(attrs, fmt.Sprintf("mtp=%d", s.MaxThroughput))
+ }
+ return strings.Join(attrs, ",")
+}
+
+func (s Session) Encode() string {
+ var attrs []string
+ if s.ID != "" {
+ attrs = append(attrs, fmt.Sprintf("sid=%q", s.ID))
+ }
+ if s.ContentID != "" {
+ attrs = append(attrs, fmt.Sprintf("cid=%q", s.ContentID))
+ }
+ // "SHOULD only be sent if not equal to 1": CTA-5004 page 10.
+ if s.PlayRate != RealTime {
+ attrs = append(attrs, fmt.Sprintf("pr=%d", s.PlayRate))
+ }
+ switch s.Format {
+ case FormatDASH, FormatHLS, FormatSmooth, FormatOther:
+ attrs = append(attrs, fmt.Sprintf("sf=%s", s.Format))
+ }
+ return strings.Join(attrs, ",")
+}
diff --git a/cmcd/parse.go b/cmcd/parse.go
@@ -0,0 +1,209 @@
+package cmcd
+
+import (
+ "fmt"
+ "net/url"
+ "strconv"
+ "strings"
+ "time"
+)
+
+func parseInfo(tokens map[string]string) (Info, error) {
+ var info Info
+ var err error
+ info.Request, err = parseRequest(tokens)
+ if err != nil {
+ // TODO
+ }
+ info.Object, err = parseObject(tokens)
+ if err != nil {
+ // TODO
+ }
+ info.Status, err = parseStatus(tokens)
+ if err != nil {
+ // TODO
+ }
+ info.Session, err = parseSession(tokens)
+ if err != nil {
+ // TODO
+ }
+ if custom := parseCustom(tokens); custom != nil {
+ info.Custom = custom
+ }
+ return info, nil
+}
+
+func parseRequest(attrs map[string]string) (Request, error) {
+ var req Request
+ for k, v := range attrs {
+ switch k {
+ case "bl":
+ i, err := strconv.Atoi(v)
+ if err != nil {
+ return req, fmt.Errorf("parse buffer length: %w", err)
+ }
+ req.BufLength = time.Duration(i) * time.Millisecond
+ case "dl":
+ i, err := strconv.Atoi(v)
+ if err != nil {
+ return req, fmt.Errorf("parse deadline: %w", err)
+ }
+ req.Deadline = time.Duration(i) * time.Millisecond
+ case "mtp":
+ i, err := strconv.Atoi(v)
+ if err != nil {
+ return req, fmt.Errorf("parse throughput: %w", err)
+ }
+ req.Throughput = i
+ case "nor":
+ dec, err := url.QueryUnescape(strings.Trim(v, `"`))
+ if err != nil {
+ return req, fmt.Errorf("decode next object request: %w", err)
+ }
+ req.Next = dec
+ case "nrr":
+ rg, err := parseRange(strings.Trim(v, `"`))
+ if err != nil {
+ return req, fmt.Errorf("parse next range: %w", err)
+ }
+ req.NextRange = rg
+ case "su":
+ req.Startup = true
+ }
+ }
+ return req, nil
+}
+
+func parseObject(attrs map[string]string) (Object, error) {
+ var obj Object
+ for k, v := range attrs {
+ if v == "" {
+ continue // stray comma, perhaps at end of line. ignore
+ }
+ switch k {
+ case "br":
+ i, err := strconv.Atoi(v)
+ if err != nil {
+ return obj, fmt.Errorf("parse bitrate: %w", err)
+ }
+ obj.Bitrate = i
+ case "d":
+ i, err := strconv.Atoi(v)
+ if err != nil {
+ return obj, fmt.Errorf("parse duration: %w", err)
+ }
+ obj.Duration = time.Duration(i) * time.Millisecond
+ case "ot":
+ // TODO validate value
+ obj.Type = ObjectType(v)
+ case "tb":
+ i, err := strconv.Atoi(v)
+ if err != nil {
+ return obj, fmt.Errorf("parse top bitrate: %w", err)
+ }
+ obj.TopBitrate = i
+ }
+ }
+ return obj, nil
+}
+
+func parseStatus(attrs map[string]string) (Status, error) {
+ var stat Status
+ for k, v := range attrs {
+ switch k {
+ case "bs":
+ stat.Starved = true
+ case "rtp":
+ i, err := strconv.Atoi(v)
+ if err != nil {
+ return stat, fmt.Errorf("parse max throughput: %w", err)
+ }
+ stat.MaxThroughput = i
+ }
+ }
+ return stat, nil
+}
+
+func parseSession(attrs map[string]string) (Session, error) {
+ var ses Session
+ for k, v := range attrs {
+ switch k {
+ case "sid":
+ ses.ID = strings.Trim(v, `"`)
+ case "st":
+ if v == "l" {
+ ses.Live = true
+ }
+ case "cid":
+ ses.ContentID = strings.Trim(v, `"`)
+ case "pr":
+ i, err := strconv.Atoi(v)
+ if err != nil {
+ // TODO
+ }
+ if i > 2 {
+ return ses, fmt.Errorf("TODO")
+ }
+ ses.PlayRate = PlayRate(i)
+ case "sf":
+ if len(v) != 1 {
+ return ses, fmt.Errorf("TODO")
+ }
+ c := StreamFormat([]byte(v)[0])
+ switch c {
+ case FormatDASH, FormatHLS, FormatSmooth, FormatOther:
+ ses.Format = c
+ default:
+ return ses, fmt.Errorf("TODO")
+ }
+ }
+ }
+ // If we didn't see playrate, we must set it to realtime as
+ // only values other than realtime should be transmitted. See
+ // CTA=5004 page 10.
+ if _, ok := attrs["pr"]; !ok {
+ ses.PlayRate = RealTime
+ }
+ return ses, nil
+}
+
+func parseCustom(attrs map[string]string) map[string]any {
+ m := make(map[string]any)
+ for k, v := range attrs {
+ switch k {
+ case "bl", "dl", "mtp", "nor", "nrr", "su":
+ continue // Request keys
+ case "br", "d", "ot", "tb":
+ continue // Object keys
+ case "bs", "rtp":
+ continue // Status keys
+ case "sid", "st", "cid", "pr", "sf":
+ continue // Session keys
+ }
+ if v == "" {
+ m[k] = true
+ } else if i, err := strconv.Atoi(v); err == nil {
+ m[k] = i
+ } else {
+ m[k] = strings.Trim(v, `"`)
+ }
+ }
+ if len(m) == 0 {
+ return nil
+ }
+ return m
+}
+
+func lex(s string) map[string]string {
+ m := make(map[string]string)
+ s = clean(s)
+ for _, attr := range strings.Split(s, ",") {
+ name, val, _ := strings.Cut(attr, "=")
+ m[name] = val
+ }
+ return m
+}
+
+// clean removes stray commas. Trailing commas are technically valid
+// but we remove them to simplify parsing.
+func clean(s string) string { return strings.Trim(s, ",") }
diff --git a/cmcd/parse_test.go b/cmcd/parse_test.go
@@ -0,0 +1,122 @@
+package cmcd
+
+import (
+ "bufio"
+ "fmt"
+ "net/http"
+ "net/url"
+ "os"
+ "path"
+ "reflect"
+ "strings"
+ "testing"
+ "time"
+)
+
+var tests = map[string]Info{
+ "testdata/simple": Info{
+ Session: Session{ID: "6e2fb550-c457-11e9-bb97-0800200c9a66", PlayRate: RealTime},
+ },
+ "testdata/all_four": Info{
+ Request: Request{Throughput: 25400},
+ Object: Object{
+ Bitrate: 3200,
+ Duration: 4004 * time.Millisecond,
+ Type: ObjTypeVideo,
+ TopBitrate: 6000,
+ },
+ Status: Status{true, 15000},
+ Session: Session{ID: "6e2fb550-c457-11e9-bb97-0800200c9a66", PlayRate: RealTime},
+ },
+ "testdata/booleans": Info{
+ Status: Status{true, 0},
+ Request: Request{Startup: true},
+ Session: Session{PlayRate: RealTime},
+ },
+ "testdata/range": Info{
+ Request: Request{NextRange: [2]int{12323, 48763}},
+ Object: Object{Duration: 4004 * time.Millisecond},
+ Session: Session{PlayRate: RealTime},
+ },
+ "testdata/custom": Info{
+ Object: Object{Duration: 4004 * time.Millisecond},
+ Session: Session{PlayRate: RealTime},
+ Custom: map[string]any{
+ "com.example.javasucks.int": 500,
+ "stringy": "yamum",
+ "aBool": true,
+ },
+ },
+}
+
+func TestParse(t *testing.T) {
+ for name, want := range tests {
+ t.Run(path.Base(name), func(t *testing.T) {
+ pt, err := readParseTest(name)
+ if err != nil {
+ t.Fatal(err)
+ }
+ info, err := ParseInfo(pt.query)
+ if err != nil {
+ t.Errorf("info from query: %v", err)
+ }
+ if !reflect.DeepEqual(want, info) {
+ t.Errorf("info from query: want %+v, got %+v", want, info)
+ t.Log(want.Encode())
+ t.Log(info.Encode())
+ }
+ })
+ }
+}
+
+type parseTest struct {
+ header http.Header
+ query string
+ json []byte
+}
+
+func readParseTest(name string) (parseTest, error) {
+ f, err := os.Open(name)
+ if err != nil {
+ return parseTest{}, err
+ }
+ defer f.Close()
+ var pt parseTest
+ pt.header = make(http.Header)
+ sc := bufio.NewScanner(f)
+ for sc.Scan() {
+ if sc.Text() == "" {
+ continue
+ } else if strings.HasPrefix(sc.Text(), "#") {
+ continue // skip comments
+ }
+ if strings.HasPrefix(sc.Text(), "?CMCD=") {
+ raw := strings.TrimPrefix(sc.Text(), "?CMCD=")
+ q, err := url.QueryUnescape(raw)
+ if err != nil {
+ return pt, fmt.Errorf("parse cmcd query: %w", err)
+ }
+ pt.query = q
+ continue
+ }
+ if strings.HasPrefix(sc.Text(), "{") {
+ pt.json = sc.Bytes()
+ }
+ before, after, found := strings.Cut(sc.Text(), ":")
+ if !found {
+ return pt, fmt.Errorf("invalid case: %s", sc.Text())
+ }
+ pt.header.Set(before, strings.TrimSpace(after))
+ }
+ return pt, sc.Err()
+}
+
+/*
+ custom := make(map[string]any) // TODO
+ if !reflect.DeepEqual(tt.want.Custom, custom) {
+ t.Errorf("custom attributes: want %+v, got %+v", tt.want.Custom, custom)
+ }
+ })
+ }
+}
+*/
diff --git a/cmcd/testdata/all_four b/cmcd/testdata/all_four
@@ -0,0 +1,8 @@
+# CMCD-Request: mtp=25400
+# CMCD-Object: br=3200,d=4004,ot=v,tb=6000
+# CMCD-Status: bs,rtp=15000
+# CMCD-Session: sid="6e2fb550-c457-11e9-bb97-0800200c9a66"
+
+?CMCD=br%3D3200%2Cbs%2Cd%3D4004%2Cmtp%3D25400%2Cot%3Dv%2Crtp%3D15000%2Csid%3D%226e2fb550-c457-11e9-bb97-0800200c9a66%22%2Ctb%3D6000
+
+# {"br": 3200,"bs":true,"d": 4004,"mtp": 25400, "ot": "v", "rtp":15000,"sid": "6e2fb550-c457-11e9-bb97-0800200c9a66","tb":6000}
diff --git a/cmcd/testdata/booleans b/cmcd/testdata/booleans
@@ -0,0 +1 @@
+?CMCD=bs%2Csu
diff --git a/cmcd/testdata/custom b/cmcd/testdata/custom
@@ -0,0 +1,2 @@
+?CMCD=d%3D4004%2Ccom.example.javasucks.int%3D500%2Cstringy%3D%22yamum%22%2CaBool
+
diff --git a/cmcd/testdata/range b/cmcd/testdata/range
@@ -0,0 +1,3 @@
+# and a trailing comma for good measure.
+
+?CMCD=nrr%3D%2212323-48763%22%2Cd=4004%2C
diff --git a/cmcd/testdata/simple b/cmcd/testdata/simple
@@ -0,0 +1,5 @@
+# CMCD-Session: sid="6e2fb550-c457-11e9-bb97-0800200c9a66"
+
+?CMCD=sid%3D%226e2fb550-c457-11e9-bb97-0800200c9a66%22
+
+# {"sid": "6e2fb550-c457-11e9-bb97-0800200c9a66"}
diff --git a/go.mod b/go.mod
@@ -0,0 +1,3 @@
+module github.com/untangledco/streaming
+
+go 1.19
