Add burst buffer (not working yet!).

app/README.md

@@ -0,0 +1,9 @@
+When a stream is fed in, generally the first thing that should be done is to
+analyze the stream to extract information about it. Information that should be
+extracted include:
+
+- Format
+
+- Streams
+	- Codec
+	- Offset

app/channels/channel.go

@@ -1,61 +1,89 @@
 package channels
 
 import (
-	"context"
+	"log"
-	"sync"
 
-	"git.icedream.tech/icedream/uplink/app/streams"
+	"github.com/cskr/pubsub"
+
+	"git.icedream.tech/icedream/uplink/app/media"
+	pubsubutil "git.icedream.tech/icedream/uplink/app/pubsub"
 )
 
 type Channel struct {
-	metadataLock    sync.RWMutex
+	InputContainer *media.MediaStreamContainer
-	metadata        map[string]string
+	InputStreams   map[string]*media.MediaStream
-	metadataChannel chan map[string]string
+
-	Id              string
+	OutputContainers map[string]*media.MediaStreamContainer
-	Name            string
+	OutputStreams    map[string]*media.MediaStream
-	Description     string
+
-	MimeType        string
+	Events *pubsub.PubSub
-	InputStream     *streams.Stream
+}
-	OutputStreams   map[string]ChannelOutputStream
+
+func (channel *Channel) AddInputStream(id string) *media.MediaStream {
+	stream := &media.MediaStream{
+		PubSubWriter: pubsubutil.NewPubSubWriter(),
+	}
+	channel.InputStreams[id] = stream
+	log.Println("New input stream", id)
+	channel.Events.Pub(id, "input_stream")
+	return stream
+}
+
+func (channel *Channel) AddOutputStream(id string) *media.MediaStream {
+	stream := &media.MediaStream{
+		PubSubWriter: pubsubutil.NewPubSubWriter(),
+	}
+	channel.OutputStreams[id] = stream
+	log.Println("New output stream", id)
+	channel.Events.Pub(id, "output_stream")
+	return stream
+}
+
+func (channel *Channel) AddOutputContainer(id string) *media.MediaStreamContainer {
+	stream := &media.MediaStreamContainer{
+		PubSubWriter: pubsubutil.NewPubSubWriter(),
+	}
+	channel.OutputContainers[id] = stream
+	log.Println("New output container", id)
+	channel.Events.Pub(id, "output_container")
+	return stream
 }
 
 func (channel *Channel) SetMetadata(data map[string]string) {
-	channel.metadataLock.Lock()
+	channel.Events.Pub(data, "metadata")
-	defer channel.metadataLock.Unlock()
-	channel.metadata = data
-	channel.metadataChannel <- data
 }
 
-func (channel *Channel) Metadata(ctx context.Context) <-chan map[string]string {
+func (channel *Channel) Metadata() chan map[string]string {
-	channel.metadataLock.Lock()
+	outC := make(chan map[string]string)
-	defer channel.metadataLock.Unlock()
-	metadataChan := make(chan map[string]string, 1)
-	if channel.metadata != nil {
-		metadataChan <- channel.metadata
-	}
 	go func() {
-		for {
+		c := channel.Events.Sub("metadata")
+	forloop:
+		for event := range c {
 			select {
-			case data, ok := <-channel.metadataChannel:
+			case _, _ = <-outC:
-				if !ok {
+				break forloop
-					return
+			case outC <- event.(map[string]string):
-				}
-				metadataChan <- data
-			case <-ctx.Done():
-				return
 			}
 		}
+		channel.Events.Unsub(c, "metadata")
 	}()
-	return metadataChan
+	return outC
 }
 
 func NewChannel() *Channel {
+	ps := pubsub.New(1)
+
+	inputContainer := pubsubutil.NewPubSubWriterForTopic(ps, "input_container")
+
 	return &Channel{
-		metadataChannel: make(chan map[string]string),
+		InputContainer: &media.MediaStreamContainer{
-		OutputStreams:   map[string]ChannelOutputStream{},
+			PubSubWriter: inputContainer,
+		},
+		InputStreams: map[string]*media.MediaStream{},
+
+		OutputContainers: map[string]*media.MediaStreamContainer{},
+		OutputStreams:    map[string]*media.MediaStream{},
+
+		Events: pubsub.New(1),
 	}
 }
-
-type ChannelOutputStream struct {
-	*streams.Stream
-}

app/channels/channel_manager.go

@@ -2,6 +2,7 @@ package channels
 
 import (
 	"errors"
+	"log"
 	"sync"
 
 	"github.com/cskr/pubsub"
@@ -21,6 +22,10 @@ func NewChannelManager() *ChannelManager {
 	return mgr
 }
 
+func (manager *ChannelManager) Events() *pubsub.PubSub {
+	return manager.pubsub
+}
+
 func (manager *ChannelManager) Channel(uuid string) *Channel {
 	manager.channelsLock.RLock()
 	defer manager.channelsLock.RUnlock()
@@ -43,7 +48,7 @@ func (manager *ChannelManager) Close(uuid string) (err error) {
 		return
 	}
 
-	manager.pubsub.Pub(manager.channels[uuid], "close")
+	manager.pubsub.Pub(uuid, "close")
 	delete(manager.channels, uuid)
 
 	return
@@ -58,10 +63,11 @@ func (manager *ChannelManager) Open(uuid string) (channel *Channel, err error) {
 		return
 	}
 
-	channel = &Channel{Id: uuid}
+	channel = NewChannel()
 	manager.channels[uuid] = channel
 
-	manager.pubsub.Pub(channel, "open")
+	log.Println("Channel opened:", uuid)
+	manager.pubsub.Pub(uuid, "open")
 
 	return
 }

app/media/demuxed_stream.go

@@ -1,18 +0,0 @@
-package media
-
-import (
-	"io"
-
-	"git.icedream.tech/icedream/uplink/app/pubsub"
-)
-
-type DemuxedStream struct {
-	StreamId  int
-	Pts       int64
-	CodecInfo StreamCodecInfo
-	pubsub    *pubsub.PubSubWriter
-}
-
-func (stream *DemuxedStream) Sub() io.ReadCloser {
-	return stream.pubsub.Sub()
-}

app/media/demuxer.go

@@ -10,15 +10,20 @@ import (
 )
 
 type Demuxer struct {
-	streams chan *DemuxedStream
+	streams       chan *MediaStream
 	err           chan error
+	containerInfo *MediaStreamContainerInfo
+}
+
+func (demuxer *Demuxer) ContainerInfo() *MediaStreamContainerInfo {
+	return demuxer.containerInfo
 }
 
 func (demuxer *Demuxer) Error() <-chan error {
 	return demuxer.err
 }
 
-func (demuxer *Demuxer) Streams() <-chan *DemuxedStream {
+func (demuxer *Demuxer) Streams() <-chan *MediaStream {
 	return demuxer.streams
 }
 
@@ -27,7 +32,7 @@ func Demux(r io.ReadCloser) (demuxer *Demuxer) {
 
 	demuxer = &Demuxer{
 		err:     make(chan error),
-		streams: make(chan *DemuxedStream),
+		streams: make(chan *MediaStream),
 	}
 
 	go func() {
@@ -66,6 +71,12 @@ func Demux(r io.ReadCloser) (demuxer *Demuxer) {
 		// ctx.Dump()
 		// fmt.Println("============================")
 
+		demuxer.containerInfo = &MediaStreamContainerInfo{
+			GlobalHeader: ctx.IsGlobalHeader(),
+			StartTime:    ctx.StartTime(),
+			//SDP: ctx.GetSDPString(),
+		}
+
 		// Find out order of streams and store info about them
 		streams := []*gmf.Stream{}
 		pubsubs := []*pubsub.PubSubWriter{}
@@ -78,20 +89,24 @@ func Demux(r io.ReadCloser) (demuxer *Demuxer) {
 			streamCodec := stream.CodecCtx()
 			streams = append(streams, stream)
 
-			if stream.IsVideo() || stream.IsAudio() {
+			switch streamCodec.Type() {
+			case gmf.AVMEDIA_TYPE_AUDIO, gmf.AVMEDIA_TYPE_VIDEO:
 				ps := pubsub.NewPubSubWriter()
-				dmxStream := &DemuxedStream{
+				dmxStream := &MediaStream{
-					CodecInfo: StreamCodecInfo{
+					MediaStreamInfo: MediaStreamInfo{
+						CodecInfo: MediaStreamCodecInfo{
 							CodecName: streamCodec.Codec().Name(),
 						},
 						Pts:      stream.Pts,
 						StreamId: i,
-					pubsub:   ps,
+					},
+					PubSubWriter: ps,
 				}
 				defer ps.Close()
-				if stream.IsVideo() {
+				switch streamCodec.Type() {
+				case gmf.AVMEDIA_TYPE_VIDEO:
 					dmxStream.CodecInfo.Type = Video
-				} else {
+				case gmf.AVMEDIA_TYPE_AUDIO:
 					dmxStream.CodecInfo.Type = Audio
 				}
 				pubsubMap[i] = ps

app/media/demuxer_test.go

@@ -17,7 +17,7 @@ func Test_Demux(t *testing.T) {
 			defer reader.Close()
 
 			demuxer := Demux(reader)
-			var audioStream *DemuxedStream
+			var audioStream *MediaStream
 			var err error
 		forloop:
 			for {
@@ -47,7 +47,7 @@ func Test_Demux(t *testing.T) {
 			defer reader.Close()
 
 			demuxer := Demux(reader)
-			var audioStream, videoStream *DemuxedStream
+			var audioStream, videoStream *MediaStream
 			var err error
 		forloop:
 			for {

app/media/media_stream.go

@@ -0,0 +1,8 @@
+package media
+
+import "git.icedream.tech/icedream/uplink/app/pubsub"
+
+type MediaStream struct {
+	*pubsub.PubSubWriter
+	MediaStreamInfo
+}

app/media/media_stream_codec_info.go

@@ -1,6 +1,6 @@
 package media
 
-type StreamCodecInfo struct {
+type MediaStreamCodecInfo struct {
 	CodecName string
 	Type      StreamMediaType
 }

app/media/media_stream_container.go

@@ -0,0 +1,8 @@
+package media
+
+import "git.icedream.tech/icedream/uplink/app/pubsub"
+
+type MediaStreamContainer struct {
+	*pubsub.PubSubWriter
+	MediaStreamContainerInfo
+}

app/media/media_stream_container_info.go

@@ -0,0 +1,7 @@
+package media
+
+type MediaStreamContainerInfo struct {
+	GlobalHeader bool
+	StartTime    int
+	//SDP string
+}

app/media/media_stream_info.go

@@ -0,0 +1,7 @@
+package media
+
+type MediaStreamInfo struct {
+	StreamId  int
+	Pts       int64
+	CodecInfo MediaStreamCodecInfo
+}

app/media/muxer_test.go

@@ -16,7 +16,7 @@ func Test_Muxer(t *testing.T) {
 			defer reader.Close()
 
 			demuxer := Demux(reader)
-			var audioStream *DemuxedStream
+			var audioStream *MediaStream
 			var err error
 		forloop:
 			for {

app/pubsub/pubsubreader.go

@@ -11,11 +11,13 @@ type PubSubReader struct {
 	channel chan interface{}
 	buf     []byte
 	closed  bool
+	topic   string
 }
 
 func NewPubSubReader(ps *cskrpubsub.PubSub, topic string) *PubSubReader {
 	return &PubSubReader{
 		pubsub:  ps,
+		topic:   topic,
 		channel: ps.Sub(topic),
 	}
 }
@@ -59,6 +61,6 @@ func (r *PubSubReader) Read(p []byte) (n int, err error) {
 
 func (r *PubSubReader) Close() (err error) {
 	r.closed = true
-	r.pubsub.Unsub(r.channel, "")
+	r.pubsub.Unsub(r.channel, r.topic)
 	return
 }

app/pubsub/pubsubwriter.go

@@ -45,14 +45,14 @@ func (pipe *PubSubWriter) Close() (err error) {
 	pipe.PubSub.Close(pipe.topic)
 	if pipe.fullControl {
 		pipe.PubSub.Shutdown()
-		pipe.closed = true
 	}
+	pipe.closed = true
 	return
 }
 
 func (pipe *PubSubWriter) Sub() io.ReadCloser {
 	return &PubSubReader{
-		channel: pipe.PubSub.Sub(""),
+		channel: pipe.PubSub.Sub(pipe.topic),
 		pubsub:  pipe.PubSub,
 		closed:  pipe.closed,
 	}

app/server.go

@@ -3,6 +3,7 @@ package app
 import (
 	"log"
 
+	"git.icedream.tech/icedream/uplink/app/authentication"
 	"git.icedream.tech/icedream/uplink/app/channels"
 	"git.icedream.tech/icedream/uplink/app/servers/http"
 	"git.icedream.tech/icedream/uplink/plugins"
@@ -10,6 +11,7 @@ import (
 
 type App struct {
 	Server         *httpserver.Server
+	Authenticator  authentication.Authenticator
 	ChannelManager *channels.ChannelManager
 
 	plugins []plugins.PluginInstance
@@ -18,6 +20,7 @@ type App struct {
 func New() *App {
 	return &App{
 		Server:         httpserver.NewServer(),
+		Authenticator:  new(authentication.DummyAuthenticator),
 		ChannelManager: channels.NewChannelManager(),
 
 		plugins: []plugins.PluginInstance{},
@@ -25,16 +28,26 @@ func New() *App {
 }
 
 func (app *App) UsePlugin(plugin *plugins.Plugin) {
-	instance := plugin.Run()
+	pluginInstance := plugin.Run()
-	app.plugins = append(app.plugins, instance)
+
-	log.Println("Plugin loaded:", plugin.Descriptor.Name)
+	if p, ok := pluginInstance.(plugins.ServerPlugin); ok {
+		p.SetServer(app.Server)
+	}
+	if p, ok := pluginInstance.(plugins.ChannelPlugin); ok {
+		p.SetChannelManager(app.ChannelManager)
+	}
+	if p, ok := pluginInstance.(plugins.AuthenticatorPlugin); ok {
+		p.SetAuthenticator(app.Authenticator)
+	}
+
+	log.Println("Plugin initialized:", plugin.Descriptor.Name)
+
+	app.plugins = append(app.plugins, pluginInstance)
 }
 
 func (app *App) Init() {
 	for _, plugin := range app.plugins {
-		if p, ok := plugin.(plugins.ServerPlugin); ok {
+		plugin.Init()
-			p.SetServer(app.Server)
-		}
 	}
 }
 

app/servers/http/server.go

@@ -6,6 +6,10 @@ import (
 	"github.com/gin-gonic/gin"
 )
 
+const (
+	ServerHeaderValue = "Uplink/0.0.0"
+)
+
 type Server struct {
 	Http   *http.Server
 	Router *gin.Engine

app/streams/metadata_injector.go

@@ -5,58 +5,63 @@ import (
 )
 
 type MetadataInjector struct {
-	io.Reader
+	io.Writer
+	offset           int
 	MetadataInterval int
-	blockOffset      int
+	metadata         Metadata
-	Metadata         Metadata
-	metadataBuf      []byte
 }
 
-func NewMetadataInjector(r io.Reader, metadataInterval int) *MetadataInjector {
+func NewMetadataInjector(w io.Writer, metadataInterval int) *MetadataInjector {
 	return &MetadataInjector{
-		Reader:           r,
+		Writer:           w,
 		MetadataInterval: metadataInterval,
 	}
 }
 
-func (mi *MetadataInjector) Read(data []byte) (n int, err error) {
+func (mi *MetadataInjector) SetMetadata(metadata Metadata) {
-	if mi.metadataBuf != nil && len(mi.metadataBuf) > 0 {
+	mi.metadata = metadata
-		bytesToRead := len(data)
+}
-		if bytesToRead < len(mi.metadataBuf) {
+
-			// only read as much as possible
+func (mi *MetadataInjector) writeMetadata() (n int, err error) {
-			copy(data, mi.metadataBuf[0:bytesToRead])
+	// the metadata generated here will be read on the next Read call
-			n = bytesToRead
+	metadataBytes := mi.metadata.Bytes()
-			mi.metadataBuf = mi.metadataBuf[bytesToRead:]
+	lenByte := (len(metadataBytes) + 15) / 16
-			return
+	metadataBuf := make([]byte, int(lenByte)*16+1)
-		}
+	metadataBuf[0] = byte(lenByte)
-		// read everything
+	copy(metadataBuf[1:], metadataBytes)
-		copy(data, mi.metadataBuf)
+
-		n = len(mi.metadataBuf)
+	if len(mi.metadata) > 0 {
-		mi.metadataBuf = nil
+		mi.metadata = Metadata{}
-		return
 	}
 
-	bytesToRead := mi.MetadataInterval - mi.blockOffset
+	return mi.Writer.Write(metadataBuf)
-	if bytesToRead > len(data) {
+}
-		bytesToRead = len(data)
+
-	}
+func (mi *MetadataInjector) Write(data []byte) (n int, err error) {
-	if bytesToRead > 0 {
+	for n < len(data) {
-		n, err = mi.Reader.Read(data[0:bytesToRead])
+		toWrite := mi.MetadataInterval - mi.offset
-		if err != nil {
+
+		if toWrite <= 0 {
+			_, cerr := mi.writeMetadata()
+			//n += cn
+			if cerr != nil {
+				err = cerr
 				return
 			}
-		mi.blockOffset += n
+			mi.offset = 0
+			// toWrite = mi.MetadataInterval
+			continue
+		}
+
+		outBytes := make([]byte, toWrite)
+		copy(outBytes, data[mi.offset:mi.offset+toWrite])
+		cn, cerr := mi.Writer.Write(outBytes)
+		n += cn
+		mi.offset += cn
+		if cerr != nil {
+			err = cerr
+			return
 		}
-	if mi.blockOffset == mi.MetadataInterval {
-		// 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")
 	}
 	return
 }

app/streams/metadata_injector_test.go

@@ -2,17 +2,30 @@ package streams
 
 import (
 	"bytes"
+	"io"
 	"testing"
 
 	. "github.com/smartystreets/goconvey/convey"
 )
 
 func Test_MetadataInjector(t *testing.T) {
-	reader := bytes.NewReader(make([]byte, 1024))
+	metadata := Metadata{
-	buffer := make([]byte, 128)
+		"StreamTitle": "Hello",
+	}
+	metadataLengthByte := (len(metadata.Bytes()) + 15) / 16
+	paddedMetadataLength := (metadataLengthByte * 16)
+	paddedMetadataBytes := make([]byte, paddedMetadataLength)
+	copy(paddedMetadataBytes, metadata.Bytes())
+
+	inputBytes := make([]byte, 1024)
+	reader := bytes.NewReader(inputBytes)
+	buffer := new(bytes.Buffer)
 
 	Convey("MetadataInjector", t, func() {
-		mi := NewMetadataInjector(reader, 192)
+		t.Log(len(metadata.Bytes()), "=>", paddedMetadataLength)
+
+		mi := NewMetadataInjector(buffer, 256)
+		mi.SetMetadata(metadata)
 
 		// 128
 		// 64
@@ -21,41 +34,38 @@ func Test_MetadataInjector(t *testing.T) {
 		// 64
 		// [metadata]
 
-		n, err := mi.Read(buffer)
+		n, err := io.Copy(mi, reader)
 		So(err, ShouldBeNil)
-		So(n, ShouldEqual, 128)
+		So(n, ShouldEqual, len(inputBytes))
+		So(buffer.Len(), ShouldEqual, len(inputBytes)+3+paddedMetadataLength)
 
-		n, err = mi.Read(buffer)
+		outBytes := buffer.Bytes()
-		So(err, ShouldBeNil)
-		So(n, ShouldEqual, 64)
 
-		n, err = mi.Read(buffer)
+		for i := 0; i < 3; i++ {
-		So(err, ShouldBeNil)
+			t.Log("part", i)
-		So(n, ShouldEqual, 1)
-		So(buffer[0], ShouldEqual, 0) // no metadata => zero length!
 
-		mi.Metadata = map[string]string{
+			contentStartOffset := i + i*256
-			"StreamTitle": "Testing",
+			if i > 0 {
+				contentStartOffset += paddedMetadataLength
 			}
+			t.Log("contentStartOffset", contentStartOffset)
 
-		n, err = mi.Read(buffer)
+			contentEndOffset := contentStartOffset + 256
-		So(err, ShouldBeNil)
+			t.Log("contentEndOffset", contentEndOffset)
-		So(n, ShouldEqual, 128)
 
-		n, err = mi.Read(buffer)
+			metadataEndOffset := contentEndOffset + 1
-		So(err, ShouldBeNil)
+			if i == 0 {
-		So(n, ShouldEqual, 64)
+				metadataEndOffset += paddedMetadataLength
+			}
+			t.Log("metadataEndOffset", metadataEndOffset)
 
-		n, err = mi.Read(buffer)
+			So(outBytes[contentStartOffset:contentEndOffset], ShouldResemble, inputBytes[0:256])
-		So(err, ShouldBeNil)
+			if i == 0 {
-		So(n, ShouldEqual, 1+32)
+				So(outBytes[contentEndOffset], ShouldEqual, metadataLengthByte)
-		So(buffer[0], ShouldEqual, 2) // "StreamTitle='Testing';" => 22 bytes => quantized to 2 * 16 bytes
+				So(outBytes[contentEndOffset+1:metadataEndOffset], ShouldResemble, paddedMetadataBytes)
-		So(string(buffer[1:23]), ShouldEqual, "StreamTitle='Testing';")
+			} else {
-		So(buffer[1:23], ShouldResemble, []byte("StreamTitle='Testing';"))
+				So(outBytes[contentEndOffset], ShouldEqual, 0)
-		So(buffer[24:32], ShouldResemble, make([]byte, 8)) // 8 zeroes
+			}
-
+		}
-		n, err = mi.Read(buffer)
-		So(err, ShouldBeNil)
-		So(n, ShouldEqual, 128)
 	})
 }

app/streams/stream.go

@@ -1,151 +0,0 @@
-package streams
-
-import (
-	"errors"
-	"io"
-	"log"
-	"runtime"
-	"sync"
-)
-
-type StreamDataChannel chan []byte
-type StreamCancelChannel chan interface{}
-
-type Stream struct {
-	burstSize int
-
-	burstLock sync.RWMutex
-	burst     []byte
-
-	subscribersLock sync.RWMutex
-	subscribers     []io.WriteCloser
-}
-
-func NewStream(burstSize int) *Stream {
-	return &Stream{
-		burstSize: burstSize,
-
-		subscribers: []io.WriteCloser{},
-	}
-}
-
-func (stream *Stream) Subscribe(wc io.WriteCloser) {
-	go func(wc io.WriteCloser) {
-		stream.subscribersLock.Lock()
-		defer stream.subscribersLock.Unlock()
-
-		// send burst data
-		stream.burstLock.RLock()
-		defer stream.burstLock.RUnlock()
-		if stream.burst != nil {
-			burstToSend := len(stream.burst)
-			for burstToSend > 0 {
-				burstSent, err := wc.Write(stream.burst)
-				if err != nil {
-					stream.unsubscribe(wc)
-					if err == io.EOF {
-						return // just end prematurely
-					}
-					log.Println("WARNING - Can not send burst data to subscriber:", err)
-					return
-				}
-				burstToSend -= burstSent
-			}
-		}
-
-		// now subscribe to live broadcast
-		stream.subscribers = append(stream.subscribers, wc)
-	}(wc)
-	runtime.Gosched()
-
-	return
-}
-
-func (stream *Stream) unsubscribe(wc io.WriteCloser) error {
-	stream.subscribersLock.Lock()
-	defer stream.subscribersLock.Unlock()
-	return stream.unsubscribeNoLock(wc)
-}
-
-func (stream *Stream) unsubscribeNoLock(wc io.WriteCloser) error {
-	// log.Println("About to remove subscriber", wc)
-	for index, subscriber := range stream.subscribers {
-		if subscriber == wc {
-			// log.Println("Removing subscriber", wc, "at", index)
-			stream.subscribers = append(stream.subscribers[0:index], stream.subscribers[index+1:]...)
-			// log.Println("We now have", len(stream.subscribers), "subscribers")
-			return subscriber.Close()
-		}
-	}
-	return errors.New("Tried to unsubscribe stream that is not registered as subscriber")
-}
-
-func (stream *Stream) SubscriberCount() int {
-	stream.subscribersLock.RLock()
-	defer stream.subscribersLock.RUnlock()
-
-	return len(stream.subscribers)
-}
-
-func (stream *Stream) Write(data []byte) (n int, err error) {
-	dataLength := len(data)
-
-	stream.burstLock.Lock()
-	defer stream.burstLock.Unlock()
-
-	stream.subscribersLock.RLock()
-	subscribers := make([]io.WriteCloser, len(stream.subscribers))
-	copy(subscribers, stream.subscribers)
-	defer stream.subscribersLock.RUnlock()
-
-	// Write data out to subscribers
-	for _, subscriber := range subscribers {
-		go func(subscriber io.WriteCloser) {
-			stream.subscribersLock.Lock()
-			defer stream.subscribersLock.Unlock()
-			// TODO - absolutely ensure data is sent in the correct order
-			totalWritten := 0
-			for totalWritten < dataLength {
-				currentWritten, err := subscriber.Write(data[totalWritten:])
-				if err != nil {
-					// just remove subscriber and go to next one
-					// log.Println("WARNING: Failed to write data to subscriber, removing subscriber:", err)
-					stream.unsubscribeNoLock(subscriber)
-					return
-				}
-				totalWritten += currentWritten
-			}
-		}(subscriber)
-	}
-	runtime.Gosched()
-
-	// Store data into burst buffer
-	if stream.burstSize > 0 {
-		if stream.burst == nil {
-			stream.burst = []byte{}
-		}
-		newBurst := append(stream.burst, data...)
-		if len(newBurst) > stream.burstSize {
-			newBurst = newBurst[len(newBurst)-stream.burstSize:]
-		}
-		stream.burst = newBurst
-	}
-
-	n = len(data)
-	err = nil
-
-	return
-}
-
-func (stream *Stream) Close() error {
-	stream.subscribersLock.RLock()
-	defer stream.subscribersLock.RUnlock()
-
-	for _, subscriber := range stream.subscribers {
-		if err := subscriber.Close(); err != nil {
-			log.Println("WARNING: Failed to close subscriber stream, ignoring:", err)
-		}
-	}
-
-	return nil
-}

app/streams/stream_test.go

@@ -1,60 +0,0 @@
-package streams
-
-import (
-	"io"
-	"runtime"
-	"testing"
-
-	. "github.com/smartystreets/goconvey/convey"
-)
-
-func Test_Stream(t *testing.T) {
-	Convey("Stream", t, func() {
-		stream := NewStream(4)
-
-		// it writes burst prefill
-		n, err := stream.Write([]byte{4, 5, 6, 7})
-		So(n, ShouldEqual, 4)
-		So(err, ShouldBeNil)
-		So(stream.burst, ShouldResemble, []byte{4, 5, 6, 7})
-
-		// it writes normally
-		n, err = stream.Write([]byte{0, 1, 2})
-		So(n, ShouldEqual, 3)
-		So(err, ShouldBeNil)
-		So(stream.burst, ShouldResemble, []byte{7, 0, 1, 2})
-
-		// it has working subscriptions
-		r, w := io.Pipe()
-		stream.Subscribe(w)
-
-		//So(target, ShouldHaveLength, 4)
-		data := make([]byte, 128)
-		n, err = r.Read(data)
-		So(err, ShouldBeNil)
-		So(n, ShouldEqual, 4)
-		So(data[0:4], ShouldResemble, []byte{7, 0, 1, 2})
-
-		n, err = stream.Write([]byte{0, 0, 0, 0, 1, 0, 255, 0})
-		So(n, ShouldEqual, 8)
-		So(err, ShouldBeNil)
-
-		//So(target, ShouldHaveLength, 8)
-		n, err = r.Read(data)
-		So(err, ShouldBeNil)
-		So(n, ShouldEqual, 8)
-		So(data[0:8], ShouldResemble, []byte{0, 0, 0, 0, 1, 0, 255, 0})
-
-		runtime.Gosched()
-
-		r.Close()
-		n, err = r.Read(data)
-		So(err, ShouldEqual, io.ErrClosedPipe)
-		So(n, ShouldEqual, 0)
-
-		n, err = stream.Write([]byte{8})
-		So(n, ShouldEqual, 1)
-		So(err, ShouldBeNil)
-		So(stream.SubscriberCount(), ShouldEqual, 0)
-	})
-}

app/streams/streamreader.go

@@ -1,13 +0,0 @@
-package streams
-
-import (
-	"io"
-)
-
-func NewStreamReader(stream *Stream) io.ReadCloser {
-	r, w := io.Pipe()
-
-	stream.Subscribe(w)
-
-	return r
-}

main.go

@@ -1,119 +1,25 @@
 package main
 
 import (
-	"io"
 	"log"
-	"net/http"
-	"strconv"
-	"time"
 
-	"git.icedream.tech/icedream/uplink/app/streams"
+	"git.icedream.tech/icedream/uplink/app"
+	"git.icedream.tech/icedream/uplink/plugins/icecast/output"
 	"git.icedream.tech/icedream/uplink/plugins/test/sine"
-
-	humanize "github.com/dustin/go-humanize"
-	"github.com/gorilla/mux"
-	"github.com/viert/lame"
 )
 
 func main() {
-	stream := streams.NewStream(128 * 1024)
+	if err := run(); err != nil {
-
+		log.Fatal(err)
-	wr := lame.NewWriter(stream)
-	wr.Encoder.SetBitrate(192)
-	wr.Encoder.SetQuality(1)
-	wr.Encoder.SetInSamplerate(44100)
-	wr.Encoder.SetNumChannels(2)
-	wr.Encoder.InitParams()
-
-	go func() {
-		log.Println("Sine stream goroutine started")
-
-		sine := new(sine.SineStream)
-		sine.Samplerate = 44100
-		sine.Frequency = 990
-		sine.Beep = true
-		sine.Timestamp = time.Now()
-
-		log.Println("Will now broadcast sine stream")
-		n, err := io.Copy(wr, sine)
-		if err != nil {
-			log.Fatal("Sine stream copy failed:", err)
 	}
-		log.Println("Sine stream finished, written", humanize.Bytes(uint64(n)), "bytes")
+}
-	}()
+
-
+func run() (err error) {
-	server := new(http.Server)
+	backend := app.New()
-	mux := mux.NewRouter()
+	// backend.UsePlugin(icecast_input.Plugin)
-	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+	backend.UsePlugin(icecast_output.Plugin)
-		log.Println("Got a listener")
+	backend.UsePlugin(sine.Plugin)
-
+	backend.Init()
-		w.Header().Set("content-type", "audio/mpeg")
+	err = backend.Run()
-		w.Header().Set("server", "Uplink/0.0.0")
+	return
-		if r.Header.Get("icy-metadata") == "1" {
-			w.Header().Set("icy-metadata", "1")
-			w.Header().Set("icy-metaint", strconv.Itoa(2*1024))
-		}
-		w.WriteHeader(200)
-
-		cancel := w.(http.CloseNotifier).CloseNotify()
-
-		sr := streams.NewStreamReader(stream)
-		var n int64
-		var err error
-		if r.Header.Get("icy-metadata") == "1" {
-			mstream := streams.NewMetadataInjector(sr, 2*1024)
-			mstream.Metadata = map[string]string{
-				"StreamTitle": "beep",
-			}
-			go func() {
-				for {
-					select {
-					case <-cancel:
-						return
-					case <-time.After(time.Second):
-						mstream.Metadata["StreamTitle"] = "beep - time: " + time.Now().String()
-					}
-				}
-			}()
-			mstream.Metadata = map[string]string{
-				"StreamTitle": "DreamNetwork - Testing",
-			}
-			n, err = io.Copy(w, mstream)
-		} else {
-			n, err = io.Copy(w, sr)
-		}
-		log.Println("Transmitted", humanize.Bytes(uint64(n)))
-		if err != nil {
-			log.Println("Client transmission error:", err)
-		}
-
-		/*notify := w.(http.CloseNotifier).CloseNotify()
-		data := make([]byte, 4096)
-
-		log.Println("Start client tx loop")
-		for {
-			select {
-			case <-notify:
-				log.Println("Stop client tx loop")
-				sr.Close()
-				return
-			default:
-				n, err := sr.Read(data)
-				if err != nil {
-					log.Println("Read from stream failed:", err)
-					return
-				}
-				n, err = w.Write(data[0:n])
-				if err != nil {
-					log.Println("Write to client failed:", err)
-					log.Println("Stop client tx loop")
-					sr.Close()
-					return
-				}
-			}
-		}*/
-	})
-	server.Handler = mux
-	server.Addr = ":8080"
-	server.ListenAndServe()
 }

plugins/icecast/output/instance.go

@@ -1,18 +1,27 @@
 package icecast_output
 
 import (
+	"bytes"
+	"fmt"
 	"io"
+	"log"
+	"runtime"
 
 	"git.icedream.tech/icedream/uplink/app/authentication"
 	"git.icedream.tech/icedream/uplink/app/channels"
+	"git.icedream.tech/icedream/uplink/app/media"
 	"git.icedream.tech/icedream/uplink/app/servers/http"
+	"git.icedream.tech/icedream/uplink/app/streams"
+	humanize "github.com/dustin/go-humanize"
 	"github.com/gin-gonic/gin"
+	"github.com/glycerine/rbuf"
 )
 
 type pluginInstance struct {
 	server         *httpserver.Server
 	authenticator  authentication.Authenticator
 	channelManager *channels.ChannelManager
+	ringBuffers    map[string]map[string]*rbuf.FixedSizeRingBuf
 }
 
 func (instance *pluginInstance) SetAuthenticator(authenticator authentication.Authenticator) {
@@ -21,28 +30,108 @@ func (instance *pluginInstance) SetAuthenticator(authenticator authentication.Au
 
 func (instance *pluginInstance) SetChannelManager(channelManager *channels.ChannelManager) {
 	instance.channelManager = channelManager
+
+	go func() {
+		channelC := channelManager.Events().Sub("open")
+		log.Println("Burst cache: Now watching")
+		for c := range channelC {
+			channelId := c.(string)
+			go func(channel *channels.Channel) {
+				streamRbufMap := map[string]*rbuf.FixedSizeRingBuf{}
+				instance.ringBuffers[channelId] = streamRbufMap
+				outputContainerC := channel.Events.Sub("output_container")
+				log.Println("Burst cache: Now watching channel", channelId)
+				for c := range outputContainerC {
+					containerId := c.(string)
+					burstCache := rbuf.NewFixedSizeRingBuf(64 * 1024)
+					streamRbufMap[containerId] = burstCache
+					go func(container *media.MediaStreamContainer) {
+						r := container.Sub()
+						log.Println("Burst cache: Now watching container", containerId, "in channel", channelId)
+						io.Copy(burstCache, r)
+					}(channel.OutputContainers[containerId])
+					runtime.Gosched()
+				}
+			}(channelManager.Channel(channelId))
+			runtime.Gosched()
+		}
+	}()
+	runtime.Gosched()
+
+	// TODO - handle channel and container closure
 }
 
 func (instance *pluginInstance) SetServer(server *httpserver.Server) {
 	instance.server = server
+}
+
+func (instance *pluginInstance) Init() {
+	instance.ringBuffers = map[string]map[string]*rbuf.FixedSizeRingBuf{}
 
 	router := instance.server.Router
 
-	router.PUT("/:channel", func(ctx *gin.Context) {
+	router.GET("/:channel/:container", func(ctx *gin.Context) {
-		channel := instance.channelManager.Channel(ctx.Param("channel"))
+		r := ctx.Request
+		var mw *streams.MetadataInjector
+
+		channelId := ctx.Param("channel")
+		containerId := ctx.Param("container")
+		sendMetadata := r.Header.Get("icy-metadata") == "1"
+		metaInt := 16 * 1024
+
+		channel := instance.channelManager.Channel(channelId)
 		if channel == nil {
 			ctx.Status(404)
 			return
 		}
-		if user, password, ok := ctx.Request.BasicAuth(); ok {
+
-			if !instance.authenticator.VerifyUsernameAndPassword(channel, user, password) {
+		container, ok := channel.OutputContainers[containerId]
-				ctx.Status(401)
+		if !ok {
+			ctx.Status(404)
+		}
+
+		ctx.Writer.Header().Set("content-type", "audio/mpeg") // TODO
+		if sendMetadata {
+			ctx.Writer.Header().Set("icy-metadata", "1")
+			ctx.Writer.Header().Set("icy-metaint", fmt.Sprintf("%d", metaInt))
+		}
+		ctx.Writer.WriteHeader(200)
+
+		w := ctx.Writer
+		var nw io.Writer = w
+
+		sr := container.Sub()
+		defer sr.Close()
+
+		log.Println("Someone tuned in to", channelId, channel)
+
+		if sendMetadata {
+			mw = streams.NewMetadataInjector(w, metaInt)
+			nw = mw
+		}
+
+		if channelRbuf, ok := instance.ringBuffers[channelId]; ok {
+			if containerRbuf, ok := channelRbuf[containerId]; ok {
+				burst := containerRbuf.Bytes()
+				log.Println("Sending", humanize.Bytes(uint64(len(burst))), "burst")
+				_, err := io.Copy(nw, bytes.NewReader(burst))
+				if err != nil {
+					log.Println(err)
 					return
 				}
 			} else {
-			ctx.Status(401)
+				log.Println("No burst cache for", channelId, "/", containerId)
-			return
+			}
+		} else {
+			log.Println("No burst cache for", channelId)
+		}
+
+		_, err := io.Copy(nw, sr)
+		if err != nil {
+			log.Println(err)
 		}
-		io.Copy(channel.InputStream, ctx.Request.Body)
 	})
+
+	// TODO - output streams
+	// TODO - dynamic transcoding targets
 }

plugins/interfaces.go

@@ -1,12 +1,20 @@
 package plugins
 
 import (
+	"git.icedream.tech/icedream/uplink/app/authentication"
+	"git.icedream.tech/icedream/uplink/app/channels"
 	"git.icedream.tech/icedream/uplink/app/servers/http"
 )
 
 type PluginRunner func() PluginInstance
 
 type PluginInstance interface {
+	Init()
+}
+
+type AuthenticatorPlugin interface {
+	PluginInstance
+	SetAuthenticator(authentication.Authenticator)
 }
 
 type ServerPlugin interface {
@@ -16,5 +24,5 @@ type ServerPlugin interface {
 
 type ChannelPlugin interface {
 	PluginInstance
-	SetChannel(id string)
+	SetChannelManager(*channels.ChannelManager)
 }

plugins/test/sine/instance.go

@@ -11,9 +11,19 @@ import (
 )
 
 type pluginInstance struct {
+	channelManager *channels.ChannelManager
 }
 
 func (instance *pluginInstance) SetChannelManager(channelManager *channels.ChannelManager) {
+	instance.channelManager = channelManager
+}
+
+func (instance *pluginInstance) Init() {
+	channelManager := instance.channelManager
+
+	go func() {
+		time.Sleep(2 * time.Second) // give burst cache a chance to realize
+
 		c, err := channelManager.Open("sine")
 		if err != nil {
 			log.Println("ERROR: sine channel could not be opened:", err)
@@ -21,14 +31,36 @@ func (instance *pluginInstance) SetChannelManager(channelManager *channels.Chann
 			return
 		}
 
-	wr := lame.NewWriter(c.InputStream)
+		go func() {
+			lastTime := time.Now()
+			for {
+				lastTime = lastTime.Add(time.Second)
+				time.Sleep(time.Until(lastTime))
+
+				c.SetMetadata(map[string]string{
+					"StreamTitle": "beep - time: " + time.Now().String(),
+				})
+			}
+		}()
+
+		outputStream := c.AddOutputStream("mp3")
+		defer outputStream.Close()
+
+		outputContainer := c.AddOutputContainer("mp3")
+		defer outputContainer.Close()
+
+		w := io.MultiWriter(
+			outputContainer,
+			outputStream,
+		)
+
+		wr := lame.NewWriter(w)
 		wr.Encoder.SetBitrate(192)
 		wr.Encoder.SetQuality(1)
 		wr.Encoder.SetInSamplerate(44100)
 		wr.Encoder.SetNumChannels(2)
 		wr.Encoder.InitParams()
 
-	go func() {
 		log.Println("Sine stream goroutine started")
 
 		sine := new(SineStream)