| 1 | // Package hub is the in-process authority for who is in a room: the |
| 2 | // participations of spec §2.4, the peer identity and limits of §4.2 and |
| 3 | // §4.4, and the resume window of §4.5. It holds no HTTP and no database; |
| 4 | // the handlers wrap it and the store answers standing for it. |
| 5 | // |
| 6 | // The hub is per process, which is fine on one host and would not |
| 7 | // survive a multi-host fleet; that is noted, not solved (spec §4.2). |
| 8 | package hub |
| 9 | |
| 10 | import ( |
| 11 | "context" |
| 12 | "crypto/rand" |
| 13 | "encoding/hex" |
| 14 | "errors" |
| 15 | "log/slog" |
| 16 | "sync" |
| 17 | "time" |
| 18 | ) |
| 19 | |
| 20 | const ( |
| 21 | // MaxParticipations is what a room holds in any state: unjoined, |
| 22 | // waiting, admitted, or held for resume (spec §4.2). |
| 23 | MaxParticipations = 16 |
| 24 | // MaxWaiting is the knock queue; a ninth waiting join is refused |
| 25 | // with full (spec §4.4). |
| 26 | MaxWaiting = 8 |
| 27 | // MaxAdmitted is the mesh cap, checked at every transition into |
| 28 | // the admitted state and never on a resume (spec §4.3). |
| 29 | MaxAdmitted = 4 |
| 30 | // MaxStreamsPerPerson counts concurrent streams across all rooms. |
| 31 | MaxStreamsPerPerson = 4 |
| 32 | |
| 33 | // UnjoinedWindow is how long a participation lives without a join. |
| 34 | UnjoinedWindow = 30 * time.Second |
| 35 | // ResumeWindow is how long a detached participation keeps its |
| 36 | // admission waiting for the browser to come back (spec §4.5). |
| 37 | ResumeWindow = 30 * time.Second |
| 38 | // TombstoneLife is how long an ended peer id is remembered so a |
| 39 | // resume can be told why it ended (spec §4.5). |
| 40 | TombstoneLife = 10 * time.Minute |
| 41 | // RingSize is the replay ring: the last targeted messages held for |
| 42 | // one participation, for the resume window only (spec §4.5). |
| 43 | RingSize = 32 |
| 44 | ) |
| 45 | |
| 46 | var ( |
| 47 | // ErrRoomFull is the sixteen-participation cap of spec §4.2, which a |
| 48 | // handler answers with 503 before anything is allocated. |
| 49 | ErrRoomFull = errors.New("hub: room holds sixteen participations") |
| 50 | // ErrMeshFull is the mesh cap of four (spec §4.3), which is not a |
| 51 | // failed request: the peer has been sent full and may retry with |
| 52 | // join, so a handler answers 2xx and lets the event carry the news. |
| 53 | ErrMeshFull = errors.New("hub: four already admitted") |
| 54 | ErrSessionHasPeer = errors.New("hub: this session already has a peer in the room") |
| 55 | ErrPersonStreams = errors.New("hub: this person has four streams open") |
| 56 | ErrUnknownPeer = errors.New("hub: no such peer") |
| 57 | ErrNotYours = errors.New("hub: peer is not bound to this session") |
| 58 | ErrNotAdmitted = errors.New("hub: peer is not admitted") |
| 59 | ErrNotMember = errors.New("hub: peer is not a member") |
| 60 | ErrNotWaiting = errors.New("hub: target is not waiting") |
| 61 | ErrNotTargeted = errors.New("hub: not a targeted message type") |
| 62 | ErrStaleGen = errors.New("hub: stale generation") |
| 63 | ErrTooManyWaiting = errors.New("hub: eight already waiting") |
| 64 | ) |
| 65 | |
| 66 | type Config struct { |
| 67 | // WaitingRoom is read at join time, outside the lock. Nil preserves the |
| 68 | // hub’s original behavior: guests wait for admission. |
| 69 | WaitingRoom func(context.Context, string) (bool, error) |
| 70 | Now func() time.Time |
| 71 | Logger *slog.Logger |
| 72 | // StandingFor re-derives a person's kind and whether they still hold |
| 73 | // entry authority for the room. ok=false means standing none. |
| 74 | StandingFor func(ctx context.Context, room string, personID int64) (kind Kind, ok bool, err error) |
| 75 | // Knocked says that someone has begun waiting. It exists because the |
| 76 | // admission queue lives only here, in memory, and reaches members |
| 77 | // over their own streams — so a member who is not in the room has |
| 78 | // never had any way to learn that somebody is at the door. |
| 79 | // |
| 80 | // Called on a goroutine of its own, never under the lock: the app's |
| 81 | // implementation reads the database and talks to Apple, and the hub's |
| 82 | // mutex is held across every peer in every room. Nil is no |
| 83 | // notification, which is every test and every deployment without a |
| 84 | // push key. |
| 85 | Knocked func(room string, personIDs []int64, name string) |
| 86 | } |
| 87 | |
| 88 | // Participation is the hub's record that a peer is in a room. It is the |
| 89 | // authority for everything a peer does after entry (spec §2.4). |
| 90 | type Participation struct { |
| 91 | Peer PeerID |
| 92 | Room string |
| 93 | Session string |
| 94 | PersonID int64 |
| 95 | Name string |
| 96 | Kind Kind |
| 97 | Admitted bool |
| 98 | Joined bool // has posted join (waiting or admitted) |
| 99 | JoinedAt time.Time |
| 100 | OpenedAt time.Time |
| 101 | // Media is the peer's last report of its own microphone and camera |
| 102 | // (§15.19), nil until it makes one. It lives and dies with the |
| 103 | // participation, so a resume keeps it and a fresh join starts unknown. |
| 104 | Media *MediaState |
| 105 | |
| 106 | sub *Sub |
| 107 | resumeUntil time.Time // zero while a stream is attached |
| 108 | seqOut map[PeerID]uint64 // next seq per target (unused by the hub; kept by browsers) |
| 109 | lastSeen map[PeerID]uint64 |
| 110 | ring []Message // targeted messages delivered to this peer, newest last, cap RingSize |
| 111 | evicted map[PeerID]uint64 // highest seq per sender the ring has dropped |
| 112 | // peerTag and reportToken are spec §15.26's: the random tag this |
| 113 | // participation's reports are logged under, and the token its client |
| 114 | // reports with (report.go). |
| 115 | peerTag string |
| 116 | reportToken string |
| 117 | } |
| 118 | |
| 119 | type pairKey struct{ a, b PeerID } |
| 120 | |
| 121 | func keyFor(x, y PeerID) pairKey { |
| 122 | if x < y { |
| 123 | return pairKey{x, y} |
| 124 | } |
| 125 | return pairKey{y, x} |
| 126 | } |
| 127 | |
| 128 | type room struct { |
| 129 | sqid string |
| 130 | peers map[PeerID]*Participation |
| 131 | gens map[pairKey]uint64 |
| 132 | ended bool |
| 133 | // call is the call instance now running here (report.go): replaced |
| 134 | // when a participation opens in a room that held nobody. |
| 135 | call *callInstance |
| 136 | } |
| 137 | |
| 138 | // tombstone remembers why a participation ended, for ten minutes, so a |
| 139 | // stream that never received its lost event learns the reason on resume |
| 140 | // (spec §4.5). It carries the room and session it was bound to because |
| 141 | // the reason is account state — removed, standing, expired — and a peer |
| 142 | // id is published in rosters and is not a credential (spec §4.2). |
| 143 | type tombstone struct { |
| 144 | room string |
| 145 | session string |
| 146 | reason Reason |
| 147 | at time.Time |
| 148 | } |
| 149 | |
| 150 | type Hub struct { |
| 151 | cfg Config |
| 152 | boot string |
| 153 | |
| 154 | mu sync.Mutex |
| 155 | rooms map[string]*room |
| 156 | bySession map[string]map[string]PeerID // room -> session -> peer |
| 157 | byPerson map[int64]int // open streams per person |
| 158 | tombstones map[PeerID]tombstone |
| 159 | reports map[string]*reportGrant // report token -> grant (report.go) |
| 160 | } |
| 161 | |
| 162 | func New(cfg Config) *Hub { |
| 163 | if cfg.Now == nil { |
| 164 | cfg.Now = time.Now |
| 165 | } |
| 166 | if cfg.Logger == nil { |
| 167 | cfg.Logger = slog.Default() |
| 168 | } |
| 169 | return &Hub{ |
| 170 | cfg: cfg, boot: randomID(), |
| 171 | rooms: map[string]*room{}, bySession: map[string]map[string]PeerID{}, |
| 172 | byPerson: map[int64]int{}, tombstones: map[PeerID]tombstone{}, |
| 173 | reports: map[string]*reportGrant{}, |
| 174 | } |
| 175 | } |
| 176 | |
| 177 | func randomID() string { |
| 178 | b := make([]byte, 16) |
| 179 | if _, err := rand.Read(b); err != nil { |
| 180 | panic(err) |
| 181 | } |
| 182 | return hex.EncodeToString(b) |
| 183 | } |
| 184 | |
| 185 | // Boot is the random id chosen when the process started. Every hello |
| 186 | // carries it so a browser can tell a restart from any other refusal |
| 187 | // (spec §4.2). |
| 188 | func (h *Hub) Boot() string { return h.boot } |
| 189 | |
| 190 | func (h *Hub) roomLocked(sqid string) *room { |
| 191 | r, ok := h.rooms[sqid] |
| 192 | if !ok { |
| 193 | r = &room{sqid: sqid, peers: map[PeerID]*Participation{}, gens: map[pairKey]uint64{}} |
| 194 | h.rooms[sqid] = r |
| 195 | } |
| 196 | return r |
| 197 | } |
| 198 | |
| 199 | type OpenResult struct { |
| 200 | Peer PeerID |
| 201 | Sub *Sub |
| 202 | Existing PeerID |
| 203 | // ReportToken is the client's authority to send quality and problem |
| 204 | // reports (spec §15.26), carried in its hello. |
| 205 | ReportToken string |
| 206 | } |
| 207 | |
| 208 | // Open is a fresh stream: a new unjoined participation, subject to the |
| 209 | // room, session and person limits, all checked before allocation. |
| 210 | func (h *Hub) Open(sqid, session string, personID int64, name string, kind Kind) (OpenResult, error) { |
| 211 | h.mu.Lock() |
| 212 | defer h.mu.Unlock() |
| 213 | r := h.rooms[sqid] // nil until this room holds someone |
| 214 | if r != nil && r.ended { |
| 215 | return OpenResult{}, ErrUnknownPeer |
| 216 | } |
| 217 | if existing, ok := h.bySession[sqid][session]; ok { |
| 218 | // A participation whose resume window has passed is not a |
| 219 | // conflict: §15.7 refuses its resume the moment the window |
| 220 | // passes, so answering 409 "open in another tab" until the tick |
| 221 | // caught up told a browser that had just been refused a resume |
| 222 | // that it was competing with itself (Codex S2-2). It is ended |
| 223 | // here, on the same terms the tick would have ended it on — |
| 224 | // nothing can be listening for the lost event, and the room is |
| 225 | // owed the departure. |
| 226 | if p := h.lapsedLocked(r, existing); p != nil { |
| 227 | wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer) |
| 228 | delete(h.tombstones, p.Peer) |
| 229 | h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting) |
| 230 | } else { |
| 231 | return OpenResult{Existing: existing}, ErrSessionHasPeer |
| 232 | } |
| 233 | } |
| 234 | if r != nil && len(r.peers) >= MaxParticipations { |
| 235 | return OpenResult{}, ErrRoomFull |
| 236 | } |
| 237 | if h.byPerson[personID] >= MaxStreamsPerPerson { |
| 238 | return OpenResult{}, ErrPersonStreams |
| 239 | } |
| 240 | r = h.roomLocked(sqid) |
| 241 | p := &Participation{ |
| 242 | Peer: PeerID(randomID()), Room: sqid, Session: session, PersonID: personID, Name: name, Kind: kind, |
| 243 | OpenedAt: h.cfg.Now(), sub: newSub(), seqOut: map[PeerID]uint64{}, lastSeen: map[PeerID]uint64{}, |
| 244 | } |
| 245 | h.enrolReportLocked(r, p) // before p joins r.peers: an empty room starts a new call |
| 246 | |
| 247 | r.peers[p.Peer] = p |
| 248 | if h.bySession[sqid] == nil { |
| 249 | h.bySession[sqid] = map[string]PeerID{} |
| 250 | } |
| 251 | h.bySession[sqid][session] = p.Peer |
| 252 | h.byPerson[personID]++ |
| 253 | return OpenResult{Peer: p.Peer, Sub: p.sub, ReportToken: p.reportToken}, nil |
| 254 | } |
| 255 | |
| 256 | // Find answers a peer only through its own room and session. |
| 257 | func (h *Hub) Find(sqid, session string, peer PeerID) (*Participation, bool) { |
| 258 | h.mu.Lock() |
| 259 | defer h.mu.Unlock() |
| 260 | p, ok := h.findLocked(sqid, session, peer) |
| 261 | return p, ok |
| 262 | } |
| 263 | |
| 264 | func (h *Hub) findLocked(sqid, session string, peer PeerID) (*Participation, bool) { |
| 265 | r, ok := h.rooms[sqid] |
| 266 | if !ok { |
| 267 | return nil, false |
| 268 | } |
| 269 | p, ok := r.peers[peer] |
| 270 | if !ok || p.Session != session { |
| 271 | return nil, false |
| 272 | } |
| 273 | return p, true |
| 274 | } |
| 275 | |
| 276 | // lapsedLocked is the participation this room holds under peer, when it |
| 277 | // is detached and past its resume window — the state §15.7 has already |
| 278 | // refused a resume in, and which only the tick used to clear. It is nil |
| 279 | // for a live one, and for a peer this room does not hold. |
| 280 | func (h *Hub) lapsedLocked(r *room, peer PeerID) *Participation { |
| 281 | if r == nil { |
| 282 | return nil |
| 283 | } |
| 284 | p, ok := r.peers[peer] |
| 285 | if !ok { |
| 286 | return nil |
| 287 | } |
| 288 | if p.sub != nil || p.resumeUntil.IsZero() || !h.cfg.Now().After(p.resumeUntil) { |
| 289 | return nil |
| 290 | } |
| 291 | return p |
| 292 | } |
| 293 | |
| 294 | // LiveForSession names the peer this session holds in the room, whether |
| 295 | // attached or within its resume window (spec §4.2). |
| 296 | // |
| 297 | // Past that window it holds nothing: the participation is still in the |
| 298 | // hub until the tick, but a resume of it is already refused (§15.7) and |
| 299 | // a fresh stream now ends it and takes its place (Open), so nothing that |
| 300 | // asks "does this session hold a peer here" may still be told yes. |
| 301 | func (h *Hub) LiveForSession(sqid, session string) (PeerID, bool) { |
| 302 | h.mu.Lock() |
| 303 | defer h.mu.Unlock() |
| 304 | id, ok := h.bySession[sqid][session] |
| 305 | if !ok || h.lapsedLocked(h.rooms[sqid], id) != nil { |
| 306 | return "", false |
| 307 | } |
| 308 | return id, true |
| 309 | } |
| 310 | |
| 311 | // LiveKindForSession is LiveForSession plus the kind that participation |
| 312 | // currently carries — the kind the last tick or the last request wrote, |
| 313 | // never a fresh standing read, which is the app's to do. |
| 314 | // |
| 315 | // It exists for the page authority of spec §2.5: a live participation |
| 316 | // stands in for standing only where §2.4 says the participation is now |
| 317 | // the authority, and that is an admitted **guest** whose grant lapsed |
| 318 | // (ruling 14). A member who lost the room is dropped by the next thing |
| 319 | // they do, so their participation must not let them keep the page. |
| 320 | func (h *Hub) LiveKindForSession(sqid, session string) (PeerID, Kind, bool) { |
| 321 | h.mu.Lock() |
| 322 | defer h.mu.Unlock() |
| 323 | id, ok := h.bySession[sqid][session] |
| 324 | if !ok || h.lapsedLocked(h.rooms[sqid], id) != nil { |
| 325 | return "", 0, false |
| 326 | } |
| 327 | p, ok := h.findLocked(sqid, session, id) |
| 328 | if !ok { |
| 329 | return "", 0, false |
| 330 | } |
| 331 | return id, p.Kind, true |
| 332 | } |
| 333 | |
| 334 | // PeerNameFor answers target's display name exactly when the caller |
| 335 | // (this session, in this room) could go on to remove it: an admitted |
| 336 | // member participation, and target present in the same room and not |
| 337 | // the caller's own. This is Remove's own authority (admission.go), |
| 338 | // read here without mutating anything, so a confirm page can never |
| 339 | // name — or offer to remove — a peer the caller could not actually |
| 340 | // remove. |
| 341 | func (h *Hub) PeerNameFor(sqid, session string, target PeerID) (string, bool) { |
| 342 | h.mu.Lock() |
| 343 | defer h.mu.Unlock() |
| 344 | actor, ok := h.bySession[sqid][session] |
| 345 | if !ok { |
| 346 | return "", false |
| 347 | } |
| 348 | a, ok := h.findLocked(sqid, session, actor) |
| 349 | if !ok || !a.Admitted || a.Kind != KindMember { |
| 350 | return "", false |
| 351 | } |
| 352 | r := h.rooms[sqid] |
| 353 | tp, ok := r.peers[target] |
| 354 | if !ok || tp.Peer == a.Peer { |
| 355 | return "", false |
| 356 | } |
| 357 | return tp.Name, true |
| 358 | } |
| 359 | |
| 360 | // Detach starts the resume window for a participation whose stream |
| 361 | // closed. Its subscriber is closed; a resume attaches a new one. |
| 362 | // |
| 363 | // sub is the subscriber the caller was reading, and the detach happens |
| 364 | // only while that is still the participation's attachment. A stream |
| 365 | // handler calls this on every exit, including the exit a Resume from a |
| 366 | // second stream caused (lost{replaced}), and that displaced handler |
| 367 | // must not detach the replacement that took its place. A nil sub |
| 368 | // detaches whatever is attached, for a caller that holds no subscriber. |
| 369 | func (h *Hub) Detach(sqid string, peer PeerID, sub *Sub) { |
| 370 | h.mu.Lock() |
| 371 | defer h.mu.Unlock() |
| 372 | r, ok := h.rooms[sqid] |
| 373 | if !ok { |
| 374 | return |
| 375 | } |
| 376 | p, ok := r.peers[peer] |
| 377 | if !ok { |
| 378 | return |
| 379 | } |
| 380 | if sub != nil && p.sub != sub { |
| 381 | return |
| 382 | } |
| 383 | if p.sub != nil { |
| 384 | p.sub.terminate(Message{Type: TypeLost, Reason: ReasonBuffer}) |
| 385 | p.sub = nil |
| 386 | } |
| 387 | p.resumeUntil = h.cfg.Now().Add(ResumeWindow) |
| 388 | } |
| 389 | |
| 390 | // endLocked removes a participation for a terminal reason, closes its |
| 391 | // stream with lost{reason}, leaves a tombstone, and returns whether it |
| 392 | // was admitted or waiting so the caller can announce the departure. |
| 393 | func (h *Hub) endLocked(r *room, p *Participation, reason Reason) (wasAdmitted, wasWaiting bool) { |
| 394 | wasAdmitted, wasWaiting = p.Admitted, p.Joined && !p.Admitted |
| 395 | if p.sub != nil { |
| 396 | p.sub.terminate(Message{Type: TypeLost, Reason: reason}) |
| 397 | p.sub = nil |
| 398 | } |
| 399 | delete(r.peers, p.Peer) |
| 400 | h.endReportLocked(p) |
| 401 | if h.bySession[r.sqid][p.Session] == p.Peer { |
| 402 | delete(h.bySession[r.sqid], p.Session) |
| 403 | } |
| 404 | if h.byPerson[p.PersonID] > 0 { |
| 405 | h.byPerson[p.PersonID]-- |
| 406 | } |
| 407 | h.tombstones[p.Peer] = tombstone{room: r.sqid, session: p.Session, reason: reason, at: h.cfg.Now()} |
| 408 | for k := range r.gens { |
| 409 | if k.a == p.Peer || k.b == p.Peer { |
| 410 | delete(r.gens, k) |
| 411 | } |
| 412 | } |
| 413 | return |
| 414 | } |
| 415 | |
| 416 | // standingCheck is one joined participation whose standing the tick must |
| 417 | // re-read, and the answer it came back with. |
| 418 | type standingCheck struct { |
| 419 | room string |
| 420 | peer PeerID |
| 421 | personID int64 |
| 422 | kind Kind |
| 423 | ok bool |
| 424 | } |
| 425 | |
| 426 | // Tick runs the fifteen-second housekeeping: unjoined and lapsed |
| 427 | // participations are dropped, tombstones expire, and every joined |
| 428 | // participation's kind is re-derived from standing, applying the |
| 429 | // transition table of spec §2.4. |
| 430 | func (h *Hub) Tick(ctx context.Context) { |
| 431 | h.mu.Lock() |
| 432 | now := h.cfg.Now() |
| 433 | for _, r := range h.rooms { |
| 434 | for _, p := range r.peers { |
| 435 | switch { |
| 436 | case !p.Joined && p.sub != nil && now.Sub(p.OpenedAt) > UnjoinedWindow: |
| 437 | h.endLocked(r, p, ReasonBuffer) // dropped quietly; the browser sees a closed stream |
| 438 | delete(h.tombstones, p.Peer) |
| 439 | case p.sub == nil && !p.resumeUntil.IsZero() && now.After(p.resumeUntil): |
| 440 | wasAdmitted, wasWaiting := h.endLocked(r, p, ReasonBuffer) |
| 441 | delete(h.tombstones, p.Peer) |
| 442 | h.announceDepartureLocked(r, p, wasAdmitted, wasWaiting) |
| 443 | } |
| 444 | } |
| 445 | } |
| 446 | for id, t := range h.tombstones { |
| 447 | if now.Sub(t.at) > TombstoneLife { |
| 448 | delete(h.tombstones, id) |
| 449 | } |
| 450 | } |
| 451 | h.sweepReportsLocked(now) |
| 452 | var checks []standingCheck |
| 453 | for _, r := range h.rooms { |
| 454 | for _, p := range r.peers { |
| 455 | if p.Joined { |
| 456 | checks = append(checks, standingCheck{room: r.sqid, peer: p.Peer, personID: p.PersonID}) |
| 457 | } |
| 458 | } |
| 459 | } |
| 460 | h.mu.Unlock() |
| 461 | |
| 462 | // Standing is a database read, so it happens with the lock down. |
| 463 | answered := make([]standingCheck, 0, len(checks)) |
| 464 | for _, c := range checks { |
| 465 | kind, ok, err := h.cfg.StandingFor(ctx, c.room, c.personID) |
| 466 | if err != nil { |
| 467 | h.cfg.Logger.Warn("hub: standing lookup failed on tick", |
| 468 | "room", c.room, "peer", c.peer, "error", err) |
| 469 | continue // this participation waits for the next tick |
| 470 | } |
| 471 | c.kind, c.ok = kind, ok |
| 472 | answered = append(answered, c) |
| 473 | } |
| 474 | |
| 475 | h.mu.Lock() |
| 476 | defer h.mu.Unlock() |
| 477 | for _, c := range answered { |
| 478 | r, found := h.rooms[c.room] |
| 479 | if !found { |
| 480 | continue |
| 481 | } |
| 482 | p, found := r.peers[c.peer] |
| 483 | if !found || !p.Joined { |
| 484 | continue // it ended while standing was being read |
| 485 | } |
| 486 | switch { |
| 487 | case !c.ok: |
| 488 | // A member loses the room, and so does a waiting guest whose |
| 489 | // grant lapsed; an admitted guest is untouched (ruling 14). |
| 490 | h.standingLostLocked(r, p) |
| 491 | case p.Admitted: |
| 492 | h.setKindLocked(r, p, c.kind) // a change refreshes the rosters |
| 493 | default: |
| 494 | // Waiting. A promotion to member is admission under the cap, |
| 495 | // exactly as an admit would be; a refusal leaves them waiting |
| 496 | // as a member, free to retry with join. |
| 497 | h.setKindLocked(r, p, c.kind) |
| 498 | if c.kind == KindMember && !h.admitLocked(r, p) { |
| 499 | h.sendLocked(p, Message{Type: TypeFull}) |
| 500 | } |
| 501 | } |
| 502 | } |
| 503 | } |
| 504 | |
| 505 | // announceDepartureLocked tells the room that a participation is gone, |
| 506 | // in the shape its state earns: an admitted peer yields gone to admitted |
| 507 | // peers, a waiting one yields knock-gone to admitted members, an |
| 508 | // unjoined one yields nothing (spec §4.3). |
| 509 | func (h *Hub) announceDepartureLocked(r *room, p *Participation, wasAdmitted, wasWaiting bool) { |
| 510 | switch { |
| 511 | case wasAdmitted: |
| 512 | h.broadcastLocked(r, Message{Type: TypeGone, From: p.Peer}, admittedPeers) |
| 513 | case wasWaiting: |
| 514 | h.broadcastLocked(r, Message{Type: TypeKnockGone, From: p.Peer}, admittedMembers) |
| 515 | } |
| 516 | } |
| 517 | |
| 518 | type audience func(*Participation) bool |
| 519 | |
| 520 | func admittedPeers(p *Participation) bool { return p.Admitted } |
| 521 | func admittedMembers(p *Participation) bool { return p.Admitted && p.Kind == KindMember } |
| 522 | |
| 523 | // sendLocked is one event to one participation, which is a no-op while |
| 524 | // its stream is detached and waiting to resume. |
| 525 | func (h *Hub) sendLocked(p *Participation, m Message) { |
| 526 | if p.sub != nil { |
| 527 | p.sub.send(m) |
| 528 | } |
| 529 | } |
| 530 | |
| 531 | func (h *Hub) broadcastLocked(r *room, m Message, who audience) { |
| 532 | for _, q := range r.peers { |
| 533 | if who(q) && q.sub != nil { |
| 534 | q.sub.send(m) |
| 535 | } |
| 536 | } |
| 537 | } |
| 538 | |
| 539 | // deliver is one event to one peer found by room and id, for the events |
| 540 | // the hub raises about the room. A targeted message never comes this |
| 541 | // way: Relay holds the recipient's participation already, because it |
| 542 | // must also ring the message (see relay.go). |
| 543 | func (h *Hub) deliver(sqid string, to PeerID, m Message) { |
| 544 | h.mu.Lock() |
| 545 | defer h.mu.Unlock() |
| 546 | h.deliverLocked(sqid, to, m) |
| 547 | } |
| 548 | |
| 549 | func (h *Hub) deliverLocked(sqid string, to PeerID, m Message) { |
| 550 | r, ok := h.rooms[sqid] |
| 551 | if !ok { |
| 552 | return |
| 553 | } |
| 554 | p, ok := r.peers[to] |
| 555 | if !ok { |
| 556 | return |
| 557 | } |
| 558 | if p.sub != nil { |
| 559 | p.sub.send(m) |
| 560 | } |
| 561 | } |
| 562 | |
amadan