feat: internal MPD-compatible player streamer (AutoDJ) alpha release

Author: Datanoise

config/config.go

@@ -90,6 +90,11 @@ type Config struct {
 	DirectoryListing bool   `json:"directory_listing"`
 	DirectoryServer  string `json:"directory_server"`
 
+	// Internal Streamer (AutoDJ)
+	MPDEnabled bool   `json:"mpd_enabled"`
+	MPDPort    string `json:"mpd_port"`
+	MusicDir   string `json:"music_dir"`
+
 	// Multi-tenant
 	Users map[string]*User `json:"users"`
 }

relay/mpd.go

@@ -0,0 +1,145 @@
+package relay
+
+import (
+	"bufio"
+	"fmt"
+	"net"
+	"path/filepath"
+	"strings"
+
+	"github.com/sirupsen/logrus"
+)
+
+type MPDServer struct {
+	addr     string
+	streamer *Streamer
+	listener net.Listener
+}
+
+func NewMPDServer(addr string, s *Streamer) *MPDServer {
+	return &MPDServer{
+		addr:     addr,
+		streamer: s,
+	}
+}
+
+func (m *MPDServer) Start() error {
+	l, err := net.Listen("tcp", m.addr)
+	if err != nil {
+		return err
+	}
+	m.listener = l
+	logrus.Infof("MPD Server listening on %s", m.addr)
+
+	go func() {
+		for {
+			conn, err := l.Accept()
+			if err != nil {
+				return
+			}
+			go m.handleConnection(conn)
+		}
+	}()
+
+	return nil
+}
+
+func (m *MPDServer) Stop() {
+	if m.listener != nil {
+		m.listener.Close()
+	}
+}
+
+func (m *MPDServer) handleConnection(conn net.Conn) {
+	defer conn.Close()
+	writer := bufio.NewWriter(conn)
+	reader := bufio.NewReader(conn)
+
+		// Greeting
+		writer.WriteString("OK MPD 0.23.5\n")
+		writer.Flush()
+	
+		for {
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				return
+			}
+					line = strings.TrimSpace(line)
+					if line == "" {
+						continue
+					}
+			
+					cmd := strings.Split(line, " ")[0]
+					// args := strings.TrimPrefix(line, cmd+" ")
+			
+					switch cmd {			case "status":
+				m.handleStatus(writer)
+			case "currentsong":
+				m.handleCurrentSong(writer)
+			case "play":
+				m.streamer.Play()
+				writer.WriteString("OK\n")
+			case "stop":
+				m.streamer.Stop()
+				writer.WriteString("OK\n")
+			case "ping":
+				writer.WriteString("OK\n")
+			case "close":
+				return
+			case "listall", "listallinfo":
+				m.handleListAll(writer)
+			case "listplaylists":
+				writer.WriteString("OK\n")
+			case "lsinfo":
+				m.handleListAll(writer)
+			case "outputs":
+				fmt.Fprintf(writer, "outputid: 0\noutputname: %s\noutputenabled: 1\n", m.streamer.OutputMount)
+				writer.WriteString("OK\n")
+			case "config":
+				writer.WriteString("OK\n")
+			case "stats":
+				fmt.Fprintf(writer, "uptime: 0\nplaytime: 0\nartists: 0\nalbums: 0\nsongs: %d\n", len(m.streamer.Playlist))
+				writer.WriteString("OK\n")
+			default:
+				logrus.Debugf("MPD: Unknown command: %s", cmd)
+				writer.WriteString("OK\n") // Be lenient for now
+			}
+			writer.Flush()
+		}}
+
+func (m *MPDServer) handleStatus(w *bufio.Writer) {
+	state := "stop"
+	if m.streamer.State == StatePlaying {
+		state = "play"
+	} else if m.streamer.State == StatePaused {
+		state = "pause"
+	}
+
+	fmt.Fprintf(w, "state: %s\n", state)
+	fmt.Fprintf(w, "volume: 100\n")
+	fmt.Fprintf(w, "repeat: 0\n")
+	fmt.Fprintf(w, "random: 0\n")
+	fmt.Fprintf(w, "single: 0\n")
+	fmt.Fprintf(w, "consume: 0\n")
+	fmt.Fprintf(w, "playlist: 1\n")
+	fmt.Fprintf(w, "playlistlength: %d\n", len(m.streamer.Playlist))
+	w.WriteString("OK\n")
+}
+
+func (m *MPDServer) handleCurrentSong(w *bufio.Writer) {
+	m.streamer.mu.RLock()
+	defer m.streamer.mu.RUnlock()
+	fmt.Fprintf(w, "file: %s\n", m.streamer.CurrentFile)
+	fmt.Fprintf(w, "Title: %s\n", m.streamer.CurrentFile)
+	fmt.Fprintf(w, "Pos: %d\n", m.streamer.CurrentPos)
+	fmt.Fprintf(w, "Id: %d\n", m.streamer.CurrentPos)
+	w.WriteString("OK\n")
+}
+
+func (m *MPDServer) handleListAll(w *bufio.Writer) {
+	// Dummy for now, just list what's in the music dir or playlist
+	for _, f := range m.streamer.Playlist {
+		fmt.Fprintf(w, "file: %s\n", filepath.Base(f))
+	}
+	w.WriteString("OK\n")
+}

relay/streamer.go

@@ -0,0 +1,230 @@
+package relay
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/DatanoiseTV/tinyice/config"
+	"github.com/sirupsen/logrus"
+)
+
+type StreamerState int
+
+const (
+	StateStopped StreamerState = iota
+	StatePlaying
+	StatePaused
+)
+
+type Streamer struct {
+	Name        string
+	OutputMount string
+	MusicDir    string
+	Playlist    []string
+	CurrentPos  int
+	State       StreamerState
+	
+	relay       *Relay
+	cancel      context.CancelFunc
+	mu          sync.RWMutex
+
+	// Stats
+	BytesStreamed int64
+	CurrentFile   string
+}
+
+type StreamerManager struct {
+	instances map[string]*Streamer // key is OutputMount
+	mu        sync.RWMutex
+	relay     *Relay
+	config    *config.Config
+}
+
+func NewStreamerManager(r *Relay, cfg *config.Config) *StreamerManager {
+	return &StreamerManager{
+		instances: make(map[string]*Streamer),
+		relay:     r,
+		config:    cfg,
+	}
+}
+
+func (s *Streamer) Play() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.State = StatePlaying
+}
+
+func (s *Streamer) TogglePlay() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.State == StatePlaying {
+		s.State = StateStopped
+	} else {
+		s.State = StatePlaying
+	}
+}
+
+func (s *Streamer) Stop() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.State = StateStopped
+	if s.cancel != nil {
+		s.cancel()
+	}
+}
+
+func (s *Streamer) ScanMusicDir() error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.MusicDir == "" {
+		return fmt.Errorf("music directory not configured")
+	}
+
+	files, err := os.ReadDir(s.MusicDir)
+	if err != nil {
+		return err
+	}
+
+	s.Playlist = []string{}
+	for _, f := range files {
+		if !f.IsDir() && filepath.Ext(f.Name()) == ".mp3" {
+			s.Playlist = append(s.Playlist, filepath.Join(s.MusicDir, f.Name()))
+		}
+	}
+	return nil
+}
+
+func (sm *StreamerManager) StartStreamer(name, mount, musicDir string) (*Streamer, error) {
+	sm.mu.Lock()
+	defer sm.mu.Unlock()
+
+	if _, ok := sm.instances[mount]; ok {
+		return nil, fmt.Errorf("streamer for mount %s already exists", mount)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	s := &Streamer{
+		Name:        name,
+		OutputMount: mount,
+		MusicDir:    musicDir,
+		State:       StateStopped,
+		relay:       sm.relay,
+		cancel:      cancel,
+	}
+	sm.instances[mount] = s
+
+	go sm.runStreamerLoop(ctx, s)
+	return s, nil
+}
+
+func (sm *StreamerManager) GetStreamers() []*Streamer {
+	sm.mu.RLock()
+	defer sm.mu.RUnlock()
+	res := make([]*Streamer, 0, len(sm.instances))
+	for _, s := range sm.instances {
+		res = append(res, s)
+	}
+	return res
+}
+
+func (sm *StreamerManager) GetStreamer(mount string) *Streamer {
+	sm.mu.RLock()
+	defer sm.mu.RUnlock()
+	return sm.instances[mount]
+}
+
+func (sm *StreamerManager) runStreamerLoop(ctx context.Context, s *Streamer) {
+	logrus.Infof("Streamer %s starting for mount %s", s.Name, s.OutputMount)
+	
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			if s.State != StatePlaying {
+				time.Sleep(100 * time.Millisecond)
+				continue
+			}
+
+			s.mu.RLock()
+			if len(s.Playlist) == 0 {
+				s.mu.RUnlock()
+				time.Sleep(1 * time.Second)
+				continue
+			}
+			
+			if s.CurrentPos >= len(s.Playlist) {
+				s.CurrentPos = 0
+			}
+			filePath := s.Playlist[s.CurrentPos]
+			s.mu.RUnlock()
+
+			err := sm.streamFile(ctx, s, filePath)
+			if err != nil {
+				logrus.WithError(err).Errorf("Streamer %s: Failed to stream %s", s.Name, filePath)
+				time.Sleep(1 * time.Second)
+			}
+
+			s.mu.Lock()
+			s.CurrentPos++
+			s.mu.Unlock()
+		}
+	}
+}
+
+func (sm *StreamerManager) streamFile(ctx context.Context, s *Streamer, path string) error {
+	f, err := os.Open(path)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	s.mu.Lock()
+	s.CurrentFile = filepath.Base(path)
+	s.mu.Unlock()
+
+	// Update stream metadata
+	output := sm.relay.GetOrCreateStream(s.OutputMount)
+	output.CurrentSong = s.CurrentFile
+	output.Name = s.Name
+
+	// For now, we only support MP3 files and we assume they are compatible
+	// with our output settings.
+	// TODO: Add proper decoding/encoding if formats mismatch.
+	
+	buf := make([]byte, 8192)
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		default:
+			if s.State != StatePlaying {
+				return nil
+			}
+
+			n, err := f.Read(buf)
+			if n > 0 {
+				output.Broadcast(buf[:n], sm.relay)
+				atomic.AddInt64(&s.BytesStreamed, int64(n))
+				
+				// Basic pacing for MP3 (very rough)
+				// We should ideally use a proper pacer or decode to PCM to pace.
+				// For now, let's just assume 128kbps for pacing estimation.
+				time.Sleep(time.Duration(n) * 8 * time.Second / 128000)
+			}
+			if err == io.EOF {
+				return nil
+			}
+			if err != nil {
+				return err
+			}
+		}
+	}
+}