Method drainTx

internal/session/session.go:294–412

(maxPayload, maxFrames int, withSnapshot bool)

}

func (s *Session) drainTx(maxPayload, maxFrames int, withSnapshot bool) ([]*frame.Frame, *DrainSnapshot) {
	s.mu.Lock()
	defer s.mu.Unlock()

	var snap *DrainSnapshot
	if withSnapshot {
		snap = &DrainSnapshot{
			synNeeded:     s.synNeeded,
			txBuf:         s.txBuf,
			txSeq:         s.txSeq,
			finSent:       s.finSent,
			finSentAt:     s.finSentAt,
			firstQueuedAt: s.firstQueuedAt,
		}
	}

	if !s.synNeeded && len(s.txBuf) == 0 && !(s.closeReq && !s.finSent) {
		return nil, nil
	}

	// Estimate capacity up front to avoid repeated slice growth under large
	// uploads/downloads that split into many payload chunks.
	estFrames := 0
	if s.synNeeded {
		estFrames++
	}
	if len(s.txBuf) > 0 {
		if maxPayload <= 0 {
			maxPayload = len(s.txBuf)
		}
		// First data chunk may ride on SYN, so payload-only frame count is
		// bounded by ceil(len(txBuf)/maxPayload).
		estFrames += (len(s.txBuf) + maxPayload - 1) / maxPayload
	}
	if s.closeReq && !s.finSent {
		estFrames++
	}
	if maxFrames > 0 && estFrames > maxFrames {
		estFrames = maxFrames
	}
	frames := make([]*frame.Frame, 0, estFrames)

	canAppend := func() bool {
		return maxFrames <= 0 || len(frames) < maxFrames
	}

	// SYN (possibly with first chunk of payload).
	if s.synNeeded && canAppend() {
		f := &frame.Frame{
			SessionID: s.ID,
			Seq:       s.txSeq,
			Flags:     frame.FlagSYN,
			Target:    s.Target,
		}
		s.txSeq++
		s.synNeeded = false
		if len(s.txBuf) > 0 {
			n := len(s.txBuf)