| 1 | // Package present is the fan-out behind a live presentation: one |
| 2 | // session moved, told to every page watching it. |
| 3 | // |
| 4 | // It is Docs' hub (its internal/docs/hub.go) with one difference worth |
| 5 | // stating, because it is the difference that decides everything else |
| 6 | // here: Docs' notification is ADVISORY — a seq that means "there is |
| 7 | // something at least this new", so a subscriber that misses ten is |
| 8 | // still correct after fetching once. A presentation event is not |
| 9 | // advisory in the same way: `{slide, blackout, ended}` is the whole |
| 10 | // state, so what a subscriber must never do is apply a STALE one. |
| 11 | // |
| 12 | // Hence the two decisions below. The channel is buffered eight rather |
| 13 | // than one, because a burst of arrow-key presses is a real thing and |
| 14 | // eight covers it without the sender ever waiting. And a subscriber |
| 15 | // that falls behind that is DROPPED — unsubscribed, its channel closed |
| 16 | // — rather than having events silently skipped: the handler holding it |
| 17 | // sees the close, ends the response, and the browser reconnects, which |
| 18 | // re-reads the session's real state from the row on connect (the |
| 19 | // stream handler's one event on connect). A skipped event, by |
| 20 | // contrast, would leave a projector showing slide 4 for the rest of a |
| 21 | // talk with nothing to notice it. |
| 22 | // |
| 23 | // Scope, like Docs': one process. Slides is one instance per team, so |
| 24 | // there is no second process to reach; this type is the seam that |
| 25 | // changes on the day there is. |
| 26 | package present |
| 27 | |
| 28 | import "sync" |
| 29 | |
| 30 | // Event is a presentation's whole visible state at one moment — never |
| 31 | // a delta, so a subscriber that applies one is exactly right rather |
| 32 | // than approximately. |
| 33 | // |
| 34 | // The JSON tags are the wire format of the stream's `event: slide` |
| 35 | // data (presentation.go), and static/present.js reads these three |
| 36 | // names. Changing one is changing the protocol. |
| 37 | type Event struct { |
| 38 | // Slide is the 0-based index of the slide being shown. |
| 39 | Slide int `json:"slide"` |
| 40 | // Blackout is whether the screen is blacked out. Carried |
| 41 | // separately from the slide because coming back from a blackout |
| 42 | // must not move the deck. |
| 43 | Blackout bool `json:"blackout"` |
| 44 | // Ended says the session is over. A page that hears it stops |
| 45 | // following rather than sitting on a slide that will never move |
| 46 | // again. |
| 47 | Ended bool `json:"ended"` |
| 48 | } |
| 49 | |
| 50 | // subscriberBuffer is how many events a subscriber may fall behind by |
| 51 | // before it is dropped. Eight is a burst of arrow presses; a page that |
| 52 | // has not read eight is not reading at all. |
| 53 | const subscriberBuffer = 8 |
| 54 | |
| 55 | // Hub is the per-process fan-out, keyed by presentation sqid. |
| 56 | type Hub struct { |
| 57 | mu sync.Mutex |
| 58 | subs map[string]map[chan Event]struct{} |
| 59 | } |
| 60 | |
| 61 | // New builds an empty hub. One per app (handlers.go). |
| 62 | func New() *Hub { |
| 63 | return &Hub{subs: make(map[string]map[chan Event]struct{})} |
| 64 | } |
| 65 | |
| 66 | // Subscribe registers interest in one session and returns the channel |
| 67 | // to wait on plus the function that unregisters it. |
| 68 | // |
| 69 | // The returned cancel is idempotent — the stream handler defers it and |
| 70 | // may also have reason to stop early — and it never closes the |
| 71 | // channel. Only Publish closes one, and only when it is dropping the |
| 72 | // subscriber, so "the channel is closed" carries exactly one meaning |
| 73 | // for a reader: you fell behind, reconnect. A cancel that closed too |
| 74 | // would make that signal ambiguous at the one moment it matters. |
| 75 | func (h *Hub) Subscribe(key string) (<-chan Event, func()) { |
| 76 | ch := make(chan Event, subscriberBuffer) |
| 77 | |
| 78 | h.mu.Lock() |
| 79 | if h.subs[key] == nil { |
| 80 | h.subs[key] = make(map[chan Event]struct{}) |
| 81 | } |
| 82 | h.subs[key][ch] = struct{}{} |
| 83 | h.mu.Unlock() |
| 84 | |
| 85 | var once sync.Once |
| 86 | return ch, func() { |
| 87 | once.Do(func() { |
| 88 | h.mu.Lock() |
| 89 | defer h.mu.Unlock() |
| 90 | h.remove(key, ch) |
| 91 | }) |
| 92 | } |
| 93 | } |
| 94 | |
| 95 | // Publish sends e to everyone watching key. |
| 96 | // |
| 97 | // It runs on whichever request goroutine just moved the session (a |
| 98 | // phone's POST, the presenter's keyboard), so it must never block: a |
| 99 | // browser on a wedged connection would otherwise hold a stranger's |
| 100 | // command open. A subscriber whose buffer is full is therefore removed |
| 101 | // and its channel closed, inside the same lock that found it, so no |
| 102 | // send can race the close. |
| 103 | func (h *Hub) Publish(key string, e Event) { |
| 104 | h.mu.Lock() |
| 105 | defer h.mu.Unlock() |
| 106 | |
| 107 | for ch := range h.subs[key] { |
| 108 | select { |
| 109 | case ch <- e: |
| 110 | default: |
| 111 | h.remove(key, ch) |
| 112 | close(ch) |
| 113 | } |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | // remove takes ch out of key's set and drops the set with its last |
| 118 | // member — otherwise this map keeps one empty entry per session ever |
| 119 | // presented for the life of the process. Callers hold h.mu. |
| 120 | func (h *Hub) remove(key string, ch chan Event) { |
| 121 | subs, ok := h.subs[key] |
| 122 | if !ok { |
| 123 | return |
| 124 | } |
| 125 | delete(subs, ch) |
| 126 | if len(subs) == 0 { |
| 127 | delete(h.subs, key) |
| 128 | } |
| 129 | } |
| 130 | |
amadan