deliverRx performs in-order reassembly and delivers payloads to RxChan. Called exclusively by rxLoop. Returns true when a FIN frame is processed and the session's rx side is done.
(f *frame.Frame)
| 494 | // Called exclusively by rxLoop. Returns true when a FIN frame is processed |
| 495 | // and the session's rx side is done. |
| 496 | func (s *Session) deliverRx(f *frame.Frame) bool { |
| 497 | s.mu.Lock() |
| 498 | if s.rxClosed { |
| 499 | s.mu.Unlock() |
| 500 | return true |
| 501 | } |
| 502 | if f.Seq < s.rxSeq { |
| 503 | s.mu.Unlock() |
| 504 | return false |
| 505 | } |
| 506 | if f.Seq > s.rxSeq { |
| 507 | s.rxQueue[f.Seq] = f |
| 508 | s.mu.Unlock() |
| 509 | return false |
| 510 | } |
| 511 | |
| 512 | var toSend [][]byte |
| 513 | var closeAfter bool |
| 514 | for { |
| 515 | if len(f.Payload) > 0 { |
| 516 | toSend = append(toSend, f.Payload) |
| 517 | } |
| 518 | s.rxSeq++ |
| 519 | if f.HasFlag(frame.FlagFIN) { |
| 520 | s.rxClosed = true |
| 521 | closeAfter = true |
| 522 | break |
| 523 | } |
| 524 | next, ok := s.rxQueue[s.rxSeq] |
| 525 | if !ok { |
| 526 | break |
| 527 | } |
| 528 | delete(s.rxQueue, s.rxSeq) |
| 529 | f = next |
| 530 | } |
| 531 | s.mu.Unlock() |
| 532 | |
| 533 | for _, p := range toSend { |
| 534 | select { |
| 535 | case s.RxChan <- p: |
| 536 | case <-s.rxDone: |
| 537 | // Session was killed (e.g. rxInbox overflow). If a FIN was already |
| 538 | // decoded, close RxChan now; otherwise rxLoop's defer handles it. |
| 539 | if closeAfter { |
| 540 | close(s.RxChan) |
| 541 | } |
| 542 | return true |
| 543 | } |
| 544 | } |
| 545 | if closeAfter { |
| 546 | close(s.RxChan) |
| 547 | s.Stop() |
| 548 | } |
| 549 | return closeAfter |
| 550 | } |