diff --git a/assets/js/store.ts b/assets/js/store.ts index d82d12cb6..8e6224552 100644 --- a/assets/js/store.ts +++ b/assets/js/store.ts @@ -31,6 +31,7 @@ const initialState: State = { offline: false, loadpoints: [], vehicles: {}, + forecast: {}, }; const state = reactive(initialState); diff --git a/assets/js/types/evcc.ts b/assets/js/types/evcc.ts index 6a93ed0f5..4e48297c7 100644 --- a/assets/js/types/evcc.ts +++ b/assets/js/types/evcc.ts @@ -52,7 +52,7 @@ export interface State { offline: boolean; startup?: boolean; loadpoints: Loadpoint[]; - forecast?: Forecast; + forecast: Forecast; currency?: CURRENCY; fatal?: FatalError[]; authProviders?: AuthProviders; diff --git a/core/site_tariffs.go b/core/site_tariffs.go index 16a1cbe4e..fb92a8667 100644 --- a/core/site_tariffs.go +++ b/core/site_tariffs.go @@ -10,6 +10,7 @@ import ( "github.com/evcc-io/evcc/core/keys" "github.com/evcc-io/evcc/server/db/settings" "github.com/evcc-io/evcc/tariff" + "github.com/evcc-io/evcc/util" "github.com/jinzhu/now" "github.com/samber/lo" ) @@ -117,7 +118,7 @@ func (site *Site) publishTariffs(greenShareHome float64, greenShareLoadpoints fl fc.Solar = lo.ToPtr(site.solarDetails(solar)) } - site.publish(keys.Forecast, fc) + site.publish(keys.Forecast, util.NewSharder(keys.Forecast, fc)) } func (site *Site) solarDetails(solar api.Rates) solarDetails { diff --git a/server/socket.go b/server/socket.go index a11990d85..f096f9d38 100644 --- a/server/socket.go +++ b/server/socket.go @@ -2,6 +2,7 @@ package server import ( "context" + "encoding/json" "net/http" "strings" "sync" @@ -110,33 +111,59 @@ func (h *SocketHub) deleteSubscriber(s *socketSubscriber) { } func (h *SocketHub) welcome(subscriber *socketSubscriber, params []util.Param) { - var msg strings.Builder - msg.WriteString("{") + msg := make(map[string]json.RawMessage, len(params)) + for _, p := range params { - if msg.Len() > 1 { - msg.WriteString(",") + k := p.Key + if p.Loadpoint != nil { + k = "loadpoints." + p.UniqueID() } - msg.WriteString(kv(p)) + + msg[k] = json.RawMessage(socketEncode(p.Val)) } - msg.WriteString("}") + + b, _ := json.Marshal(msg) // should not block - subscriber.send <- []byte(msg.String()) + subscriber.send <- b } func (h *SocketHub) broadcast(p util.Param) { h.mu.RLock() defer h.mu.RUnlock() - if len(h.subscribers) > 0 { - msg := "{" + kv(p) + "}" + if len(h.subscribers) == 0 { + return + } - for s := range h.subscribers { - select { - case s.send <- []byte(msg): - default: - s.closeSlow() - } + msg := make(map[string]json.RawMessage) + + k := p.Key + if p.Loadpoint != nil { + k = "loadpoints." + p.UniqueID() + } + + // Sharder splits data into chunks + if sp, ok := (p.Val).(util.Sharder); ok { + shards := sp.Shards() + if len(shards) == 0 { + return // nothing changed, skip broadcast + } + + for _, shard := range shards { + msg[k+"."+shard.Key] = json.RawMessage(socketEncode(shard.Value)) + } + } else { + msg[k] = json.RawMessage(socketEncode(p.Val)) + } + + b, _ := json.Marshal(msg) + + for s := range h.subscribers { + select { + case s.send <- b: + default: + s.closeSlow() } } } diff --git a/server/socket_helper.go b/server/socket_helper.go index e16052e15..f59be6837 100644 --- a/server/socket_helper.go +++ b/server/socket_helper.go @@ -6,7 +6,6 @@ import ( "reflect" "strings" - "github.com/evcc-io/evcc/util" "github.com/evcc-io/evcc/util/encode" ) @@ -31,36 +30,22 @@ func encodeSliceAsString(v any) (string, error) { return fmt.Sprintf("[%s]", strings.Join(res, ",")), nil } -func kv(p util.Param) string { +func socketEncode(pval any) string { var ( val string err error ) // unwrap slices - if p.Val != nil && reflect.TypeOf(p.Val).Kind() == reflect.Slice { - val, err = encodeSliceAsString(p.Val) + if rv := reflect.ValueOf(pval); pval != nil && rv.Kind() == reflect.Slice && !rv.IsNil() { + val, err = encodeSliceAsString(pval) } else { - val, err = encodeAsString(p.Val) + val, err = encodeAsString(pval) } if err != nil { panic(err) } - if p.Key == "" && val == "" { - log.ERROR.Printf("invalid key/val for %+v, please report to https://github.com/evcc-io/evcc/issues/6439", p) - return "\"foo\":\"bar\"" - } - - var msg strings.Builder - msg.WriteString("\"") - if p.Loadpoint != nil { - msg.WriteString(fmt.Sprintf("loadpoints.%d.", *p.Loadpoint)) - } - msg.WriteString(p.Key) - msg.WriteString("\":") - msg.WriteString(val) - - return msg.String() + return val } diff --git a/tests/plan.evcc.yaml b/tests/plan.evcc.yaml index a20db16fc..344bbedca 100755 --- a/tests/plan.evcc.yaml +++ b/tests/plan.evcc.yaml @@ -1,4 +1,4 @@ -interval: 0.25s +interval: 0.1s site: title: Plan diff --git a/util/param.go b/util/param.go index 453fb3d67..252fdcc56 100644 --- a/util/param.go +++ b/util/param.go @@ -4,7 +4,6 @@ import ( "maps" "slices" "strconv" - "strings" "sync" "github.com/evcc-io/evcc/util/encode" @@ -19,15 +18,11 @@ type Param struct { // UniqueID returns unique identifier for parameter Loadpoint/Key combination func (p Param) UniqueID() string { - var b strings.Builder - if p.Loadpoint != nil { - b.WriteString(strconv.Itoa(*p.Loadpoint) + ".") + return strconv.Itoa(*p.Loadpoint) + "." + p.Key } - b.WriteString(p.Key) - - return b.String() + return p.Key } // ParamCache is a data store diff --git a/util/param_shard.go b/util/param_shard.go new file mode 100644 index 000000000..ac78fc297 --- /dev/null +++ b/util/param_shard.go @@ -0,0 +1,81 @@ +package util + +import ( + "crypto/sha256" + "encoding/json" + "fmt" + "strings" + "sync" + + "github.com/fatih/structs" +) + +// Sharder splits data into chunks, omitting unmodified chunks +type Sharder interface { + Shards() []Shard +} + +type Shard struct { + Key string + Value any +} + +type sharderImpl struct { + prefix string + struc any +} + +// shared shard cache +var ( + shardCache = make(map[string][32]byte) + shardMu sync.Mutex +) + +func (s *sharderImpl) MarshalJSON() ([]byte, error) { + return json.Marshal(s.struc) +} + +func (s *sharderImpl) Shards() []Shard { + ff := structs.Fields(s.struc) + res := make([]Shard, 0, len(ff)) + + shardMu.Lock() + defer shardMu.Unlock() + + for _, f := range ff { + key := f.Name() + if t := f.Tag("json"); t != "" { + if n := strings.Split(t, ",")[0]; n != "" { + key = n + } + } + + // Use JSON for stable hashing (fmt.Append includes pointer addresses) + b, err := json.Marshal(f.Value()) + if err != nil { + // Fallback to fmt.Append if JSON fails + b = fmt.Append(nil, f.Value()) + } + + hash := sha256.Sum256(b) + if cached, ok := shardCache[s.prefix+key]; ok && hash == cached { + continue + } + shardCache[s.prefix+key] = hash + + res = append(res, Shard{ + Key: key, + Value: f.Value(), + }) + } + + return res +} + +var _ Sharder = (*sharderImpl)(nil) + +// NewSharder creates a Sharder that splits structs into sub-structs for space-efficient socket publishing +// Passing anything else than a struct will panic +func NewSharder(prefix string, struc any) Sharder { + return &sharderImpl{prefix, struc} +} diff --git a/util/tee.go b/util/tee.go index 81973f75f..54375a58f 100644 --- a/util/tee.go +++ b/util/tee.go @@ -1,7 +1,6 @@ package util import ( - "reflect" "sync" "github.com/evcc-io/evcc/api" @@ -37,11 +36,12 @@ func (t *Tee) add(out chan<- Param) { // Run starts parameter distribution func (t *Tee) Run(in <-chan Param) { for msg := range in { - if val := reflect.ValueOf(msg.Val); val.Kind() == reflect.Ptr { - if ptr := reflect.Indirect(val); ptr.IsValid() { - msg.Val = ptr.Addr().Elem().Interface() - } - } + // TODO MUST NOT PUBLISH POINTERS (WHO'S VALUES ARE LATER MODIFIED) + // if val := reflect.ValueOf(msg.Val); val.Kind() == reflect.Ptr { + // if ptr := reflect.Indirect(val); ptr.IsValid() { + // fmt.Println("DANGER pointer value:", msg.Key) + // } + // } if val, ok := (msg.Val).(api.Redactor); ok { msg.Val = val.Redacted()