mirror of
https://github.com/hanzoai/agent.git
synced 2026-09-11 21:08:51 +00:00
zeekay
9556032ab1
Multi-platform cloud bot provisioning (Linux K8s pods, Windows/macOS EC2), billing authorization via bootnode API, usage metering, and real-time monitoring. Includes REST + UI APIs, SSE events, dedicated host pooling for macOS, and per-tier compute quotas.
129 lines
2.9 KiB
Go
129 lines
2.9 KiB
Go
package cloud
|
|
|
|
import (
|
|
"encoding/json"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
)
|
|
|
|
// EventBus provides publish/subscribe for cloud events.
|
|
type EventBus struct {
|
|
mu sync.RWMutex
|
|
subscribers map[string]chan CloudEvent
|
|
buffer []CloudEvent
|
|
bufferSize int
|
|
}
|
|
|
|
// NewEventBus creates a new event bus with the given buffer size.
|
|
func NewEventBus(bufferSize int) *EventBus {
|
|
if bufferSize <= 0 {
|
|
bufferSize = 100
|
|
}
|
|
return &EventBus{
|
|
subscribers: make(map[string]chan CloudEvent),
|
|
buffer: make([]CloudEvent, 0, bufferSize),
|
|
bufferSize: bufferSize,
|
|
}
|
|
}
|
|
|
|
// Publish sends an event to all subscribers.
|
|
func (eb *EventBus) Publish(event CloudEvent) {
|
|
if event.ID == "" {
|
|
event.ID = uuid.New().String()
|
|
}
|
|
if event.Timestamp.IsZero() {
|
|
event.Timestamp = time.Now().UTC()
|
|
}
|
|
|
|
eb.mu.Lock()
|
|
// Add to ring buffer.
|
|
if len(eb.buffer) >= eb.bufferSize {
|
|
eb.buffer = eb.buffer[1:]
|
|
}
|
|
eb.buffer = append(eb.buffer, event)
|
|
|
|
// Copy subscribers to avoid holding lock during send.
|
|
subs := make(map[string]chan CloudEvent, len(eb.subscribers))
|
|
for k, v := range eb.subscribers {
|
|
subs[k] = v
|
|
}
|
|
eb.mu.Unlock()
|
|
|
|
for _, ch := range subs {
|
|
select {
|
|
case ch <- event:
|
|
default:
|
|
// Drop if subscriber is slow.
|
|
}
|
|
}
|
|
}
|
|
|
|
// Subscribe returns a channel for receiving events.
|
|
func (eb *EventBus) Subscribe() (string, <-chan CloudEvent) {
|
|
id := uuid.New().String()
|
|
ch := make(chan CloudEvent, 32)
|
|
|
|
eb.mu.Lock()
|
|
eb.subscribers[id] = ch
|
|
eb.mu.Unlock()
|
|
|
|
return id, ch
|
|
}
|
|
|
|
// Unsubscribe removes a subscriber.
|
|
func (eb *EventBus) Unsubscribe(id string) {
|
|
eb.mu.Lock()
|
|
if ch, ok := eb.subscribers[id]; ok {
|
|
close(ch)
|
|
delete(eb.subscribers, id)
|
|
}
|
|
eb.mu.Unlock()
|
|
}
|
|
|
|
// Recent returns the most recent events from the buffer.
|
|
func (eb *EventBus) Recent(limit int) []CloudEvent {
|
|
eb.mu.RLock()
|
|
defer eb.mu.RUnlock()
|
|
|
|
if limit <= 0 || limit > len(eb.buffer) {
|
|
limit = len(eb.buffer)
|
|
}
|
|
|
|
start := len(eb.buffer) - limit
|
|
result := make([]CloudEvent, limit)
|
|
copy(result, eb.buffer[start:])
|
|
return result
|
|
}
|
|
|
|
// EmitInstanceEvent is a convenience method for publishing instance lifecycle events.
|
|
func (eb *EventBus) EmitInstanceEvent(eventType, instanceID string, data interface{}) {
|
|
var rawData json.RawMessage
|
|
if data != nil {
|
|
if b, err := json.Marshal(data); err == nil {
|
|
rawData = b
|
|
}
|
|
}
|
|
|
|
eb.Publish(CloudEvent{
|
|
Type: eventType,
|
|
InstanceID: instanceID,
|
|
Timestamp: time.Now().UTC(),
|
|
Data: rawData,
|
|
})
|
|
}
|
|
|
|
// Common event types.
|
|
const (
|
|
EventInstanceRequested = "instance.requested"
|
|
EventInstanceProvisioning = "instance.provisioning"
|
|
EventInstanceRunning = "instance.running"
|
|
EventInstanceStopped = "instance.stopped"
|
|
EventInstanceTerminated = "instance.terminated"
|
|
EventInstanceFailed = "instance.failed"
|
|
EventInstanceConnected = "instance.connected" // agent registered
|
|
EventHostAllocated = "host.allocated"
|
|
EventHostReleased = "host.released"
|
|
)
|