mirrors/uquic
1
0
Fork 0
mirror of https://github.com/refraction-networking/uquic.git synced 2025-04-04 20:57:36 +03:00
Code Issues Projects Releases Packages Wiki Activity

remove the error return value from Stream.CancelWrite

Browse source
This commit is contained in:
Marten Seemann 2019-01-24 09:37:28 +07:00
parent ca939df44e
commit 3372fa794c
7 changed files with 24 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

2
h2quic/response_writer_test.go
View file

@ -45,7 +45,7 @@ func newMockStream(id protocol.StreamID) *mockStream {
func (s *mockStream) Close() error { s.closed = true; s.ctxCancel(); return nil }
func (s *mockStream) CancelRead(quic.ErrorCode) { s.canceledRead = true }
func (s *mockStream) CancelWrite(quic.ErrorCode) error { s.canceledWrite = true; return nil }
func (s *mockStream) CancelWrite(quic.ErrorCode) { s.canceledWrite = true }
func (s *mockStream) CloseRemote(offset protocol.ByteCount) { s.remoteClosed = true; s.ctxCancel() }
func (s mockStream) StreamID() protocol.StreamID { return s.id }
func (s *mockStream) Context() context.Context { return s.ctx }

8
integrationtests/self/cancelation_test.go
View file

@ -207,7 +207,7 @@ var _ = Describe("Stream Cancelations", func() {
Expect(err).ToNot(HaveOccurred())
// cancel about 2/3 of the streams
if rand.Int31()%3 != 0 {
Expect(str.CancelWrite(quic.ErrorCode(str.StreamID()))).To(Succeed())
str.CancelWrite(quic.ErrorCode(str.StreamID()))
atomic.AddInt32(&canceledCounter, 1)
return
}
@ -241,7 +241,7 @@ var _ = Describe("Stream Cancelations", func() {
length := int(rand.Int31n(int32(len(testserver.PRData) - 1)))
_, err = str.Write(testserver.PRData[:length])
Expect(err).ToNot(HaveOccurred())
Expect(str.CancelWrite(quic.ErrorCode(str.StreamID()))).To(Succeed())
str.CancelWrite(quic.ErrorCode(str.StreamID()))
atomic.AddInt32(&canceledCounter, 1)
return
}
@ -277,7 +277,7 @@ var _ = Describe("Stream Cancelations", func() {
Expect(err).ToNot(HaveOccurred())
// cancel about half of the streams
if rand.Int31()%2 == 0 {
Expect(str.CancelWrite(quic.ErrorCode(str.StreamID()))).To(Succeed())
str.CancelWrite(quic.ErrorCode(str.StreamID()))
return
}
if _, err = str.Write(testserver.PRData); err != nil {
@ -359,7 +359,7 @@ var _ = Describe("Stream Cancelations", func() {
return
}
if length < len(testserver.PRData) {
Expect(str.CancelWrite(quic.ErrorCode(str.StreamID()))).To(Succeed())
str.CancelWrite(quic.ErrorCode(str.StreamID()))
} else {
Expect(str.Close()).To(Succeed())
}

4
interface.go
View file

@ -53,7 +53,7 @@ type Stream interface {
// 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 or after closing the stream it is a no-op.
CancelWrite(ErrorCode) error
CancelWrite(ErrorCode)
// CancelRead aborts receiving on this stream.
// It will ask the peer to stop transmitting stream data.
// Read will unblock immediately, and future Read calls will fail.
@ -100,7 +100,7 @@ type SendStream interface {
// see Stream.Close
io.Closer
// see Stream.CancelWrite
CancelWrite(ErrorCode) error
CancelWrite(ErrorCode)
// see Stream.Context
Context() context.Context
// see Stream.SetWriteDeadline

6
mock_send_stream_internal_test.go
View file

@ -38,10 +38,8 @@ func (m *MockSendStreamI) EXPECT() *MockSendStreamIMockRecorder {
}
// CancelWrite mocks base method
func (m *MockSendStreamI) CancelWrite(arg0 protocol.ApplicationErrorCode) error {
ret := m.ctrl.Call(m, "CancelWrite", arg0)
ret0, _ := ret[0].(error)
return ret0
func (m *MockSendStreamI) CancelWrite(arg0 protocol.ApplicationErrorCode) {
m.ctrl.Call(m, "CancelWrite", arg0)
}
// CancelWrite indicates an expected call of CancelWrite

6
mock_stream_internal_test.go
View file

@ -48,10 +48,8 @@ func (mr *MockStreamIMockRecorder) CancelRead(arg0 interface{}) *gomock.Call {
}
// CancelWrite mocks base method
func (m *MockStreamI) CancelWrite(arg0 protocol.ApplicationErrorCode) error {
ret := m.ctrl.Call(m, "CancelWrite", arg0)
ret0, _ := ret[0].(error)
return ret0
func (m *MockStreamI) CancelWrite(arg0 protocol.ApplicationErrorCode) {
m.ctrl.Call(m, "CancelWrite", arg0)
}
// CancelWrite indicates an expected call of CancelWrite

15
send_stream.go
View file

@ -241,21 +241,20 @@ func (s *sendStream) Close() error {
return nil
}
func (s *sendStream) CancelWrite(errorCode protocol.ApplicationErrorCode) error {
func (s *sendStream) CancelWrite(errorCode protocol.ApplicationErrorCode) {
s.mutex.Lock()
completed, err := s.cancelWriteImpl(errorCode, fmt.Errorf("Write on stream %d canceled with error code %d", s.streamID, errorCode))
completed := s.cancelWriteImpl(errorCode, fmt.Errorf("Write on stream %d canceled with error code %d", s.streamID, errorCode))
s.mutex.Unlock()
if completed {
s.sender.onStreamCompleted(s.streamID) // must be called without holding the mutex
}
return err
}
// must be called after locking the mutex
func (s *sendStream) cancelWriteImpl(errorCode protocol.ApplicationErrorCode, writeErr error) (bool /*completed */, error) {
func (s *sendStream) cancelWriteImpl(errorCode protocol.ApplicationErrorCode, writeErr error) bool /*completed */ {
if s.canceledWrite || s.finishedWriting {
return false, nil
return false
}
s.canceledWrite = true
s.cancelWriteErr = writeErr
@ -267,7 +266,7 @@ func (s *sendStream) cancelWriteImpl(errorCode protocol.ApplicationErrorCode, wr
})
// TODO(#991): cancel retransmissions for this stream
s.ctxCancel()
return true, nil
return true
}
func (s *sendStream) handleMaxStreamDataFrame(frame *wire.MaxStreamDataFrame) {
@ -295,9 +294,7 @@ func (s *sendStream) handleStopSendingFrameImpl(frame *wire.StopSendingFrame) bo
errorCode: frame.ErrorCode,
error: fmt.Errorf("Stream %d was reset with error code %d", s.streamID, frame.ErrorCode),
}
errorCode := errorCodeStopping
completed, _ := s.cancelWriteImpl(errorCode, writeErr)
return completed
return s.cancelWriteImpl(errorCodeStopping, writeErr)
}
func (s *sendStream) Context() context.Context {

15
send_stream_test.go
View file

@ -527,8 +527,7 @@ var _ = Describe("Send Stream", func() {
})
mockSender.EXPECT().onStreamCompleted(streamID)
str.writeOffset = 1234
err := str.CancelWrite(9876)
Expect(err).ToNot(HaveOccurred())
str.CancelWrite(9876)
})
It("unblocks Write", func() {
@ -549,7 +548,7 @@ var _ = Describe("Send Stream", func() {
waitForWrite()
frame, _ := str.popStreamFrame(50)
Expect(frame).ToNot(BeNil())
Expect(str.CancelWrite(1234)).To(Succeed())
str.CancelWrite(1234)
Eventually(writeReturned).Should(BeClosed())
Expect(n).To(BeEquivalentTo(frame.DataLen()))
})
@ -571,7 +570,7 @@ var _ = Describe("Send Stream", func() {
frame, hasMoreData := str.popStreamFrame(50)
Expect(hasMoreData).To(BeTrue())
Expect(frame).ToNot(BeNil())
Expect(str.CancelWrite(1234)).To(Succeed())
str.CancelWrite(1234)
frame, hasMoreData = str.popStreamFrame(10)
Expect(hasMoreData).To(BeFalse())
Expect(frame).To(BeNil())
@ -589,7 +588,7 @@ var _ = Describe("Send Stream", func() {
It("doesn't allow further calls to Write", func() {
mockSender.EXPECT().queueControlFrame(gomock.Any())
mockSender.EXPECT().onStreamCompleted(streamID)
Expect(str.CancelWrite(1234)).To(Succeed())
str.CancelWrite(1234)
_, err := strWithTimeout.Write([]byte("foobar"))
Expect(err).To(MatchError("Write on stream 1337 canceled with error code 1234"))
})
@ -597,15 +596,15 @@ var _ = Describe("Send Stream", func() {
It("only cancels once", func() {
mockSender.EXPECT().queueControlFrame(&wire.ResetStreamFrame{StreamID: streamID, ErrorCode: 1234})
mockSender.EXPECT().onStreamCompleted(streamID)
Expect(str.CancelWrite(1234)).To(Succeed())
Expect(str.CancelWrite(4321)).To(Succeed())
str.CancelWrite(1234)
str.CancelWrite(4321)
})
It("doesn't do anything when the stream was already closed", func() {
mockSender.EXPECT().onHasStreamData(streamID)
Expect(str.Close()).To(Succeed())
// don't EXPECT any calls to queueControlFrame
Expect(str.CancelWrite(123)).To(Succeed())
str.CancelWrite(123)
})
})