Allow priority sheduling if supported by QUIC connection

examples/interop/main.go

@@ -88,7 +88,7 @@ func setupAndRun() error {
 	if keyLog != nil {
 		defer func() {
 			log.Printf("closing keylog")
-			keyLog.Close()
+			keyLog.Close() //nolint
 		}()
 	}
 	conn, err := connect(context.Background(), f, keyLog)
@@ -142,7 +142,7 @@ func runSender(f flags, conn *quic.Conn) error {
 	}
 	qlogfile := getQLOGWriter("roq", role)
 	if qlogfile != nil {
-		defer qlogfile.Close()
+		defer qlogfile.Close() //nolint
 	}
 	var qlogger *mqlog.Logger
 	if qlogfile != nil {
@@ -178,7 +178,9 @@ func runSender(f flags, conn *quic.Conn) error {
 		return ffmpeg.Process.Kill()
 	}
 	time.Sleep(time.Second)
-	s.Close()
+	if err = s.Close(); err != nil {
+		return err
+	}
 	return ffmpeg.Wait()
 }
 
@@ -189,7 +191,7 @@ func runReceiver(f flags, conn *quic.Conn) error {
 	}
 	qlogfile := getQLOGWriter("roq", role)
 	if qlogfile != nil {
-		defer qlogfile.Close()
+		defer qlogfile.Close() //nolint
 	}
 	var qlogger *mqlog.Logger
 	if qlogfile != nil {
@@ -221,7 +223,7 @@ func runReceiver(f flags, conn *quic.Conn) error {
 			log.Println("WAITING FOR FFPLAY")
 			//state, err := ffplay.Process.Wait()
 			//log.Printf("ffmpeg returned %v, err: %v", state, err)
-			ffplay.Process.Kill()
+			ffplay.Process.Kill() //nolint
 		}()
 		writer, err = newIVFWriterWith(stdout, f.Codec)
 		if err != nil {

examples/interop/receiver.go

@@ -7,10 +7,6 @@ import (
 	"github.com/mengelbart/roq"
 )
 
-const (
-	bufferSize = 1 << 15
-)
-
 type receiver struct {
 	session *roq.Session
 }
@@ -30,8 +26,8 @@ func (r *receiver) receive(flowID uint64, writer io.WriteCloser) error {
 	if err != nil {
 		return err
 	}
-	defer flow.Close()
-	defer writer.Close()
+	defer flow.Close()   //nolint
+	defer writer.Close() //nolint
 	_, err = io.Copy(writer, flow)
 	return err
 }

examples/interop/sender.go

@@ -42,13 +42,13 @@ func (s *sender) send(flowID uint64, reader FrameReader, packetizer rtp.Packetiz
 	}
 	var singleStream *roq.RTPSendStream
 	if sendMode == singleStreamMode {
-		singleStream, err = flow.NewSendStream(context.Background())
+		singleStream, err = flow.NewSendStream(context.Background(), 0, false)
 		if err != nil {
 			return err
 		}
-		defer singleStream.Close()
+		defer singleStream.Close() //nolint
 	}
-	defer flow.Close()
+	defer flow.Close() //nolint
 	for {
 		frame, err := reader.Read()
 		if err != nil {
@@ -64,7 +64,7 @@ func (s *sender) send(flowID uint64, reader FrameReader, packetizer rtp.Packetiz
 				return err
 			}
 		case streamPerFrameMode:
-			stream, err := flow.NewSendStream(context.Background())
+			stream, err := flow.NewSendStream(context.Background(), 0, false)
 			if err != nil {
 				return err
 			}

examples/playfromdisk/main.go

@@ -104,7 +104,7 @@ func main() {
 			panic(ivfErr)
 		}
 
-		stream, err := flow.NewSendStream(context.Background())
+		stream, err := flow.NewSendStream(context.Background(), 0, false)
 		if err != nil {
 			panic(err)
 		}
@@ -120,6 +120,6 @@ func main() {
 				panic(err)
 			}
 		}
-		stream.Close()
+		stream.Close() //nolint
 	}
 }

examples/savetodisk/main.go

@@ -42,12 +42,12 @@ func main() {
 	if err != nil {
 		panic(err)
 	}
-	defer flow.Close()
+	defer flow.Close() //nolint
 	fileWriter, err := ivfwriter.New("output.ivf")
 	if err != nil {
 		panic(err)
 	}
-	defer fileWriter.Close()
+	defer fileWriter.Close() //nolint
 	for {
 		buf := make([]byte, 4096)
 		n, err := flow.Read(buf)

go.mod

@@ -1,6 +1,6 @@
 module github.com/mengelbart/roq
 
-go 1.24
+go 1.24.0
 
 require (
 	github.com/mengelbart/qlog v0.0.0-20250304144032-13728b0b6fae
@@ -20,5 +20,6 @@ require (
 	golang.org/x/crypto v0.41.0 // indirect
 	golang.org/x/net v0.43.0 // indirect
 	golang.org/x/sys v0.35.0 // indirect
+	golang.org/x/time v0.14.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -169,8 +169,8 @@ golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.11.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.12.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
-golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE=
-golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg=
+golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
+golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=

integrationtests/integration_test.go

@@ -48,7 +48,7 @@ func TestIntegration(t *testing.T) {
 			sender := accept(t, ctx, l)
 			f, err := sender.NewSendFlow(0)
 			assert.NoError(t, err)
-			s, err := f.NewSendStream(ctx)
+			s, err := f.NewSendStream(ctx, 0, false)
 			assert.NoError(t, err)
 			for i := 0; i < 10; i++ {
 				pkt := rtp.Packet{

send_flow.go

@@ -2,6 +2,7 @@ package roq
 
 import (
 	"context"
+	"log/slog"
 	"sync"
 
 	"github.com/mengelbart/qlog"
@@ -64,14 +65,21 @@ 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) (*RTPSendStream, error) {
+func (f *SendFlow) NewSendStream(ctx context.Context, priority uint32, incremantal bool) (*RTPSendStream, error) {
 	if err := f.isClosed(); err != nil {
 		return nil, err
 	}
 	s, err := f.conn.OpenUniStreamSync(ctx)
 	if err != nil {
 		return nil, err
 	}
+	priorityStream, ok := s.(PrioritySendStream)
+	if ok {
+		slog.Info("roq setting stream priority", "priority", priority, "incremental", incremantal)
+		priorityStream.SetPriority(priority)
+		priorityStream.SetIncremental(incremantal)
+	}
+
 	stream, err := newRTPSendStream(s, f.id, f.flowID, f.qlog)
 	if err != nil {
 		return nil, err

session.go

@@ -21,6 +21,11 @@ type SendStream interface {
 	CancelWrite(uint64)
 }
 
+type PrioritySendStream interface {
+	SetPriority(p uint32)
+	SetIncremental(b bool)
+}
+
 type ReceiveStream interface {
 	io.Reader
 	ID() int64
@@ -144,8 +149,8 @@ func (s *Session) NewReceiveFlow(id uint64) (*ReceiveFlow, error) {
 func (s *Session) close(code uint64, reason string) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
-	s.sendFlows.rangeFn(func(_ uint64, v *SendFlow) { v.Close() })
-	s.receiveFlows.rangeFn(func(_ uint64, v *ReceiveFlow) { v.Close() })
+	s.sendFlows.rangeFn(func(_ uint64, v *SendFlow) { _ = v.Close() })
+	s.receiveFlows.rangeFn(func(_ uint64, v *ReceiveFlow) { _ = v.Close() })
 	_ = s.conn.CloseWithError(code, reason)
 	s.closedErr = SessionError{
 		code:   code,