route/vendor/github.com/lucas-clemente/quic-go/stream.go

181 lines
5.2 KiB
Go
Raw Normal View History

2017-12-12 02:51:45 +00:00
package quic
import (
2018-01-03 19:19:49 +00:00
"net"
2017-12-12 02:51:45 +00:00
"sync"
2018-01-03 19:19:49 +00:00
"time"
2017-12-12 02:51:45 +00:00
2018-01-03 19:19:49 +00:00
"github.com/lucas-clemente/quic-go/internal/flowcontrol"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/wire"
2017-12-12 02:51:45 +00:00
)
2018-01-20 18:07:01 +00:00
const (
errorCodeStopping protocol.ApplicationErrorCode = 0
errorCodeStoppingGQUIC protocol.ApplicationErrorCode = 7
)
2018-01-03 19:19:49 +00:00
2018-01-20 18:07:01 +00:00
// The streamSender is notified by the stream about various events.
type streamSender interface {
queueControlFrame(wire.Frame)
onHasWindowUpdate(protocol.StreamID)
onHasStreamData(protocol.StreamID)
onStreamCompleted(protocol.StreamID)
2018-01-03 19:19:49 +00:00
}
2018-01-20 18:07:01 +00:00
// Each of the both stream halves gets its own uniStreamSender.
// This is necessary in order to keep track when both halves have been completed.
type uniStreamSender struct {
streamSender
onStreamCompletedImpl func()
2018-01-03 19:19:49 +00:00
}
2018-01-20 18:07:01 +00:00
func (s *uniStreamSender) queueControlFrame(f wire.Frame) {
s.streamSender.queueControlFrame(f)
}
2017-12-12 02:51:45 +00:00
2018-01-20 18:07:01 +00:00
func (s *uniStreamSender) onHasWindowUpdate(id protocol.StreamID) {
s.streamSender.onHasWindowUpdate(id)
}
func (s *uniStreamSender) onHasStreamData(id protocol.StreamID) {
s.streamSender.onHasStreamData(id)
}
2018-01-03 19:19:49 +00:00
2018-01-20 18:07:01 +00:00
func (s *uniStreamSender) onStreamCompleted(protocol.StreamID) {
s.onStreamCompletedImpl()
}
2017-12-12 02:51:45 +00:00
2018-01-20 18:07:01 +00:00
var _ streamSender = &uniStreamSender{}
2017-12-12 02:51:45 +00:00
2018-01-20 18:07:01 +00:00
type streamI interface {
Stream
closeForShutdown(error)
// for receiving
handleStreamFrame(*wire.StreamFrame) error
handleRstStreamFrame(*wire.RstStreamFrame) error
getWindowUpdate() protocol.ByteCount
// for sending
handleStopSendingFrame(*wire.StopSendingFrame)
popStreamFrame(maxBytes protocol.ByteCount) (*wire.StreamFrame, bool)
handleMaxStreamDataFrame(*wire.MaxStreamDataFrame)
}
2017-12-12 02:51:45 +00:00
2018-01-20 18:07:01 +00:00
var _ receiveStreamI = (streamI)(nil)
var _ sendStreamI = (streamI)(nil)
2017-12-12 02:51:45 +00:00
2018-01-20 18:07:01 +00:00
// A Stream assembles the data from StreamFrames and provides a super-convenient Read-Interface
//
// Read() and Write() may be called concurrently, but multiple calls to Read() or Write() individually must be synchronized manually.
type stream struct {
receiveStream
sendStream
2017-12-12 02:51:45 +00:00
2018-01-20 18:07:01 +00:00
completedMutex sync.Mutex
sender streamSender
receiveStreamCompleted bool
sendStreamCompleted bool
2017-12-12 02:51:45 +00:00
2018-01-20 18:07:01 +00:00
version protocol.VersionNumber
2017-12-12 02:51:45 +00:00
}
2018-01-03 19:19:49 +00:00
var _ Stream = &stream{}
type deadlineError struct{}
func (deadlineError) Error() string { return "deadline exceeded" }
func (deadlineError) Temporary() bool { return true }
func (deadlineError) Timeout() bool { return true }
var errDeadline net.Error = &deadlineError{}
2018-01-20 18:07:01 +00:00
type streamCanceledError struct {
error
errorCode protocol.ApplicationErrorCode
}
func (streamCanceledError) Canceled() bool { return true }
func (e streamCanceledError) ErrorCode() protocol.ApplicationErrorCode { return e.errorCode }
var _ StreamError = &streamCanceledError{}
2017-12-12 02:51:45 +00:00
// newStream creates a new Stream
2018-01-20 18:07:01 +00:00
func newStream(streamID protocol.StreamID,
sender streamSender,
2018-01-03 19:19:49 +00:00
flowController flowcontrol.StreamFlowController,
version protocol.VersionNumber,
) *stream {
2018-01-20 18:07:01 +00:00
s := &stream{sender: sender}
senderForSendStream := &uniStreamSender{
streamSender: sender,
onStreamCompletedImpl: func() {
s.completedMutex.Lock()
s.sendStreamCompleted = true
s.checkIfCompleted()
s.completedMutex.Unlock()
},
}
s.sendStream = *newSendStream(streamID, senderForSendStream, flowController, version)
senderForReceiveStream := &uniStreamSender{
streamSender: sender,
onStreamCompletedImpl: func() {
s.completedMutex.Lock()
s.receiveStreamCompleted = true
s.checkIfCompleted()
s.completedMutex.Unlock()
},
}
s.receiveStream = *newReceiveStream(streamID, senderForReceiveStream, flowController)
2018-01-03 19:19:49 +00:00
return s
2017-12-12 02:51:45 +00:00
}
2018-01-20 18:07:01 +00:00
// need to define StreamID() here, since both receiveStream and readStream have a StreamID()
func (s *stream) StreamID() protocol.StreamID {
// the result is same for receiveStream and sendStream
return s.sendStream.StreamID()
2017-12-12 02:51:45 +00:00
}
func (s *stream) Close() error {
2018-01-20 18:07:01 +00:00
if err := s.sendStream.Close(); err != nil {
2017-12-12 02:51:45 +00:00
return err
}
2018-01-20 18:07:01 +00:00
// in gQUIC, we need to send a RST_STREAM with the final offset if CancelRead() was called
s.receiveStream.onClose(s.sendStream.getWriteOffset())
2018-01-03 19:19:49 +00:00
return nil
}
func (s *stream) SetDeadline(t time.Time) error {
_ = s.SetReadDeadline(t) // SetReadDeadline never errors
_ = s.SetWriteDeadline(t) // SetWriteDeadline never errors
2017-12-12 02:51:45 +00:00
return nil
}
2018-01-20 18:07:01 +00:00
// CloseForShutdown closes a stream abruptly.
// It makes Read and 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 (s *stream) closeForShutdown(err error) {
s.sendStream.closeForShutdown(err)
s.receiveStream.closeForShutdown(err)
2017-12-12 02:51:45 +00:00
}
2018-01-20 18:07:01 +00:00
func (s *stream) handleRstStreamFrame(frame *wire.RstStreamFrame) error {
if err := s.receiveStream.handleRstStreamFrame(frame); err != nil {
2018-01-03 19:19:49 +00:00
return err
2017-12-12 02:51:45 +00:00
}
2018-01-20 18:07:01 +00:00
if !s.version.UsesIETFFrameFormat() {
s.handleStopSendingFrame(&wire.StopSendingFrame{
StreamID: s.StreamID(),
ErrorCode: frame.ErrorCode,
})
2017-12-12 02:51:45 +00:00
}
2018-01-03 19:19:49 +00:00
return nil
2017-12-12 02:51:45 +00:00
}
2018-01-20 18:07:01 +00:00
// checkIfCompleted is called from the uniStreamSender, when one of the stream halves is completed.
// It makes sure that the onStreamCompleted callback is only called if both receive and send side have completed.
func (s *stream) checkIfCompleted() {
if s.sendStreamCompleted && s.receiveStreamCompleted {
s.sender.onStreamCompleted(s.StreamID())
}
2018-01-03 19:19:49 +00:00
}