func (t *ConnectionState) disconnect() error {
	t.config.Log.Info().Msgf("Target %s: Disconnecting", t.name)
	t.stopped = true
	return t.client.Close()  
}
Disconnect from the target or stop trying to connect.
package connections
import (
	"context"
	"crypto/tls"
	"fmt"
	"github.com/go-zookeeper/zk"
	"github.com/openconfig/gnmi/errlist"
	"sync"
	"time"
	"github.com/Netflix/spectator-go"
	"github.com/Netflix/spectator-go/histogram"
	"github.com/golang/protobuf/proto"
	"github.com/openconfig/gnmi/cache"
	"github.com/openconfig/gnmi/client"
	gnmiclient "github.com/openconfig/gnmi/client/gnmi"
	gnmipb "github.com/openconfig/gnmi/proto/gnmi"
	targetpb "github.com/openconfig/gnmi/proto/target"
	"golang.org/x/sync/semaphore"
	"github.com/openconfig/gnmi-gateway/gateway/configuration"
	"github.com/openconfig/gnmi-gateway/gateway/locking"
	"github.com/openconfig/gnmi-gateway/gateway/stats"
	"github.com/openconfig/gnmi-gateway/gateway/utils"
)
type ConnectionState struct {
	ConnectionLockAcquired bool
	client                 *client.ReconnectClient
	clientCancel           context.CancelFunc
	clusterMember          bool
	config                 *configuration.GatewayConfig
	connected bool
	connecting  bool
	connManager ConnectionManager
	lock locking.DistributedLocker
	name string
	noTLSWarning bool
	queryTarget  string
	request      *gnmipb.SubscribeRequest
	seen      map[string]bool
	seenMutex sync.Mutex
	stopped bool
	synced      bool
	target      *targetpb.Target
	targetCache *cache.Target
	useLock     bool
	metricTags           map[string]string
	counterCoalesced     *spectator.Counter
	counterNotifications *spectator.Counter
	counterRejected      *spectator.Counter
	counterStale         *spectator.Counter
	counterSync          *spectator.Counter
	gaugeSynced          *spectator.Gauge
	timerLatency         *histogram.PercentileTimer
}
func (t *ConnectionState) InitializeMetrics() {
	t.metricTags = map[string]string{"gnmigateway.client.target": t.name}
	t.counterCoalesced = stats.Registry.Counter("gnmigateway.client.subscribe.coalesced", t.metricTags)
	t.counterNotifications = stats.Registry.Counter("gnmigateway.client.subscribe.notifications", t.metricTags)
	t.counterRejected = stats.Registry.Counter("gnmigateway.client.subscribe.rejected", t.metricTags)
	t.counterStale = stats.Registry.Counter("gnmigateway.client.subscribe.stale", t.metricTags)
	t.counterSync = stats.Registry.Counter("gnmigateway.client.subscribe.sync", t.metricTags)
	t.gaugeSynced = stats.Registry.Gauge("gnmigateway.client.subscribe.synced", t.metricTags)
	t.timerLatency = histogram.NewPercentileTimer(stats.Registry, "gnmigateway.client.subscribe.latency", t.metricTags)
}
func (t *ConnectionState) Equal(other *targetpb.Target) bool {
	if len(t.target.Addresses) != len(other.Addresses) {
		return false
	}
	for i, addr := range t.target.Addresses {
		if other.Addresses[i] != addr {
			return false
		}
	}
	if t.target.Credentials == nil && other.Credentials != nil || t.target.Credentials != nil && other.Credentials == nil {
		return false
	}
	if t.target.Credentials != nil && other.Credentials != nil {
		if t.target.Credentials.Username != other.Credentials.Username {
			return false
		}
		if t.target.Credentials.Password != other.Credentials.Password {
			return false
		}
	}
	return true
}
func (t *ConnectionState) Seen(target string) bool {
	t.seenMutex.Lock()
	defer t.seenMutex.Unlock()
	return t.seen[target]
}
func (t *ConnectionState) doConnect() {
	t.connecting = true
	t.config.Log.Info().Msgf("Target %s: Connecting", t.name)
	query, err := client.NewQuery(t.request)
	if err != nil {
		t.config.Log.Error().Msgf("Target %s: unable to create query: NewQuery(%s): %v", t.name, t.request.String(), err)
		return
	}
	query.Addrs = t.target.Addresses
	if t.target.Credentials != nil {
		query.Credentials = &client.Credentials{
			Username: t.target.Credentials.Username,
			Password: t.target.Credentials.Password,
		}
	}
	_, NoTLS := t.target.Meta["NoTLS"]
	if NoTLS && !t.noTLSWarning {
		t.noTLSWarning = true
		t.config.Log.Warn().Msg("DEPRECATED: The 'NoTLS' target flag has been deprecated and will be removed in a future release. Please use 'NoTLSVerify' instead.")
	}
	_, NoTLSVerify := t.target.Meta["NoTLSVerify"]
	if t.config.ClientTLSConfig != nil && !NoTLS && !NoTLSVerify {
		query.TLS = t.config.ClientTLSConfig
	} else {
		query.TLS = &tls.Config{
			InsecureSkipVerify: true,
		}
	}
	var prefixTarget string
	if t.request.GetSubscribe() != nil && t.request.GetSubscribe().GetPrefix() != nil {
		prefixTarget = t.request.GetSubscribe().GetPrefix().GetTarget()
	}
	if prefixTarget != "" {
		query.Target = prefixTarget
		t.queryTarget = prefixTarget
	} else {
		query.Target = t.name
		t.queryTarget = t.name
	}
	query.Timeout = t.config.TargetDialTimeout
	query.ProtoHandler = t.handleUpdate
	if err := query.Validate(); err != nil {
		t.config.Log.Error().Err(err).Msgf("query.Validate(): %v", err)
		return
	}
	var ctx context.Context
	ctx, t.clientCancel = context.WithCancel(context.Background())
	t.config.Log.Info().Msgf("Target %s: Subscribing", t.name)
	t.client = client.Reconnect(&client.BaseClient{}, t.disconnected, t.reset)
	if err := t.client.Subscribe(ctx, query, gnmiclient.Type); err != nil {
		t.config.Log.Info().Msgf("Target %s: Subscribe stopped: %v", t.name, err)
	}
}
func (t *ConnectionState) connect(connectionSlot *semaphore.Weighted) {
	var connectionSlotAcquired = false
	for !t.stopped {
		if !connectionSlotAcquired {
			connectionSlotAcquired = connectionSlot.TryAcquire(1)
		}
		if connectionSlotAcquired {
			t.doConnect()
		}
	}
	if connectionSlotAcquired {
		connectionSlot.Release(1)
	}
}
func (t *ConnectionState) connectWithLock(connectionSlot *semaphore.Weighted) {
	var connectionSlotAcquired = false
	for !t.stopped {
		if !connectionSlotAcquired {
			t.config.Log.Info().Msgf("Target %s: Acquiring connection slot", t.name)
			connectionSlotAcquired = connectionSlot.TryAcquire(1)
		}
		if connectionSlotAcquired {
			if !t.ConnectionLockAcquired {
				t.config.Log.Info().Msgf("Target %s: Acquiring lock", t.name)
				var err error
				t.ConnectionLockAcquired, err = t.lock.Try()
				if err != nil {
					t.config.Log.Error().Msgf("Target %s: error while trying to acquire lock: %v", t.name, err)
					time.Sleep(2 * time.Second)
				}
			}
			if t.ConnectionLockAcquired {
				t.config.Log.Info().Msgf("Target %s: Lock acquired", t.name)
				t.doConnect()
				if t.lock.LockAcquired() {
					err := t.lock.Unlock()
					if err != nil && err != zk.ErrNotLocked {
						t.config.Log.Error().Msgf("Target %s: error while releasing lock: %v", t.name, err)
					}
				}
				t.ConnectionLockAcquired = false
				t.config.Log.Info().Msgf("Target %s: Lock released", t.name)
			} else {
				time.Sleep(1 * time.Second)
			}
		}
	}
	t.config.Log.Info().Msgf("Target %s: Stopped", t.name)
	if connectionSlotAcquired {
		connectionSlot.Release(1)
	}
}