func matchPath(path []*gnmipb.PathElem, toMatch []*gnmipb.PathElem) bool {
	if len(path) < len(toMatch) {
		return false
	}
	for i, elem := range toMatch {
		if path[i].Name != elem.Name {
			return false
		}
		if elem.Key != nil {
			if path[i].Key == nil {
				return false
			}
			for k, v := range elem.Key {
				ov, exists := path[i].Key[k]
				if !exists {
					return false
				}
				if v != ov {
					return false
				}
			}
		}
	}
	return true
}
Return true if all of the elements in toMatch are found in path.
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)
	}
}
func (t *ConnectionState) disconnect() error {
	t.config.Log.Info().Msgf("Target %s: Disconnecting", t.name)
	t.stopped = true
	return t.client.Close()  
}
func (t *ConnectionState) reset() {
	t.config.Log.Info().Msgf("Target %s: gNMI client will reconnect", t.name)
}
func (t *ConnectionState) disconnected() {
	t.connected = false
	t.synced = false
	t.seenMutex.Lock()
	t.seen = map[string]bool{}
	t.seenMutex.Unlock()
	if t.queryTarget != "*" {
		t.targetCache.Reset()
	}
	t.config.Log.Info().Msgf("Target %s: Disconnected", t.name)
}
func (t *ConnectionState) reconnect() error {
	t.config.Log.Info().Msgf("Target %s: Reconnecting", t.name)
	return t.client.Close()
}
func (t *ConnectionState) unlock() error {
	t.config.Log.Info().Msgf("Target %s: Unlocking", t.name)
	t.clientCancel()
	return nil
}
func (t *ConnectionState) handleUpdate(msg proto.Message) error {
	t.counterNotifications.Increment()
	if !t.connected {
		if t.queryTarget != "*" {
			t.targetCache.Connect()
		}
		t.connected = true
		t.config.Log.Info().Msgf("Target %s: Connected", t.name)
	}
	resp, ok := msg.(*gnmipb.SubscribeResponse)
	if !ok {
		return fmt.Errorf("failed to type assert message %#v", msg)
	}
	switch v := resp.Response.(type) {
	case *gnmipb.SubscribeResponse_Update:
		if t.rejectUpdate(v.Update) {
			t.counterRejected.Increment()
			return nil
		}
		if t.synced {
			for _, u := range v.Update.Update {
				t.counterCoalesced.Add(int64(u.Duplicates))
			}
			t.timerLatency.Record(time.Duration(time.Now().UnixNano() - v.Update.Timestamp))
		}
		switch t.queryTarget {
		case "*":
			targetCache := t.connManager.Cache().GetTarget(v.Update.Prefix.Target)
			if targetCache == nil {
				targetCache = t.connManager.Cache().Add(v.Update.Prefix.Target)
			}
			t.seenMutex.Lock()
			t.seen[v.Update.Prefix.Target] = true
			t.seenMutex.Unlock()
			err := t.updateTargetCache(targetCache, v.Update)
			if err != nil {
				return err
			}
		default:
			if v.Update.GetPrefix() == nil {
				v.Update.Prefix = &gnmipb.Path{}
			}
			if v.Update.Prefix.Target == "" {
				v.Update.Prefix.Target = t.queryTarget
			}
			err := t.updateTargetCache(t.targetCache, v.Update)
			if err != nil {
				return err
			}
		}
	case *gnmipb.SubscribeResponse_SyncResponse:
		t.sync()
		switch t.queryTarget {
		case "*":
		default:
			t.targetCache.Sync()
		}
	case *gnmipb.SubscribeResponse_Error:
		return fmt.Errorf("error in response: %s", v)
	default:
		return fmt.Errorf("unknown response %T: %v", v, v)
	}
	return nil
}
func (t *ConnectionState) sync() {
	t.config.Log.Info().Msgf("Target %s: Synced", t.name)
	t.synced = true
	t.counterSync.Increment()
	go func() {
		ticker := time.NewTicker(30 * time.Second)
		for t.synced {
			select {
			case <-ticker.C:
				t.gaugeSynced.Set(1)
			}
		}
	}()
}
func (t *ConnectionState) updateTargetCache(cache *cache.Target, update *gnmipb.Notification) error {
	var hasError bool
	err := cache.GnmiUpdate(update)
	if err != nil {
		errList, isList := err.(errlist.Error)
		if isList {
			for _, err := range errList.Errors() {
				hasError = t.handleCacheError(err)
				if hasError {
					break
				}
			}
		} else {
			hasError = t.handleCacheError(err)
		}
		if hasError {
			return fmt.Errorf("target '%s' cache update error: %v: %+v", t.name, err, utils.GNMINotificationPrettyString(update))
		}
	}
	return nil
}
func (t *ConnectionState) handleCacheError(err error) bool {
	switch err.Error() {
	case "suppressed duplicate value":
	case "update is stale":
		t.counterStale.Increment()
		return false
	}
	return true
}
func (t *ConnectionState) rejectUpdate(notification *gnmipb.Notification) bool {
	for _, update := range notification.GetUpdate() {
		path := update.GetPath().GetElem()
		for _, rejectionPath := range t.config.UpdateRejections {
			if matchPath(path, rejectionPath) {
				return true
			}
		}
	}
	return false
}