1 files changed, 166 insertions, 40 deletions

diff --git a/inform/server.go b/inform/server.go
index 7e33a86..710b8ac 100644
--- a/inform/server.go
+++ b/inform/server.go
@@ -1,55 +1,155 @@
 package inform
 
 import (
+        "container/list"
+        "errors"
         "fmt"
         "log"
         "net/http"
+        "sync"
         "time"
 )
 
-type StateTree struct {
-        states map[string]map[int]int
+func Log(handler http.Handler) http.Handler {
+        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+                handler.ServeHTTP(w, r)
+                t := time.Now().Format("02/Jan/2006 15:04:05")
+                log.Printf("%s - - [%s] \"%s %s %s\"", r.RemoteAddr, t, r.Method, r.URL, r.Proto)
+        })
 }
 
-func NewStateTree() *StateTree {
-        return &StateTree{make(map[string]map[int]int)}
+type Device struct {
+        Initialized  bool
+        CurrentState bool
+        DesiredState bool
+        *sync.Mutex
+}
+
+type InformHandler struct {
+        Codec *Codec
+        ports map[string]map[int]*Device
+        queue map[string]*list.List
+        *sync.RWMutex
+}
+
+func NewInformHandler(c *Codec) *InformHandler {
+        return &InformHandler{
+                Codec:   c,
+                ports:   make(map[string]map[int]*Device),
+                queue:   make(map[string]*list.List),
+                RWMutex: &sync.RWMutex{},
+        }
 }
 
-func (t *StateTree) ensureNode(device string, port int) {
-        _, ok := t.states[device]
+func (h *InformHandler) AddPort(dev string, port int) {
+        h.Lock()
+        defer h.Unlock()
+
+        _, ok := h.ports[dev]
         if !ok {
-                t.states[device] = make(map[int]int)
+                h.ports[dev] = make(map[int]*Device)
         }
 
-        _, ok = t.states[device][port]
+        _, ok = h.queue[dev]
         if !ok {
-                t.states[device][port] = 0
+                log.Printf("Adding queue for %s", dev)
+                h.queue[dev] = list.New()
+        }
+
+        log.Printf("Adding %s port %d", dev, port)
+        h.ports[dev][port] = &Device{
+                Mutex: &sync.Mutex{},
         }
 }
 
-func (t *StateTree) GetState(device string, port int) int {
-        t.ensureNode(device, port)
-        return t.states[device][port]
+func (h *InformHandler) getPort(dev string, port int) (*Device, error) {
+        h.RLock()
+        defer h.RUnlock()
+
+        _, ok := h.ports[dev]
+        if !ok {
+                return nil, errors.New("No device found")
+        }
+
+        p, ok := h.ports[dev][port]
+        if !ok {
+                return nil, errors.New("No port found")
+        }
+
+        return p, nil
 }
 
-func (t *StateTree) SetState(device string, port, value int) {
-        t.ensureNode(device, port)
-        t.states[device][port] = value
+func (h *InformHandler) SetState(dev string, port int, state bool) error {
+        p, err := h.getPort(dev, port)
+        if err != nil {
+                return err
+        }
+
+        p.Lock()
+        defer p.Unlock()
+
+        log.Printf("Set state to %t for %s port %d", state, dev, port)
+        p.DesiredState = state
+        return nil
 }
 
-func Log(handler http.Handler) http.Handler {
-        return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-                handler.ServeHTTP(w, r)
-                t := time.Now().Format("02/Jan/2006 15:04:05")
-                log.Printf("%s - - [%s] \"%s %s %s\"", r.RemoteAddr, t, r.Method, r.URL, r.Proto)
-                // Addr - - [D/M/Y H:M:S] "Method RequestURI Proto" Code Size
-                // 127.0.0.1 - - [24/Sep/2016 14:30:35] "GET / HTTP/1.1" 200 -
-        })
+func (h *InformHandler) buildCommands(dev string, pl *DeviceMessage) error {
+        for _, o := range pl.Outputs {
+                ds, err := h.getPort(dev, o.Port)
+                if err != nil {
+                        return err
+                }
+                ds.Lock()
+
+                // Get initial state
+                if !ds.Initialized {
+                        ds.CurrentState = o.OutputState
+                        ds.Initialized = true
+                        return nil
+                }
+
+                // State didn't change at the sensor
+                if ds.CurrentState == o.OutputState {
+                        if ds.DesiredState != o.OutputState {
+                                log.Printf("Toggle state %t for %s port %d", ds.DesiredState, dev, o.Port)
+                                // Generate change command
+                                // TODO: Don't lock the whole handler
+                                h.Lock()
+                                h.queue[dev].PushFront(NewOutputCommand(o.Port, ds.DesiredState, 0))
+                                h.Unlock()
+                        }
+                } else { // Sensor caused the change, leave it alone
+                        log.Printf("Sensor state changed %s port %d", dev, o.Port)
+                        ds.DesiredState = o.OutputState
+                }
+
+                ds.CurrentState = o.OutputState
+                ds.Unlock() // Don't hold the lock the entire loop
+        }
+
+        return nil
 }
 
-type InformHandler struct {
-        Codec     *Codec
-        StateTree *StateTree
+func (h *InformHandler) pop(dev string) *CommandMessage {
+        // TODO: Don't lock the whole handler
+        h.Lock()
+        defer h.Unlock()
+
+        q, ok := h.queue[dev]
+        if !ok {
+                log.Printf("No queue for %s", dev)
+                return nil
+        }
+
+        e := q.Front()
+        if e != nil {
+                h.queue[dev].Remove(e)
+                cmd := e.Value.(*CommandMessage)
+                cmd.Freshen()
+                return cmd
+        }
+
+        return nil
 }
 
 func (h *InformHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -58,35 +158,61 @@ func (h *InformHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
                 return
         }
 
-        if r.URL != nil && r.URL.Path != "/inform" {
-                http.Error(w, "404 Not Found", http.StatusNotFound)
+        msg, err := h.Codec.Unmarshal(r.Body)
+        if err != nil {
+                log.Printf("Unmarshal message: %s", err.Error())
+                http.Error(w, "400 Bad Request", http.StatusBadRequest)
                 return
         }
 
-        msg, err := h.Codec.Unmarshal(r.Body)
+        pl, err := msg.UnmarshalPayload()
         if err != nil {
-                http.Error(w, "Bad Request", http.StatusBadRequest)
+                log.Printf("Unmarshal payload: %s", err.Error())
+                http.Error(w, "400 Bad Request", http.StatusBadRequest)
                 return
         }
 
-        pl := NewInformWrapper()
-        copy(pl.MacAddr, msg.MacAddr)
-        pl.SetEncrypted(true)
-
-        // TODO: compare current state to tree and update
+        dev := msg.FormattedMac()
+        ret := NewInformWrapperResponse(msg)
+        log.Printf("Inform from %s", dev)
+
+        // Send a command until the queue is empty
+        if cmd := h.pop(dev); cmd != nil {
+                ret.UpdatePayload(cmd)
+        } else {
+                // Update internal state vs reality
+                if err := h.buildCommands(msg.FormattedMac(), pl); err != nil {
+                        http.Error(w, "500 Server Error", 500)
+                        return
+                }
+
+                // If that generated a command send it
+                if cmd = h.pop(dev); cmd != nil {
+                        ret.UpdatePayload(cmd)
+                } else {
+                        // Otherwise noop
+                        ret.UpdatePayload(NewNoop(10))
+                }
+        }
 
-        res, err := h.Codec.Marshal(pl)
+        res, err := h.Codec.Marshal(ret)
         if err != nil {
-                http.Error(w, "Server Error", 500)
+                http.Error(w, "500 Server Error", 500)
                 return
         }
 
         fmt.Fprintf(w, "%s", res)
 }
 
-func NewServer(handler *InformHandler) *http.Server {
+// Create a new server, returns the mux so users can add other methods (for
+// example, if they want to share a process to build a console that also
+// accepts informs)
+func NewServer(handler *InformHandler) (*http.Server, *http.ServeMux) {
+        mux := http.NewServeMux()
+        mux.Handle("/inform", handler)
+
         return &http.Server{
                 Addr:    ":6080",
-                Handler: Log(handler),
-        }
+                Handler: Log(mux),
+        }, mux
 }