rxLoop is a per-session goroutine that delivers frames from rxInbox to RxChan in sequence order. Running it independently from poll workers means a slow SOCKS reader on one session cannot stall frame delivery for any other session.
()
| 104 | // in sequence order. Running it independently from poll workers means a slow |
| 105 | // SOCKS reader on one session cannot stall frame delivery for any other session. |
| 106 | func (s *Session) rxLoop() { |
| 107 | defer func() { |
| 108 | // Guarantee RxChan is closed when rxLoop exits for any reason (rxDone |
| 109 | // fired, FIN processed, or session killed via ProcessRx overflow). This |
| 110 | // unblocks any goroutine ranging over RxChan without a separate close call. |
| 111 | s.mu.Lock() |
| 112 | if !s.rxClosed { |
| 113 | s.rxClosed = true |
| 114 | close(s.RxChan) |
| 115 | } |
| 116 | s.mu.Unlock() |
| 117 | }() |
| 118 | for { |
| 119 | select { |
| 120 | case f := <-s.rxInbox: |
| 121 | if s.deliverRx(f) { |
| 122 | return |
| 123 | } |
| 124 | case <-s.rxDone: |
| 125 | return |
| 126 | } |
| 127 | } |
| 128 | } |
| 129 | |
| 130 | // EnqueueTx appends bytes to the session's tx buffer. Blocks while the buffer |
| 131 | // exceeds TxBufHighWater. Safe to call concurrently with DrainTx. |