Introduce sample plugin code.

.gitignore

@@ -0,0 +1,43 @@
+# Compiled Object files, Static and Dynamic libs (Shared Objects)
+*.o
+*.a
+*.so
+
+# Folders
+_obj
+_test
+
+# Architecture specific extensions/prefixes
+*.[568vq]
+[568vq].out
+
+*.cgo1.go
+*.cgo2.c
+_cgo_defun.c
+_cgo_gotypes.go
+_cgo_export.*
+
+_testmain.go
+
+# built binaries
+uplink
+*.exe
+*.test
+
+###############################################################################
+
+# Windows image file caches
+Thumbs.db
+ehthumbs.db
+
+# Folder config file
+Desktop.ini
+
+# Recycle Bin used on file shares
+$RECYCLE.BIN/
+
+# Windows Installer files
+*.cab
+*.msi
+*.msm
+*.msp

app/authentication/authenticator.go

@@ -1,7 +1,7 @@
 package authentication
 
 import (
-	"git.icedream.tech/icedream/uplink/internal/channels"
+	"git.icedream.tech/icedream/uplink/app/channels"
 )
 
 type Authenticator interface {

app/authentication/dummy_authenticator.go

@@ -1,7 +1,7 @@
 package authentication
 
 import (
-	"git.icedream.tech/icedream/uplink/internal/channels"
+	"git.icedream.tech/icedream/uplink/app/channels"
 )
 
 type DummyAuthenticator struct{}

app/channels/channel.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"sync"
 
-	"git.icedream.tech/icedream/uplink/internal"
+	"git.icedream.tech/icedream/uplink/app/streams"
 )
 
 type Channel struct {
@@ -14,7 +14,7 @@ type Channel struct {
 	Name            string
 	Description     string
 	MimeType        string
-	InputStream     *internal.Stream
+	InputStream     *streams.Stream
 	OutputStreams   map[string]ChannelOutputStream
 }
 
@@ -35,20 +35,26 @@ func (channel *Channel) Metadata(ctx context.Context) <-chan map[string]string {
 	go func() {
 		for {
 			select {
-			case data := <-channel.metadataChannel:
+			case data, ok := <-channel.metadataChannel:
+				if !ok {
+					return
+				}
+				metadataChan <- data
 			case <-ctx.Done():
+				return
 			}
 		}
 	}()
+	return metadataChan
 }
 
 func NewChannel() *Channel {
 	return &Channel{
 		metadataChannel: make(chan map[string]string),
-		OutputStreams: map[string]ChannelOutputStream
+		OutputStreams:   map[string]ChannelOutputStream{},
 	}
 }
 
 type ChannelOutputStream struct {
-	*internal.Stream
+	*streams.Stream
 }

app/channels/channel_manager.go

@@ -8,9 +8,6 @@ import (
 type ChannelManager struct {
 	channels     map[string]*Channel
 	channelsLock sync.RWMutex
-
-	channelStreams     map[string]*ChannelStreams
-	channelStreamsLock sync.RWMutex
 }
 
 func (manager *ChannelManager) Channel(uuid string) *Channel {
@@ -25,25 +22,10 @@ func (manager *ChannelManager) Channel(uuid string) *Channel {
 	return channel
 }
 
-func (manager *ChannelManager) Streams(uuid string) *ChannelStreams {
-	manager.channelStreamsLock.RLock()
-	defer manager.channelStreamsLock.RUnlock()
-
-	streams, ok := manager.channelStreams[uuid]
-	if !ok {
-		return nil
-	}
-
-	return streams
-}
-
 func (manager *ChannelManager) Close(uuid string) (err error) {
 	manager.channelsLock.Lock()
 	defer manager.channelsLock.Unlock()
 
-	manager.channelStreamsLock.Lock()
-	defer manager.channelStreamsLock.Unlock()
-
 	_, ok := manager.channels[uuid]
 	if !ok {
 		err = errors.New("channel uuid is not known")
@@ -51,7 +33,6 @@ func (manager *ChannelManager) Close(uuid string) (err error) {
 	}
 
 	delete(manager.channels, uuid)
-	delete(manager.channelStreams, uuid)
 
 	return
 }
@@ -65,13 +46,8 @@ func (manager *ChannelManager) Open(uuid string) (channel *Channel, err error) {
 		return
 	}
 
-	manager.channelStreamsLock.Lock()
-	defer manager.channelStreamsLock.Unlock()
-
 	channel = new(Channel)
 	manager.channels[uuid] = channel
 
-	manager.channelStreams[uuid] = new(ChannelStreams)
-
 	return
 }

app/media/demuxed_stream.go

@@ -3,7 +3,7 @@ package media
 import (
 	"io"
 
-	"git.icedream.tech/icedream/uplink/internal/pubsub"
+	"git.icedream.tech/icedream/uplink/app/pubsub"
 )
 
 type DemuxedStream struct {

app/media/demuxer.go

@@ -4,7 +4,7 @@ import (
 	"io"
 	"log"
 
-	"git.icedream.tech/icedream/uplink/internal/pubsub"
+	"git.icedream.tech/icedream/uplink/app/pubsub"
 
 	"github.com/3d0c/gmf"
 )
@@ -59,6 +59,7 @@ func Demux(r io.ReadCloser) (demuxer *Demuxer) {
 		}
 		defer avioCtx.Release()
 		ctx.SetPb(avioCtx)
+		ctx.SetFlag(0x0080) // AVFMT_FLAG_CUSTOM_IO
 		ctx.OpenInput("")
 
 		// fmt.Println("=== FFMPEG DUMP OF INPUT ===")

app/media/demuxer_test.go

@@ -11,7 +11,7 @@ import (
 )
 
 func Test_Demux(t *testing.T) {
-	Convey("Demux", t, func() {
+	Convey("Demuxer", t, func() {
 		Convey("audio-only", func() {
 			reader, _ := os.Open("mpthreetest.mp3")
 			defer reader.Close()

app/media/init.go

@@ -2,11 +2,9 @@ package media
 
 import (
 	"runtime"
-
-	"github.com/3d0c/gmf"
 )
 
 func init() {
 	runtime.GOMAXPROCS(runtime.NumCPU())
-	gmf.LogSetLevel(gmf.AV_LOG_DEBUG)
+	// gmf.LogSetLevel(gmf.AV_LOG_DEBUG)
 }

app/media/muxer.go

@@ -1,8 +1,8 @@
 package media
 
 import (
+	"errors"
 	"io"
-	"log"
 	"reflect"
 
 	"github.com/3d0c/gmf"
@@ -39,15 +39,14 @@ func Mux(muxer string, readers ...io.ReadCloser) (retval io.Reader) {
 		defer output.Ctx.CloseOutputAndRelease()
 		if output.AvioCtx, err = gmf.NewAVIOContext(output.Ctx, &gmf.AVIOHandlers{
 			WritePacket: func(p []byte) {
-				log.Println("WritePacket:", p)
+				w.Write(p)
-				n, err := w.Write(p)
-				log.Println("WritePacket:", n, err)
 			},
 		}); err != nil {
 			return
 		}
 		defer output.AvioCtx.Release()
 		output.Ctx.SetPb(output.AvioCtx)
+		output.Ctx.SetFlag(0x0080) // AVFMT_FLAG_CUSTOM_IO
 
 		inputs := make([]Instance, len(readers))
 
@@ -75,6 +74,10 @@ func Mux(muxer string, readers ...io.ReadCloser) (retval io.Reader) {
 			}
 
 			inputs[i] = input
+			if input.Ctx.StreamsCnt() > 1 {
+				err = errors.New("Too many streams found in input")
+				return
+			}
 
 			var stream *gmf.Stream
 			if stream, err = input.Ctx.GetStream(0); err != nil {
@@ -99,9 +102,19 @@ func Mux(muxer string, readers ...io.ReadCloser) (retval io.Reader) {
 				Chan: reflect.ValueOf(c.Ctx.GetNewPackets()),
 			}
 		}
+		var closedStreamIndex = 0
+		defer func() {
+			for i, r := range readers {
+				if i == closedStreamIndex {
+					continue
+				}
+				r.Close()
+			}
+		}()
 		for err == nil {
 			streamIndex, packetVal, ok := reflect.Select(cases)
 			if !ok {
+				closedStreamIndex = streamIndex
 				break // some stream has been closed, just close them all
 			}
 			packet := packetVal.Interface().(*gmf.Packet)
@@ -109,10 +122,6 @@ func Mux(muxer string, readers ...io.ReadCloser) (retval io.Reader) {
 			err = output.Ctx.WritePacket(packet)
 			packet.Release()
 		}
-		log.Println("Bailing out")
-		for _, r := range readers {
-			r.Close()
-		}
 		output.Ctx.WriteTrailer()
 	}()
 

app/server.go

@@ -0,0 +1,11 @@
+package app
+
+import (
+	"git.icedream.tech/icedream/uplink/app/channels"
+	"github.com/gin-gonic/gin"
+)
+
+type Server struct {
+	*gin.Engine
+	*channels.ChannelManager
+}

app/servers/http/server.go

@@ -1,10 +1,10 @@
 package httpserver
 
 import (
-	_ "git.icedream.tech/icedream/uplink/internal"
+	_ "git.icedream.tech/icedream/uplink/app"
-	"git.icedream.tech/icedream/uplink/internal/authentication"
+	"git.icedream.tech/icedream/uplink/app/authentication"
-	channels "git.icedream.tech/icedream/uplink/internal/channels"
+	channels "git.icedream.tech/icedream/uplink/app/channels"
-	_ "git.icedream.tech/icedream/uplink/internal/transcoders"
+	_ "git.icedream.tech/icedream/uplink/app/transcoders"
 
 	"net/http"
 

app/streams/metadata_injector.go

@@ -1,4 +1,4 @@
-package internal
+package streams
 
 import (
 	"fmt"

app/streams/metadata_injector_test.go

@@ -1,4 +1,4 @@
-package internal
+package streams
 
 import (
 	"bytes"

app/streams/stream.go

@@ -1,4 +1,4 @@
-package internal
+package streams
 
 import (
 	"errors"

app/streams/stream_test.go

@@ -1,4 +1,4 @@
-package internal
+package streams
 
 import (
 	"io"

app/streams/streamreader.go

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

app/transcoders/instance.go

@@ -0,0 +1,12 @@
+package transcoders
+
+import (
+	"io"
+
+	"git.icedream.tech/icedream/uplink/app"
+)
+
+type TranscoderInstance interface {
+	io.WriteCloser
+	Init(out *app.Stream)
+}

app/transcoders/lame/transcoder_instance.go

@@ -3,9 +3,9 @@ package lametranscoder
 import (
 	"github.com/viert/lame"
 
-	"git.icedream.tech/icedream/uplink/internal"
+	"git.icedream.tech/icedream/uplink/app"
-	"git.icedream.tech/icedream/uplink/internal/transcoders"
+	"git.icedream.tech/icedream/uplink/app/transcoders"
-	"git.icedream.tech/icedream/uplink/internal/transcoders/options"
+	"git.icedream.tech/icedream/uplink/app/transcoders/options"
 )
 
 var transcoderOptions = map[string]options.TranscoderOptionType{
@@ -20,9 +20,7 @@ func (transcoder *Transcoder) Options() map[string]options.TranscoderOptionType
 }
 
 func (transcoder *Transcoder) New(options map[string]interface{}) transcoders.TranscoderInstance {
-	return &TranscoderInstance{
+	return nil
-		options: options,
-	}
 }
 
 type TranscoderInstance struct {
@@ -30,7 +28,7 @@ type TranscoderInstance struct {
 	*lame.LameWriter
 }
 
-func (instance *TranscoderInstance) Init(out *internal.Stream) {
+func (instance *TranscoderInstance) Init(out *app.Stream, samplerate int, channels int) {
 	instance.LameWriter = lame.NewWriter(out)
 	instance.LameWriter.Encoder.SetBitrate(int(instance.options["bitrate"].(int64)))
 	instance.LameWriter.Encoder.SetQuality(int(instance.options["quality"].(int64)))

app/transcoders/transcoder.go

@@ -0,0 +1,10 @@
+package transcoders
+
+import (
+	"git.icedream.tech/icedream/uplink/app/transcoders/options"
+)
+
+type Transcoder interface {
+	Options() map[string]options.TranscoderOptionType
+	New(options map[string]interface{}) *TranscoderInstance
+}

internal/streamreader.go

@@ -1,54 +0,0 @@
-package internal
-
-import (
-	"io"
-)
-
-type StreamReader struct {
-	dataChan   <-chan []byte
-	cancelChan chan<- interface{}
-	extraData  []byte
-}
-
-func NewStreamReader(stream *Stream) io.ReadCloser {
-
-	r, w := io.Pipe()
-
-	stream.Subscribe(w)
-
-	return r
-}
-
-func (reader *StreamReader) Close() error {
-	reader.cancelChan <- nil
-	return nil
-}
-
-func (reader *StreamReader) Read(data []byte) (n int, err error) {
-	n = 0
-	ok := false
-
-	// Do we have a buffer to read data from?
-	if reader.extraData == nil {
-		// Fill our buffer with new data.
-		reader.extraData, ok = <-reader.dataChan
-		if !ok { // EOF?
-			err = io.EOF
-			return
-		}
-	}
-
-	// Target array too small to fit all of our data? Keep the rest.
-	if len(reader.extraData) > len(data) {
-		copy(data, reader.extraData[0:len(data)])
-		reader.extraData = reader.extraData[len(data):]
-		n = len(data)
-		return
-	}
-
-	// Copy all of the buffer and reset the buffer.
-	copy(data, reader.extraData)
-	n = len(reader.extraData)
-	reader.extraData = nil
-	return
-}

internal/transcoders/instance.go

@@ -1,18 +0,0 @@
-package transcoders
-
-import (
-	"io"
-
-	"git.icedream.tech/icedream/uplink/internal"
-	"git.icedream.tech/icedream/uplink/internal/transcoders/options"
-)
-
-type Transcoder interface {
-	Options() map[string]options.TranscoderOptionType
-	New(options map[string]interface{}) *TranscoderInstance
-}
-
-type TranscoderInstance interface {
-	io.WriteCloser
-	Init(out *internal.Stream)
-}

internal/transcoders/packetizer.go

@@ -1 +0,0 @@
-package transcoders

main.go

@@ -7,8 +7,8 @@ import (
 	"strconv"
 	"time"
 
-	"git.icedream.tech/icedream/uplink/internal"
+	"git.icedream.tech/icedream/uplink/app/sources"
-	"git.icedream.tech/icedream/uplink/internal/sources"
+	"git.icedream.tech/icedream/uplink/app/streams"
 
 	humanize "github.com/dustin/go-humanize"
 	"github.com/gorilla/mux"
@@ -16,7 +16,7 @@ import (
 )
 
 func main() {
-	stream := internal.NewStream(128 * 1024)
+	stream := streams.NewStream(128 * 1024)
 
 	wr := lame.NewWriter(stream)
 	wr.Encoder.SetBitrate(192)
@@ -57,11 +57,11 @@ func main() {
 
 		cancel := w.(http.CloseNotifier).CloseNotify()
 
-		sr := internal.NewStreamReader(stream)
+		sr := streams.NewStreamReader(stream)
 		var n int64
 		var err error
 		if r.Header.Get("icy-metadata") == "1" {
-			mstream := internal.NewMetadataInjector(sr, 2*1024)
+			mstream := streams.NewMetadataInjector(sr, 2*1024)
 			mstream.Metadata = map[string]string{
 				"StreamTitle": "beep",
 			}

plugins/descriptor.go

@@ -0,0 +1,7 @@
+package plugins
+
+type PluginDescriptor struct {
+	Name        string
+	Version     string
+	Description string
+}

plugins/icecast/input/instance.go

@@ -0,0 +1,11 @@
+package main
+
+import "git.icedream.tech/icedream/uplink/app"
+
+type pluginInstance struct {
+	server *app.Server
+}
+
+func (instance *pluginInstance) SetServer(server *app.Server) {
+	instance.server = server
+}

plugins/icecast/input/plugin.go

@@ -0,0 +1,14 @@
+package main
+
+import (
+	"git.icedream.tech/icedream/uplink/plugins"
+)
+
+var Descriptor = plugins.PluginDescriptor{
+	Name:        "Icecast Input",
+	Description: "Allows for Icecast clients to stream to the server.",
+}
+
+func Run() *pluginInstance {
+	return &pluginInstance{}
+}