From ce403eccceb4f19787ad404928a90b390d089695 Mon Sep 17 00:00:00 2001 From: Mathis Engelbart Date: Fri, 28 Aug 2026 17:53:16 +0200 Subject: [PATCH] Document priority API --- send_flow.go | 11 ++++++--- send_flow_test.go | 61 +++++++++++++++++++++++++++++++++++++++++++++++ session.go | 5 ++++ 3 files changed, 74 insertions(+), 3 deletions(-) diff --git a/send_flow.go b/send_flow.go index 2bb979f..e4b639b 100644 --- a/send_flow.go +++ b/send_flow.go @@ -67,7 +67,12 @@ func (f *SendFlow) WriteRTPBytes(packet []byte) error { // NewSendStream creates a new Stream for sending outgoing RTP and RTCP packets // over a QUIC stream. -func (f *SendFlow) NewSendStream(ctx context.Context, priority uint32, incremantal bool) (*RTPSendStream, error) { +// +// priority and incremental are applied to the new stream only if the +// Connection of the session opens SendStreams that implement +// PrioritySendStream. They are ignored otherwise, which is the case for +// QUICGoConnection, because quic-go does not implement stream priorities. +func (f *SendFlow) NewSendStream(ctx context.Context, priority uint32, incremental bool) (*RTPSendStream, error) { if err := f.isClosed(); err != nil { return nil, err } @@ -78,9 +83,9 @@ func (f *SendFlow) NewSendStream(ctx context.Context, priority uint32, incremant priorityStream, ok := s.(PrioritySendStream) if ok { f.logger.Debug("setting stream priority", - "flowID", f.id, "streamID", s.ID(), "priority", priority, "incremental", incremantal) + "flowID", f.id, "streamID", s.ID(), "priority", priority, "incremental", incremental) priorityStream.SetPriority(priority) - priorityStream.SetIncremental(incremantal) + priorityStream.SetIncremental(incremental) } stream, err := newRTPSendStream(s, f.id, f.flowID, f.qlog) diff --git a/send_flow_test.go b/send_flow_test.go index 850240d..8239fee 100644 --- a/send_flow_test.go +++ b/send_flow_test.go @@ -3,6 +3,7 @@ package roq import ( "context" "errors" + "sync" "testing" ) @@ -53,3 +54,63 @@ func TestSendFlowCloseContinuesAfterStreamError(t *testing.T) { t.Error("flow was not removed from the session after Close") } } + +// prioritySendStream is a SendStream that supports stream priorities, like the +// streams of a Connection implementation built on a QUIC stack that has them. +type prioritySendStream struct { + *stubSendStream + mutex sync.Mutex + priority uint32 + incremental bool +} + +func (s *prioritySendStream) SetPriority(p uint32) { + s.mutex.Lock() + defer s.mutex.Unlock() + s.priority = p +} + +func (s *prioritySendStream) SetIncremental(b bool) { + s.mutex.Lock() + defer s.mutex.Unlock() + s.incremental = b +} + +func (s *prioritySendStream) get() (uint32, bool) { + s.mutex.Lock() + defer s.mutex.Unlock() + return s.priority, s.incremental +} + +// priorityConn opens send streams that implement PrioritySendStream. +type priorityConn struct { + *stubConn + stream *prioritySendStream +} + +func (c *priorityConn) OpenUniStreamSync(context.Context) (SendStream, error) { + return c.stream, nil +} + +// The priority and incremental arguments of NewSendStream are passed on if the +// connection opens streams that implement PrioritySendStream, and ignored if +// it does not. +func TestNewSendStreamPriority(t *testing.T) { + ps := &prioritySendStream{stubSendStream: &stubSendStream{id: 1}} + c := &priorityConn{stubConn: newStubConn(), stream: ps} + f := newFlow(c, 1, func() {}, nil, nil) + if _, err := f.NewSendStream(context.Background(), 7, true); err != nil { + t.Fatal(err) + } + if priority, incremental := ps.get(); priority != 7 || !incremental { + t.Errorf("priority = %v, incremental = %v, want 7, true", priority, incremental) + } + + // A stream without priority support must not keep NewSendStream from + // returning a usable stream. + plain := newStubConn() + f = newFlow(plain, 1, func() {}, nil, nil) + if _, err := f.NewSendStream(context.Background(), 7, true); err != nil { + t.Fatal(err) + } +} diff --git a/session.go b/session.go index a79c154..02a9499 100644 --- a/session.go +++ b/session.go @@ -19,6 +19,11 @@ type SendStream interface { CancelWrite(uint64) } +// PrioritySendStream is implemented by SendStreams that support stream +// priorities. quic-go does not implement stream priorities, so the SendStreams +// of QUICGoConnection do not implement it either: the priority and incremental +// arguments of SendFlow.NewSendStream are only honored by Connection +// implementations that provide it. type PrioritySendStream interface { SetPriority(p uint32) SetIncremental(b bool)