diff --git a/cmd/root.go b/cmd/root.go index 7962a0b3df..05ea34d3e2 100644 --- a/cmd/root.go +++ b/cmd/root.go @@ -5,7 +5,6 @@ import ( "os" "time" - "github.com/andig/evcc/core" "github.com/andig/evcc/provider" "github.com/andig/evcc/server" "github.com/andig/evcc/util" @@ -129,15 +128,6 @@ func checkVersion() { } } -// handle UI update requests -func handleUI(triggerChan <-chan struct{}, loadPoints []*core.LoadPoint) { - for range triggerChan { - for _, lp := range loadPoints { - lp.Update() - } - } -} - func run(cmd *cobra.Command, args []string) { util.LogLevel(viper.GetString("log")) log.INFO.Printf("evcc %s (%s)", server.Version, server.Commit) @@ -165,6 +155,9 @@ func run(cmd *cobra.Command, args []string) { // start broadcasting values tee := &Tee{} + cache := util.NewCache() + go cache.Run(tee.Attach()) + // setup influx if viper.Get("influx") != nil { influx := server.NewInfluxClient( @@ -190,13 +183,8 @@ func run(cmd *cobra.Command, args []string) { socketHub := server.NewSocketHub() httpd := server.NewHTTPd(uri, conf.Menu, loadPoints[0], socketHub) - triggerChan := make(chan struct{}) - - // handle UI update requests whenever browser connects - go handleUI(triggerChan, loadPoints) - // publish to UI - go socketHub.Run(tee.Attach(), triggerChan) + go socketHub.Run(tee.Attach(), cache) // setup values channel valueChan := make(chan util.Param) diff --git a/server/http.go b/server/http.go index 8c627b95c9..0e62ce8f46 100644 --- a/server/http.go +++ b/server/http.go @@ -73,7 +73,6 @@ func indexHandler(links []MenuConfig, liveAssets bool) http.HandlerFunc { _, debug := _escData["/js/debug.js"] return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - log.TRACE.Println("index") // w.Header().Set("Content-Type", "text/html; charset=UTF-8") // w.WriteHeader(http.StatusOK) diff --git a/server/socket.go b/server/socket.go index d9d2100a4a..830b4f1bf7 100644 --- a/server/socket.go +++ b/server/socket.go @@ -3,6 +3,7 @@ package server import ( "fmt" "net/http" + "strings" "time" "github.com/andig/evcc/util" @@ -98,13 +99,40 @@ func encode(v interface{}) (string, error) { return s, nil } -func (h *SocketHub) broadcast(i util.Param) { - if len(h.clients) > 0 { - val, err := encode(i.Val) - if err != nil { - log.FATAL.Fatal(err) +func kv(i util.Param) string { + val, err := encode(i.Val) + if err != nil { + log.FATAL.Fatal(err) + } + + return "\"" + i.Key + "\":" + val +} + +func (h *SocketHub) welcome(client *SocketClient, params []util.Param) { + var msg strings.Builder + + // build json object + _, _ = msg.WriteString("{") + for _, p := range params { + if msg.Len() > 1 { + _, _ = msg.WriteString(",") } - message := fmt.Sprintf("{\"%s\": %s}", i.Key, val) + msg.WriteString(kv(p)) + } + _, _ = msg.WriteString("}") + + // add client if send successful + select { + case client.send <- []byte(msg.String()): + h.clients[client] = true + default: + close(client.send) + } +} + +func (h *SocketHub) broadcast(p util.Param) { + if len(h.clients) > 0 { + message := fmt.Sprintf("{%s}", kv(p)) for client := range h.clients { select { @@ -117,13 +145,17 @@ func (h *SocketHub) broadcast(i util.Param) { } } +// Cacher gives access to current cache state +type Cacher interface { + All() []util.Param +} + // Run starts data and status distribution -func (h *SocketHub) Run(in <-chan util.Param, triggerChan chan<- struct{}) { +func (h *SocketHub) Run(in <-chan util.Param, cache Cacher) { for { select { case client := <-h.register: - h.clients[client] = true - triggerChan <- struct{}{} // trigger loadpoint update + h.welcome(client, cache.All()) case client := <-h.unregister: if _, ok := h.clients[client]; ok { close(client.send) diff --git a/util/cache.go b/util/cache.go new file mode 100644 index 0000000000..513aef5360 --- /dev/null +++ b/util/cache.go @@ -0,0 +1,58 @@ +package util + +import "sync" + +// Cache is a data store +type Cache struct { + sync.Mutex + val map[string]interface{} +} + +// NewCache creates cache +func NewCache() *Cache { + return &Cache{val: make(map[string]interface{})} +} + +// Run adds input channel's values to cache +func (c *Cache) Run(in <-chan Param) { + for p := range in { + c.Add(p.Key, p) + } +} + +// All provides a copy of the cached values +func (c *Cache) All() []Param { + c.Lock() + defer c.Unlock() + + copy := make([]Param, 0, len(c.val)) + for _, v := range c.val { + if param, ok := v.(Param); ok { + copy = append(copy, param) + } + } + + return copy +} + +// Add entry to cache +func (c *Cache) Add(key string, param Param) { + c.Lock() + defer c.Unlock() + + c.val[key] = param +} + +// Get entry from cache +func (c *Cache) Get(key string) Param { + c.Lock() + defer c.Unlock() + + if val, ok := c.val[key]; ok { + if param, ok := val.(Param); ok { + return param + } + } + + return Param{} +}