uplink/plugins/icecast/output/instance.go

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) {
	instance.authenticator = authenticator
}

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.GET("/:channel/:container", func(ctx *gin.Context) {
		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
		}

		container, ok := channel.OutputContainers[containerId]
		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 {
				log.Println("No burst cache for", channelId, "/", containerId)
			}
		} else {
			log.Println("No burst cache for", channelId)
		}

		_, err := io.Copy(nw, sr)
		if err != nil {
			log.Println(err)
		}
	})

	// TODO - output streams
	// TODO - dynamic transcoding targets
}