transport: unexport transport opt field

pkg/upstream/transport/pipeline.go

@@ -30,13 +30,14 @@ import (
 // PipelineTransport will pipeline queries as RFC 7766 6.2.1.1 suggested.
 // It also can reuse udp socket. Since dns over udp is some kind of "pipeline".
 type PipelineTransport struct {
-	PipelineOpts
-
 	m      sync.Mutex // protect following fields
 	closed bool
 	conns  map[*lazyDnsConn]struct{}
 
-	logger *zap.Logger // not nil
+	dialFunc         func(ctx context.Context) (DnsConn, error)
+	dialTimeout      time.Duration
+	maxLazyConnQueue int
+	logger           *zap.Logger // not nil
 }
 
 type PipelineOpts struct {
@@ -59,14 +60,13 @@ type PipelineOpts struct {
 
 func NewPipelineTransport(opt PipelineOpts) *PipelineTransport {
 	t := &PipelineTransport{
-		PipelineOpts: opt,
-		conns:        make(map[*lazyDnsConn]struct{}),
-	}
-	if opt.Logger != nil {
-		t.logger = opt.Logger
-	} else {
-		t.logger = nopLogger
+		conns: make(map[*lazyDnsConn]struct{}),
 	}
+	t.dialFunc = opt.DialContext
+	setDefaultGZ(&t.dialTimeout, opt.DialTimeout, defaultDialTimeout)
+	setDefaultGZ(&t.maxLazyConnQueue, opt.MaxConcurrentQueryWhileDialing, defaultMaxLazyConnQueue)
+	setNonNilLogger(&t.logger, opt.Logger)
+
 	return t
 }
 
@@ -108,11 +108,6 @@ func (t *PipelineTransport) Close() error {
 }
 
 func (t *PipelineTransport) getReservedExchanger() (_ ReservedExchanger, isNewConn bool, err error) {
-	maxConcurrentQueryWhileDialing := t.MaxConcurrentQueryWhileDialing
-	if maxConcurrentQueryWhileDialing <= 0 {
-		maxConcurrentQueryWhileDialing = defaultLazyConnMaxConcurrentQuery
-	}
-
 	t.m.Lock()
 	if t.closed {
 		err = ErrClosedTransport
@@ -141,7 +136,7 @@ func (t *PipelineTransport) getReservedExchanger() (_ ReservedExchanger, isNewCo
 
 	// Dial a new connection
 	if rxc == nil {
-		c := newLazyDnsConn(t.DialContext, t.DialTimeout, t.MaxConcurrentQueryWhileDialing, t.logger)
+		c := newLazyDnsConn(t.dialFunc, t.dialTimeout, t.maxLazyConnQueue, t.logger)
 		rxc, _ = c.ReserveNewQuery() // ignore the closed error for new lazy connection
 		isNewConn = true
 		t.conns[c] = struct{}{}

pkg/upstream/transport/reuse.go

@@ -31,11 +31,12 @@ import (
 
 // ReuseConnTransport is for old tcp protocol. (no pipelining)
 type ReuseConnTransport struct {
-	ReuseConnOpts
-
-	logger    *zap.Logger // non-nil
-	ctx       context.Context
-	ctxCancel context.CancelCauseFunc
+	dialFunc    func(ctx context.Context) (NetConn, error)
+	dialTimeout time.Duration
+	idleTimeout time.Duration
+	logger      *zap.Logger // non-nil
+	ctx         context.Context
+	ctxCancel   context.CancelCauseFunc
 
 	m         sync.Mutex // protect following fields
 	closed    bool
@@ -61,17 +62,16 @@ type ReuseConnOpts struct {
 func NewReuseConnTransport(opt ReuseConnOpts) *ReuseConnTransport {
 	ctx, cancel := context.WithCancelCause(context.Background())
 	t := &ReuseConnTransport{
-		ctx:           ctx,
-		ctxCancel:     cancel,
-		ReuseConnOpts: opt,
-		idleConns:     make(map[*reusableConn]struct{}),
-		conns:         make(map[*reusableConn]struct{}),
-	}
-	if opt.Logger != nil {
-		t.logger = opt.Logger
-	} else {
-		t.logger = nopLogger
+		ctx:       ctx,
+		ctxCancel: cancel,
+		idleConns: make(map[*reusableConn]struct{}),
+		conns:     make(map[*reusableConn]struct{}),
 	}
+	t.dialFunc = opt.DialContext
+	setDefaultGZ(&t.dialTimeout, opt.DialTimeout, defaultDialTimeout)
+	setDefaultGZ(&t.idleTimeout, opt.IdleTimeout, defaultIdleTimeout)
+	setNonNilLogger(&t.logger, opt.Logger)
+
 	return t
 }
 
@@ -136,25 +136,16 @@ func (t *ReuseConnTransport) wantNewConn(ctx context.Context) (*reusableConn, er
 
 	dialChan := make(chan dialRes)
 	go func() {
-		dialTimeout := t.DialTimeout
-		if dialTimeout <= 0 {
-			dialTimeout = defaultDialTimeout
-		}
-		idleTimeout := t.IdleTimeout
-		if idleTimeout <= 0 {
-			idleTimeout = defaultIdleTimeout
-		}
-
-		ctx, cancel := context.WithTimeout(t.ctx, dialTimeout)
+		ctx, cancel := context.WithTimeout(t.ctx, t.dialTimeout)
 		defer cancel()
 
 		var rc *reusableConn
-		c, err := t.DialContext(ctx)
+		c, err := t.dialFunc(ctx)
 		if err != nil {
 			t.logger.Check(zap.WarnLevel, "fail to dial reusable conn").Write(zap.Error(err))
 		}
 		if c != nil {
-			rc = newReusableConn(c, idleTimeout)
+			rc = newReusableConn(c, t.idleTimeout)
 			t.trackNewReusableConn(rc)
 		}
 

pkg/upstream/transport/transport.go

@@ -26,7 +26,6 @@ import (
 	"time"
 
 	"github.com/IrineSistiana/mosdns/v5/pkg/pool"
-	"go.uber.org/zap"
 )
 
 var (
@@ -36,8 +35,6 @@ var (
 	ErrLazyConnCannotReserveQueryExchanger = errors.New("lazy connection failed to reserve query exchanger")
 )
 
-var nopLogger = zap.NewNop()
-
 func ReleaseResp(b *[]byte) {
 	pool.ReleaseBuf(b)
 }
@@ -55,8 +52,8 @@ const (
 	// something goes wrong with the connection or the server. The connection will be closed.
 	waitingReplyTimeout = time.Second * 10
 
-	defaultTdcMaxConcurrentQuery      = 32
-	defaultLazyConnMaxConcurrentQuery = 16
+	defaultTdcMaxConcurrentQuery = 32
+	defaultMaxLazyConnQueue      = 16
 )
 
 // One method MUST be called in ReservedExchanger.

pkg/upstream/transport/utils.go

@@ -26,6 +26,8 @@ import (
 
 	"github.com/IrineSistiana/mosdns/v5/pkg/pool"
 	"github.com/miekg/dns"
+	"go.uber.org/zap"
+	"golang.org/x/exp/constraints"
 )
 
 const (
@@ -87,3 +89,21 @@ readAgain:
 	*payload = (*payload)[:n]
 	return payload, err
 }
+
+func setDefaultGZ[T constraints.Float | constraints.Integer](i *T, s, d T) {
+	if s > 0 {
+		*i = s
+	} else {
+		*i = d
+	}
+}
+
+var nopLogger = zap.NewNop()
+
+func setNonNilLogger(i **zap.Logger, s *zap.Logger) {
+	if s != nil {
+		*i = s
+	} else {
+		*i = nopLogger
+	}
+}