func (c *Connection) Expire() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.connection == nil {
		return nil
	}
	_ = c.close()
	err := c.pool.put(c.connection)
	c.connection = nil
	return err
}
Expire closes this connection and will closeConnection the underlying socket.
package topology
import (
	"context"
	"crypto/tls"
	"errors"
	"fmt"
	"io"
	"net"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"go.mongodb.org/mongo-driver/event"
	"go.mongodb.org/mongo-driver/internal"
	"go.mongodb.org/mongo-driver/mongo/address"
	"go.mongodb.org/mongo-driver/mongo/description"
	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
	"go.mongodb.org/mongo-driver/x/mongo/driver"
	"go.mongodb.org/mongo-driver/x/mongo/driver/ocsp"
	"go.mongodb.org/mongo-driver/x/mongo/driver/wiremessage"
)
var globalConnectionID uint64 = 1
var (
	defaultMaxMessageSize uint32 = 48000000
	errResponseTooLarge   error  = errors.New("length of read message too large")
)
func nextConnectionID() uint64 { return atomic.AddUint64(&globalConnectionID, 1) }
type connection struct {
	id                   string
	nc                   net.Conn  
	addr                 address.Address
	idleTimeout          time.Duration
	idleDeadline         atomic.Value  
	readTimeout          time.Duration
	writeTimeout         time.Duration
	desc                 description.Server
	isMasterRTT          time.Duration
	compressor           wiremessage.CompressorID
	zliblevel            int
	zstdLevel            int
	connected            int32  
	connectDone          chan struct{}
	connectErr           error
	config               *connectionConfig
	cancelConnectContext context.CancelFunc
	connectContextMade   chan struct{}
	canStream            bool
	currentlyStreaming   bool
	connectContextMutex  sync.Mutex
	cancellationListener cancellationListener
	pool         *pool
	poolID       uint64
	generation   uint64
	expireReason string
	poolMonitor  *event.PoolMonitor
}
func newConnection(addr address.Address, opts ...ConnectionOption) (*connection, error) {
	cfg, err := newConnectionConfig(opts...)
	if err != nil {
		return nil, err
	}
	id := fmt.Sprintf("%s[-%d]", addr, nextConnectionID())
	c := &connection{
		id:                   id,
		addr:                 addr,
		idleTimeout:          cfg.idleTimeout,
		readTimeout:          cfg.readTimeout,
		writeTimeout:         cfg.writeTimeout,
		connectDone:          make(chan struct{}),
		config:               cfg,
		connectContextMade:   make(chan struct{}),
		cancellationListener: internal.NewCancellationListener(),
		poolMonitor:          cfg.poolMonitor,
	}
	atomic.StoreInt32(&c.connected, initialized)
	return c, nil
}
func (c *connection) processInitializationError(err error) {
	atomic.StoreInt32(&c.connected, disconnected)
	if c.nc != nil {
		_ = c.nc.Close()
	}
	c.connectErr = ConnectionError{Wrapped: err, init: true}
	if c.config.errorHandlingCallback != nil {
		c.config.errorHandlingCallback(c.connectErr, c.generation)
	}
}
func (c *connection) connect(ctx context.Context) {
	if !atomic.CompareAndSwapInt32(&c.connected, initialized, connected) {
		return
	}
	defer close(c.connectDone)
	c.connectContextMutex.Lock()
	var handshakeCtx context.Context
	handshakeCtx, c.cancelConnectContext = context.WithCancel(ctx)
	c.connectContextMutex.Unlock()
	dialCtx := handshakeCtx
	var dialCancel context.CancelFunc
	if c.config.connectTimeout != 0 {
		dialCtx, dialCancel = context.WithTimeout(handshakeCtx, c.config.connectTimeout)
		defer dialCancel()
	}
	defer func() {
		var cancelFn context.CancelFunc
		c.connectContextMutex.Lock()
		cancelFn = c.cancelConnectContext
		c.cancelConnectContext = nil
		c.connectContextMutex.Unlock()
		if cancelFn != nil {
			cancelFn()
		}
	}()
	close(c.connectContextMade)
	var err error
	var tempNc net.Conn
	tempNc, err = c.config.dialer.DialContext(dialCtx, c.addr.Network(), c.addr.String())
	if err != nil {
		c.processInitializationError(err)
		return
	}
	c.nc = tempNc
	if c.config.tlsConfig != nil {
		tlsConfig := c.config.tlsConfig.Clone()
		ocspOpts := &ocsp.VerifyOptions{
			Cache:                   c.config.ocspCache,
			DisableEndpointChecking: c.config.disableOCSPEndpointCheck,
		}
		tlsNc, err := configureTLS(dialCtx, c.config.tlsConnectionSource, c.nc, c.addr, tlsConfig, ocspOpts)
		if err != nil {
			c.processInitializationError(err)
			return
		}
		c.nc = tlsNc
	}
	c.bumpIdleDeadline()
	handshaker := c.config.handshaker
	if handshaker == nil {
		if c.poolMonitor != nil {
			c.poolMonitor.Event(&event.PoolEvent{
				Type:         event.ConnectionReady,
				Address:      c.addr.String(),
				ConnectionID: c.poolID,
			})
		}
		return
	}
	var handshakeInfo driver.HandshakeInformation
	handshakeStartTime := time.Now()
	handshakeConn := initConnection{c}
	handshakeInfo, err = handshaker.GetHandshakeInformation(handshakeCtx, c.addr, handshakeConn)
	if err == nil {
		c.desc = handshakeInfo.Description
		c.isMasterRTT = time.Since(handshakeStartTime)
		err = handshaker.FinishHandshake(handshakeCtx, handshakeConn)
	}
	if err != nil {
		c.processInitializationError(err)
		return
	}
	if len(c.desc.Compression) > 0 {
	clientMethodLoop:
		for _, method := range c.config.compressors {
			for _, serverMethod := range c.desc.Compression {
				if method != serverMethod {
					continue
				}
				switch strings.ToLower(method) {
				case "snappy":
					c.compressor = wiremessage.CompressorSnappy
				case "zlib":
					c.compressor = wiremessage.CompressorZLib
					c.zliblevel = wiremessage.DefaultZlibLevel
					if c.config.zlibLevel != nil {
						c.zliblevel = *c.config.zlibLevel
					}
				case "zstd":
					c.compressor = wiremessage.CompressorZstd
					c.zstdLevel = wiremessage.DefaultZstdLevel
					if c.config.zstdLevel != nil {
						c.zstdLevel = *c.config.zstdLevel
					}
				}
				break clientMethodLoop
			}
		}
	}
	if c.poolMonitor != nil {
		c.poolMonitor.Event(&event.PoolEvent{
			Type:         event.ConnectionReady,
			Address:      c.addr.String(),
			ConnectionID: c.poolID,
		})
	}
}
func (c *connection) wait() error {
	if c.connectDone != nil {
		<-c.connectDone
	}
	return c.connectErr
}
func (c *connection) closeConnectContext() {
	<-c.connectContextMade
	var cancelFn context.CancelFunc
	c.connectContextMutex.Lock()
	cancelFn = c.cancelConnectContext
	c.cancelConnectContext = nil
	c.connectContextMutex.Unlock()
	if cancelFn != nil {
		cancelFn()
	}
}
func transformNetworkError(ctx context.Context, originalError error, contextDeadlineUsed bool) error {
	if originalError == nil {
		return nil
	}
	if ctx.Err() == context.Canceled {
		return context.Canceled
	}
	if !contextDeadlineUsed {
		return originalError
	}
	if netErr, ok := originalError.(net.Error); ok && netErr.Timeout() {
		return context.DeadlineExceeded
	}
	return originalError
}
func (c *connection) cancellationListenerCallback() {
	_ = c.close()
}
func (c *connection) writeWireMessage(ctx context.Context, wm []byte) error {
	var err error
	if atomic.LoadInt32(&c.connected) != connected {
		return ConnectionError{ConnectionID: c.id, message: "connection is closed"}
	}
	select {
	case <-ctx.Done():
		return ConnectionError{ConnectionID: c.id, Wrapped: ctx.Err(), message: "failed to write"}
	default:
	}
	var deadline time.Time
	if c.writeTimeout != 0 {
		deadline = time.Now().Add(c.writeTimeout)
	}
	var contextDeadlineUsed bool
	if dl, ok := ctx.Deadline(); ok && (deadline.IsZero() || dl.Before(deadline)) {
		contextDeadlineUsed = true
		deadline = dl
	}
	if err := c.nc.SetWriteDeadline(deadline); err != nil {
		return ConnectionError{ConnectionID: c.id, Wrapped: err, message: "failed to set write deadline"}
	}
	err = c.write(ctx, wm)
	if err != nil {
		c.close()
		return ConnectionError{
			ConnectionID: c.id,
			Wrapped:      transformNetworkError(ctx, err, contextDeadlineUsed),
			message:      "unable to write wire message to network",
		}
	}
	c.bumpIdleDeadline()
	return nil
}
func (c *connection) write(ctx context.Context, wm []byte) (err error) {
	go c.cancellationListener.Listen(ctx, c.cancellationListenerCallback)
	defer func() {
		if aborted := c.cancellationListener.StopListening(); aborted && err == nil {
			err = context.Canceled
		}
	}()
	_, err = c.nc.Write(wm)
	return err
}
func (c *connection) readWireMessage(ctx context.Context, dst []byte) ([]byte, error) {
	if atomic.LoadInt32(&c.connected) != connected {
		return dst, ConnectionError{ConnectionID: c.id, message: "connection is closed"}
	}
	select {
	case <-ctx.Done():
		c.close()
		return nil, ConnectionError{ConnectionID: c.id, Wrapped: ctx.Err(), message: "failed to read"}
	default:
	}
	var deadline time.Time
	if c.readTimeout != 0 {
		deadline = time.Now().Add(c.readTimeout)
	}
	var contextDeadlineUsed bool
	if dl, ok := ctx.Deadline(); ok && (deadline.IsZero() || dl.Before(deadline)) {
		contextDeadlineUsed = true
		deadline = dl
	}
	if err := c.nc.SetReadDeadline(deadline); err != nil {
		return nil, ConnectionError{ConnectionID: c.id, Wrapped: err, message: "failed to set read deadline"}
	}
	dst, errMsg, err := c.read(ctx, dst)
	if err != nil {
		c.close()
		message := errMsg
		if err == io.EOF {
			message = "socket was unexpectedly closed"
		}
		return nil, ConnectionError{
			ConnectionID: c.id,
			Wrapped:      transformNetworkError(ctx, err, contextDeadlineUsed),
			message:      message,
		}
	}
	c.bumpIdleDeadline()
	return dst, nil
}
func (c *connection) read(ctx context.Context, dst []byte) (bytesRead []byte, errMsg string, err error) {
	go c.cancellationListener.Listen(ctx, c.cancellationListenerCallback)
	defer func() {
		if aborted := c.cancellationListener.StopListening(); aborted && err == nil {
			errMsg = "unable to read server response"
			err = context.Canceled
		}
	}()
	var sizeBuf [4]byte
	_, err = io.ReadFull(c.nc, sizeBuf[:])
	if err != nil {
		return nil, "incomplete read of message header", err
	}
	size := (int32(sizeBuf[0])) | (int32(sizeBuf[1]) << 8) | (int32(sizeBuf[2]) << 16) | (int32(sizeBuf[3]) << 24)
	maxMessageSize := c.desc.MaxMessageSize
	if maxMessageSize == 0 {
		maxMessageSize = defaultMaxMessageSize
	}
	if uint32(size) > maxMessageSize {
		return nil, errResponseTooLarge.Error(), errResponseTooLarge
	}
	if int(size) > cap(dst) {
		dst = make([]byte, 0, size)
	}
	dst = dst[:size]
	copy(dst, sizeBuf[:])
	_, err = io.ReadFull(c.nc, dst[4:])
	if err != nil {
		return nil, "incomplete read of full message", err
	}
	return dst, "", nil
}
func (c *connection) close() error {
	if !atomic.CompareAndSwapInt32(&c.connected, connected, disconnected) {
		return nil
	}
	var err error
	if c.nc != nil {
		err = c.nc.Close()
	}
	return err
}
func (c *connection) closed() bool {
	return atomic.LoadInt32(&c.connected) == disconnected
}
func (c *connection) idleTimeoutExpired() bool {
	now := time.Now()
	if c.idleTimeout > 0 {
		idleDeadline, ok := c.idleDeadline.Load().(time.Time)
		if ok && now.After(idleDeadline) {
			return true
		}
	}
	return false
}
func (c *connection) bumpIdleDeadline() {
	if c.idleTimeout > 0 {
		c.idleDeadline.Store(time.Now().Add(c.idleTimeout))
	}
}
func (c *connection) setCanStream(canStream bool) {
	c.canStream = canStream
}
func (c initConnection) supportsStreaming() bool {
	return c.canStream
}
func (c *connection) setStreaming(streaming bool) {
	c.currentlyStreaming = streaming
}
func (c *connection) getCurrentlyStreaming() bool {
	return c.currentlyStreaming
}
func (c *connection) setSocketTimeout(timeout time.Duration) {
	c.readTimeout = timeout
	c.writeTimeout = timeout
}
func (c *connection) ID() string {
	return c.id
}
type initConnection struct{ *connection }
var _ driver.Connection = initConnection{}
var _ driver.StreamerConnection = initConnection{}
func (c initConnection) Description() description.Server {
	if c.connection == nil {
		return description.Server{}
	}
	return c.connection.desc
}
func (c initConnection) Close() error             { return nil }
func (c initConnection) ID() string               { return c.id }
func (c initConnection) Address() address.Address { return c.addr }
func (c initConnection) Stale() bool              { return false }
func (c initConnection) LocalAddress() address.Address {
	if c.connection == nil || c.nc == nil {
		return address.Address("0.0.0.0")
	}
	return address.Address(c.nc.LocalAddr().String())
}
func (c initConnection) WriteWireMessage(ctx context.Context, wm []byte) error {
	return c.writeWireMessage(ctx, wm)
}
func (c initConnection) ReadWireMessage(ctx context.Context, dst []byte) ([]byte, error) {
	return c.readWireMessage(ctx, dst)
}
func (c initConnection) SetStreaming(streaming bool) {
	c.setStreaming(streaming)
}
func (c initConnection) CurrentlyStreaming() bool {
	return c.getCurrentlyStreaming()
}
func (c initConnection) SupportsStreaming() bool {
	return c.supportsStreaming()
}
type Connection struct {
	*connection
	mu sync.RWMutex
}
var _ driver.Connection = (*Connection)(nil)
var _ driver.Expirable = (*Connection)(nil)
func (c *Connection) WriteWireMessage(ctx context.Context, wm []byte) error {
	c.mu.RLock()
	defer c.mu.RUnlock()
	if c.connection == nil {
		return ErrConnectionClosed
	}
	return c.writeWireMessage(ctx, wm)
}
func (c *Connection) ReadWireMessage(ctx context.Context, dst []byte) ([]byte, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	if c.connection == nil {
		return dst, ErrConnectionClosed
	}
	return c.readWireMessage(ctx, dst)
}
func (c *Connection) CompressWireMessage(src, dst []byte) ([]byte, error) {
	c.mu.RLock()
	defer c.mu.RUnlock()
	if c.connection == nil {
		return dst, ErrConnectionClosed
	}
	if c.connection.compressor == wiremessage.CompressorNoOp {
		return append(dst, src...), nil
	}
	_, reqid, respto, origcode, rem, ok := wiremessage.ReadHeader(src)
	if !ok {
		return dst, errors.New("wiremessage is too short to compress, less than 16 bytes")
	}
	idx, dst := wiremessage.AppendHeaderStart(dst, reqid, respto, wiremessage.OpCompressed)
	dst = wiremessage.AppendCompressedOriginalOpCode(dst, origcode)
	dst = wiremessage.AppendCompressedUncompressedSize(dst, int32(len(rem)))
	dst = wiremessage.AppendCompressedCompressorID(dst, c.connection.compressor)
	opts := driver.CompressionOpts{
		Compressor: c.connection.compressor,
		ZlibLevel:  c.connection.zliblevel,
		ZstdLevel:  c.connection.zstdLevel,
	}
	compressed, err := driver.CompressPayload(rem, opts)
	if err != nil {
		return nil, err
	}
	dst = wiremessage.AppendCompressedCompressedMessage(dst, compressed)
	return bsoncore.UpdateLength(dst, idx, int32(len(dst[idx:]))), nil
}
func (c *Connection) Description() description.Server {
	c.mu.RLock()
	defer c.mu.RUnlock()
	if c.connection == nil {
		return description.Server{}
	}
	return c.desc
}
func (c *Connection) Close() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.connection == nil {
		return nil
	}
	err := c.pool.put(c.connection)
	c.connection = nil
	return err
}