diff --git a/pmolog/prettyxml.go b/pmolog/prettyxml.go
new file mode 100644
index 00000000..1c7ced96
--- /dev/null
+++ b/pmolog/prettyxml.go
@@ -0,0 +1,24 @@
+package pmolog
+
+import (
+ "bytes"
+ "encoding/xml"
+)
+
+func PrettyPrintXML(raw string) string {
+ var out bytes.Buffer
+ dec := xml.NewDecoder(bytes.NewReader([]byte(raw)))
+ enc := xml.NewEncoder(&out)
+ enc.Indent("", " ")
+ for {
+ t, err := dec.Token()
+ if err != nil {
+ break
+ }
+ if err := enc.EncodeToken(t); err != nil {
+ break
+ }
+ }
+ enc.Flush()
+ return out.String()
+}
diff --git a/soap/parseSoap.go b/soap/parseSoap.go
index f8cc4d65..66b4fc6b 100644
--- a/soap/parseSoap.go
+++ b/soap/parseSoap.go
@@ -48,24 +48,6 @@ type Fault struct {
// ----- Utils -----
-func prettyPrintXML(raw string) string {
- var out bytes.Buffer
- dec := xml.NewDecoder(bytes.NewReader([]byte(raw)))
- enc := xml.NewEncoder(&out)
- enc.Indent("", " ")
- for {
- t, err := dec.Token()
- if err != nil {
- break
- }
- if err := enc.EncodeToken(t); err != nil {
- break
- }
- }
- enc.Flush()
- return out.String()
-}
-
// ----- Parseurs -----
func ParseSOAPEnvelope(body []byte) (*Envelope, error) {
diff --git a/upnp/devices/services/statevariables/statevalueinstance.go b/upnp/devices/services/statevariables/statevalueinstance.go
index 7c06b518..2b465211 100644
--- a/upnp/devices/services/statevariables/statevalueinstance.go
+++ b/upnp/devices/services/statevariables/statevalueinstance.go
@@ -12,8 +12,14 @@ import (
"github.com/beevik/etree"
"github.com/google/uuid"
+ log "github.com/sirupsen/logrus"
)
+type notifiable interface {
+ Name() string
+ EventToBeSent(name string, value interface{})
+}
+
type StateVarInstance struct {
model *StateVariable
name string
@@ -27,12 +33,12 @@ type StateVarInstance struct {
sendEvents bool
parse StringValueParser
marshal ValueSerializer
-
- value interface{}
- previousValue interface{}
- lastChange time.Time
- lastEvent time.Time
- mu sync.RWMutex
+ service notifiable
+ value interface{}
+ previousValue interface{}
+ lastChange time.Time
+ lastEvent time.Time
+ mu sync.RWMutex
}
func (instance *StateVarInstance) Name() string {
@@ -102,13 +108,11 @@ func (instance *StateVarInstance) ParseValue(value string) (interface{}, error)
// IsValueInRange checks if a value falls within the defined range.
// Always returns true if no range is set.
-
// Parameters:
-
-// value: Value to check
-
+//
+// value: Value to check
+//
// Returns:
-
// bool: True if within range or no range defined
func (instance *StateVarInstance) IsValueInRange(value interface{}) (bool, error) {
return instance.model.valueType.InRange(value, instance.valueRange)
@@ -186,27 +190,73 @@ func (instance *StateVarInstance) SetValue(val interface{}) error {
return err
}
+ if ok, err := instance.IsValidValue(cval); !ok || err != nil {
+ if err != nil {
+ return err
+ }
+ return fmt.Errorf("Not valid value %v for variable %s", cval, instance.Name())
+ }
+
instance.mu.Lock()
defer instance.mu.Unlock()
instance.previousValue = instance.value
instance.value = cval
+
+ if instance.ShouldTriggerEvent() {
+ instance.service.EventToBeSent(instance.Name(), instance.Value())
+ }
+
return nil
}
-func (instance *StateVarInstance) Incr() {
- instance.mu.Lock()
- defer instance.mu.Unlock()
+func (instance *StateVarInstance) Incr() error {
+ if instance.HasStep() {
+ value, err := instance.model.valueType.Add(instance.Value(), instance.Step())
+ if err != nil {
+ return err
+ }
+ return instance.SetValue(value)
+ }
+ return fmt.Errorf(
+ "no step for variable %s:%s",
+ instance.service.Name(),
+ instance.Name(),
+ )
+}
+func (instance *StateVarInstance) Decr() error {
+ if instance.HasStep() {
+ value, err := instance.model.valueType.Sub(instance.Value(), instance.Step())
+ if err != nil {
+ return err
+ }
+ return instance.SetValue(value)
+ }
+ return fmt.Errorf(
+ "no step for variable %s:%s",
+ instance.service.Name(),
+ instance.Name(),
+ )
}
// ShouldTriggerEvent vérifie toutes les conditions
func (instance *StateVarInstance) ShouldTriggerEvent() bool {
- for _, condition := range instance.model.eventConditions {
- if !condition(instance) {
- return false
+ if instance.IsSendingEvents() {
+ for name, condition := range instance.model.eventConditions {
+ if !condition(instance) {
+ log.Debugf(
+ "State variable %s:%s event condition %s not true",
+ instance.service.Name(),
+ instance.Name(),
+ name,
+ )
+ return false
+ }
}
+
+ return true
}
- return true
+ return false
}
func (sv *StateVarInstance) GenerateEvent() *etree.Element {
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..410e384f 100644
--- a/upnp/serviceinstance.go
+++ b/upnp/serviceinstance.go
@@ -1,14 +1,21 @@
package upnp
import (
+ "context"
"fmt"
"io"
"net/http"
+ "net/url"
+ "strings"
+ "sync"
+ "time"
+ "gargoton.petite-maison-orange.fr/eric/pmomusic/pmolog"
"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 +27,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 +132,81 @@ 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
+ }
+
+ changed := make(map[string]interface{})
+ for name, sv := range svc.statevariables {
+ if sv.IsSendingEvents() { // sendEvents="yes"
+ changed[name] = sv.Value()
+ }
+ }
+
+ if len(changed) == 0 {
+ return
+ }
+
+ go func() {
+ callback = strings.TrimSpace(callback)
+ callback = strings.Trim(callback, "<>")
+
+ body := ``
+ for name, val := range changed {
+ body += fmt.Sprintf("<%s>%v", name, val, name)
+ }
+ body += ""
+
+ req, _ := http.NewRequest("NOTIFY", callback, strings.NewReader(body))
+ 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") // initial event
+
+ 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\n\n\n```xml\n%s\n```\n \n",
+ callback,
+ resp.Status,
+ pmolog.PrettyPrintXML(body),
+ )
+ }()
+}
+
func (svc *ServiceInstance) ToXMLElement() *etree.Element {
elem := etree.NewElement("service")
@@ -136,11 +228,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))
+ }
}
}