0911d5ad7b
core/go was violating its own RFC-025 Principle 2: every exported function must have a comment showing HOW with real values. 37 functions had no comments — mostly one-liner accessors on Core, Config, ServiceRuntime, IPC, and Options. Now every exported function in every source file has a usage-example comment. AX Principle 2 compliance: 0/37 → 37/37 (100%). Co-Authored-By: Virgil <virgil@lethean.io>
113 lines
2.8 KiB
Go
113 lines
2.8 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}
|
|
}
|
|
|
|
// Query dispatches a request — first handler to return OK wins.
|
|
//
|
|
// r := c.Query(MyQuery{})
|
|
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{}
|
|
}
|
|
|
|
// QueryAll dispatches a request — collects all OK responses.
|
|
//
|
|
// r := c.QueryAll(countQuery{})
|
|
// results := r.Value.([]any)
|
|
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}
|
|
}
|
|
|
|
// RegisterQuery registers a handler for QUERY dispatch.
|
|
//
|
|
// c.RegisterQuery(func(_ *core.Core, q core.Query) core.Result { ... })
|
|
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()
|
|
}
|
|
|