feat(ui): Structured observability/status view (#40)
* refactor(ui): Make message status SSE name more specific Signed-off-by: Richard Palethorpe <io@richiejp.com> * feat(ui): Add structured observability events Signed-off-by: Richard Palethorpe <io@richiejp.com> --------- Signed-off-by: Richard Palethorpe <io@richiejp.com>
This commit is contained in:
committed by
GitHub
parent
70e749b53a
commit
b5a12a1da6
87
core/agent/observer.go
Normal file
87
core/agent/observer.go
Normal file
@@ -0,0 +1,87 @@
|
||||
package agent
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/mudler/LocalAGI/core/sse"
|
||||
"github.com/mudler/LocalAGI/core/types"
|
||||
"github.com/mudler/LocalAGI/pkg/xlog"
|
||||
)
|
||||
|
||||
type Observer interface {
|
||||
NewObservable() *types.Observable
|
||||
Update(types.Observable)
|
||||
History() []types.Observable
|
||||
}
|
||||
|
||||
type SSEObserver struct {
|
||||
agent string
|
||||
maxID int32
|
||||
manager sse.Manager
|
||||
|
||||
mutex sync.Mutex
|
||||
history []types.Observable
|
||||
historyLast int
|
||||
}
|
||||
|
||||
func NewSSEObserver(agent string, manager sse.Manager) *SSEObserver {
|
||||
return &SSEObserver{
|
||||
agent: agent,
|
||||
maxID: 1,
|
||||
manager: manager,
|
||||
history: make([]types.Observable, 100),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SSEObserver) NewObservable() *types.Observable {
|
||||
id := atomic.AddInt32(&s.maxID, 1)
|
||||
return &types.Observable{
|
||||
ID: id - 1,
|
||||
Agent: s.agent,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SSEObserver) Update(obs types.Observable) {
|
||||
data, err := json.Marshal(obs)
|
||||
if err != nil {
|
||||
xlog.Error("Error marshaling observable", "error", err)
|
||||
return
|
||||
}
|
||||
msg := sse.NewMessage(string(data)).WithEvent("observable_update")
|
||||
s.manager.Send(msg)
|
||||
|
||||
s.mutex.Lock()
|
||||
defer s.mutex.Unlock()
|
||||
|
||||
for i, o := range s.history {
|
||||
if o.ID == obs.ID {
|
||||
s.history[i] = obs
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
s.history[s.historyLast] = obs
|
||||
s.historyLast += 1
|
||||
if s.historyLast >= len(s.history) {
|
||||
s.historyLast = 0
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SSEObserver) History() []types.Observable {
|
||||
h := make([]types.Observable, 0, 20)
|
||||
|
||||
s.mutex.Lock()
|
||||
defer s.mutex.Unlock()
|
||||
|
||||
for _, obs := range s.history {
|
||||
if obs.ID == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
h = append(h, obs)
|
||||
}
|
||||
|
||||
return h
|
||||
}
|
||||
Reference in New Issue
Block a user