1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
package guestcontrolplane
import (
"context"
"errors"
"fmt"
"github.com/johnandersen777/socialweb-computer-kcp/internal/bus"
"github.com/johnandersen777/socialweb-computer-kcp/internal/contract/guestcp"
)
func (s *Service) openQueue(ctx context.Context, inputs guestcp.Source) error {
if s.stateDir == "" {
return errors.New("guestcontrolplane: the guest event queue needs a state dir")
}
self := s.self()
ctx = bus.WithCallerLabels(ctx, self, s.deps.Labels)
queue, err := bus.Resolve[guestcp.Queue](ctx, s.deps.Bus, self, self, bus.Capability{Name: guestcp.CapabilityQueue})
if err != nil {
return fmt.Errorf("guestcontrolplane: resolving the guest event queue: %w", err)
}
if err := queue.Open(s.stateDir); err != nil {
return err
}
unwatch := inputs.Watch(func(in guestcp.Inputs) {
names := make([]string, 0, len(in.ComputeVMs))
for i := range in.ComputeVMs {
names = append(names, in.ComputeVMs[i].Name)
}
queue.Retain(names)
})
s.mu.Lock()
s.queue, s.unwatchQueue = queue, unwatch
s.mu.Unlock()
return nil
}
func (s *Service) publishGuestEvent(ctx context.Context, event guestcp.GuestEvent) error {
s.mu.Lock()
queue := s.queue
s.mu.Unlock()
if queue == nil {
return guestcp.ErrQueueNotOpen
}
return queue.Publish(ctx, event)
}
