S.O.N.G
/
MeteorLight
3
0
Fork 0
Code Issues Pull Requests Packages Releases Activity

Refactor queue / mount sections and split into multiple files and interfaces
continuous-integration/drone/push Build is passing Details

Browse Source
master
DataHoarder 7 months ago
parent 143bb916a8
commit b8610799c8
Signed by: DataHoarder
SSH Key Fingerprint: SHA256:OLTRf6Fl87G52SiR7sWLGNzlJt4WOX+tfI2yxo0z7xk
13 changed files with 1347 additions and 1064 deletions
Show Stats Download Patch File Download Diff File

17
MeteorLight.go
Unescape View File

@ -3,6 +3,9 @@ package main
import (
"flag"
"fmt"
"git.gammaspectra.live/S.O.N.G/MeteorLight/api"
"git.gammaspectra.live/S.O.N.G/MeteorLight/config"
"git.gammaspectra.live/S.O.N.G/MeteorLight/queue"
"git.gammaspectra.live/S.O.N.G/MeteorLight/util"
"log"
"net/http"
@ -15,22 +18,22 @@ func main() {
flag.Parse()
config, err := GetConfig(*configPath)
conf, err := config.GetConfig(*configPath)
if err != nil {
log.Panic(err)
}
var wg sync.WaitGroup
queue := NewQueue(config)
queueInstance := queue.NewQueue(conf)
wg.Add(1)
go func() {
defer wg.Done()
server := http.Server{
Addr: fmt.Sprintf("%s:%d", config.Radio.Host, config.Radio.Port),
Handler: http.HandlerFunc(queue.HandleRadioRequest),
Addr: fmt.Sprintf("%s:%d", conf.Radio.Host, conf.Radio.Port),
Handler: http.HandlerFunc(queueInstance.HandleRadioRequest),
}
server.SetKeepAlivesEnabled(false)
@ -45,9 +48,9 @@ func main() {
}
}()
api := NewAPI(config, queue)
apiInstance := api.NewAPI(conf, queueInstance)
api.Wait()
apiInstance.Wait()
wg.Wait()
queue.Wait()
queueInstance.Wait()
}

43
api.go → api/api.go
Unescape View File

@ -1,4 +1,4 @@
package main
package api
import (
"bytes"
@ -6,6 +6,9 @@ import (
"errors"
"fmt"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/queue"
config2 "git.gammaspectra.live/S.O.N.G/MeteorLight/config"
queue2 "git.gammaspectra.live/S.O.N.G/MeteorLight/queue"
"git.gammaspectra.live/S.O.N.G/MeteorLight/queue/track"
"io"
"log"
"net/http"
@ -16,13 +19,13 @@ import (
)
type API struct {
config *Config
queue *Queue
config *config2.Config
queue *queue2.Queue
wg sync.WaitGroup
nr *QueueTrackEntry
nr *track.Entry
}
func NewAPI(config *Config, queue *Queue) *API {
func NewAPI(config *config2.Config, queue *queue2.Queue) *API {
api := &API{
config: config,
queue: queue,
@ -38,9 +41,9 @@ func (a *API) Wait() {
a.wg.Wait()
}
func (a *API) getQueueEntryFromBody(body []byte) (*QueueTrackEntry, error) {
entry := &QueueTrackEntry{}
err := json.Unmarshal(body, &entry.original)
func (a *API) getQueueEntryFromBody(body []byte) (*track.Entry, error) {
entry := &track.Entry{}
err := json.Unmarshal(body, &entry.Original)
if err != nil {
return nil, err
}
@ -53,11 +56,11 @@ func (a *API) getQueueEntryFromBody(body []byte) (*QueueTrackEntry, error) {
var strVal string
var ok bool
if val, ok = entry.original["hash"]; ok && a.config.Queue.SongFetchUrl != "" {
if val, ok = entry.Original["hash"]; ok && a.config.Queue.SongFetchUrl != "" {
if strVal, ok = val.(string); ok {
entry.Path = a.config.Queue.SongFetchUrl + strVal
}
} else if val, ok = entry.original["path"]; ok {
} else if val, ok = entry.Original["path"]; ok {
if strVal, ok = val.(string); ok {
entry.Path = strVal
}
@ -70,16 +73,16 @@ func (a *API) getQueueEntryFromBody(body []byte) (*QueueTrackEntry, error) {
return nil, errors.New("could not create queue entry")
}
func (a *API) getFallbackTrack() (*QueueTrackEntry, error) {
func (a *API) getFallbackTrack() (*track.Entry, error) {
m := make(map[string]interface{})
m["path"] = a.config.Queue.FallbackPath
return &QueueTrackEntry{
return &track.Entry{
Path: a.config.Queue.FallbackPath,
original: m,
Original: m,
}, nil
}
func (a *API) getRandomTrack() (*QueueTrackEntry, error) {
func (a *API) getRandomTrack() (*track.Entry, error) {
response, err := (&http.Client{
Timeout: time.Second * 60,
@ -97,10 +100,10 @@ func (a *API) getRandomTrack() (*QueueTrackEntry, error) {
return a.getQueueEntryFromBody(body)
}
func (a *API) setNowRandom(nr *QueueTrackEntry) {
func (a *API) setNowRandom(nr *track.Entry) {
a.nr = nr
if a.config.Queue.NowRandom != "" {
jsonData, _ := json.Marshal(nr.original)
jsonData, _ := json.Marshal(nr.Original)
response, err := http.DefaultClient.Post(a.config.Queue.NowRandom, "application/json; charset=utf-8", bytes.NewReader(jsonData))
if err != nil {
log.Print(err)
@ -155,7 +158,7 @@ func (a *API) handleQueue() {
go func() {
defer a.wg.Done()
for np := range a.queue.NowPlaying {
jsonData, _ := json.Marshal(np.original)
jsonData, _ := json.Marshal(np.Original)
response, err := http.DefaultClient.Post(a.config.Queue.NowPlaying, "application/json; charset=utf-8", bytes.NewReader(jsonData))
if err != nil {
log.Print(err)
@ -220,7 +223,7 @@ func (a *API) listen() {
return
case "np":
if e := a.queue.GetNowPlaying(); e != nil {
jsonData, _ := json.Marshal(e.original)
jsonData, _ := json.Marshal(e.Original)
writer.Write(jsonData)
} else {
writer.Write([]byte{'{', '}'})
@ -229,7 +232,7 @@ func (a *API) listen() {
return
case "random":
if a.nr != nil {
jsonData, _ := json.Marshal(a.nr.original)
jsonData, _ := json.Marshal(a.nr.Original)
writer.Write(jsonData)
} else {
writer.Write([]byte{'{', '}'})
@ -243,7 +246,7 @@ func (a *API) listen() {
}
var blobs = make([]map[string]interface{}, 0, 1)
for _, e := range a.queue.GetQueue() {
blobs = append(blobs, e.original)
blobs = append(blobs, e.Original)
}
jsonData, _ := json.Marshal(blobs)
writer.Write(jsonData)

4
config.go → config/config.go
Unescape View File

@ -1,4 +1,4 @@
package main
package config
import (
"errors"
@ -6,6 +6,8 @@ import (
"strconv"
)
const MaxBufferSize = 10
type Config struct {
Api struct {
Host string `toml:"host"`

211
listener/aps1/aps1.go
Unescape View File

@ -0,0 +1,211 @@
package aps1
import (
"bytes"
"encoding/binary"
"encoding/json"
"errors"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/packetizer"
"git.gammaspectra.live/S.O.N.G/MeteorLight/listener"
"git.gammaspectra.live/S.O.N.G/MeteorLight/queue/metadata"
"git.gammaspectra.live/S.O.N.G/MeteorLight/util"
"sync/atomic"
)
type Listener struct {
information listener.Information
started atomic.Bool
streamStartOffset int64
streamSamplesBuffer int64
ctx *util.RequestDone
sliceWriter chan []byte
offsetStart bool
headerBytes []byte
}
func NewListener(info listener.Information, ctx *util.RequestDone, sliceWriter chan []byte, samplesToBuffer int64, offsetStart bool, channels, sampleRate int, mimeType string) (*Listener, map[string]string) {
headers := make(map[string]string)
headers["x-audio-packet-stream"] = "1"
headers["content-type"] = "application/x-audio-packet-stream"
headerBytes := new(bytes.Buffer)
binary.Write(headerBytes, binary.LittleEndian, int64(channels))
binary.Write(headerBytes, binary.LittleEndian, int64(sampleRate))
binary.Write(headerBytes, binary.LittleEndian, int32(len(mimeType)))
headerBytes.Write([]byte(mimeType))
return &Listener{
information: info,
ctx: ctx,
sliceWriter: sliceWriter,
offsetStart: offsetStart,
streamSamplesBuffer: samplesToBuffer,
streamStartOffset: -1,
headerBytes: headerBytes.Bytes(),
}, headers
}
func (l *Listener) Identifier() string {
return l.information.Identifier
}
func (l *Listener) Information() *listener.Information {
return &l.information
}
func (l *Listener) HasStarted() bool {
return l.started.Load()
}
func (l *Listener) Start(packets []packetizer.Packet) error {
if l.started.Swap(true) {
return nil
}
l.sliceWriter <- (&packetStreamFrame{
Type: Header,
Category: 0,
StartSampleNumber: 0,
DurationInSamples: 0,
Data: l.headerBytes,
}).Encode()
if len(packets) > 0 {
sampleBufferMin := packets[len(packets)-1].GetStartSampleNumber() - l.streamSamplesBuffer
for _, p := range packets {
if p.KeepMode() != packetizer.Discard || p.GetEndSampleNumber() >= sampleBufferMin {
if err := l.Write(p); err != nil {
return err
}
}
}
}
return nil
}
func (l *Listener) Write(packet packetizer.Packet) error {
if l.ctx.Done() {
return l.ctx.Error()
}
if metadataPacket, ok := packet.(*metadata.Packet); ok {
queueInfoBuf := make([]byte, binary.MaxVarintLen64)
n := binary.PutVarint(queueInfoBuf, int64(metadataPacket.TrackEntry.QueueIdentifier))
if len(l.sliceWriter) >= (cap(l.sliceWriter) - 1) {
l.ctx.Fail(errors.New("client ran out of writer buffer"))
return l.ctx.Error()
}
l.sliceWriter <- (&packetStreamFrame{
Type: TrackIdentifier,
Category: packet.Category(),
StartSampleNumber: packet.GetStartSampleNumber(),
DurationInSamples: packet.GetEndSampleNumber() - packet.GetStartSampleNumber(),
Data: queueInfoBuf[:n],
}).Encode()
if metadataBytes, err := json.Marshal(metadataPacket.TrackEntry.Metadata); err == nil {
if len(l.sliceWriter) >= (cap(l.sliceWriter) - 1) {
l.ctx.Fail(errors.New("client ran out of writer buffer"))
return l.ctx.Error()
}
l.sliceWriter <- (&packetStreamFrame{
Type: TrackMetadata,
Category: packet.Category(),
StartSampleNumber: packet.GetStartSampleNumber(),
DurationInSamples: packet.GetEndSampleNumber() - packet.GetStartSampleNumber(),
Data: metadataBytes,
}).Encode()
}
return nil
}
if len(l.sliceWriter) >= (cap(l.sliceWriter) - 1) {
l.ctx.Fail(errors.New("client ran out of writer buffer"))
return l.ctx.Error()
}
var frameType PacketStreamType
switch packet.KeepMode() {
case packetizer.KeepLast:
frameType = DataKeepLast
case packetizer.Keep:
frameType = DataKeep
case packetizer.GroupKeep:
frameType = DataGroupKeep
case packetizer.GroupDiscard:
frameType = DataGroupDiscard
case packetizer.Discard:
frameType = DataDiscard
default:
return errors.New("unknown KeepMode")
}
l.sliceWriter <- (&packetStreamFrame{
Type: frameType,
Category: packet.Category(),
StartSampleNumber: packet.GetStartSampleNumber(),
DurationInSamples: packet.GetEndSampleNumber() - packet.GetStartSampleNumber(),
Data: packet.GetData(),
}).Encode()
return nil
}
func (l *Listener) Close() {
close(l.sliceWriter)
}
type PacketStreamType uint64
// PacketStreamType The order of these fields is important and set on-wire protocol
const (
Header = PacketStreamType(iota)
DataKeepLast
DataKeep
DataGroupKeep
DataGroupDiscard
DataDiscard
TrackIdentifier
TrackMetadata
)
type packetStreamFrame struct {
Type PacketStreamType
Category int64
StartSampleNumber int64
DurationInSamples int64
//automatically filled based on Data
//Size uint64
Data []byte
}
func (p *packetStreamFrame) Encode() []byte {
buf := new(bytes.Buffer)
varBuf := make([]byte, binary.MaxVarintLen64)
n := binary.PutUvarint(varBuf, uint64(p.Type))
buf.Write(varBuf[:n])
n = binary.PutUvarint(varBuf, uint64(p.Category))
buf.Write(varBuf[:n])
n = binary.PutVarint(varBuf, p.StartSampleNumber)
buf.Write(varBuf[:n])
n = binary.PutVarint(varBuf, p.DurationInSamples)
buf.Write(varBuf[:n])
n = binary.PutUvarint(varBuf, uint64(len(p.Data)))
buf.Write(varBuf[:n])
buf.Write(p.Data)
return buf.Bytes()
}

162
listener/icy/icy.go
Unescape View File

@ -0,0 +1,162 @@
package icy
import (
"errors"
"fmt"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/packetizer"
"git.gammaspectra.live/S.O.N.G/MeteorLight/listener"
"git.gammaspectra.live/S.O.N.G/MeteorLight/queue/metadata"
"git.gammaspectra.live/S.O.N.G/MeteorLight/util"
"strconv"
"sync/atomic"
)
// icyInterval weird clients might not support other numbers than this
const icyInterval = 8192
type Listener struct {
information listener.Information
started atomic.Bool
streamStartOffset int64
streamSamplesBuffer int64
counter int
ctx *util.RequestDone
sliceWriter chan []byte
offsetStart bool
pendingMetadata map[string]string
}
func NewListener(info listener.Information, ctx *util.RequestDone, sliceWriter chan []byte, samplesToBuffer int64, offsetStart bool) (*Listener, map[string]string) {
headers := make(map[string]string)
headers["icy-metaint"] = strconv.Itoa(icyInterval)
return &Listener{
information: info,
ctx: ctx,
sliceWriter: sliceWriter,
offsetStart: offsetStart,
streamSamplesBuffer: samplesToBuffer,
streamStartOffset: -1,
pendingMetadata: make(map[string]string),
}, headers
}
func (l *Listener) Identifier() string {
return l.information.Identifier
}
func (l *Listener) Information() *listener.Information {
return &l.information
}
func (l *Listener) HasStarted() bool {
return l.started.Load()
}
func (l *Listener) Start(packets []packetizer.Packet) error {
if l.started.Swap(true) {
return nil
}
if len(packets) > 0 {
sampleBufferMin := packets[len(packets)-1].GetStartSampleNumber() - l.streamSamplesBuffer
for _, p := range packets {
if p.KeepMode() != packetizer.Discard || p.GetEndSampleNumber() >= sampleBufferMin {
if err := l.Write(p); err != nil {
return err
}
}
}
}
return nil
}
func (l *Listener) writeIcy() []byte {
packetContent := make([]byte, 1, 16*255+1)
for k, v := range l.pendingMetadata {
packetContent = append(packetContent, []byte(fmt.Sprintf("%s='%s';", k, v))...)
delete(l.pendingMetadata, k)
//shouldn't send multiple properties in same packet if we want working single quotes, wait until next ICY frame
break
}
contentLength := len(packetContent) - 1
if contentLength > 16*255 {
//cannot send long titles
return make([]byte, 1)
}
if (contentLength % 16) == 0 { //already padded
packetContent[0] = byte(contentLength / 16)
} else {
packetContent[0] = byte(contentLength/16) + 1
packetContent = append(packetContent, make([]byte, 16-(contentLength%16))...)
}
return packetContent
}
func (l *Listener) Write(packet packetizer.Packet) error {
if l.ctx.Done() {
return l.ctx.Error()
}
if metadataPacket, ok := packet.(*metadata.Packet); ok {
if len(metadataPacket.TrackEntry.Artist()) > 0 {
l.pendingMetadata["StreamTitle"] = fmt.Sprintf("%s - %s", metadataPacket.TrackEntry.Artist(), metadataPacket.TrackEntry.Title())
} else {
l.pendingMetadata["StreamTitle"] = metadataPacket.TrackEntry.Title()
}
if len(metadataPacket.TrackEntry.Metadata.Art) > 0 {
l.pendingMetadata["StreamURL"] = metadataPacket.TrackEntry.Metadata.Art
}
return nil
}
var p []byte
if offsetable, ok := packet.(packetizer.OffsetablePacket); l.offsetStart && ok {
if l.streamStartOffset <= -1 {
if offsetable.KeepMode() != packetizer.Keep {
l.streamStartOffset = offsetable.GetStartSampleNumber()
p = offsetable.GetDataOffset(l.streamStartOffset)
} else {
p = packet.GetData()
}
} else {
p = offsetable.GetDataOffset(l.streamStartOffset)
}
} else {
p = packet.GetData()
}
var data []byte
for len(p) > 0 {
length := icyInterval - l.counter
if length <= len(p) {
data = append(data, p[:length]...)
data = append(data, l.writeIcy()...)
l.counter = 0
p = p[length:]
} else {
data = append(data, p...)
l.counter += len(p)
p = p[:0]
}
}
if len(l.sliceWriter) >= (cap(l.sliceWriter) - 1) {
l.ctx.Fail(errors.New("client ran out of writer buffer"))
return l.ctx.Error()
}
l.sliceWriter <- data
return nil
}
func (l *Listener) Close() {
close(l.sliceWriter)
}

35
listener/listener.go
Unescape View File

@ -0,0 +1,35 @@
package listener
import "git.gammaspectra.live/S.O.N.G/Kirika/audio/packetizer"
type Information struct {
Identifier string `json:"identifier"`
Mount string `json:"mount"`
Path string `json:"path"`
Start int64 `json:"start"`
Headers []HeaderEntry `json:"headers"`
}
type HeaderEntry struct {
Name string `json:"name"`
Value string `json:"value"`
}
type Listener interface {
Identifier() string
Information() *Information
HasStarted() bool
Start(packets []packetizer.Packet) error
Write(packet packetizer.Packet) error
Close()
}
/*
type Listener struct {
Information ListenerInformation
Start func(packets []packetizer.Packet) error
Write func(packet packetizer.Packet) error
Close func()
}
*/

98
listener/plain/plain.go
Unescape View File

@ -0,0 +1,98 @@
package plain
import (
"errors"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/packetizer"
"git.gammaspectra.live/S.O.N.G/MeteorLight/listener"
"git.gammaspectra.live/S.O.N.G/MeteorLight/queue/metadata"
"git.gammaspectra.live/S.O.N.G/MeteorLight/util"
"sync/atomic"
)
type Listener struct {
information listener.Information
started atomic.Bool
streamStartOffset int64
streamSamplesBuffer int64
ctx *util.RequestDone
sliceWriter chan []byte
offsetStart bool
}
func NewListener(info listener.Information, ctx *util.RequestDone, sliceWriter chan []byte, samplesToBuffer int64, offsetStart bool) (*Listener, map[string]string) {
return &Listener{
information: info,
ctx: ctx,
sliceWriter: sliceWriter,
offsetStart: offsetStart,
streamSamplesBuffer: samplesToBuffer,
streamStartOffset: -1,
}, nil
}
func (l *Listener) Identifier() string {
return l.information.Identifier
}
func (l *Listener) Information() *listener.Information {
return &l.information
}
func (l *Listener) HasStarted() bool {
return l.started.Load()
}
func (l *Listener) Start(packets []packetizer.Packet) error {
if l.started.Swap(true) {
return nil
}
if len(packets) > 0 {
sampleBufferMin := packets[len(packets)-1].GetStartSampleNumber() - l.streamSamplesBuffer
for _, p := range packets {
if p.KeepMode() != packetizer.Discard || p.GetEndSampleNumber() >= sampleBufferMin {
if err := l.Write(p); err != nil {
return err
}
}
}
}
return nil
}
func (l *Listener) Write(packet packetizer.Packet) error {
if l.ctx.Done() {
return l.ctx.Error()
}
if _, ok := packet.(*metadata.Packet); ok {
return nil
}
if len(l.sliceWriter) >= (cap(l.sliceWriter) - 1) {
l.ctx.Fail(errors.New("client ran out of writer buffer"))
return l.ctx.Error()
}
if offsetable, ok := packet.(packetizer.OffsetablePacket); l.offsetStart && ok {
if l.streamStartOffset <= -1 {
if offsetable.KeepMode() != packetizer.Keep {
l.streamStartOffset = offsetable.GetStartSampleNumber()
l.sliceWriter <- offsetable.GetDataOffset(l.streamStartOffset)
} else {
l.sliceWriter <- packet.GetData()
}
} else {
l.sliceWriter <- offsetable.GetDataOffset(l.streamStartOffset)
}
} else {
l.sliceWriter <- packet.GetData()
}
return nil
}
func (l *Listener) Close() {
close(l.sliceWriter)
}

995
queue.go
Unescape View File

@ -1,995 +0,0 @@
package main
import (
"bytes"
"crypto/sha256"
"encoding/binary"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"git.gammaspectra.live/S.O.N.G/Kirika/audio"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/filter"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/format/guess"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/packetizer"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/queue"
"git.gammaspectra.live/S.O.N.G/Kirika/audio/replaygain"
"git.gammaspectra.live/S.O.N.G/MeteorLight/util"
"github.com/dhowden/tag"
"io"
"log"
"net/http"
"os"
"path"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
)
const maxBufferSize = 10
type QueueTrackEntry struct {
QueueIdentifier queue.QueueIdentifier
Path string
Metadata struct {
Title interface{} `json:"title"`
Album interface{} `json:"album"`
Artist interface{} `json:"artist"`
Art string `json:"art"`
ReplayGain struct {
TrackPeak float64 `json:"track_peak"`
TrackGain float64 `json:"track_gain"`
AlbumPeak float64 `json:"album_peak"`
AlbumGain float64 `json:"album_gain"`
} `json:"replay_gain,omitempty"`
}
reader io.ReadSeekCloser
source audio.Source
original map[string]interface{}
}
func (e *QueueTrackEntry) Title() string {
if strVal, ok := e.Metadata.Title.(string); ok {
return strVal
} else if intVal, ok := e.Metadata.Title.(int); ok {
return strconv.Itoa(intVal)
}
return ""
}
func (e *QueueTrackEntry) Album() string {
if strVal, ok := e.Metadata.Album.(string); ok {
return strVal
} else if intVal, ok := e.Metadata.Album.(int); ok {
return strconv.Itoa(intVal)
}
return ""
}
func (e *QueueTrackEntry) Artist() string {
if strVal, ok := e.Metadata.Artist.(string); ok {
return strVal
} else if intVal, ok := e.Metadata.Artist.(int); ok {
return strconv.Itoa(intVal)
}
return ""
}
func (e *QueueTrackEntry) Load() error {
if e.source != nil {
return nil
}
fileName := path.Base(e.Path)
if len(e.Path) > 4 && e.Path[:4] == "http" {
s, err := util.NewRangeReadSeekCloser(e.Path)
if err != nil {
return err
}
fileName = s.GetFileName()
runtime.SetFinalizer(s, (*util.RangeReadSeekCloser).Close)
e.reader = s
} else {
f, err := os.Open(e.Path)
if err != nil {
return err
}
runtime.SetFinalizer(f, (*os.File).Close)
e.reader = f
}
if e.reader == nil {
return errors.New("could not find stream opener")
}
meta, err := tag.ReadFrom(e.reader)
if err != nil {
err = nil
}
if _, err = e.reader.Seek(0, io.SeekStart); err != nil {
return err
}
decoders, err := guess.GetDecoders(e.reader, fileName)
if err != nil {
return err
}
source, err := guess.Open(e.reader, decoders)
if err != nil {
return err
}
if source == nil {
return fmt.Errorf("could not find decoder for %s", e.Path)
}
e.source = source
//apply tags found on file
if meta != nil {
if e.Title() == "" {
e.Metadata.Title = meta.Title()
}
if e.Album() == "" {
e.Metadata.Album = meta.Album()
}
if e.Artist() == "" {
e.Metadata.Artist = meta.Artist()
}
if e.Artist() == "" {
e.Metadata.Artist = meta.AlbumArtist()
}
tags := meta.Raw()
var strValue string
var value interface{}
var ok bool
getDb := func(strValue string) (ret float64) {
ret, _ = strconv.ParseFloat(strings.TrimSpace(strings.TrimSuffix(strings.ToLower(strValue), "db")), 64)
return
}
if e.Metadata.ReplayGain.TrackPeak == 0 {
if value, ok = tags["replaygain_track_gain"]; ok {
if strValue, ok = value.(string); ok {
e.Metadata.ReplayGain.TrackGain = getDb(strValue)
}
}
if value, ok = tags["replaygain_track_peak"]; ok {
if strValue, ok = value.(string); ok {
e.Metadata.ReplayGain.TrackPeak = getDb(strValue)
}
}
if value, ok = tags["replaygain_album_gain"]; ok {
if strValue, ok = value.(string); ok {
e.Metadata.ReplayGain.AlbumGain = getDb(strValue)
}
}
if value, ok = tags["replaygain_album_peak"]; ok {
if strValue, ok = value.(string); ok {
e.Metadata.ReplayGain.AlbumPeak = getDb(strValue)
}
}
}
}
return nil
}
type QueueMetadataPacket struct {
sampleNumber int64
TrackEntry *QueueTrackEntry
}
func (p *QueueMetadataPacket) KeepMode() packetizer.KeepMode {
return packetizer.KeepLast
}
func (p *QueueMetadataPacket) GetStartSampleNumber() int64 {
return p.sampleNumber
}
func (p *QueueMetadataPacket) GetEndSampleNumber() int64 {
return p.sampleNumber
}
func (p *QueueMetadataPacket) Category() int64 {
return -1
}
func (p *QueueMetadataPacket) GetData() []byte {
return nil
}
type Queue struct {
NowPlaying chan *QueueTrackEntry
QueueEmpty chan *QueueTrackEntry
duration atomic.Int64
durationError int64
audioQueue *queue.Queue
mounts []*StreamMount
queue []*QueueTrackEntry
mutex sync.RWMutex
config *Config
wg sync.WaitGroup
}
func NewQueue(config *Config) *Queue {
if config.Queue.SampleRate <= 0 {
config.Queue.SampleRate = 44100
}
sampleFormat := audio.SourceInt16
bitDepth := 16
switch config.Queue.SampleFormat {
case "f32", "float", "float32", "f32le":
sampleFormat = audio.SourceFloat32
bitDepth = 0
case "i32", "s32", "int32", "int", "s32le":
sampleFormat = audio.SourceInt32
bitDepth = 32
case "i16", "s16", "int16", "s16le":
sampleFormat = audio.SourceInt16
bitDepth = 16
}
if config.Queue.BitDepth > 0 {
bitDepth = config.Queue.BitDepth
}
q := &Queue{
NowPlaying: make(chan *QueueTrackEntry, 1),
QueueEmpty: make(chan *QueueTrackEntry),
config: config,
audioQueue: queue.NewQueue(sampleFormat, bitDepth, config.Queue.SampleRate, 2),
}
blocksPerSecond := 20
sources := filter.NewFilterChain(q.audioQueue.GetSource(), filter.NewBufferFilter(16), filter.NewRealTimeFilter(blocksPerSecond), filter.NewBufferFilter(maxBufferSize*blocksPerSecond)).Split(len(config.Streams))
for i, s := range q.config.Streams {
mount := NewStreamMount(sources[i], s)
if mount == nil {
log.Panicf("could not initialize %s\n", s.MountPath)
}
q.mounts = append(q.mounts, mount)
q.wg.Add(1)
go mount.Process(&q.wg)
}
return q
}
func (q *Queue) GetDuration() time.Duration {
return time.Duration(q.duration.Load())
}
func (q *Queue) Wait() {
q.wg.Wait()
close(q.NowPlaying)
}
func (q *Queue) AddTrack(entry *QueueTrackEntry, tail bool) error {
if err := entry.Load(); err != nil {
return err
}
startCallback := func(queue *queue.Queue, queueEntry *queue.QueueEntry) {
if e := q.Get(queueEntry.Identifier); e != nil { //is this needed?
log.Printf("now playing \"%s\": %s - %s (%s)\n", e.Path, e.Metadata.Title, e.Metadata.Artist, e.Metadata.Album)
q.NowPlaying <- e
for _, mount := range q.mounts {
_ = mount.MetadataQueue.Enqueue(&QueueMetadataPacket{
//TODO: carry sample rate error
sampleNumber: (q.duration.Load() * int64(queue.GetSampleRate())) / int64(time.Second),
TrackEntry: e,
})
}
} else {
log.Printf("now playing \"%s\": %s - %s (%s)\n", entry.Path, entry.Metadata.Title, entry.Metadata.Artist, entry.Metadata.Album)
}
}
endCallback := func(queue *queue.Queue, entry *queue.QueueEntry) {
}
removeCallback := func(queue *queue.Queue, entry *queue.QueueEntry) {
//TODO: carry sample rate error
q.duration.Add(int64((time.Second * time.Duration(entry.ReadSamples.Load())) / time.Duration(entry.Source.GetSampleRate())))
q.Remove(entry.Identifier)
q.HandleQueue()
}
q.mutex.Lock()
defer q.mutex.Unlock()
if q.config.Queue.Length > 0 && len(q.queue) >= q.config.Queue.Length {
return errors.New("queue too long")
}
source := entry.source
if q.config.Queue.ReplayGain {
if entry.Metadata.ReplayGain.TrackPeak != 0 {
source = replaygain.NewReplayGainFilter(entry.Metadata.ReplayGain.TrackGain, entry.Metadata.ReplayGain.TrackPeak, 0).Process(source)
} else {
source = replaygain.NewNormalizationFilter(5).Process(source)
}
}
if tail {
entry.QueueIdentifier = q.audioQueue.AddTail(source, startCallback, endCallback, removeCallback)
} else {
entry.QueueIdentifier = q.audioQueue.AddHead(source, startCallback, endCallback, removeCallback)
}
entry.original["queue_id"] = entry.QueueIdentifier
if tail || len(q.queue) == 0 {
q.queue = append(q.queue, entry)
} else {
q.queue = append(q.queue[:1], append([]*QueueTrackEntry{entry}, q.queue[1:]...)...)
}
return nil
}
func (q *Queue) HandleQueue() {
if q.audioQueue.GetQueueSize() == 0 {
if err := q.AddTrack(<-q.QueueEmpty, true); err != nil {
log.Printf("track addition error: \"%s\"", err)
//TODO: maybe fail after n tries
time.Sleep(time.Second)
q.HandleQueue()
}
}
}
func (q *Queue) GetQueue() (result []*QueueTrackEntry) {
q.mutex.RLock()
defer q.mutex.RUnlock()
if len(q.queue) > 1 {
result = make([]*QueueTrackEntry, len(q.queue)-1)
copy(result, q.queue[1:])
}
return
}
func (q *Queue) Get(identifier queue.QueueIdentifier) *QueueTrackEntry {
q.mutex.RLock()
defer q.mutex.RUnlock()
for _, e := range q.queue {
if e.QueueIdentifier == identifier {
return e
}
}
return nil
}
func (q *Queue) GetNowPlaying() *QueueTrackEntry {
if e := q.audioQueue.GetQueueHead(); e != nil {
return q.Get(e.Identifier)
}
return nil
}
func (q *Queue) SkipNowPlaying() bool {
if e := q.audioQueue.GetQueueHead(); e != nil {
return q.Remove(e.Identifier)
}
return false
}
func (q *Queue) GetIndex(index int) *QueueTrackEntry {
if e := q.audioQueue.GetQueueIndex(index + 1); e != nil {
return q.Get(e.Identifier)
}
return nil
}
func (q *Queue) GetHead() *QueueTrackEntry {
if e := q.audioQueue.GetQueueIndex(1); e != nil {
return q.Get(e.Identifier)
}
return nil
}
func (q *Queue) GetTail() *QueueTrackEntry {
if i, e := q.audioQueue.GetQueueTail(); i != 0 && e != nil {
return q.Get(e.Identifier)
}
return nil
}
func (q *Queue) Remove(identifier queue.QueueIdentifier) bool {
q.mutex.Lock()
for i, e := range q.queue {
if e.QueueIdentifier == identifier {
q.queue = append(q.queue[:i], q.queue[i+1:]...)
q.mutex.Unlock()
q.audioQueue.Remove(identifier)
if e.reader != nil {
e.reader.Close()
}
return true
}
}
q.mutex.Unlock()
q.audioQueue.Remove(identifier)
return false
}
func (q *Queue) RemoveListener(identifier string) bool {
q.mutex.RLock()
defer q.mutex.RUnlock()
for _, mount := range q.mounts {
if mount.RemoveListener(identifier) {
return true
}
}
return false
}
func (q *Queue) GetListeners() (listeners []*ListenerInformation) {
q.mutex.RLock()
defer q.mutex.RUnlock()
listeners = make([]*ListenerInformation, 0, 1)
for _, mount := range q.mounts {
listeners = append(listeners, mount.GetListeners()...)
}
return
}
type PacketStreamType uint64
// PacketStreamType The order of these fields is important and set on-wire protocol
const (
Header = PacketStreamType(iota)
DataKeepLast
DataKeep
DataGroupKeep
DataGroupDiscard
DataDiscard
TrackIdentifier
TrackMetadata
)
type packetStreamFrame struct {
Type PacketStreamType
Category int64
StartSampleNumber int64
DurationInSamples int64
//automatically filled based on Data
//Size uint64
Data []byte
}
func (p *packetStreamFrame) Encode() []byte {
buf := new(bytes.Buffer)
varBuf := make([]byte, binary.MaxVarintLen64)
n := binary.PutUvarint(varBuf, uint64(p.Type))
buf.Write(varBuf[:n])
n = binary.PutUvarint(varBuf, uint64(p.Category))
buf.Write(varBuf[:n])
n = binary.PutVarint(varBuf, p.StartSampleNumber)
buf.Write(varBuf[:n])
n = binary.PutVarint(varBuf, p.DurationInSamples)
buf.Write(varBuf[:n])
n = binary.PutUvarint(varBuf, uint64(len(p.Data)))
buf.Write(varBuf[:n])
buf.Write(p.Data)
return buf.Bytes()
}
func (q *Queue) HandleRadioRequest(writer http.ResponseWriter, request *http.Request) {
writer.Header().Set("Server", "MeteorLight/radio")
writer.Header().Set("Connection", "close")
writer.Header().Set("X-Content-Type-Options", "nosniff")
writer.Header().Set("Access-Control-Allow-Origin", "*")
writer.Header().Set("Access-Control-Allow-Headers", "Accept, Content-Type, Icy-Metadata")
writer.Header().Set("Accept-Ranges", "none")
writer.Header().Set("Connection", "close")
if strings.HasSuffix(request.URL.Path, "mounts") {
writer.Header().Set("Content-Type", "application/json; charset=utf-8")
writer.Header().Set("Access-Control-Expose-Headers", "Accept-Ranges, Server, Content-Type")
type mountData struct {
Path string `json:"mount"`
MimeType string `json:"mime"`
FormatDescription string `json:"formatDescription"`
SampleRate int `json:"sampleRate"`
Channels int `json:"channels"`
Listeners int `json:"listeners"`
Options map[string]interface{} `json:"options"`
}
var mounts []mountData
for _, mount := range q.mounts {
mounts = append(mounts, mountData{
Path: strings.TrimSuffix(request.URL.Path, "mounts") + mount.Mount,
MimeType: mount.MimeType,
SampleRate: mount.SampleRate,
FormatDescription: mount.FormatDescription,
Channels: mount.Channels,
Listeners: len(mount.listeners),
Options: mount.Options,
})
}
jsonBytes, _ := json.MarshalIndent(mounts, "", " ")
writer.WriteHeader(http.StatusOK)
writer.Write(jsonBytes)
return
}
for _, mount := range q.mounts {
if strings.HasSuffix(request.URL.Path, mount.Mount) {
writer.Header().Set("Content-Type", mount.MimeType)
writer.Header().Set("Cache-Control", "no-store, max-age=604800")
writer.Header().Set("Access-Control-Expose-Headers", "Accept-Ranges, Server, Content-Type, Icy-MetaInt, X-Listener-Identifier")
writer.Header().Set("Vary", "*")
rangeHeader := request.Header.Get("range")
if rangeHeader != "" && rangeHeader != "bytes=0-" {
//TODO: maybe should fail in case bytes are requested
if strings.Index(request.UserAgent(), " Safari/") != -1 && mount.MimeType == "audio/flac" {
//Safari special case, fake Range check so it decodes afterwards.
//Safari creates a request with Range for 0-1, specifically for FLAC, and expects a result supporting range. Afterwards it requests the whole file.
//However the decoder is able to decode FLAC livestreams. If we fake the initial range response, then afterwards serve normal responses, Safari will happily work.
//TODO: remove this AS SOON as safari works on its own
//safariLargeFileValue arbitrary large value, cannot be that large or iOS Safari fails.
safariLargeFileValue := 1024 * 1024 * 1024 * 1024 * 16 // 16 TiB
if rangeHeader == "bytes=0-1" {
//first request
writer.Header().Set("Accept-Ranges", "bytes")
writer.Header().Set("Content-Range", fmt.Sprintf("bytes 0-1/%d", safariLargeFileValue)) //64 TiB max fake size
writer.Header().Set("Content-Length", "2")
writer.WriteHeader(http.StatusPartialContent)
writer.Write([]byte{'f', 'L'})
return
} else if rangeHeader == fmt.Sprintf("bytes=0-%d", safariLargeFileValue-1) {
//second request, serve status 200 to keep retries to a minimum
writer.Header().Set("Content-Length", fmt.Sprintf("%d", safariLargeFileValue))
writer.WriteHeader(http.StatusOK)
} else if strings.HasPrefix(rangeHeader, "bytes=") && strings.HasSuffix(rangeHeader, fmt.Sprintf("-%d", safariLargeFileValue-1)) {
//any other requests, these should fail
writer.Header().Set("Content-Range", fmt.Sprintf("bytes %s/%d", strings.TrimPrefix(rangeHeader, "bytes="), safariLargeFileValue))
writer.Header().Set("Accept-Ranges", "bytes")
writer.WriteHeader(http.StatusRequestedRangeNotSatisfiable)
return
}
}
}
bitrate := 0
if value, ok := mount.Options["bitrate"]; ok {
if intValue, ok := value.(int); ok {
bitrate = intValue
} else if int64Value, ok := value.(int64); ok {
bitrate = int(int64Value)
}
}
//set some audiocast/icy radio headers
writer.Header().Set("x-audiocast-name", q.config.Radio.Name)
writer.Header().Set("x-audiocast-bitrate", fmt.Sprintf("%d", bitrate))
writer.Header().Set("icy-name", q.config.Radio.Name)
writer.Header().Set("icy-version", "2")
writer.Header().Set("icy-index-metadata", "1")
if q.config.Radio.Description != "" {
writer.Header().Set("x-audiocast-description", q.config.Radio.Description)
writer.Header().Set("icy-description", q.config.Radio.Description)
}
if q.config.Radio.URL != "" {
writer.Header().Set("x-audiocast-url", q.config.Radio.URL)
writer.Header().Set("icy-url", q.config.Radio.URL)
}
if q.config.Radio.Logo != "" {
writer.Header().Set("icy-logo", q.config.Radio.Logo)
}
writer.Header().Set("icy-br", fmt.Sprintf("%d", bitrate))
writer.Header().Set("icy-sr", fmt.Sprintf("%d", mount.SampleRate))
writer.Header().Set("icy-audio-info", fmt.Sprintf("ice-channels=%d;ice-samplerate=%d;ice-bitrate=%d", mount.Channels, mount.SampleRate, bitrate))
if q.config.Radio.Private {
writer.Header().Set("icy-pub", "0")
writer.Header().Set("icy-do-not-index", "1")
writer.Header().Set("x-audiocast-public", "0")
writer.Header().Set("x-robots-tag", "noindex, nofollow")
} else {
writer.Header().Set("icy-pub", "1")
writer.Header().Set("icy-do-not-index", "0")
writer.Header().Set("x-audiocast-public", "1")
}
var packetWriteCallback func(packet packetizer.Packet) error
//buffer a bit, drop channels when buffer grows to not lock others. They will get disconnected elsewhere
const byteSliceChannelBuffer = 1024 * 16
writeChannel := make(chan []byte, byteSliceChannelBuffer)
requestDone := struct {
Done atomic.Bool
Lock sync.Mutex
Error error
}{}
var wgClient sync.WaitGroup
//set X-Audio-Packet-Stream for strictly timed packets and metadata
if numberValue, err := strconv.Atoi(request.Header.Get("x-audio-packet-stream")); err == nil && numberValue == 1 {
//version 1
writer.Header().Set("x-audio-packet-stream", "1")
writer.Header().Set("Content-Type", "application/x-audio-packet-stream")
packetWriteCallback = func(packet packetizer.Packet) error {
if requestDone.Done.Load() {
return requestDone.Error
}
if metadataPacket, ok := packet.(*QueueMetadataPacket); ok {
queueInfoBuf := make([]byte, binary.MaxVarintLen64)
n := binary.PutVarint(queueInfoBuf, int64(metadataPacket.TrackEntry.QueueIdentifier))
if len(writeChannel) >= (byteSliceChannelBuffer - 1) {
requestDone.Lock.Lock()
defer requestDone.Lock.Unlock()
requestDone.Error = errors.New("client ran out of buffer")
requestDone.Done.Store(true)
return requestDone.Error
}
writeChannel <- (&packetStreamFrame{
Type: TrackIdentifier,
Category: packet.Category(),
StartSampleNumber: packet.GetStartSampleNumber(),
DurationInSamples: packet.GetEndSampleNumber() - packet.GetStartSampleNumber(),
Data: queueInfoBuf[:n],
}).Encode()
if metadataBytes, err := json.Marshal(metadataPacket.TrackEntry.Metadata); err == nil {
if len(writeChannel) >= (byteSliceChannelBuffer - 1) {
requestDone.Lock.Lock()
defer requestDone.Lock.Unlock()
requestDone.Error = errors.New("client ran out of buffer")
requestDone.Done.Store(true)
return requestDone.Error
}
writeChannel <- (&packetStreamFrame{
Type: TrackMetadata,
Category: packet.Category(),
StartSampleNumber: packet.GetStartSampleNumber(),
DurationInSamples: packet.GetEndSampleNumber() - packet.GetStartSampleNumber(),
Data: metadataBytes,
}).Encode()
}
return nil
}
if len(writeChannel) >= (byteSliceChannelBuffer - 1) {
requestDone.Lock.Lock()
defer requestDone.Lock.Unlock()
requestDone.Error = errors.New("client ran out of buffer")
requestDone.Done.Store(true)
return requestDone.Error
}
var frameType PacketStreamType
switch packet.KeepMode() {
case packetizer.KeepLast:
frameType = DataKeepLast
case packetizer.Keep:
frameType = DataKeep
case packetizer.GroupKeep:
frameType = DataGroupKeep
case packetizer.GroupDiscard:
frameType = DataGroupDiscard
case packetizer.Discard:
frameType = DataDiscard
default:
return errors.New("unknown KeepMode")
}
writeChannel <- (&packetStreamFrame{
Type: frameType,
Category: packet.Category(),