Basic working segmentation of MP3 audio stream

go.mod

@@ -0,0 +1,8 @@
+module segmento
+
+go 1.14
+
+require (
+	github.com/stretchr/testify v1.6.1
+	github.com/tcolgate/mp3 v0.0.0-20170426193717-e79c5a46d300
+)

go.sum

@@ -0,0 +1,13 @@
+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/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 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/tcolgate/mp3 v0.0.0-20170426193717-e79c5a46d300 h1:XQdibLKagjdevRB6vAjVY4qbSr8rQ610YzTkWcxzxSI=
+github.com/tcolgate/mp3 v0.0.0-20170426193717-e79c5a46d300/go.mod h1:FNa/dfN95vAYCNFrIKRrlRo+MBLbwmR9Asa5f2ljmBI=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

main.go

@@ -1,8 +1,8 @@
 package main
 
 import (
+	"bytes"
 	"flag"
-	"fmt"
 	"io"
 	"log"
 	"net/http"
@@ -12,33 +12,52 @@ import (
 	"github.com/tcolgate/mp3"
 )
 
-// TODO what should be the behaviour when adding a Frame to a Segment causes it to over-shoot
-// the TargetDuration?
-// should the frame be partially added, or not added at all?
-// typical frame duration 36ms
+const DefaultTargetDuration = 3 * time.Second
+
 type Segment struct {
-	TargetDuration time.Duration
-	Duration       time.Duration
-	Data           []byte
+	targetDuration time.Duration
+	duration       time.Duration
+	data           *bytes.Buffer
 }
 
-func newSegment(capacity int) *Segment {
+// Initialize a new Segment struct. The capacity is the initial maximum capacity of the internal
+// buffer, in bytes. It should be initialized with a value greater than the expected maximum buffer size,
+// depending on the implementation.
+func newSegment(targetDuration time.Duration, capacity int) *Segment {
 	return &Segment{
-		Data: make([]byte, 0, capacity),
+		data:           bytes.NewBuffer(make([]byte, 0, capacity)),
+		duration:       0,
+		targetDuration: targetDuration,
 	}
 }
 
-func (s *Segment) Write(p []byte) (n int, err error) {
-	fmt.Println("write", len(p), "bytes")
-	return len(p), nil
+func (s *Segment) ReadFrom(r io.Reader) (n int64, err error) {
+	return s.data.ReadFrom(r)
+}
+
+func (s *Segment) IncrementDuration(d time.Duration) time.Duration {
+	s.duration += d
+	return s.duration
+}
+
+func (s *Segment) CanWrite(d time.Duration) bool {
+	return s.targetDuration-s.duration >= d
+}
+
+func (s *Segment) Duration() time.Duration {
+	return s.duration
+}
+
+func (s *Segment) Len() int {
+	return s.data.Len()
 }
 
 type MP3HTTPSegmenter struct {
 	decoder *mp3.Decoder
 }
 
-func (s *MP3HTTPSegmenter) Segment(r io.Reader) (chan Segment, error) {
-	c := make(chan Segment)
+func (s *MP3HTTPSegmenter) Segment(r io.Reader) (chan *Segment, error) {
+	c := make(chan *Segment)
 
 	go func() {
 		d := mp3.NewDecoder(r)
@@ -48,12 +67,39 @@ func (s *MP3HTTPSegmenter) Segment(r io.Reader) (chan Segment, error) {
 			skipped int
 		)
 
+		var (
+			s *Segment
+		)
+
 		for {
 			if err := d.Decode(&v, &skipped); err != nil {
+				if err == io.EOF {
+					break
+				}
 				log.Fatal(err)
 			}
 
-			fmt.Println("decoded frame of", v.Size(), "bytes, duration", v.Duration(), ", skipped", skipped, "bytes")
+			if s != nil && !s.CanWrite(v.Duration()) {
+				// publish the current segment, and initialize a new one.
+				c <- s
+				s = nil
+			}
+
+			if s == nil {
+				s = newSegment(DefaultTargetDuration, 1024)
+			}
+
+			n, err := s.ReadFrom(v.Reader())
+			if err != nil {
+				log.Fatal(err) // TODO: some proper error handling
+			}
+
+			if n != int64(v.Size()) {
+				// TODO would this ever happen?
+				log.Fatal("unexpeced frame size, want = ", v.Size(), "got = ", n)
+			}
+
+			s.IncrementDuration(v.Duration())
 		}
 	}()
 
@@ -91,8 +137,6 @@ func main() {
 		os.Exit(-1)
 	}
 
-	log.Println("Open URL:", url)
-
 	client := &http.Client{}
 	resp, err := client.Get(url)
 
@@ -109,8 +153,8 @@ func main() {
 	}
 
 	go func() {
-		for segment := range segments {
-			log.Println("got segment", segment)
+		for s := range segments {
+			log.Println("got segment with duration", s.Duration(), "and len", s.Len(), "bytes")
 		}
 	}()
 

main_test.go

@@ -0,0 +1,22 @@
+package main
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSegment(t *testing.T) {
+	segment := newSegment(10*time.Second, 0)
+
+	require.Equal(t, time.Duration(0), segment.Duration())
+	require.True(t, segment.CanWrite(9*time.Second))
+	require.True(t, segment.CanWrite(10*time.Second))
+	require.False(t, segment.CanWrite(11*time.Second))
+
+	d := segment.IncrementDuration(10 * time.Second)
+	require.Equal(t, segment.Duration(), d)
+	require.Equal(t, 10*time.Second, segment.Duration())
+	require.False(t, segment.CanWrite(1*time.Millisecond))
+}