| 1 | package hub |
| 2 | |
| 3 | import "sync/atomic" |
| 4 | |
| 5 | const subBuffer = 64 |
| 6 | |
| 7 | // Sub is one stream's outbound channel. A subscriber that falls 64 |
| 8 | // events behind is declared lost and told so (spec §4.4), as in Sheets. |
| 9 | type Sub struct { |
| 10 | ch chan Message |
| 11 | lost atomic.Bool |
| 12 | done chan struct{} |
| 13 | } |
| 14 | |
| 15 | func newSub() *Sub { return &Sub{ch: make(chan Message, subBuffer), done: make(chan struct{})} } |
| 16 | |
| 17 | func (s *Sub) C() <-chan Message { return s.ch } |
| 18 | func (s *Sub) Lost() bool { return s.lost.Load() } |
| 19 | |
| 20 | // send is non-blocking; a full buffer declares the subscriber lost, |
| 21 | // drains room for the lost event, and closes. |
| 22 | func (s *Sub) send(m Message) { |
| 23 | select { |
| 24 | case <-s.done: |
| 25 | return |
| 26 | default: |
| 27 | } |
| 28 | select { |
| 29 | case s.ch <- m: |
| 30 | default: |
| 31 | s.terminate(Message{Type: TypeLost, Reason: ReasonBuffer}) |
| 32 | } |
| 33 | } |
| 34 | |
| 35 | func (s *Sub) terminate(final Message) { |
| 36 | if s.lost.Swap(true) { |
| 37 | return |
| 38 | } |
| 39 | close(s.done) |
| 40 | // The final event must land, but nothing already queued may be |
| 41 | // dropped to make room for it: a peer ended with a real reason |
| 42 | // still gets the events it had not read yet. Only a full buffer |
| 43 | // costs one, and by then the subscriber is lost anyway. |
| 44 | // |
| 45 | // Every operation here is non-blocking, and that is the whole point |
| 46 | // (Codex S2-4). This runs with h.mu held, and a failed send proves |
| 47 | // the buffer was full at that instant only: the stream's own reader |
| 48 | // is not excluded by the lock and can drain the channel between the |
| 49 | // two statements. An unconditional receive would then block on an |
| 50 | // empty channel, holding the hub's one lock against every other |
| 51 | // room in the process, until the reader happened to send something |
| 52 | // nobody is sending. So: try, evict one if there is one, try again. |
| 53 | // Two attempts is the ceiling — the hub is the only sender and it |
| 54 | // holds the lock, so a send that fails after a successful eviction |
| 55 | // cannot happen, and dropping the final event is still better than |
| 56 | // deadlocking the hub if it ever does. |
| 57 | for i := 0; i < 2; i++ { |
| 58 | select { |
| 59 | case s.ch <- final: |
| 60 | close(s.ch) |
| 61 | return |
| 62 | default: |
| 63 | } |
| 64 | select { |
| 65 | case <-s.ch: // full a moment ago; one event makes room |
| 66 | default: // the reader drained it: the retry above will fit |
| 67 | } |
| 68 | } |
| 69 | close(s.ch) |
| 70 | } |
| 71 | |
| 72 | func (s *Sub) Close() { s.terminate(Message{Type: TypeLost, Reason: ReasonBuffer}) } |
| 73 | |
amadan