func (t *Topology) Unsubscribe(sub *driver.Subscription) error {
	t.subLock.Lock()
	defer t.subLock.Unlock()
	if t.subscriptionsClosed {
		return nil
	}
	ch, ok := t.subscribers[sub.ID]
	if !ok {
		return nil
	}
	close(ch)
	delete(t.subscribers, sub.ID)
	return nil
}
Unsubscribe unsubscribes the given subscription from the topology and closes the subscription channel.
Unsubscribe implements the driver.Subscriber interface.
package topology  
import (
	"context"
	"errors"
	"math/rand"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"fmt"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/event"
	"go.mongodb.org/mongo-driver/mongo/address"
	"go.mongodb.org/mongo-driver/mongo/description"
	"go.mongodb.org/mongo-driver/x/mongo/driver"
	"go.mongodb.org/mongo-driver/x/mongo/driver/dns"
)
var ErrSubscribeAfterClosed = errors.New("cannot subscribe after closeConnection")
var ErrTopologyClosed = errors.New("topology is closed")
var ErrTopologyConnected = errors.New("topology is connected or connecting")
var ErrServerSelectionTimeout = errors.New("server selection timeout")
type MonitorMode uint8
const (
	AutomaticMode MonitorMode = iota
	SingleMode
)
type Topology struct {
	connectionstate int32
	cfg *config
	desc atomic.Value  
	dnsResolver *dns.Resolver
	done chan struct{}
	pollingRequired   bool
	pollingDone       chan struct{}
	pollingwg         sync.WaitGroup
	rescanSRVInterval time.Duration
	pollHeartbeatTime atomic.Value  
	updateCallback updateTopologyCallback
	fsm            *fsm
	subscribers         map[uint64]chan description.Topology
	currentSubscriberID uint64
	subscriptionsClosed bool
	subLock             sync.Mutex
	serversLock   sync.Mutex
	serversClosed bool
	servers       map[address.Address]*Server
	id primitive.ObjectID
}
var _ driver.Deployment = &Topology{}
var _ driver.Subscriber = &Topology{}
type serverSelectionState struct {
	selector    description.ServerSelector
	timeoutChan <-chan time.Time
}
func newServerSelectionState(selector description.ServerSelector, timeoutChan <-chan time.Time) serverSelectionState {
	return serverSelectionState{
		selector:    selector,
		timeoutChan: timeoutChan,
	}
}
func New(opts ...Option) (*Topology, error) {
	cfg, err := newConfig(opts...)
	if err != nil {
		return nil, err
	}
	t := &Topology{
		cfg:               cfg,
		done:              make(chan struct{}),
		pollingDone:       make(chan struct{}),
		rescanSRVInterval: 60 * time.Second,
		fsm:               newFSM(),
		subscribers:       make(map[uint64]chan description.Topology),
		servers:           make(map[address.Address]*Server),
		dnsResolver:       dns.DefaultResolver,
		id:                primitive.NewObjectID(),
	}
	t.desc.Store(description.Topology{})
	t.updateCallback = func(desc description.Server) description.Server {
		return t.apply(context.TODO(), desc)
	}
	if t.cfg.uri != "" {
		t.pollingRequired = strings.HasPrefix(t.cfg.uri, "mongodb+srv://")
	}
	t.publishTopologyOpeningEvent()
	return t, nil
}
func (t *Topology) Connect() error {
	if !atomic.CompareAndSwapInt32(&t.connectionstate, disconnected, connecting) {
		return ErrTopologyConnected
	}
	t.desc.Store(description.Topology{})
	var err error
	t.serversLock.Lock()
	if t.cfg.replicaSetName != "" {
		t.fsm.SetName = t.cfg.replicaSetName
		t.fsm.Kind = description.ReplicaSetNoPrimary
	}
	if t.cfg.mode == SingleMode {
		t.fsm.Kind = description.Single
	}
	for _, a := range t.cfg.seedList {
		addr := address.Address(a).Canonicalize()
		t.fsm.Servers = append(t.fsm.Servers, description.NewDefaultServer(addr))
	}
	newDesc := description.Topology{
		Kind:                  t.fsm.Kind,
		Servers:               t.fsm.Servers,
		SessionTimeoutMinutes: t.fsm.SessionTimeoutMinutes,
	}
	t.desc.Store(newDesc)
	t.publishTopologyDescriptionChangedEvent(description.Topology{}, t.fsm.Topology)
	for _, a := range t.cfg.seedList {
		addr := address.Address(a).Canonicalize()
		err = t.addServer(addr)
		if err != nil {
			return err
		}
	}
	t.serversLock.Unlock()
	if t.pollingRequired {
		go t.pollSRVRecords()
		t.pollingwg.Add(1)
	}
	t.subscriptionsClosed = false  
	atomic.StoreInt32(&t.connectionstate, connected)
	return nil
}
func (t *Topology) Disconnect(ctx context.Context) error {
	if !atomic.CompareAndSwapInt32(&t.connectionstate, connected, disconnecting) {
		return ErrTopologyClosed
	}
	servers := make(map[address.Address]*Server)
	t.serversLock.Lock()
	t.serversClosed = true
	for addr, server := range t.servers {
		servers[addr] = server
	}
	t.serversLock.Unlock()
	for _, server := range servers {
		_ = server.Disconnect(ctx)
		t.publishServerClosedEvent(server.address)
	}
	t.subLock.Lock()
	for id, ch := range t.subscribers {
		close(ch)
		delete(t.subscribers, id)
	}
	t.subscriptionsClosed = true
	t.subLock.Unlock()
	if t.pollingRequired {
		t.pollingDone <- struct{}{}
		t.pollingwg.Wait()
	}
	t.desc.Store(description.Topology{})
	atomic.StoreInt32(&t.connectionstate, disconnected)
	t.publishTopologyClosedEvent()
	return nil
}
func (t *Topology) Description() description.Topology {
	td, ok := t.desc.Load().(description.Topology)
	if !ok {
		td = description.Topology{}
	}
	return td
}
func (t *Topology) Kind() description.TopologyKind { return t.Description().Kind }
func (t *Topology) Subscribe() (*driver.Subscription, error) {
	if atomic.LoadInt32(&t.connectionstate) != connected {
		return nil, errors.New("cannot subscribe to Topology that is not connected")
	}
	ch := make(chan description.Topology, 1)
	td, ok := t.desc.Load().(description.Topology)
	if !ok {
		td = description.Topology{}
	}
	ch <- td
	t.subLock.Lock()
	defer t.subLock.Unlock()
	if t.subscriptionsClosed {
		return nil, ErrSubscribeAfterClosed
	}
	id := t.currentSubscriberID
	t.subscribers[id] = ch
	t.currentSubscriberID++
	return &driver.Subscription{
		Updates: ch,
		ID:      id,
	}, nil
}