diff --git a/upnp/server.go b/upnp/server.go index 8419401c..32b70f88 100644 --- a/upnp/server.go +++ b/upnp/server.go @@ -131,8 +131,11 @@ func (s *Server) Run(ctx context.Context) error { } for d := range s.devices.All() { - log.Infof("coucou from %s", d.Name()) d.RegisterSSPD() + + for svc := range d.services.All() { + svc.StartNotifier(ctx, 1*time.Second) + } } // attente d’annulation du contexte diff --git a/upnp/service.go b/upnp/service.go index f4e148b8..21dcb5cb 100644 --- a/upnp/service.go +++ b/upnp/service.go @@ -81,6 +81,10 @@ func (svc *Service) NewInstance() *ServiceInstance { statevariables: make(sv.StateVarInstanceSet), actions: make(actions.ActionInstanceSet), + + subscribers: make(map[string]string), + changedBuffer: make(map[string]interface{}), + seqid: make(map[string]uint32), } for v := range svc.stateTable.All() { diff --git a/upnp/serviceinstance.go b/upnp/serviceinstance.go index a3ffbf47..5063f936 100644 --- a/upnp/serviceinstance.go +++ b/upnp/serviceinstance.go @@ -1,14 +1,20 @@ package upnp import ( + "context" "fmt" "io" "net/http" + "net/url" + "strings" + "sync" + "time" "gargoton.petite-maison-orange.fr/eric/pmomusic/soap" "gargoton.petite-maison-orange.fr/eric/pmomusic/upnp/devices/services/actions" "gargoton.petite-maison-orange.fr/eric/pmomusic/upnp/devices/services/statevariables" "github.com/beevik/etree" + "github.com/google/uuid" log "github.com/sirupsen/logrus" ) @@ -20,8 +26,18 @@ type ServiceInstance struct { device *DeviceInstance statevariables statevariables.StateVarInstanceSet actions actions.ActionInstanceSet + subscribers map[string]string // SID → Callback URL + changedBuffer map[string]interface{} + seqid map[string]uint32 + cbmu sync.Mutex + mu sync.Mutex } +const ( + MethodSubscribe = "SUBSCRIBE" + MethodUnsubscribe = "UNSUBSCRIBE" +) + func (si *ServiceInstance) Name() string { return si.name } @@ -115,6 +131,71 @@ func (svc *ServiceInstance) SPCDElement() *etree.Element { return elem } +func (svc *ServiceInstance) AddSubscriber(sid, callback string) { + svc.mu.Lock() + defer svc.mu.Unlock() + if svc.subscribers == nil { + svc.subscribers = make(map[string]string) + } + svc.subscribers[sid] = callback +} + +func (svc *ServiceInstance) RenewSubscriber(sid, timeout string) { + // Pour l'instant juste log, on peut étendre avec expiration + svc.mu.Lock() + defer svc.mu.Unlock() + log.Infof("Renewed SID %s for timeout %s", sid, timeout) +} + +func (svc *ServiceInstance) RemoveSubscriber(sid string) { + svc.mu.Lock() + defer svc.mu.Unlock() + delete(svc.subscribers, sid) +} + +// Envoi d'un événement initial (optionnel) +func (svc *ServiceInstance) SendInitialEvent(sid string) { + svc.mu.Lock() + callback := svc.subscribers[sid] + svc.mu.Unlock() + if callback == "" { + return + } + + // Ici tu peux construire un SOAP Event XML minimal + body := `Initial` + + callback = strings.TrimSpace(callback) + callback = strings.Trim(callback, "<>") // retire les < et > + + // parser pour valider + u, err := url.Parse(callback) + if err != nil { + log.Errorf("Invalid callback URL: %v", err) + return + } + + req, err := http.NewRequest("NOTIFY", u.String(), strings.NewReader(body)) + if err != nil { + log.Errorf("Failed to create NOTIFY request: %v", err) + return + } + + req.Header.Set("Content-Type", "text/xml; charset=\"utf-8\"") + req.Header.Set("NT", "upnp:event") + req.Header.Set("NTS", "upnp:propchange") + req.Header.Set("SID", sid) + req.Header.Set("SEQ", "0") + client := &http.Client{} + resp, err := client.Do(req) + if err != nil { + log.Errorf("Failed to send initial event to %s: %v", callback, err) + return + } + defer resp.Body.Close() + log.Infof("✅ Initial event sent to %s, status=%s", callback, resp.Status) +} + func (svc *ServiceInstance) ToXMLElement() *etree.Element { elem := etree.NewElement("service") @@ -136,11 +217,146 @@ func (svc *ServiceInstance) ToXMLElement() *etree.Element { return elem } +func (svc *ServiceInstance) EventToBeSent(name string, value interface{}) { + svc.cbmu.Lock() + defer svc.cbmu.Unlock() + + svc.changedBuffer[name] = value +} + +func (svc *ServiceInstance) nextSeq(sid string) string { + svc.mu.Lock() + defer svc.mu.Unlock() + + svc.seqid[sid]++ + return fmt.Sprintf("%d", svc.seqid[sid]) +} + +func (svc *ServiceInstance) NotifySubscribers() { + svc.cbmu.Lock() + if len(svc.subscribers) == 0 || len(svc.changedBuffer) == 0 { + svc.cbmu.Unlock() + return + } + + // Copier et réinitialiser le buffer + changed := svc.changedBuffer + svc.changedBuffer = make(map[string]interface{}) + svc.cbmu.Unlock() + + for sid, callback := range svc.subscribers { + go func(sid, callback string, changed map[string]interface{}) { + callback = strings.TrimSpace(callback) + callback = strings.Trim(callback, "<>") + + u, err := url.Parse(callback) + if err != nil { + log.Errorf("Invalid callback URL %s: %v", callback, err) + return + } + + body := `` + for name, val := range changed { + body += fmt.Sprintf("<%s>%v", name, val, name) + } + body += "" + + req, err := http.NewRequest("NOTIFY", u.String(), strings.NewReader(body)) + if err != nil { + log.Errorf("Failed to create NOTIFY request to %s: %v", callback, err) + return + } + + req.Header.Set("Content-Type", `text/xml; charset="utf-8"`) + req.Header.Set("NT", "upnp:event") + req.Header.Set("NTS", "upnp:propchange") + req.Header.Set("SID", sid) + req.Header.Set("SEQ", svc.nextSeq(sid)) + + client := &http.Client{} + resp, err := client.Do(req) + if err != nil { + log.Errorf("Failed to notify subscriber %s: %v", callback, err) + return + } + defer resp.Body.Close() + + log.Infof("✅ Notified subscriber %s of changes: %v", callback, changed) + }(sid, callback, changed) + } +} + +func (svc *ServiceInstance) StartNotifier(ctx context.Context, interval time.Duration) { + log.Infof("✅ Starting notifier for %s:%s every %.2f s", svc.device.Name(), svc.Name(), interval.Seconds()) + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + log.Infof("✅ Notifier stopped for %s:%s", svc.device.Name(), svc.Name()) + return + case <-ticker.C: + svc.NotifySubscribers() + } + } + }() +} + func (svc *ServiceInstance) EventSubHandler() func(w http.ResponseWriter, r *http.Request) { return func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - log.Infof("Event Subscription handler for %s:%s", svc.device.Name(), svc.Name()) - // corps vide volontairement + log.Infof("📡 Event Subscription request for %s:%s", svc.device.Name(), svc.Name()) + + sid := r.Header.Get("SID") + timeout := r.Header.Get("Timeout") + callback := r.Header.Get("Callback") + + switch r.Method { + case MethodSubscribe: + if sid == "" { + // Nouvelle subscription + sid = fmt.Sprintf("uuid:%s", uuid.New().String()) + if callback != "" { + svc.AddSubscriber(sid, callback) + } + if timeout == "" { + timeout = "Second-1800" + } + log.Infof("🔔 New subscription: SID=%s, Callback=%s, Timeout=%s", sid, callback, timeout) + go svc.SendInitialEvent(sid) // envoyer l’état initial + } else { + // Renouvellement + svc.RenewSubscriber(sid, timeout) + log.Infof("♻️ Renew subscription: SID=%s, Timeout=%s", sid, timeout) + } + + w.Header().Set("SID", sid) + w.Header().Set("Timeout", timeout) + w.WriteHeader(http.StatusOK) + + case MethodUnsubscribe: + if sid != "" { + svc.RemoveSubscriber(sid) + log.Infof("❌ Unsubscribe SID=%s", sid) + } + w.WriteHeader(http.StatusOK) + + default: + log.Warnf("Unsupported EventSub method: %s", r.Method) + http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed) + } + + // Debug log headers + for k, v := range r.Header { + log.Debugf("Header: %s=%v", k, v) + } + + // Lire le body au besoin + body, err := io.ReadAll(r.Body) + if err == nil && len(body) > 0 { + log.Debugf("Body: %s", string(body)) + } } }