diff --git a/examples/interop/main.go b/examples/interop/main.go index 33bbec9..b9d5a4a 100644 --- a/examples/interop/main.go +++ b/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 { diff --git a/examples/interop/receiver.go b/examples/interop/receiver.go index f16e6de..5cb4090 100644 --- a/examples/interop/receiver.go +++ b/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 } diff --git a/examples/interop/sender.go b/examples/interop/sender.go index 87fe1d7..23ad792 100644 --- a/examples/interop/sender.go +++ b/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 } diff --git a/examples/playfromdisk/main.go b/examples/playfromdisk/main.go index 57cdd1e..3ff3232 100644 --- a/examples/playfromdisk/main.go +++ b/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 } } diff --git a/examples/savetodisk/main.go b/examples/savetodisk/main.go index 61c396a..81fbbd3 100644 --- a/examples/savetodisk/main.go +++ b/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) diff --git a/go.mod b/go.mod index 0326a11..2e4181e 100644 --- a/go.mod +++ b/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 ) diff --git a/go.sum b/go.sum index c82a269..b4399ba 100644 --- a/go.sum +++ b/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= diff --git a/integrationtests/integration_test.go b/integrationtests/integration_test.go index 0a9a139..ec4a7ce 100644 --- a/integrationtests/integration_test.go +++ b/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{ diff --git a/send_flow.go b/send_flow.go index df4a066..9a846e9 100644 --- a/send_flow.go +++ b/send_flow.go @@ -2,6 +2,7 @@ package roq import ( "context" + "log/slog" "sync" "github.com/mengelbart/qlog" @@ -64,7 +65,7 @@ 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 } @@ -72,6 +73,13 @@ func (f *SendFlow) NewSendStream(ctx context.Context) (*RTPSendStream, error) { 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 diff --git a/session.go b/session.go index f17e2a7..c9fd931 100644 --- a/session.go +++ b/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,