attachment_routes.go RawHistoryFolder

1package client
2
3import "github.com/bnema/vev/internal/protocol"
4
5// Route publication and daemon navigation seams of the attached foreground.
6// The worker owns the stream, so the supervisor reaches the
7// daemon only through these bounded channels: the newest route snapshot
8// replaces an unsent one, and navigation failures are dropped rather than
9// blocking when the worker is gone. In the other direction the worker hands
10// every daemon navigation request to the supervisor, which resolves it
11// through the broker catalogue exactly like a picker commit.
12
13// attachmentReplyCapacity bounds queued navigation failures per foreground.
14const attachmentReplyCapacity = 4
15
16// demandRoutes wakes the supervisor to republish routes for this foreground.
17func (f *attachmentForeground) demandRoutes() {
18	if f == nil || f.host == nil || !f.authority.actionAuthorized(f) {
19		return
20	}
21	select {
22	case f.host.routeDemand <- struct{}{}:
23	default:
24	}
25}
26
27func (f *attachmentForeground) routeSnapshots() <-chan protocol.RecentRouteSnapshot {
28	if f == nil {
29		return nil
30	}
31	return f.routes
32}
33
34func (f *attachmentForeground) navigationReplies() <-chan protocol.ClientMessage {
35	if f == nil {
36		return nil
37	}
38	return f.replies
39}
40
41// requestNavigation hands one daemon navigation request to the supervisor.
42// The newest request replaces one the supervisor has not taken yet.
43func (f *attachmentForeground) requestNavigation(message protocol.ServerMessage) {
44	if f == nil || f.host == nil || !f.authority.actionAuthorized(f) {
45		return
46	}
47	for {
48		select {
49		case f.host.navigations <- message:
50			return
51		default:
52		}
53		select {
54		case <-f.host.navigations:
55		default:
56		}
57	}
58}
59
60// RouteDemands wakes the supervisor when the attached session changed.
61func (h *attachmentHost) RouteDemands() <-chan struct{} {
62	if h == nil {
63		return nil
64	}
65	return h.routeDemand
66}
67
68// Navigations carries the daemon's navigation requests to the supervisor.
69func (h *attachmentHost) Navigations() <-chan protocol.ServerMessage {
70	if h == nil {
71		return nil
72	}
73	return h.navigations
74}
75
76// takeNavigation returns a navigation request left behind by a worker that
77// already settled, if any.
78func (h *attachmentHost) takeNavigation() (protocol.ServerMessage, bool) {
79	if h == nil {
80		return nil, false
81	}
82	select {
83	case message := <-h.navigations:
84		return message, true
85	default:
86		return nil, false
87	}
88}
89
90// publishRoutes hands the newest route snapshot to the current foreground's
91// worker. It reports false when token no longer owns the foreground.
92func (h *attachmentHost) publishRoutes(token AttachmentToken, snapshot protocol.RecentRouteSnapshot) bool {
93	fg := h.currentForeground(token)
94	if fg == nil {
95		return false
96	}
97	for {
98		select {
99		case fg.routes <- snapshot:
100			return true
101		default:
102		}
103		select {
104		case <-fg.routes:
105		default:
106		}
107	}
108}
109
110// replyNavigation queues one navigation failure for the current worker.
111func (h *attachmentHost) replyNavigation(token AttachmentToken, message protocol.ClientMessage) bool {
112	fg := h.currentForeground(token)
113	if fg == nil {
114		return false
115	}
116	select {
117	case fg.replies <- message:
118		return true
119	default:
120		return false
121	}
122}
123
124func (h *attachmentHost) currentForeground(token AttachmentToken) *attachmentForeground {
125	if h == nil {
126		return nil
127	}
128	fg := h.authority.foreground()
129	if fg == nil || fg.token != token {
130		return nil
131	}
132	return fg
133}