func (m *ClientConnectionManager) GetClientConnectionByRemote(nsm *registry.NetworkServiceManager) []*model.ClientConnection {
	clientConnections := m.getClientConnections()
	var result []*model.ClientConnection
	for _, clientConnection := range clientConnections {
		if clientConnection.RemoteNsm != nil && clientConnection.RemoteNsm.GetName() == nsm.GetName() {
			result = append(result, clientConnection)
		}
	}
	return result
}
GetClientConnectionByRemote - return client conndction by remote name
package services
import (
	"context"
	"sync"
	"time"
	mechanismCommon "github.com/networkservicemesh/networkservicemesh/controlplane/api/connection/mechanisms/common"
	"github.com/networkservicemesh/networkservicemesh/controlplane/api/connection/mechanisms/kernel"
	"github.com/pkg/errors"
	"github.com/networkservicemesh/networkservicemesh/controlplane/pkg/common"
	"github.com/networkservicemesh/networkservicemesh/pkg/tools/spanhelper"
	"github.com/sirupsen/logrus"
	"github.com/networkservicemesh/networkservicemesh/controlplane/api/connection"
	"github.com/networkservicemesh/networkservicemesh/controlplane/api/crossconnect"
	"github.com/networkservicemesh/networkservicemesh/controlplane/api/registry"
	"github.com/networkservicemesh/networkservicemesh/controlplane/pkg/api/nsm"
	"github.com/networkservicemesh/networkservicemesh/controlplane/pkg/model"
	"github.com/networkservicemesh/networkservicemesh/controlplane/pkg/serviceregistry"
)
const (
	deletedConnectionLifetime    = time.Minute * 3
	connectionStateCheckInterval = 5 * time.Second
)
type managedClientConnection struct {
	sync.RWMutex
	deleteTime       time.Time
	deleted          bool
	clientConnection *model.ClientConnection
}
type ClientConnectionManager struct {
	model           model.Model
	serviceRegistry serviceregistry.ServiceRegistry
	manager         nsm.NetworkServiceManager
	clientConnections sync.Map
}
func NewClientConnectionManager(model model.Model, manager nsm.NetworkServiceManager, serviceRegistry serviceregistry.ServiceRegistry) *ClientConnectionManager {
	return &ClientConnectionManager{
		model:           model,
		serviceRegistry: serviceRegistry,
		manager:         manager,
	}
}
func (m *ClientConnectionManager) GetNsmName() string {
	return m.model.GetNsm().Name
}
func (m *ClientConnectionManager) UpdateXcon(ctx context.Context, cc nsm.ClientConnection, newXcon *crossconnect.CrossConnect) {
	span := spanhelper.FromContext(ctx, "UpdateXcon")
	defer span.Finish()
	ctx = span.Context()
	span.LogObject("connection", cc)
	span.LogObject("newXCon", newXcon)
	logger := span.Logger()
	if upd := m.model.ApplyClientConnectionChanges(ctx, cc.GetID(), func(cc *model.ClientConnection) {
		cc.Xcon = newXcon
	}); upd != nil {
		cc = upd
	} else {
		err := errors.Errorf("trying to update not existing connection: %v", cc.GetID())
		span.LogError(err)
		return
	}
	if src := newXcon.GetLocalSource(); src != nil && src.State == connection.State_DOWN {
		logger.Info("ClientConnection src state is down. Closing.")
		go func() {
			for {
				actualConn := m.model.GetClientConnection(cc.GetID())
				if actualConn == nil || !m.isConnectionPending(actualConn) {
					err := m.manager.CloseConnection(ctx, cc)
					span.LogError(err)
					return
				}
				logger.Warnf("Trying to close connection %v in bad state (%v)...", cc.GetID(), m.model.GetClientConnection(cc.GetID()).ConnectionState)
				<-time.After(connectionStateCheckInterval)
			}
		}()
		return
	}
	if dst := newXcon.GetLocalDestination(); dst != nil && dst.State == connection.State_DOWN {
		logger.Info("ClientConnection dst state is down. calling Heal.")
		m.manager.Heal(ctx, cc, nsm.HealStateDstDown)
		return
	}
}
func (m *ClientConnectionManager) DestinationDown(ctx context.Context, cc nsm.ClientConnection, nsmdDie bool) {
	span := spanhelper.FromContext(ctx, "DestinationDown")
	defer span.Finish()
	ctx = span.Context()
	span.LogValue("nsmgr.die", nsmdDie)
	if nsmdDie {
		m.manager.Heal(ctx, cc, nsm.HealStateDstNmgrDown)
	} else {
		m.manager.Heal(ctx, cc, nsm.HealStateDstDown)
	}
}
func (m *ClientConnectionManager) ForwarderDown(ctx context.Context, forwarder *model.Forwarder) {
	ccs := m.model.GetAllClientConnections()
	for _, cc := range ccs {
		if cc.ForwarderRegisteredName == forwarder.RegisteredName {
			span := common.SpanHelperFromConnection(ctx, cc, "ForwarderDown")
			span.LogObject("connection", cc)
			span.LogObject("forwarder", forwarder)
			defer span.Finish()
			m.manager.Heal(span.Context(), cc, nsm.HealStateForwarderDown)
		}
	}
}
func (m *ClientConnectionManager) LocalDestinationUpdated(ctx context.Context, cc *model.ClientConnection, localDst *connection.Connection) {
	span := spanhelper.FromContext(ctx, "LocalDestinationUpdated")
	defer span.Finish()
	ctx = span.Context()
	span.LogObject("connection", cc)
	span.LogObject("destination", localDst)
	if cc.ConnectionState != model.ClientConnectionReady {
		return
	}
	localDst.GetMechanism().GetParameters()[mechanismCommon.Workspace] =
		cc.Xcon.GetLocalDestination().GetMechanism().GetParameters()[mechanismCommon.Workspace]
	localDst.GetMechanism().GetParameters()[kernel.WorkspaceNSEName] =
		cc.Xcon.GetLocalDestination().GetMechanism().GetParameters()[kernel.WorkspaceNSEName]
	m.destinationUpdated(ctx, cc, localDst)
}
func (m *ClientConnectionManager) RemoteDestinationUpdated(ctx context.Context, cc *model.ClientConnection, remoteDst *connection.Connection) {
	span := spanhelper.FromContext(ctx, "RemoteDestinationUpdated")
	defer span.Finish()
	ctx = span.Context()
	span.LogObject("connection", cc)
	span.LogObject("remoteDst", remoteDst)
	logger := span.Logger()
	if cc.ConnectionState != model.ClientConnectionReady {
		logger.Infof("Event not send... %v", cc.GetID())
		return
	}
	if remoteDst.State == connection.State_UP {
		logger.Infof("State is already UP do not send")
		return
	}
	logger.Infof("Event send... %v", cc.GetID())
	m.destinationUpdated(ctx, cc, remoteDst)
}
func (m *ClientConnectionManager) destinationUpdated(ctx context.Context, cc nsm.ClientConnection, dst *connection.Connection) {
	span := spanhelper.FromContext(ctx, "destinationUpdated")
	defer span.Finish()
	ctx = span.Context()
	span.LogObject("connection", cc)
	span.LogObject("destination", dst)
	logger := span.Logger()
	if dst.Equals(cc.GetConnectionDestination()) {
		logger.Infof("No event same destination objects")
		return
	}
	if upd := m.model.ApplyClientConnectionChanges(ctx, cc.GetID(), func(cc *model.ClientConnection) {
		cc.Xcon.Destination = dst.Clone()
	}); upd != nil {
		cc = upd
	} else {
		err := errors.Errorf("trying to update not existing connection: %v", cc.GetID())
		span.LogError(err)
		return
	}
	m.manager.Heal(ctx, cc, nsm.HealStateDstUpdate)
}
func (m *ClientConnectionManager) GetClientConnectionByXcon(xcon *crossconnect.CrossConnect) *model.ClientConnection {
	id := xcon.GetId()
	result := m.model.GetClientConnection(id)
	if result != nil {
		return result
	}
	if v, ok := m.clientConnections.Load(id); ok {
		if deleted, ok := v.(*managedClientConnection); ok {
			return deleted.clientConnection
		}
	}
	return nil
}
func (m *ClientConnectionManager) GetClientConnectionByLocalDst(dstID string) *model.ClientConnection {
	clientConnections := m.getClientConnections()
	for _, clientConnection := range clientConnections {
		xCon := clientConnection.Xcon
		if xCon != nil {
			logrus.Infof("checking existing connection: %v to match %v", xCon, dstID)
			if dst := xCon.GetLocalDestination(); dst != nil && dst.GetId() == dstID {
				return clientConnection
			}
		}
	}
	return nil
}
func (m *ClientConnectionManager) GetClientConnectionByRemoteDst(dstID, remoteName string) *model.ClientConnection {
	clientConnections := m.getClientConnections()
	for _, clientConnection := range clientConnections {
		xCon := clientConnection.Xcon
		if xCon != nil {
			logrus.Infof("checking existing connection: %v to match %v %v", xCon, dstID, remoteName)
			if dst := xCon.GetRemoteDestination(); dst != nil && dst.GetId() == dstID && dst.GetDestinationNetworkServiceManagerName() == remoteName {
				logrus.Infof("found remote connection %v", clientConnection)
				return clientConnection
			}
		}
	}
	logrus.Infof("NO DST found to match %v to match %v", dstID, remoteName)
	return nil
}
func (m *ClientConnectionManager) getClientConnections() []*model.ClientConnection {
	clientConnections := m.model.GetAllClientConnections()
	m.clientConnections.Range(func(k, v interface{}) bool {
		if cc, ok := v.(*managedClientConnection); ok {
			cc.Lock()
			clientConnections = append(clientConnections, cc.clientConnection)
			cc.Unlock()
		}
		return true
	})
	return clientConnections
}
type channelConnectionListener struct {
	model.ListenerImpl
	channel chan *model.ClientConnection
}
func (c *channelConnectionListener) ClientConnectionUpdated(ctx context.Context, old, new *model.ClientConnection) {
	c.channel <- new
}
func (c *channelConnectionListener) ClientConnectionDeleted(ctx context.Context, clientConnection *model.ClientConnection) {
	c.channel <- clientConnection
}
func (m *ClientConnectionManager) WaitPendingConnections(ctx context.Context, id, remoteName string) (*model.ClientConnection, error) {
	clientConnections := m.model.GetAllClientConnections()
	var pendingConnections []*model.ClientConnection
	for _, cc := range clientConnections {
		if m.isConnectionPending(cc) {
			pendingConnections = append(pendingConnections, cc)
		}
	}
	if len(pendingConnections) > 0 {
		listener := &channelConnectionListener{
			channel: make(chan *model.ClientConnection, 100),
		}
		m.model.AddListener(listener)
		defer m.model.RemoveListener(listener)
		for len(pendingConnections) > 0 {
			select {
			case <-ctx.Done():
				return nil, errors.Errorf("timeout during wait for connection with connectionId=%v and remoteName=%s", id, remoteName)
			case c := <-listener.channel:
				if !m.isConnectionPending(c) || m.model.GetClientConnection(c.GetID()) == nil {
					for idx, cc := range pendingConnections {
						if cc.ConnectionID == c.ConnectionID {
							pendingConnections = append(pendingConnections[:idx], pendingConnections[idx+1:]...)
							break
						}
					}
					if cc := m.GetClientConnectionByRemoteDst(id, remoteName); cc != nil {
						return cc, nil
					}
				}
			}
		}
		return nil, nil
	}
	return nil, errors.Errorf("no connection with id=%v and remoteName=%v are found", id, remoteName)
}
func (m *ClientConnectionManager) isConnectionPending(cc *model.ClientConnection) bool {
	return cc.ConnectionState == model.ClientConnectionRequesting || cc.ConnectionState == model.ClientConnectionHealing || cc.ConnectionState == model.ClientConnectionHealingBegin
}