okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

Aly Raffauffeat(api)!: migrate daemon and clients to protocol 361c43ca

main
branchesmainall refs →
3.0 KiB124 linesraw
1package daemon
2
3import (
4	"context"
5	"errors"
6	"fmt"
7	"net/http"
8	"strings"
9	"sync"
10	"sync/atomic"
11)
12
13type ConnectionSnapshot struct {
14	InstanceID string `json:"instance_id,omitempty"`
15	Token      string `json:"token"`
16	Build      string `json:"build,omitempty"`
17	Digest     string `json:"digest,omitempty"`
18	Port       int    `json:"port"`
19	Pid        int    `json:"pid"`
20	Version    int    `json:"version"`
21}
22
23type Connection struct {
24	state       atomic.Pointer[connectionState]
25	httpClient  *http.Client
26	refreshGate chan struct{}
27	rediscover  Rediscovery
28	httpOnce    sync.Once
29	refreshOnce sync.Once
30}
31
32type connectionState struct {
33	endpoint     ConnectionSnapshot
34	capabilities map[string]int64
35}
36
37// Rediscovery obtains an updated endpoint without owning API attachment.
38type Rediscovery func(context.Context) (ConnectionSnapshot, error)
39
40// NewConnection creates an unattached client for endpoint inspection. Attach
41// validates the daemon and installs the capabilities needed by optional APIs.
42// A nil rediscover disables recovery.
43func NewConnection(snap ConnectionSnapshot, rediscover Rediscovery) *Connection {
44	c := &Connection{rediscover: rediscover}
45	c.state.Store(&connectionState{endpoint: snap})
46	return c
47}
48
49func (c *Connection) Snapshot() ConnectionSnapshot {
50	if c == nil {
51		return ConnectionSnapshot{}
52	}
53	p := c.state.Load()
54	if p == nil {
55		return ConnectionSnapshot{}
56	}
57	return p.endpoint
58}
59
60func (c *Connection) Port() int {
61	return c.Snapshot().Port
62}
63
64func (c *Connection) BaseURL() string {
65	port := c.Port()
66	if port <= 0 {
67		return ""
68	}
69	return fmt.Sprintf("http://127.0.0.1:%d", port)
70}
71
72// Local says the daemon runs on this machine, so an ssh master opened here
73// is one it can ride.
74func (c *Connection) Local() bool {
75	return strings.HasPrefix(c.BaseURL(), "http://127.0.0.1:")
76}
77
78func (c *Connection) Update(other *Connection) {
79	if c == nil || other == nil || c == other {
80		return
81	}
82	c.state.Store(other.state.Load())
83}
84
85// Refresh serializes rediscovery and installs only a compatible live endpoint.
86func (c *Connection) Refresh(ctx context.Context) error {
87	if c == nil {
88		return errors.New("not connected to Albedo")
89	}
90	if ctx == nil {
91		ctx = context.Background()
92	}
93	c.refreshOnce.Do(func() { c.refreshGate = make(chan struct{}, 1) })
94	select {
95	case c.refreshGate <- struct{}{}:
96	case <-ctx.Done():
97		return ctx.Err()
98	}
99	defer func() { <-c.refreshGate }()
100	if err := ctx.Err(); err != nil {
101		return err
102	}
103	if c.rediscover == nil {
104		return errors.New("daemon rediscovery is unavailable")
105	}
106	snap, err := c.rediscover(ctx)
107	if err != nil {
108		return err
109	}
110	attached, err := Attach(ctx, snap, nil)
111	if err != nil {
112		return err
113	}
114	defer attached.HTTPClient().CloseIdleConnections()
115	c.Update(attached)
116	return nil
117}
118
119// HTTPClient returns the reusable HTTP client owned by this Connection.
120// Address and token updates retain the same client and pool.
121func (c *Connection) HTTPClient() *http.Client {
122	c.httpOnce.Do(func() { c.httpClient = newHTTPClient() })
123	return c.httpClient
124}