Tidy logger usage and configuration

options.go

@@ -2,6 +2,7 @@ package roq
 
 import (
 	"fmt"
+	"log/slog"
 
 	"github.com/quic-go/quic-go/qlogwriter"
 )
@@ -21,6 +22,7 @@ type Option func(*sessionConfig) error
 type sessionConfig struct {
 	receiveBufferSize     int
 	unknownFlowBufferSize int
+	logger                *slog.Logger
 	qlogTrace             qlogwriter.Trace
 	qlogPacketData        bool
 	qlogPacketDataLimit   int
@@ -31,6 +33,7 @@ func newSessionConfig(opts []Option) (*sessionConfig, error) {
 	c := &sessionConfig{
 		receiveBufferSize:     defaultReceiveBufferSize,
 		unknownFlowBufferSize: defaultUnknownFlowBufferSize,
+		logger:                discardLogger(),
 	}
 	for _, opt := range opts {
 		if opt == nil {
@@ -104,3 +107,23 @@ func WithQlogPacketData(maxBytes int) Option {
 		return nil
 	}
 }
+
+// WithLogger sets the logger the session and its flows write to.The session
+// discards all log records unless the application passes one in. Pass
+// slog.Default() to send them to the standard logger. The logger must not be
+// nil.
+func WithLogger(logger *slog.Logger) Option {
+	return func(c *sessionConfig) error {
+		if logger == nil {
+			return fmt.Errorf("%w: logger must not be nil", errInvalidOption)
+		}
+		c.logger = logger
+		return nil
+	}
+}
+
+// discardLogger returns a logger that drops every record. It is what a session
+// uses when the application did not pass WithLogger.
+func discardLogger() *slog.Logger {
+	return slog.New(slog.DiscardHandler)
+}

options_test.go

@@ -1,7 +1,11 @@
 package roq
 
 import (
+	"bytes"
 	"errors"
+	"log"
+	"log/slog"
+	"os"
 	"testing"
 	"time"
 
@@ -94,6 +98,7 @@ func TestInvalidOptions(t *testing.T) {
 		{"zero unknown flow buffer", WithUnknownFlowBufferSize(0)},
 		{"negative unknown flow buffer", WithUnknownFlowBufferSize(-1)},
 		{"negative qlog packet data limit", WithQlogPacketData(-1)},
+		{"nil logger", WithLogger(nil)},
 		{"nil option", nil},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
@@ -107,6 +112,71 @@ func TestInvalidOptions(t *testing.T) {
 	}
 }
 
+func TestWithLogger(t *testing.T) {
+	var logs bytes.Buffer
+	logger := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{
+		Level: slog.LevelDebug,
+	}))
+
+	c := newStubConn()
+	s, err := NewSession(c, true, WithLogger(logger))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer closeWithin(t, s, 5*time.Second)
+
+	rf, err := s.NewReceiveFlow(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if rf.logger != logger {
+		t.Error("receive flow does not use the logger passed with WithLogger")
+	}
+	sf, err := s.NewSendFlow(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if sf.logger != logger {
+		t.Error("send flow does not use the logger passed with WithLogger")
+	}
+
+	c.datagrams <- quicvarint.Append(nil, 2)
+	if bf := awaitBufferedFlow(t, s, 2); bf.logger != logger {
+		t.Error("buffered flow does not use the logger passed with WithLogger")
+	}
+}
+
+func TestSessionDoesNotUseGlobalLoggerByDefault(t *testing.T) {
+	var logs bytes.Buffer
+	log.SetOutput(&logs)
+	defer log.SetOutput(os.Stderr)
+	defaultLogger := slog.Default()
+	slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{
+		Level: slog.LevelDebug,
+	})))
+	defer slog.SetDefault(defaultLogger)
+
+	s, err := NewSession(newStubConn(), true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer closeWithin(t, s, 5*time.Second)
+
+	f, err := s.NewReceiveFlow(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.readStream(&stubReceiveStream{
+		id:        1,
+		readErr:   errors.New("boom"),
+		cancelled: make(chan struct{}),
+	})
+
+	if logs.Len() != 0 {
+		t.Errorf("session wrote %q to the global logger, want nothing", logs.String())
+	}
+}
+
 // awaitBufferedFlow waits for the session to buffer a flow for an unclaimed
 // flow ID and returns it.
 func awaitBufferedFlow(t *testing.T, s *Session, id uint64) *ReceiveFlow {

receive_flow.go

@@ -5,7 +5,7 @@ import (
 	"context"
 	"errors"
 	"io"
-	"log"
+	"log/slog"
 	"os"
 	"sync"
 	"time"
@@ -38,10 +38,14 @@ type ReceiveFlow struct {
 	deadlineExpired bool
 	deadlineTimer   *time.Timer
 
-	qlog *qlogger
+	qlog   *qlogger
+	logger *slog.Logger
 }
 
-func newReceiveFlow(id uint64, receiveBufferSize int, qlog *qlogger) *ReceiveFlow {
+func newReceiveFlow(id uint64, receiveBufferSize int, qlog *qlogger, logger *slog.Logger) *ReceiveFlow {
+	if logger == nil {
+		logger = discardLogger()
+	}
 	ctx, cancel := context.WithCancel(context.Background())
 	return &ReceiveFlow{
 		id:     id,
@@ -58,6 +62,7 @@ func newReceiveFlow(id uint64, receiveBufferSize int, qlog *qlogger) *ReceiveFlo
 		closed:     false,
 		deadlineCh: make(chan struct{}),
 		qlog:       qlog,
+		logger:     logger,
 	}
 }
 
@@ -110,10 +115,12 @@ func (f *ReceiveFlow) readStream(rs ReceiveStream) {
 			}
 			var streamErr *quic.StreamError
 			if errors.As(err, &streamErr) {
-				log.Printf("got stream error while reading length: %v", streamErr)
+				f.logger.Debug("got stream error while reading length",
+					"flowID", f.id, "streamID", rs.ID(), "error", streamErr)
 				return
 			}
-			log.Printf("got unexpected error while reading length: %v", err)
+			f.logger.Warn("got unexpected error while reading length",
+				"flowID", f.id, "streamID", rs.ID(), "error", err)
 			return
 		}
 		r := io.LimitReader(reader, int64(length))
@@ -126,10 +133,12 @@ func (f *ReceiveFlow) readStream(rs ReceiveStream) {
 		if err != nil {
 			var streamErr *quic.StreamError
 			if errors.As(err, &streamErr) {
-				log.Printf("got stream error after reading %v bytes of payload: %v", n, streamErr)
+				f.logger.Debug("got stream error while reading payload",
+					"flowID", f.id, "streamID", rs.ID(), "read", n, "error", streamErr)
 				return
 			}
-			log.Printf("got unexpected error after reading %v bytes of payload: %v", n, err)
+			f.logger.Warn("got unexpected error while reading payload",
+				"flowID", f.id, "streamID", rs.ID(), "read", n, "error", err)
 			return
 		}
 		if f.qlog != nil {

receive_flow_buffer.go

@@ -1,6 +1,7 @@
 package roq
 
 import (
+	"log/slog"
 	"slices"
 	"sync"
 )
@@ -26,7 +27,7 @@ func newReceiveFlowBuffer(maxLen int) *receiveFlowBuffer {
 // flow if the buffer holds none. Lookup and creation happen under a single
 // lock hold, so concurrent callers for the same unknown flow ID all get the
 // same flow.
-func (b *receiveFlowBuffer) getOrCreate(id uint64, receiveBufferSize int, qlog *qlogger) *ReceiveFlow {
+func (b *receiveFlowBuffer) getOrCreate(id uint64, receiveBufferSize int, qlog *qlogger, logger *slog.Logger) *ReceiveFlow {
 	b.mutex.Lock()
 	defer b.mutex.Unlock()
 	if f, ok := b.buffer[id]; ok {
@@ -41,7 +42,7 @@ func (b *receiveFlowBuffer) getOrCreate(id uint64, receiveBufferSize int, qlog *
 		delete(b.buffer, front)
 		b.queue = b.queue[1:]
 	}
-	f := newReceiveFlow(id, receiveBufferSize, qlog)
+	f := newReceiveFlow(id, receiveBufferSize, qlog, logger)
 	b.queue = append(b.queue, id)
 	b.buffer[id] = f
 	return f

receive_flow_buffer_test.go

@@ -7,7 +7,7 @@ import (
 
 func TestReceiveFlowBufferPopRemovesFlow(t *testing.T) {
 	b := newReceiveFlowBuffer(16)
-	f := b.getOrCreate(1, 1, nil)
+	f := b.getOrCreate(1, 1, nil, nil)
 
 	if got := b.pop(1); got != f {
 		t.Fatalf("pop returned %v, want %v", got, f)
@@ -21,7 +21,7 @@ func TestReceiveFlowBufferPopRemovesFlow(t *testing.T) {
 	if got := b.pop(1); got != nil {
 		t.Errorf("second pop returned %v, want nil", got)
 	}
-	if got := b.getOrCreate(1, 1, nil); got == f {
+	if got := b.getOrCreate(1, 1, nil, nil); got == f {
 		t.Error("getOrCreate returned the popped flow, want a new one")
 	}
 }
@@ -31,7 +31,7 @@ func TestReceiveFlowBufferPopRemovesFlow(t *testing.T) {
 func TestReceiveFlowBufferPoppedFlowIsNotEvicted(t *testing.T) {
 	const maxLen = 16
 	b := newReceiveFlowBuffer(maxLen)
-	f := b.getOrCreate(1, 1, nil)
+	f := b.getOrCreate(1, 1, nil, nil)
 
 	rs := &stubReceiveStream{id: 1, cancelled: make(chan struct{})}
 	f.streams[rs.ID()] = rs
@@ -40,7 +40,7 @@ func TestReceiveFlowBufferPoppedFlowIsNotEvicted(t *testing.T) {
 		t.Fatalf("pop returned %v, want %v", got, f)
 	}
 	for i := range uint64(maxLen + 1) {
-		b.getOrCreate(100+i, 1, nil)
+		b.getOrCreate(100+i, 1, nil, nil)
 	}
 
 	select {
@@ -71,7 +71,7 @@ func TestReceiveFlowBufferGetOrCreateIsAtomic(t *testing.T) {
 		go func() {
 			defer done.Done()
 			start.Wait()
-			flows[i] = b.getOrCreate(1, 1, nil)
+			flows[i] = b.getOrCreate(1, 1, nil, nil)
 		}()
 	}
 	start.Done()