hierarchical event topics and TCP support
Event topics now follow session hierarchy:
session.{sid}.new/kill/pause/resume/rename
session.{sid}.agent.{aid}.new/kill/state
Wildcard routing publishes to all ancestors:
session.{sid}.agent.{aid}.state publishes to:
session.{sid}.agent.{aid}.*
session.{sid}.agent.*
session.{sid}.*
session.*
*
GUI can subscribe to session.{sid}.* to get all events
for one session including its agents.
Also:
- ollie-9p -a flag now supports TCP (host:port or tcp!host:port)
- Add Session.ID() accessor method
This commit is contained in:
parent
4e817f5de9
commit
73de0b3cc0
|
|
@ -149,8 +149,20 @@ func dial() *client.Fsys {
|
|||
}
|
||||
|
||||
// mountWithAname connects and attaches with a specific aname.
|
||||
// Address format: "unix!/path" or "/path" for unix sockets, "tcp!host:port" or "host:port" for TCP.
|
||||
func mountWithAname(address, aname string) (*client.Fsys, error) {
|
||||
conn, err := client.Dial("unix", address)
|
||||
network := "unix"
|
||||
addr := address
|
||||
if strings.HasPrefix(address, "tcp!") {
|
||||
network = "tcp"
|
||||
addr = strings.TrimPrefix(address, "tcp!")
|
||||
} else if strings.HasPrefix(address, "unix!") {
|
||||
addr = strings.TrimPrefix(address, "unix!")
|
||||
} else if strings.Contains(address, ":") && !strings.HasPrefix(address, "/") {
|
||||
// Looks like host:port, assume TCP
|
||||
network = "tcp"
|
||||
}
|
||||
conn, err := client.Dial(network, addr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
|
|||
|
|
@ -415,7 +415,7 @@ func handleAgentCtl(a *agent.Agent, s *Session, data []byte) error {
|
|||
case "kill":
|
||||
if s.Core != nil {
|
||||
s.Core.RemoveAgent(a.ID())
|
||||
PublishEvent("agent."+a.ID()+".kill", "")
|
||||
PublishEvent("session."+s.ID()+".agent."+a.ID()+".kill", "")
|
||||
}
|
||||
case "stop":
|
||||
a.Interrupt(agent.ErrInterrupted)
|
||||
|
|
|
|||
|
|
@ -49,14 +49,26 @@ type Event struct {
|
|||
}
|
||||
|
||||
// PublishEvent publishes an event to a specific topic.
|
||||
// Topic format: "session.{id}.{action}" or "agent.{id}.{action}"
|
||||
// Topic hierarchy:
|
||||
// session.{sid}.new/kill/pause/resume/rename
|
||||
// session.{sid}.agent.{aid}.new/kill/state
|
||||
//
|
||||
// Wildcard routing publishes to all ancestor wildcards:
|
||||
// session.{sid}.agent.{aid}.state -> also publishes to:
|
||||
// session.{sid}.agent.{aid}.*
|
||||
// session.{sid}.agent.*
|
||||
// session.{sid}.*
|
||||
// session.*
|
||||
// *
|
||||
func PublishEvent(topic, payload string) {
|
||||
ev := Event{Topic: topic, Payload: payload}
|
||||
eventBus.Publish(topic, ev)
|
||||
// Also publish to wildcard topics for subscribers who want everything
|
||||
|
||||
// Publish to hierarchical wildcards
|
||||
parts := strings.Split(topic, ".")
|
||||
if len(parts) >= 1 {
|
||||
eventBus.Publish(parts[0]+".*", ev)
|
||||
for i := len(parts) - 1; i >= 1; i-- {
|
||||
wildcard := strings.Join(parts[:i], ".") + ".*"
|
||||
eventBus.Publish(wildcard, ev)
|
||||
}
|
||||
eventBus.Publish("*", ev)
|
||||
}
|
||||
|
|
@ -213,10 +225,10 @@ func Kill(rs *rootState, id string) {
|
|||
}
|
||||
removePersistedAgent(rs, sess.id)
|
||||
rs.cfg.Log.Info("killed session %s", id)
|
||||
for _, name := range agentNames {
|
||||
PublishEvent("agent."+name+".kill", "")
|
||||
for _, agentID := range agentNames {
|
||||
PublishEvent("session."+sess.id+".agent."+agentID+".kill", "")
|
||||
}
|
||||
PublishEvent("session."+id+".kill", "")
|
||||
PublishEvent("session."+sess.id+".kill", "")
|
||||
}
|
||||
|
||||
// Rename renames a session by updating its Name and the sessions map.
|
||||
|
|
@ -726,9 +738,10 @@ func CreateAgent(rs *rootState, sessName string, args []string) error {
|
|||
FlushSave: flushFn,
|
||||
})
|
||||
sess.Core.AddAgent(ag)
|
||||
// Wire up state change events
|
||||
// Wire up state change events - capture sessID for hierarchical topic
|
||||
sid := sessID
|
||||
ag.SetOnStateChange(func(agentID, state string) {
|
||||
PublishEvent("agent."+agentID+".state", state)
|
||||
PublishEvent("session."+sid+".agent."+agentID+".state", state)
|
||||
})
|
||||
sess.mu.Lock()
|
||||
sess.setAgentLogLocked(uname, al)
|
||||
|
|
@ -749,7 +762,7 @@ func CreateAgent(rs *rootState, sessName string, args []string) error {
|
|||
|
||||
rs.cfg.Log.Info("created agent in session %s (backend=%s model=%s agent=%s)",
|
||||
sessName, ag.BackendName(), ag.ModelName(), ag.Name())
|
||||
PublishEvent("agent."+ag.ID()+".new", sessName)
|
||||
PublishEvent("session."+sessID+".agent."+ag.ID()+".new", ag.Name())
|
||||
return nil
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -206,6 +206,7 @@ func (sess *Session) SetDisallowTools(names []string) {
|
|||
}
|
||||
|
||||
func (sess *Session) RunnableID() string { return sess.id }
|
||||
func (sess *Session) ID() string { return sess.id }
|
||||
func (sess *Session) Uname() string { return sess.uname }
|
||||
func (sess *Session) Name() string {
|
||||
sess.mu.RLock()
|
||||
|
|
|
|||
Loading…
Reference in New Issue