package quic

import (
	
	
	
	
	
	

	
	
	
	
	
)

// A SendStream is a unidirectional Send Stream.
type SendStream struct {
	mutex sync.Mutex

	numOutstandingFrames int64 // outstanding STREAM and RESET_STREAM frames
	retransmissionQueue  []*wire.StreamFrame

	ctx       context.Context
	ctxCancel context.CancelCauseFunc

	streamID protocol.StreamID
	sender   streamSender

	// reliableSize is the portion of the stream that needs to be transmitted reliably,
	// even if the stream is cancelled.
	// This requires the peer to support RESET_STREAM_AT.
	// This value should not be accessed directly, but only through the reliableOffset method.
	// This method returns 0 if the peer doesn't support the RESET_STREAM_AT extension.
	reliableSize protocol.ByteCount
	writeOffset  protocol.ByteCount

	shutdownErr            error
	resetErr               *StreamError
	queuedResetStreamFrame *wire.ResetStreamFrame

	supportsResetStreamAt bool
	finishedWriting       bool // set once Close() is called
	finSent               bool // set when a STREAM_FRAME with FIN bit has been sent
	// Set when the application knows about the cancellation.
	// This can happen because the application called CancelWrite,
	// or because Write returned the error (for remote cancellations).
	cancellationFlagged bool
	completed           bool // set when this stream has been reported to the streamSender as completed

	dataForWriting  []byte // during a Write() call, this slice is the part of p that still needs to be sent out
	writeBuffer     []byte
	writeBufferHead int
	// writeBufferLimit is the demand-grown buffer limit; zero means
	// sendStreamWriteBufferSize. It doubles toward
	// sendStreamWriteBufferMaxSize only on the full-buffer write path.
	writeBufferLimit int
	active           bool
	writesInEpisode  uint16
	burstUntil       monotime.Time
	corkPending      bool
	activationTimer  *time.Timer
	activationGen    uint64

	writeChan   chan struct{}
	writeActive bool
	writeWake   chan struct{}
	deadline    monotime.Time

	flowController flowcontrol.StreamFlowController
}

const (
	// sendStreamWriteBufferSize is the initial write buffer limit.
	sendStreamWriteBufferSize = 4096
	// sendStreamWriteBufferMaxSize caps the demand-grown write buffer
	// limit. Only streams under sustained full-buffer pressure reach it;
	// it bounds the worst-case buffered-unsent bytes on cancel.
	sendStreamWriteBufferMaxSize = 65536
	// maxBufferedWriteSize is the largest single write that buffers (and
	// grows the buffer). Larger writes keep the direct blocked-writer
	// path, whose frames copy straight from the caller's slice — routing
	// them through the buffer would add a full extra copy per write.
	maxBufferedWriteSize = sendStreamWriteBufferMaxSize / 4
)

var (
	_ io.ReaderFrom            = &SendStream{}
	_ streamControlFrameGetter = &SendStream{}
	_ outgoingStream           = &SendStream{}
	_ sendStreamFrameHandler   = &SendStream{}
)

func newSendStream(
	 context.Context,
	 protocol.StreamID,
	 streamSender,
	 flowcontrol.StreamFlowController,
	 bool,
) *SendStream {
	 := &SendStream{
		streamID:              ,
		sender:                ,
		flowController:        ,
		writeChan:             make(chan struct{}, 1),
		writeWake:             make(chan struct{}, 1),
		supportsResetStreamAt: ,
	}
	.ctx, .ctxCancel = context.WithCancelCause()
	return 
}

// StreamID returns the stream ID.
func ( *SendStream) () StreamID {
	return .streamID // same for receiveStream and sendStream
}

// Write writes data to the stream.
// Write can be made to time out using [SendStream.SetWriteDeadline].
// If the stream was canceled, the error is a [StreamError].
func ( *SendStream) ( []byte) (int, error) {
	// Concurrent use of Write is not permitted (and doesn't make any sense),
	// but sometimes people do it anyway.
	// Make sure that we only execute one call at any given time to avoid hard to debug failures.
	.mutex.Lock()
	// Steady-state fast path: no concurrent write episode, no error or
	// deadline condition, stream already active, and the write fits in the
	// buffer. The fast path never sets writeActive, so it creates no
	// waiters and must wake none; any condition failing falls through to
	// the general path unchanged.
	if !.writeActive && .resetErr == nil && .shutdownErr == nil &&
		!.finishedWriting && .deadline.IsZero() && len() > 0 &&
		len() <= maxBufferedWriteSize &&
		.active && .growWriteBufferFor(len()) {
		.appendWriteBuffer()
		if .writesInEpisode < ^uint16(0) {
			.writesInEpisode++
		}
		.mutex.Unlock()
		return len(), nil
	}
	for .writeActive {
		.mutex.Unlock()
		<-.writeWake
		.mutex.Lock()
	}
	.writeActive = true

	, ,  := .writeLocked()
	if  {
		.sender.onStreamCompleted(.streamID)
	}
	return , 
}

// ReadFrom implements [io.ReaderFrom]. It reads from r until EOF or error,
// writing to the stream in buffer-sized chunks so each write stays on the
// buffered fast path. Data is copied into stream-owned storage before each
// chunk write returns.
func ( *SendStream) ( io.Reader) (int64, error) {
	var  int64
	 := make([]byte, sendStreamWriteBufferSize)
	for {
		,  := .Read()
		if  > 0 {
			,  := .Write([:])
			 += int64()
			if  != nil {
				return , 
			}
		}
		if  == io.EOF {
			return , nil
		}
		if  != nil {
			return , 
		}
	}
}

// Writev writes the buffers in order, amortizing the per-call lock and
// bookkeeping across the vector. It returns the total number of bytes
// written and advances bufs to reflect exactly what was consumed,
// including a partially written element, so the caller can resume after a
// short write. The stream copies data into owned storage before Writev
// returns; the caller may reuse the underlying slices immediately.
//
// Writev does not hold the stream for the whole vector. Elements too large
// for the write buffer are delegated to Write, and another writer may
// interleave between elements. The delivered byte stream is identical to
// the equivalent sequence of Write calls.
//
// To send a [net.Buffers], call Writev directly: [net.Buffers] implements
// [io.WriterTo], which [io.Copy] prefers over [io.ReaderFrom], so
// io.Copy(stream, &bufs) degrades to one Write call per element and never
// batches.
//
// For burst accounting, a Writev counts as one write for the group of
// elements appended to the write buffer plus one per delegated element, so
// a vector of mixed sizes is not accounted the same as either one Write or
// N Writes.
//
// Writev is intended to be no slower than the equivalent sequence of Write
// calls; whether it is faster depends on the batch depth and on how many
// elements exceed the write buffer.
func ( *SendStream) ( *net.Buffers) (int64, error) {
	,  := .writeVectored(*)
	 := 
	for  > 0 && len(*) > 0 {
		if  := int64(len((*)[0]));  >=  {
			 -= 
			* = (*)[1:]
			continue
		}
		(*)[0] = (*)[0][:]
		 = 0
	}
	if  > 0 && len(*) > 0 &&  == nil {
		// Unreachable today (writeVectored only stops early on error),
		// kept so a future short write cannot silently desynchronize bufs.
		 = io.ErrShortWrite
	}
	return , 
}

// writeVectored writes the elements of bufs in order. Elements that fit the
// write buffer in the steady state are appended under a single mutex hold;
// any element that does not is delegated to Write, which handles activation,
// blocking, deadlines, and errors. The delivered byte stream is identical to
// the equivalent sequence of Write calls.
func ( *SendStream) ( [][]byte) (int64, error) {
	var  int64
	.mutex.Lock()
	 := false
	for  := 0;  < len(); {
		 := []
		if len() == 0 {
			++
			continue
		}
		if !.writeActive && .resetErr == nil && .shutdownErr == nil &&
			!.finishedWriting && .deadline.IsZero() &&
			.active && len() <= maxBufferedWriteSize && .growWriteBufferFor(len()) {
			.appendWriteBuffer()
			 = true
			 += int64(len())
			++
			continue
		}
		.mutex.Unlock()
		,  := .Write()
		 += int64()
		if  != nil {
			return , 
		}
		++
		.mutex.Lock()
	}
	if  && .writesInEpisode < ^uint16(0) {
		.writesInEpisode++
	}
	.mutex.Unlock()
	return , nil
}

// writeLocked writes p. The caller holds s.mutex and owns the write operation.
func ( *SendStream) ( []byte) (bool /* is newly completed */, int, error) {
	if .resetErr != nil {
		.cancellationFlagged = true
		 := .isNewlyCompleted()
		 := .resetErr
		.finishWriteLocked()
		return , 0, 
	}
	if .shutdownErr != nil {
		 := .shutdownErr
		.finishWriteLocked()
		return false, 0, 
	}
	if .finishedWriting {
		.finishWriteLocked()
		return false, 0, fmt.Errorf("write on closed stream %d", .streamID)
	}
	if !.deadline.IsZero() && !monotime.Now().Before(.deadline) {
		.finishWriteLocked()
		return false, 0, errDeadline
	}
	if len() == 0 {
		.finishWriteLocked()
		return false, 0, nil
	}

	.dataForWriting = 

	var (
		  *time.Timer
		   int
		 bool
	)
	for {
		var  bool
		var  monotime.Time
		if .shutdownErr != nil || .resetErr != nil {
			break
		}
		// Copy a bounded tail so Write can return before the bytes are packetized.
		// Larger writes retain the direct blocked-writer path.
		 := .canBufferWrite()
		if ! && len() <= maxBufferedWriteSize {
			 = .growWriteBufferFor(len(.dataForWriting))
		}
		if  && len(.dataForWriting) > 0 {
			.appendWriteBuffer(.dataForWriting)
			.dataForWriting = nil
			 = len()
			 = true
			if .writesInEpisode < ^uint16(0) {
				.writesInEpisode++
			}
		} else {
			 = len() - len(.dataForWriting)
			 = .deadline
			if !.IsZero() {
				if !monotime.Now().Before() {
					.dataForWriting = nil
					.finishWriteLocked()
					return false, , errDeadline
				}
				if  == nil {
					 = time.NewTimer(monotime.Until())
					defer .Stop()
				} else {
					.Reset(monotime.Until())
				}
			}
			if .dataForWriting == nil || .shutdownErr != nil || .resetErr != nil {
				break
			}
		}

		 := false
		if ! {
			 = true
			if !.active {
				 = .activateOrDelayLocked()
			}
		}
		if  {
			.writeActive = false
		}
		.mutex.Unlock()
		if  {
			.sender.onHasStreamData(.streamID, )
		}
		if  {
			.wakeWriter()
			return false, , nil
		}
		if .IsZero() {
			<-.writeChan
		} else {
			select {
			case <-.writeChan:
			case <-.C:
			}
		}
		.mutex.Lock()
	}

	if  == len() {
		.finishWriteLocked()
		return false, , nil
	}
	if .shutdownErr != nil {
		 := .shutdownErr
		.finishWriteLocked()
		return false, , 
	}
	if .resetErr != nil {
		.cancellationFlagged = true
		 := .isNewlyCompleted()
		 := .resetErr
		.finishWriteLocked()
		return , , 
	}
	.finishWriteLocked()
	return false, , nil
}

func ( *SendStream) () {
	.writeActive = false
	.mutex.Unlock()
	.wakeWriter()
}

func ( *SendStream) () {
	select {
	case .writeWake <- struct{}{}:
	default:
	}
}

func ( *SendStream) () int {
	return len(.writeBuffer) - .writeBufferHead
}

// writeBufferLimitLocked returns the current demand-grown buffer limit
// without growing it. The caller holds s.mutex.
func ( *SendStream) () int {
	if .writeBufferLimit == 0 {
		return sendStreamWriteBufferSize
	}
	return .writeBufferLimit
}

// growWriteBufferFor reports whether n more bytes fit the write buffer,
// doubling the demand-grown limit toward the cap when they do not.
// Growth happens only here — on the full-buffer write path — never on
// stream open or on the drain side. The caller holds s.mutex.
func ( *SendStream) ( int) bool {
	 := .writeBufferLimitLocked()
	 := .bufferedWriteLen() + 
	for  >  &&  < sendStreamWriteBufferMaxSize {
		 *= 2
	}
	.writeBufferLimit = 
	return  <= 
}

// canBufferWrite reports room under the current limit without growing;
// the drain side uses it to decide when to wake a blocked writer.
func ( *SendStream) () bool {
	return .bufferedWriteLen()+len(.dataForWriting) <= .writeBufferLimitLocked()
}

func ( *SendStream) ( []byte) {
	if .writeBuffer == nil {
		.writeBuffer = make([]byte, 0, sendStreamWriteBufferSize)
	}
	if cap(.writeBuffer)-len(.writeBuffer) < len() {
		copy(.writeBuffer, .writeBuffer[.writeBufferHead:])
		.writeBuffer = .writeBuffer[:.bufferedWriteLen()]
		.writeBufferHead = 0
	}
	.writeBuffer = append(.writeBuffer, ...)
}

func ( *SendStream) () bool {
	if !.corkPending {
		if .burstUntil.IsZero() || !monotime.Now().Before(.burstUntil) {
			.burstUntil = 0
			.active = true
			return true
		}
		.corkPending = true
	}
	if sendStreamTailDelay <= 0 || .bufferedWriteLen() >= sendStreamActivationThreshold {
		.stopActivationTimerLocked()
		.burstUntil = 0
		.corkPending = false
		.active = true
		return true
	}
	if .activationTimer == nil {
		.activationGen++
		 := .activationGen
		.activationTimer = time.AfterFunc(sendStreamTailDelay, func() {
			.activateAfterDelay()
		})
	}
	return false
}

func ( *SendStream) () {
	if .activationTimer == nil {
		return
	}
	.activationTimer.Stop()
	.activationTimer = nil
	.activationGen++
}

func ( *SendStream) ( uint64) {
	.mutex.Lock()
	if  != .activationGen {
		.mutex.Unlock()
		return
	}
	.activationTimer = nil
	if .active || .shutdownErr != nil || .resetErr != nil ||
		(.bufferedWriteLen() == 0 && .dataForWriting == nil) {
		.mutex.Unlock()
		return
	}
	.burstUntil = 0
	.corkPending = false
	.active = true
	.mutex.Unlock()
	recordCorkTimerActivation()
	.sender.onHasStreamData(.streamID, )
}

// popStreamFrame returns the next STREAM frame that is supposed to be sent on this stream
// maxBytes is the maximum length this frame (including frame header) will have.
func ( *SendStream) ( protocol.ByteCount,  protocol.Version) ( ackhandler.StreamFrame,  *wire.StreamDataBlockedFrame,  bool) {
	.mutex.Lock()
	, ,  := .popNewOrRetransmittedStreamFrame(, )
	if  != nil {
		.numOutstandingFrames++
	}
	if ! {
		if .writesInEpisode >= sendStreamBurstMinWrites {
			.burstUntil = monotime.Now().Add(sendStreamBurstFreshness)
		} else {
			.burstUntil = 0
		}
		.writesInEpisode = 0
		.corkPending = false
		.active = false
	}
	.mutex.Unlock()

	if  == nil {
		return ackhandler.StreamFrame{}, , 
	}
	return ackhandler.StreamFrame{
		Frame:   ,
		Handler: (*sendStreamAckHandler)(),
	}, , 
}

func ( *SendStream) () {
	.mutex.Lock()
	.stopActivationTimerLocked()
	.burstUntil = 0
	.corkPending = false
	if .active {
		.mutex.Unlock()
		return
	}
	.active = true
	.mutex.Unlock()
	.sender.onHasStreamData(.streamID, )
}

func ( *SendStream) ( protocol.ByteCount,  protocol.Version) ( *wire.StreamFrame,  *wire.StreamDataBlockedFrame,  bool) {
	if .shutdownErr != nil {
		return nil, nil, false
	}
	if .resetErr != nil {
		 := .reliableOffset()
		if  == 0 || (.writeOffset >=  && len(.retransmissionQueue) == 0) {
			return nil, nil, false
		}
	}

	if len(.retransmissionQueue) > 0 {
		,  := .maybeGetRetransmission(, )
		if  != nil ||  {
			if  == nil {
				return nil, nil, true
			}
			// We always claim that we have more data to send.
			// This might be incorrect, in which case there'll be a spurious call to popStreamFrame in the future.
			return , nil, true
		}
	}

	if len(.dataForWriting) == 0 && .bufferedWriteLen() == 0 {
		if .finishedWriting && !.finSent {
			.finSent = true
			return &wire.StreamFrame{
				StreamID:       .streamID,
				Offset:         .writeOffset,
				DataLenPresent: true,
				Fin:            true,
			}, nil, false
		}
		return nil, nil, false
	}

	 := .flowController.SendWindowSize()
	if  == 0 {
		return nil, nil, true
	}

	// if the stream is canceled, only data up to the reliable size needs to be sent
	 := .reliableOffset()
	if .resetErr != nil &&  > 0 {
		 = min(, -.writeOffset)
	}
	,  := .popNewStreamFrame(, , )
	if  == nil {
		return nil, nil, 
	}
	if .DataLen() > 0 {
		.writeOffset += .DataLen()
		.flowController.AddBytesSent(.DataLen())
	}
	if .resetErr != nil && .writeOffset >=  {
		 = false
	}
	var  *wire.StreamDataBlockedFrame
	// If the entire send window is used, the stream might have become blocked on stream-level flow control.
	// This is not guaranteed though, because the stream might also have been blocked on connection-level flow control.
	if .DataLen() ==  && .flowController.IsNewlyBlocked() {
		 = &wire.StreamDataBlockedFrame{StreamID: .streamID, MaximumStreamData: .writeOffset}
	}
	.Fin = .finishedWriting && .dataForWriting == nil && .bufferedWriteLen() == 0 && !.finSent
	if .Fin {
		.finSent = true
	}
	return , , 
}

// popNewStreamFrame returns a new STREAM frame to send for this stream
// hasMoreData says if there's more data to send, *not* taking into account the reliable size
func ( *SendStream) (,  protocol.ByteCount,  protocol.Version) ( *wire.StreamFrame,  bool) {
	 := wire.GetStreamFrame()
	.Fin = false
	.StreamID = .streamID
	.Offset = .writeOffset
	.DataLenPresent = true
	.Data = .Data[:0]

	 = min(, .MaxDataLen(, ))
	if  == 0 {
		.PutBack()
		return nil, true
	}
	if  := min(.bufferedWriteLen(), int());  > 0 {
		.Data = .Data[:]
		copy(.Data, .writeBuffer[.writeBufferHead:.writeBufferHead+])
		.writeBufferHead += 
		if .writeBufferHead == len(.writeBuffer) {
			.writeBuffer = .writeBuffer[:0]
			.writeBufferHead = 0
		}
		.signalWrite()
		 = .bufferedWriteLen() > 0 || .dataForWriting != nil
	} else {
		.getDataForWriting(, )
		 = .dataForWriting != nil || .finishedWriting
	}
	if len(.Data) == 0 && !.Fin {
		.PutBack()
		return nil, 
	}
	return , 
}

func ( *SendStream) ( protocol.ByteCount,  protocol.Version) (*wire.StreamFrame, bool /* has more retransmissions */) {
	 := .retransmissionQueue[0]
	,  := .MaybeSplitOffFrame(, )
	if  {
		return , true
	}
	.retransmissionQueue = .retransmissionQueue[1:]
	return , len(.retransmissionQueue) > 0
}

func ( *SendStream) ( *wire.StreamFrame,  protocol.ByteCount) {
	if protocol.ByteCount(len(.dataForWriting)) <=  {
		.Data = .Data[:len(.dataForWriting)]
		copy(.Data, .dataForWriting)
		.dataForWriting = nil
		.signalWrite()
		return
	}
	.Data = .Data[:]
	copy(.Data, .dataForWriting)
	.dataForWriting = .dataForWriting[:]
	if .canBufferWrite() {
		.signalWrite()
	}
}

func ( *SendStream) () bool {
	if .completed {
		return false
	}
	if .bufferedWriteLen() > 0 {
		return false
	}
	// We need to keep the stream around until all frames have been sent and acknowledged.
	if .numOutstandingFrames > 0 || len(.retransmissionQueue) > 0 || .queuedResetStreamFrame != nil {
		return false
	}
	// The stream is completed if we sent the FIN.
	if .finSent {
		.completed = true
		return true
	}
	// The stream is also completed if:
	// 1. the application called CancelWrite, or
	// 2. we received a STOP_SENDING, and
	// 		* the application consumed the error via Write, or
	//		* the application called Close
	if .resetErr != nil && (.cancellationFlagged || .finishedWriting) {
		.completed = true
		return true
	}
	return false
}

// Close closes the write-direction of the stream.
// Future calls to Write are not permitted after calling Close.
// It must not be called concurrently with Write.
// It must not be called after calling CancelWrite.
func ( *SendStream) () error {
	.mutex.Lock()
	if .shutdownErr != nil || .finishedWriting {
		.mutex.Unlock()
		return nil
	}
	.finishedWriting = true
	 := .resetErr != nil
	if  {
		.cancellationFlagged = true
	}
	 := .isNewlyCompleted()
	.mutex.Unlock()

	if  {
		.sender.onStreamCompleted(.streamID)
	}
	if  {
		return fmt.Errorf("close called for canceled stream %d", .streamID)
	}
	.notifyHasStreamData()

	.ctxCancel(nil)
	return nil
}

// SetReliableBoundary marks the data written to this stream so far as reliable.
// It is valid to call this function multiple times, thereby increasing the reliable size.
// It only has an effect if the peer enabled support for the RESET_STREAM_AT extension,
// otherwise, it is a no-op.
func ( *SendStream) () {
	.mutex.Lock()
	defer .mutex.Unlock()

	.reliableSize = .writeOffset
	.reliableSize += protocol.ByteCount(.bufferedWriteLen())
}

// returnFramesToPool returns all queued frames to the sync.Pool
func ( *SendStream) () {
	.stopActivationTimerLocked()
	.burstUntil = 0
	.corkPending = false
	for ,  := range .retransmissionQueue {
		.PutBack()
	}
	clear(.retransmissionQueue)
	.retransmissionQueue = nil
	.writeBuffer = nil
	.writeBufferHead = 0
}

// CancelWrite aborts sending on this stream.
// Data already written, but not yet delivered to the peer is not guaranteed to be delivered reliably.
// Write will unblock immediately, and future calls to Write will fail.
// When called multiple times it is a no-op.
// When called after Close, it aborts reliable delivery of outstanding stream data.
// Note that there is no guarantee if the peer will receive the FIN or the cancellation error first.
func ( *SendStream) ( StreamErrorCode) {
	.mutex.Lock()
	if .shutdownErr != nil {
		.mutex.Unlock()
		return
	}

	.cancellationFlagged = true

	if .resetErr != nil {
		 := .isNewlyCompleted()
		.mutex.Unlock()
		// The user has called CancelWrite. If the previous cancellation was because of a
		// STOP_SENDING, we don't need to flag the error to the user anymore.
		if  {
			.sender.onStreamCompleted(.streamID)
		}
		return
	}
	.resetErr = &StreamError{StreamID: .streamID, ErrorCode: , Remote: false}
	.ctxCancel(.resetErr)

	 := .reliableOffset()
	if  == 0 {
		.numOutstandingFrames = 0
		.returnFramesToPool()
	}
	.queuedResetStreamFrame = &wire.ResetStreamFrame{
		StreamID:  .streamID,
		FinalSize: max(.writeOffset, ),
		ErrorCode: ,
		// if the peer doesn't support the extension, the reliable offset will always be 0
		ReliableSize: ,
	}
	if  > 0 {
		if  := protocol.ByteCount(.bufferedWriteLen());  > 0 {
			 :=  - .writeOffset
			if  <= 0 {
				.writeBuffer = .writeBuffer[:0]
				.writeBufferHead = 0
			} else if  <  {
				.writeBuffer = .writeBuffer[:.writeBufferHead+int()]
			}
		}
		if len(.retransmissionQueue) > 0 {
			 := make([]*wire.StreamFrame, 0, len(.retransmissionQueue))
			for ,  := range .retransmissionQueue {
				if .Offset >=  {
					.PutBack()
					continue
				}
				if .Offset+.DataLen() <=  {
					 = append(, )
				} else {
					.Data = .Data[:-.Offset]
					 = append(, )
				}
			}
			.retransmissionQueue = 
		}
	}
	.mutex.Unlock()

	.signalWrite()
	.sender.onHasStreamControlFrame(.streamID, )
}

func ( *SendStream) () {
	.mutex.Lock()
	.supportsResetStreamAt = true
	.mutex.Unlock()
}

func ( *SendStream) ( protocol.ByteCount) {
	 := .flowController.UpdateSendWindow()
	if ! { // duplicate or reordered MAX_STREAM_DATA frame
		return
	}
	.mutex.Lock()
	 := .dataForWriting != nil || .bufferedWriteLen() > 0
	.mutex.Unlock()
	if  {
		.notifyHasStreamData()
	}
}

func ( *SendStream) () {
	.mutex.Lock()
	 := .dataForWriting != nil || .bufferedWriteLen() > 0
	.mutex.Unlock()
	if  {
		.notifyHasStreamData()
	}
}

func ( *SendStream) ( *wire.StopSendingFrame) {
	.mutex.Lock()
	if .shutdownErr != nil {
		.mutex.Unlock()
		return
	}

	// If the stream was already cancelled (either locally, or due to a previous STOP_SENDING frame),
	// there's nothing else to do.
	if .resetErr != nil && .reliableOffset() == 0 {
		.mutex.Unlock()
		return
	}
	// if the peer stopped reading from the stream, there's no need to transmit any data reliably
	.reliableSize = 0
	.numOutstandingFrames = 0
	.returnFramesToPool()
	if .resetErr == nil {
		.resetErr = &StreamError{StreamID: .streamID, ErrorCode: .ErrorCode, Remote: true}
		.ctxCancel(.resetErr)
	}
	.queuedResetStreamFrame = &wire.ResetStreamFrame{
		StreamID:  .streamID,
		FinalSize: .writeOffset,
		ErrorCode: .resetErr.ErrorCode,
	}
	.mutex.Unlock()

	.signalWrite()
	.sender.onHasStreamControlFrame(.streamID, )
}

func ( *SendStream) (monotime.Time) ( ackhandler.Frame, ,  bool) {
	.mutex.Lock()
	defer .mutex.Unlock()

	if .queuedResetStreamFrame == nil {
		return ackhandler.Frame{}, false, false
	}
	.numOutstandingFrames++
	 := ackhandler.Frame{
		Frame:   .queuedResetStreamFrame,
		Handler: (*sendStreamResetStreamHandler)(),
	}
	.queuedResetStreamFrame = nil
	return , true, false
}

func ( *SendStream) () protocol.ByteCount {
	if !.supportsResetStreamAt {
		return 0
	}
	return .reliableSize
}

// The Context is canceled as soon as the write-side of the stream is closed.
// This happens when Close() or CancelWrite() is called, or when the peer
// cancels the read-side of their stream.
// The cancellation cause is set to the error that caused the stream to
// close, or `context.Canceled` in case the stream is closed without error.
func ( *SendStream) () context.Context {
	return .ctx
}

// SetWriteDeadline sets the deadline for future Write calls
// and any currently-blocked Write call.
// Even if write times out, it may return n > 0, indicating that
// some data was successfully written.
// A zero value for t means Write will not time out.
func ( *SendStream) ( time.Time) error {
	.mutex.Lock()
	.deadline = monotime.FromTime()
	.mutex.Unlock()
	.signalWrite()
	return nil
}

// CloseForShutdown closes a stream abruptly.
// It makes Write unblock (and return the error) immediately.
// The peer will NOT be informed about this: the stream is closed without sending a FIN or RST.
func ( *SendStream) ( error) {
	.mutex.Lock()
	if .shutdownErr == nil && !.finishedWriting {
		.shutdownErr = 
		.returnFramesToPool()
	}
	.mutex.Unlock()
	.signalWrite()
}

// signalWrite performs a non-blocking send on the writeChan
func ( *SendStream) () {
	select {
	case .writeChan <- struct{}{}:
	default:
	}
}

type sendStreamAckHandler SendStream

var _ ackhandler.FrameHandler = &sendStreamAckHandler{}

func ( *sendStreamAckHandler) ( wire.Frame) {
	 := .(*wire.StreamFrame)
	.PutBack()

	.mutex.Lock()
	if .resetErr != nil && (*SendStream)().reliableOffset() == 0 {
		.mutex.Unlock()
		return
	}
	.numOutstandingFrames--
	if .numOutstandingFrames < 0 {
		panic("numOutStandingFrames negative")
	}
	 := (*SendStream)().isNewlyCompleted()
	.mutex.Unlock()

	if  {
		.sender.onStreamCompleted(.streamID)
	}
}

func ( *sendStreamAckHandler) ( wire.Frame) {
	 := .(*wire.StreamFrame)
	.mutex.Lock()
	// If the reliable size was 0 when the stream was cancelled,
	// the number of outstanding frames was immediately set to 0, and the retransmission queue was dropped.
	if .resetErr != nil && (*SendStream)().reliableOffset() == 0 {
		// Return the frame to pool since it won't be retransmitted
		.PutBack()
		.mutex.Unlock()
		return
	}
	.numOutstandingFrames--
	if .numOutstandingFrames < 0 {
		panic("numOutStandingFrames negative")
	}

	if .resetErr != nil && (*SendStream)().reliableOffset() > 0 {
		// If the stream was reset, and this frame is beyond the reliable offset,
		// it doesn't need to be retransmitted.
		if .Offset >= (*SendStream)().reliableOffset() {
			.PutBack()
			// If this frame was the last one tracked, losing it might cause the stream to be completed.
			 := (*SendStream)().isNewlyCompleted()
			.mutex.Unlock()
			if  {
				.sender.onStreamCompleted(.streamID)
			}
			return
		}
		// If the payload of the frame extends beyond the reliable size,
		// truncate the frame to the reliable size.
		if .Offset+.DataLen() > (*SendStream)().reliableOffset() {
			.Data = .Data[:(*SendStream)().reliableOffset()-.Offset]
		}
	}

	.DataLenPresent = true
	.retransmissionQueue = append(.retransmissionQueue, )
	.mutex.Unlock()

	(*SendStream)().notifyHasStreamData()
}

type sendStreamResetStreamHandler SendStream

var _ ackhandler.FrameHandler = &sendStreamResetStreamHandler{}

func ( *sendStreamResetStreamHandler) ( wire.Frame) {
	 := .(*wire.ResetStreamFrame)
	.mutex.Lock()
	// If the peer sent a STOP_SENDING after we sent a RESET_STREAM_AT frame,
	// we sent 1. reduced the reliable size to 0 and 2. sent a RESET_STREAM frame.
	// In this case, we don't care about the acknowledgment of this frame.
	if .ReliableSize != (*SendStream)().reliableOffset() {
		.mutex.Unlock()
		return
	}
	.numOutstandingFrames--
	if .numOutstandingFrames < 0 {
		panic("numOutStandingFrames negative")
	}
	 := (*SendStream)().isNewlyCompleted()
	.mutex.Unlock()

	if  {
		.sender.onStreamCompleted(.streamID)
	}
}

func ( *sendStreamResetStreamHandler) ( wire.Frame) {
	 := .(*wire.ResetStreamFrame)
	.mutex.Lock()
	// If the peer sent a STOP_SENDING after we sent a RESET_STREAM_AT frame,
	// we sent 1. reduced the reliable size to 0 and 2. sent a RESET_STREAM frame.
	// In this case, the loss of the RESET_STREAM_AT frame can be ignored.
	if .ReliableSize != (*SendStream)().reliableOffset() {
		.mutex.Unlock()
		return
	}
	.queuedResetStreamFrame = 
	.numOutstandingFrames--
	.mutex.Unlock()
	.sender.onHasStreamControlFrame(.streamID, (*SendStream)())
}