Introduce cache to immediately update UI when socket client connects (#176)
This commit is contained in:
parent
0490db10fa
commit
45b3ef33e4
4 changed files with 103 additions and 26 deletions
20
cmd/root.go
20
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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
58
util/cache.go
Normal file
58
util/cache.go
Normal file
|
|
@ -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{}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue