pmomusic/pmostream/streamer.go

package pmostream

import (
	"encoding/binary"
	"log"
	"net/http"
	"sync"
	"time"

	"github.com/gopxl/beep"
)

// StreamManager gère plusieurs AudioProcessor et clients HTTP
type StreamManager struct {
	processors map[string]*AudioProcessor
	mu         sync.RWMutex
}

// NewStreamManager crée un gestionnaire de flux audio
func NewStreamManager() *StreamManager {
	return &StreamManager{
		processors: make(map[string]*AudioProcessor),
	}
}

// AddProcessor ajoute un processeur audio et démarre sa boucle Process
func (m *StreamManager) AddProcessor(id string, processor *AudioProcessor) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if _, exists := m.processors[id]; exists {
		log.Printf("Processor with id %s already exists", id)
		return
	}
	m.processors[id] = processor

	go func() {
		if err := processor.Process(); err != nil {
			log.Printf("Processor error for %s: %v", id, err)
		}
	}()
}

// RemoveProcessor arrête et supprime un processeur audio
func (m *StreamManager) RemoveProcessor(id string) {
	m.mu.Lock()
	defer m.mu.Unlock()

	if processor, exists := m.processors[id]; exists {
		processor.Stop()
		processor.Close()
		delete(m.processors, id)
	}
}

// GetHandler retourne un handler HTTP pour streamer un flux audio
func (m *StreamManager) GetHandler(id string) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		m.mu.RLock()
		processor, exists := m.processors[id]
		m.mu.RUnlock()

		if !exists {
			http.Error(w, "Stream not found", http.StatusNotFound)
			return
		}

		buffer := processor.GetBuffer()
		config := processor.config

		w.Header().Set("Content-Type", "audio/wav")
		w.Header().Set("Cache-Control", "no-cache")
		w.Header().Set("Connection", "keep-alive")
		w.Header().Set("Access-Control-Allow-Origin", "*")

		flusher, ok := w.(http.Flusher)
		if !ok {
			http.Error(w, "Streaming not supported", http.StatusInternalServerError)
			return
		}

		// Écrire l'en-tête WAV pour streaming
		if err := writeWavHeader(w, config.TargetSampleRate, config.Format); err != nil {
			log.Printf("Failed to write WAV header: %v", err)
			return
		}
		flusher.Flush()

		clientDone := r.Context().Done()

		for {
			select {
			case <-clientDone:
				log.Printf("Client disconnected: %s", id)
				return
			default:
			}

			// Attendre les données avec timeout
			if !buffer.WaitForData(1, 100*time.Millisecond) {
				continue
			}

			chunk := buffer.Read()
			if chunk == nil {
				continue
			}

			// Envoyer le chunk au client
			if _, err := w.Write(chunk); err != nil {
				log.Printf("Write error: %v", err)
				return
			}
			flusher.Flush()
		}
	}
}

// writeWavHeader écrit un en-tête WAV pour streaming
func writeWavHeader(w http.ResponseWriter, sampleRate beep.SampleRate, format SampleFormat) error {
	var audioFormat uint16 = 1 // PCM
	var bitsPerSample uint16 = 16

	if format == Float32 {
		audioFormat = 3 // IEEE_FLOAT
		bitsPerSample = 32
	}

	numChannels := uint16(2)
	blockAlign := numChannels * bitsPerSample / 8
	byteRate := uint32(sampleRate) * uint32(blockAlign)

	header := make([]byte, 44)
	copy(header[0:4], "RIFF")
	binary.LittleEndian.PutUint32(header[4:8], 0xFFFFFFFF) // Taille inconnue pour streaming
	copy(header[8:12], "WAVE")
	copy(header[12:16], "fmt ")
	binary.LittleEndian.PutUint32(header[16:20], 16)
	binary.LittleEndian.PutUint16(header[20:22], audioFormat)
	binary.LittleEndian.PutUint16(header[22:24], numChannels)
	binary.LittleEndian.PutUint32(header[24:28], uint32(sampleRate))
	binary.LittleEndian.PutUint32(header[28:32], byteRate)
	binary.LittleEndian.PutUint16(header[32:34], blockAlign)
	binary.LittleEndian.PutUint16(header[34:36], bitsPerSample)
	copy(header[36:40], "data")
	binary.LittleEndian.PutUint32(header[40:44], 0xFFFFFFFF) // Taille inconnue pour streaming

	_, err := w.Write(header)
	return err
}