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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
package guestcp
import (
"context"
"encoding/json"
"errors"
)
const (
CapabilityEvents = "guestcp.events"
CapabilityQueue = "guestcp.queue"
)
var (
ErrQueueNotOpen = errors.New("guestcp: the guest event queue is not open")
ErrQueueFull = errors.New("guestcp: the guest event queue holds too many unacknowledged events")
ErrQueueGone = errors.New("guestcp: the guest event queue is gone")
ErrFutureCursor = errors.New("guestcp: the cursor is past the last event")
ErrUnauthenticated = errors.New("guestcp: the guest bearer token is missing or refused")
ErrUnknownGuest = errors.New("guestcp: no live ComputeVM holds this guest")
ErrNoPublisher = errors.New("guestcp: the bidder does not publish guest events yet")
)
type QueuedEvent struct {
Seq int64 `json:"seq"`
URI string `json:"uri"`
CID string `json:"cid"`
Record json.RawMessage `json:"record,omitempty"`
}
type GuestEvent struct {
ComputeVM string
Collection string
Tag string
Kind string
Record json.RawMessage
}
type Publish func(ctx context.Context, event GuestEvent) error
type Events interface {
Append(ctx context.Context, computeVM string, event QueuedEvent) (bool, error)
Serve(publish Publish) (cancel func())
}
type EventStream interface {
Next(ctx context.Context) (QueuedEvent, error)
Close()
}
type Queue interface {
Open(dir string) error
Retain(computeVMs []string)
Subscribe(computeVM string, cursor *int64) (EventStream, error)
Publish(ctx context.Context, event GuestEvent) error
}
