028ec84c5e
The old IPC Task system passed `any` through TaskHandler and
PerformAsync. Now that named Actions exist with typed signatures
(ActionHandler func(context.Context, Options) Result), the untyped
layer is dead weight.
Changes:
- type Task any removed (was in contract.go)
- type Task struct is now the composed sequence (action.go)
- PerformAsync takes (action string, opts Options) not (t Task)
- TaskHandler type removed — use c.Action("name", handler)
- RegisterTask removed — use c.Action("name", handler)
- PERFORM sugar removed — use c.Action("name").Run()
- ActionTaskStarted/Progress/Completed carry typed fields
(Action string, Options, Result) not any
ActionDef → Action rename also in this commit (same principle:
DTOs don't have Run() methods).
Co-Authored-By: Virgil <virgil@lethean.io>
103 lines
2.4 KiB
Go
103 lines
2.4 KiB
Go
// SPDX-License-Identifier: EUPL-1.2
|
|
|
|
// Message bus for the Core framework.
|
|
// Dispatches actions (fire-and-forget), queries (first responder),
|
|
// and tasks (first executor) between registered handlers.
|
|
|
|
package core
|
|
|
|
import (
|
|
"slices"
|
|
"sync"
|
|
)
|
|
|
|
// Ipc holds IPC dispatch data and the named action registry.
|
|
//
|
|
// ipc := (&core.Ipc{}).New()
|
|
type Ipc struct {
|
|
ipcMu sync.RWMutex
|
|
ipcHandlers []func(*Core, Message) Result
|
|
|
|
queryMu sync.RWMutex
|
|
queryHandlers []QueryHandler
|
|
|
|
actions *Registry[*Action] // named action registry
|
|
tasks *Registry[*Task] // named task registry
|
|
}
|
|
|
|
// broadcast dispatches a message to all registered IPC handlers.
|
|
// Each handler is wrapped in panic recovery. All handlers fire regardless of individual results.
|
|
func (c *Core) broadcast(msg Message) Result {
|
|
c.ipc.ipcMu.RLock()
|
|
handlers := slices.Clone(c.ipc.ipcHandlers)
|
|
c.ipc.ipcMu.RUnlock()
|
|
|
|
for _, h := range handlers {
|
|
func() {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
Error("ACTION handler panicked", "panic", r)
|
|
}
|
|
}()
|
|
h(c, msg)
|
|
}()
|
|
}
|
|
return Result{OK: true}
|
|
}
|
|
|
|
func (c *Core) Query(q Query) Result {
|
|
c.ipc.queryMu.RLock()
|
|
handlers := slices.Clone(c.ipc.queryHandlers)
|
|
c.ipc.queryMu.RUnlock()
|
|
|
|
for _, h := range handlers {
|
|
r := h(c, q)
|
|
if r.OK {
|
|
return r
|
|
}
|
|
}
|
|
return Result{}
|
|
}
|
|
|
|
func (c *Core) QueryAll(q Query) Result {
|
|
c.ipc.queryMu.RLock()
|
|
handlers := slices.Clone(c.ipc.queryHandlers)
|
|
c.ipc.queryMu.RUnlock()
|
|
|
|
var results []any
|
|
for _, h := range handlers {
|
|
r := h(c, q)
|
|
if r.OK && r.Value != nil {
|
|
results = append(results, r.Value)
|
|
}
|
|
}
|
|
return Result{results, true}
|
|
}
|
|
|
|
func (c *Core) RegisterQuery(handler QueryHandler) {
|
|
c.ipc.queryMu.Lock()
|
|
c.ipc.queryHandlers = append(c.ipc.queryHandlers, handler)
|
|
c.ipc.queryMu.Unlock()
|
|
}
|
|
|
|
// --- IPC Registration (handlers) ---
|
|
|
|
// RegisterAction registers a broadcast handler for ACTION messages.
|
|
//
|
|
// c.RegisterAction(func(c *core.Core, msg core.Message) core.Result {
|
|
// if ev, ok := msg.(AgentCompleted); ok { ... }
|
|
// return core.Result{OK: true}
|
|
// })
|
|
func (c *Core) RegisterAction(handler func(*Core, Message) Result) {
|
|
c.ipc.ipcMu.Lock()
|
|
c.ipc.ipcHandlers = append(c.ipc.ipcHandlers, handler)
|
|
c.ipc.ipcMu.Unlock()
|
|
}
|
|
|
|
// RegisterActions registers multiple broadcast handlers.
|
|
func (c *Core) RegisterActions(handlers ...func(*Core, Message) Result) {
|
|
c.ipc.ipcMu.Lock()
|
|
c.ipc.ipcHandlers = append(c.ipc.ipcHandlers, handlers...)
|
|
c.ipc.ipcMu.Unlock()
|
|
}
|
|
|