orchard

 1package main
 2
 3// What every open tab is told, whichever one asked for it.
 4//
 5// A turn already outlives the tab that started it, but nothing told the other
 6// tabs about it. So a question asked on the desktop never reached the phone
 7// without a reload, and switching to another conversation and back was the only
 8// way to see an answer that had finished while you were elsewhere. Isaac asked
 9// for a notification, a queue and cross device history, and the queue was the
10// only one of the three that already existed.
11//
12// This is the meta channel and deliberately not a second copy of the answer
13// stream. It carries which conversation changed and nothing about what was
14// said, so a tab knows what to go and read. The answer itself still comes from
15// the run, which is the one place it is assembled.
16
17import (
18	"encoding/json"
19	"sync"
20)
21
22// A slow reader is dropped rather than waited on. A tab on a locked phone can
23// stop reading for minutes, and a broadcast that blocks on it would stall every
24// other tab and the turn doing the publishing.
25const hubBuffer = 16
26
27type HubEvent struct {
28	Kind   string `json:"kind"` // started, finished, changed
29	ConvID string `json:"conversation_id,omitempty"`
30	Title  string `json:"title,omitempty"`
31}
32
33type Hub struct {
34	mu   sync.Mutex
35	subs map[chan HubEvent]struct{}
36}
37
38func NewHub() *Hub { return &Hub{subs: map[chan HubEvent]struct{}{}} }
39
40// Subscribe returns a channel of events and the function that closes it. The
41// caller must call cancel, and may call it more than once.
42func (h *Hub) Subscribe() (<-chan HubEvent, func()) {
43	ch := make(chan HubEvent, hubBuffer)
44	h.mu.Lock()
45	h.subs[ch] = struct{}{}
46	h.mu.Unlock()
47
48	var once sync.Once
49	return ch, func() {
50		once.Do(func() {
51			h.mu.Lock()
52			delete(h.subs, ch)
53			h.mu.Unlock()
54			close(ch)
55		})
56	}
57}
58
59// Publish tells every open tab. It never blocks: a subscriber whose buffer is
60// full has stopped reading, and the events are hints to go and re-read rather
61// than a record that has to arrive.
62func (h *Hub) Publish(ev HubEvent) {
63	if h == nil {
64		return
65	}
66	h.mu.Lock()
67	defer h.mu.Unlock()
68	for ch := range h.subs {
69		select {
70		case ch <- ev:
71		default:
72		}
73	}
74}
75
76// Len is how many tabs are listening. It exists so nothing has to reach into
77// the map without the lock to find out.
78func (h *Hub) Len() int {
79	h.mu.Lock()
80	defer h.mu.Unlock()
81	return len(h.subs)
82}
83
84func (ev HubEvent) frame() []byte {
85	b, err := json.Marshal(ev)
86	if err != nil {
87		return nil
88	}
89	return b
90}