Document priority API

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)

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)
+	}
+}

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)