| 1 | package live |
| 2 | |
| 3 | import ( |
| 4 | "testing" |
| 5 | "time" |
| 6 | ) |
| 7 | |
| 8 | // drain reads every event already queued on ch without blocking. |
| 9 | func drain(ch <-chan Event) []Event { |
| 10 | var out []Event |
| 11 | for { |
| 12 | select { |
| 13 | case e, ok := <-ch: |
| 14 | if !ok { |
| 15 | return out |
| 16 | } |
| 17 | out = append(out, e) |
| 18 | default: |
| 19 | return out |
| 20 | } |
| 21 | } |
| 22 | } |
| 23 | |
| 24 | // next takes the one event a test just caused, failing rather than |
| 25 | // hanging when the hub said nothing. |
| 26 | func next(t *testing.T, ch <-chan Event) Event { |
| 27 | t.Helper() |
| 28 | select { |
| 29 | case e, ok := <-ch: |
| 30 | if !ok { |
| 31 | t.Fatal("the subscriber was dropped") |
| 32 | } |
| 33 | return e |
| 34 | default: |
| 35 | t.Fatal("no event") |
| 36 | } |
| 37 | return Event{} |
| 38 | } |
| 39 | |
| 40 | // TestSeqIsAdvisoryAndOnlyReachesItsOwnDeck: a published seq goes to |
| 41 | // everyone watching that deck and to nobody watching another one. |
| 42 | func TestSeqIsAdvisoryAndOnlyReachesItsOwnDeck(t *testing.T) { |
| 43 | h := New() |
| 44 | defer h.Close() |
| 45 | |
| 46 | mine, stop := h.Subscribe("deck-a") |
| 47 | defer stop() |
| 48 | theirs, stopTheirs := h.Subscribe("deck-b") |
| 49 | defer stopTheirs() |
| 50 | |
| 51 | h.PublishSeq("deck-a", 7) |
| 52 | |
| 53 | e := next(t, mine) |
| 54 | if e.Seq != 7 || e.Presence != nil { |
| 55 | t.Errorf("the subscriber heard %+v, want seq 7 and no presence", e) |
| 56 | } |
| 57 | if got := drain(theirs); len(got) != 0 { |
| 58 | t.Errorf("a subscriber on another deck heard %+v", got) |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | // TestPresenceIsAnnouncedAndExpires: posting presence publishes the |
| 63 | // WHOLE set for that deck, and a person who stops heartbeating is |
| 64 | // swept out and the remaining set republished. |
| 65 | func TestPresenceIsAnnouncedAndExpires(t *testing.T) { |
| 66 | h := New() |
| 67 | defer h.Close() |
| 68 | |
| 69 | now := time.Date(2026, 9, 7, 12, 0, 0, 0, time.UTC) |
| 70 | h.now = func() time.Time { return now } |
| 71 | |
| 72 | ch, stop := h.Subscribe("deck-a") |
| 73 | defer stop() |
| 74 | |
| 75 | h.PublishPresence("deck-a", Presence{Client: "tab-1", Name: "Paul", Colour: "#111", Slide: "s1"}) |
| 76 | e := next(t, ch) |
| 77 | if e.Presence == nil { |
| 78 | t.Fatalf("presence published a seq event: %+v", e) |
| 79 | } |
| 80 | if set := *e.Presence; len(set) != 1 || set[0].Client != "tab-1" || set[0].Name != "Paul" { |
| 81 | t.Fatalf("the announced set is %+v, want one entry for tab-1", set) |
| 82 | } |
| 83 | |
| 84 | // A second tab: the set, never a delta. |
| 85 | h.PublishPresence("deck-a", Presence{Client: "tab-2", Name: "Sam", Colour: "#222"}) |
| 86 | if set := *next(t, ch).Presence; len(set) != 2 { |
| 87 | t.Fatalf("the announced set is %+v, want both tabs", set) |
| 88 | } |
| 89 | |
| 90 | // tab-1 keeps heartbeating; tab-2 goes quiet. |
| 91 | now = now.Add(30 * time.Second) |
| 92 | h.PublishPresence("deck-a", Presence{Client: "tab-1", Name: "Paul", Colour: "#111", Slide: "s2"}) |
| 93 | drain(ch) |
| 94 | |
| 95 | // Past two missed heartbeats the sweeper drops tab-2 and says so. |
| 96 | now = now.Add(20 * time.Second) |
| 97 | h.sweep() |
| 98 | e = next(t, ch) |
| 99 | if e.Presence == nil { |
| 100 | t.Fatalf("the sweep published a seq event: %+v", e) |
| 101 | } |
| 102 | set := *e.Presence |
| 103 | if len(set) != 1 || set[0].Client != "tab-1" { |
| 104 | t.Fatalf("after the sweep the set is %+v, want tab-1 alone", set) |
| 105 | } |
| 106 | if got := h.Snapshot("deck-a"); len(got) != 1 || got[0].Slide != "s2" { |
| 107 | t.Errorf("Snapshot is %+v, want tab-1 on s2", got) |
| 108 | } |
| 109 | |
| 110 | // A sweep that changes nothing says nothing. |
| 111 | h.sweep() |
| 112 | if got := drain(ch); len(got) != 0 { |
| 113 | t.Errorf("an idle sweep published %+v", got) |
| 114 | } |
| 115 | |
| 116 | // Everyone gone: the deck is forgotten rather than kept as an |
| 117 | // empty entry for the life of the process. |
| 118 | now = now.Add(time.Minute) |
| 119 | h.sweep() |
| 120 | if set := *next(t, ch).Presence; len(set) != 0 { |
| 121 | t.Errorf("the final sweep announced %+v, want an empty set", set) |
| 122 | } |
| 123 | if n := h.presenceDecks(); n != 0 { |
| 124 | t.Errorf("the hub still holds presence for %d decks, want 0", n) |
| 125 | } |
| 126 | } |
| 127 | |
| 128 | // TestSubscriberThatFallsBehindIsDropped: presence is whole state, so |
| 129 | // a reader that cannot keep up is closed and left to reconnect rather |
| 130 | // than fed a set with holes in it. |
| 131 | func TestSubscriberThatFallsBehindIsDropped(t *testing.T) { |
| 132 | h := New() |
| 133 | defer h.Close() |
| 134 | |
| 135 | ch, stop := h.Subscribe("deck-a") |
| 136 | defer stop() |
| 137 | for i := 0; i < subscriberBuffer+2; i++ { |
| 138 | h.PublishSeq("deck-a", int64(i+1)) |
| 139 | } |
| 140 | for i := 0; i < subscriberBuffer; i++ { |
| 141 | if _, ok := <-ch; !ok { |
| 142 | t.Fatalf("the subscriber closed after %d events, want %d buffered first", i, subscriberBuffer) |
| 143 | } |
| 144 | } |
| 145 | if _, ok := <-ch; ok { |
| 146 | t.Fatal("a subscriber that fell behind is still open") |
| 147 | } |
| 148 | if n := h.deckCount(); n != 0 { |
| 149 | t.Errorf("the dropped subscriber left %d decks behind", n) |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | // TestUnsubscribeForgetsTheDeck: the subscriber map must not grow one |
| 154 | // empty entry per deck ever opened. |
| 155 | func TestUnsubscribeForgetsTheDeck(t *testing.T) { |
| 156 | h := New() |
| 157 | defer h.Close() |
| 158 | |
| 159 | _, stop := h.Subscribe("deck-a") |
| 160 | if n := h.deckCount(); n != 1 { |
| 161 | t.Fatalf("deckCount = %d after one subscribe, want 1", n) |
| 162 | } |
| 163 | stop() |
| 164 | stop() // idempotent |
| 165 | if n := h.deckCount(); n != 0 { |
| 166 | t.Errorf("deckCount = %d after unsubscribing, want 0", n) |
| 167 | } |
| 168 | } |
| 169 | |
amadan