func memberAddressInMemberList(addr string, list []clustering.MemberID) bool {
	ipParts := strings.Split(addr, ":")
	if len(ipParts) < 1 || ipParts[0] == "" {
		return false
	}
	for _, v := range list {
		memberParts := strings.Split(string(v), ":")
		if len(memberParts) < 1 || memberParts[0] == "" {
			continue
		}
		if ipParts[0] == memberParts[0] {
			return true
		}
	}
	return false
}
memberAddressInMemberList will return true of the IP portion of the supplied address is present in the member list.
package server
import (
	"context"
	"errors"
	"fmt"
	"io"
	"strings"
	"time"
	"github.com/openconfig/gnmi-gateway/gateway/clustering"
	"github.com/openconfig/gnmi-gateway/gateway/configuration"
	"github.com/openconfig/gnmi-gateway/gateway/connections"
	"github.com/openconfig/gnmi-gateway/gateway/stats"
	"github.com/golang/protobuf/proto"
	"github.com/openconfig/gnmi/cache"
	"github.com/openconfig/gnmi/coalesce"
	"github.com/openconfig/gnmi/ctree"
	"github.com/openconfig/gnmi/match"
	"github.com/openconfig/gnmi/path"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/peer"
	"google.golang.org/grpc/status"
	pb "github.com/openconfig/gnmi/proto/gnmi"
)
var (
	flowControlTest = func() {}
	Timeout = time.Minute
	SubscriptionLimit = 0
	subscriptionLimitTest = func() {}
)
type aclStub struct{}
func (a *aclStub) Check(string) bool {
	return true
}
type RPCACL interface {
	Check(string) bool
}
type ACL interface {
	NewRPCACL(context.Context) (RPCACL, error)
	Check(string, string) bool
}
type Server struct {
	pb.UnimplementedGNMIServer  
	c       *cache.Cache  
	m       *match.Match  
	a       ACL           
	config  *configuration.GatewayConfig
	connMgr connections.ConnectionManager
	cluster clustering.ClusterMember
	subscribeSlots chan struct{}
	timeout        time.Duration
}
type GNMIServerOpts struct {
	Config  *configuration.GatewayConfig
	Cache   *cache.Cache
	Cluster clustering.ClusterMember
	ConnMgr connections.ConnectionManager
}
func NewServer(opts *GNMIServerOpts) (*Server, error) {
	s := &Server{
		c:       opts.Cache,
		m:       match.New(),
		config:  opts.Config,
		cluster: opts.Cluster,
		connMgr: opts.ConnMgr,
		timeout: Timeout,
	}
	if SubscriptionLimit > 0 {
		s.subscribeSlots = make(chan struct{}, SubscriptionLimit)
	}
	return s, nil
}
func (s *Server) SetACL(a ACL) {
	s.a = a
}
func (s *Server) Update(n *ctree.Leaf) {
	switch v := n.Value().(type) {
	case *pb.Notification:
		p := path.ToStrings(v.Prefix, true)
		if len(v.Update) > 0 {
			p = append(p, path.ToStrings(v.Update[0].Path, false)...)
		} else if len(v.Delete) > 0 {
			p = append(p, path.ToStrings(v.Delete[0], false)...)
		}
		s.m.Update(n, p)
	default:
		s.config.Log.Error().Msgf("update is not a known type; type is %T", v)
	}
}
func addSubscription(m *match.Match, s *pb.SubscriptionList, c *matchClient) (remove func()) {
	var removes []func()
	prefix := path.ToStrings(s.Prefix, true)
	for _, p := range s.Subscription {
		if p.Path == nil {
			continue
		}
		p := append(prefix, path.ToStrings(p.Path, false)...)
		removes = append(removes, m.AddQuery(p, c))
	}
	return func() {
		for _, remove := range removes {
			remove()
		}
	}
}
func (s *Server) Subscribe(stream pb.GNMI_SubscribeServer) error {
	ctxPeer, _ := peer.FromContext(stream.Context())
	tags := map[string]string{"gnmigateway.server.subscribe.peer": ctxPeer.Addr.String()}
	stats.Registry.Counter("gnmigateway.server.subscribe.request", tags).Increment()
	c := streamClient{stream: stream, acl: &aclStub{}}
	var err error
	if s.a != nil {
		a, err := s.a.NewRPCACL(stream.Context())
		if err != nil {
			s.config.Log.Error().Msgf("NewRPCACL fails due to %v", err)
			tags["gnmigateway.server.subscribe.error_desc"] = "unauthenticated"
			stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
			return status.Error(codes.Unauthenticated, "no authentication/authorization for requested operation")
		}
		c.acl = a
	}
	c.sr, err = stream.Recv()
	switch {
	case err == io.EOF:
		return nil
	case err != nil:
		return err
	case c.sr.GetSubscribe() == nil:
		tags["gnmigateway.server.subscribe.error_desc"] = "bad_request"
		stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
		return status.Errorf(codes.InvalidArgument, "request must contain a subscription %#v", c.sr)
	case c.sr.GetSubscribe().GetPrefix() == nil:
		tags["gnmigateway.server.subscribe.error_desc"] = "bad_request"
		stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
		return status.Errorf(codes.InvalidArgument, "request subscription must contain a prefix %#v", c.sr)
	case c.sr.GetSubscribe().GetPrefix().GetTarget() == "":
		tags["gnmigateway.server.subscribe.error_desc"] = "bad_request"
		stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
		return status.Errorf(codes.InvalidArgument, "request subscription prefix must contain a target %#v", c.sr)
	}
	c.target = c.sr.GetSubscribe().GetPrefix().GetTarget()
	if !s.c.HasTarget(c.target) {
		tags["gnmigateway.server.subscribe.error_desc"] = "target_not_found"
		stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
		return status.Errorf(codes.NotFound, "no such target: %q", c.target)
	}
	mode := c.sr.GetSubscribe().Mode
	var clusterMember = false
	var memberList []clustering.MemberID
	if s.cluster != nil {
		memberList, err = s.cluster.MemberList()
		if err != nil {
			tags["gnmigateway.server.subscribe.error_desc"] = "internal"
			stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
			return fmt.Errorf("unable to retrieve current cluster member list: %v", err)
		}
	}
	if memberAddressInMemberList(ctxPeer.Addr.String(), memberList) {
		clusterMember = true
		s.config.Log.Info().Msgf(`subscribe: cluster member peer: %v
								  target: %q subscription: %s`, ctxPeer.Addr, c.target, c.sr)
		defer s.config.Log.Info().Msgf("subscribe: cluster member peer: %v target %q subscription: end: %q", ctxPeer.Addr, c.target, c.sr)
	} else {
		s.config.Log.Info().Msgf("subscribe: client: %v target: %q subscription: %s", ctxPeer.Addr, c.target, c.sr)
		defer s.config.Log.Info().Msgf("subscribe: client: %v target %q subscription: end: %q", ctxPeer.Addr, c.target, c.sr)
	}
	c.queue = coalesce.NewQueue()
	defer c.queue.Close()
	if c.target != "*" && !c.acl.Check(c.target) {
		tags["gnmigateway.server.subscribe.error_desc"] = "permission_denied"
		stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
		return status.Errorf(codes.PermissionDenied, "not authorized for target %q", c.target)
	}
	errC := make(chan error, 3)
	c.errC = errC
	switch mode {
	case pb.SubscriptionList_ONCE:
		go func() {
			s.processSubscription(&c)
			c.queue.Close()
		}()
	case pb.SubscriptionList_POLL:
		go s.processPollingSubscription(&c)
	case pb.SubscriptionList_STREAM:
		if c.sr.GetSubscribe().GetUpdatesOnly() {
			_, err = c.queue.Insert(syncMarker{})
			if err != nil {
				tags["gnmigateway.server.subscribe.error_desc"] = "internal"
				stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
				return fmt.Errorf("unable to insert sync marker: %v", err)
			}
		}
		remove := addSubscription(s.m, c.sr.GetSubscribe(), &matchClient{acl: c.acl, q: c.queue})
		defer remove()
		if !c.sr.GetSubscribe().GetUpdatesOnly() {
			go s.processSubscription(&c)
		}
	default:
		tags["gnmigateway.server.subscribe.error_desc"] = "bad_request"
		stats.Registry.Counter("gnmigateway.server.subscribe.error", tags).Increment()
		return status.Errorf(codes.InvalidArgument, "Subscription mode %v not recognized", mode)
	}
	go s.sendStreamingResults(&c, s.connMgr, clusterMember)
	return <-errC
}
type resp struct {
	stream pb.GNMI_SubscribeServer
	n      *ctree.Leaf
	dup    uint32
	t      *time.Timer  
}
func (s *Server) sendSubscribeResponse(r *resp, c *streamClient) error {
	notification, err := MakeSubscribeResponse(r.n.Value(), r.dup)
	if err != nil {
		return status.Errorf(codes.Unknown, err.Error())
	}
	if pre := notification.GetUpdate().GetPrefix(); pre != nil {
		if !c.acl.Check(pre.GetTarget()) {
			return nil
		}
	}
	r.t.Reset(s.timeout)
	defer r.t.Stop()
	flowControlTest()
	return r.stream.Send(notification)
}
var subscribeSync = &pb.SubscribeResponse{Response: &pb.SubscribeResponse_SyncResponse{true}}
type syncMarker struct{}
type matchClient struct {
	acl RPCACL
	q   *coalesce.Queue
	err error
}
func (c matchClient) Update(n interface{}) {
	if c.err != nil {
		return
	}
	_, c.err = c.q.Insert(n)
}
type streamClient struct {
	acl    RPCACL
	target string
	sr     *pb.SubscribeRequest
	queue  *coalesce.Queue
	stream pb.GNMI_SubscribeServer
	errC   chan<- error
}
func (s *Server) processSubscription(c *streamClient) {
	var err error
	s.config.Log.Debug().Msgf("start processSubscription for %p", c)
	defer func() {
		if err != nil {
			s.config.Log.Error().Msgf("processSubscription error: %v", err)
			c.queue.Close()
			c.errC <- err
		}
		s.config.Log.Debug().Msgf("end processSubscription for %p", c)
	}()
	if s.subscribeSlots != nil {
		select {
		case s.subscribeSlots <- struct{}{}:
		default:
			s.config.Log.Debug().Msgf("subscription %s delayed waiting for 1 of %d subscription slots.", c.sr, len(s.subscribeSlots))
			s.subscribeSlots <- struct{}{}
			s.config.Log.Debug().Msgf("subscription %s resumed", c.sr)
		}
		defer func() {
			subscriptionLimitTest()
			<-s.subscribeSlots
		}()
	}
	if !c.sr.GetSubscribe().GetUpdatesOnly() {
		for _, subscription := range c.sr.GetSubscribe().Subscription {
			var fullPath []string
			fullPath, err = path.CompletePath(c.sr.GetSubscribe().GetPrefix(), subscription.GetPath())
			if err != nil {
				return
			}
			err = s.c.Query(c.target, fullPath, func(path []string, l *ctree.Leaf, val interface{}) error {
				if err != nil {
					return err
				}
				if val == nil {
					return nil
				}
				_, err = c.queue.Insert(l)
				return nil
			})
			if err != nil {
				return
			}
		}
	}
	_, err = c.queue.Insert(syncMarker{})
}
func (s *Server) processPollingSubscription(c *streamClient) {
	s.processSubscription(c)
	s.config.Log.Info().Msgf("polling subscription: first complete response: %q", c.sr)
	for {
		if c.queue.IsClosed() {
			return
		}
		_, err := c.stream.Recv()
		if err == io.EOF {
			return
		}
		if err != nil {
			s.config.Log.Error().Msgf("processPollingSubscription error: %v", err)
			c.errC <- err
			return
		}
		s.config.Log.Info().Msgf("polling subscription: repoll: %q", c.sr)
		s.processSubscription(c)
		s.config.Log.Info().Msgf("polling subscription: repoll complete: %q", c.sr)
	}
}
func (s *Server) sendStreamingResults(c *streamClient, connMgr connections.ConnectionManager, clusterMember bool) {
	ctx := c.stream.Context()
	ctxPeer, _ := peer.FromContext(ctx)
	t := time.NewTimer(s.timeout)
	t.Stop()
	done := make(chan struct{})
	defer close(done)
	go func() {
		select {
		case <-t.C:
			err := errors.New("subscription timed out while sending")
			c.errC <- err
			s.config.Log.Error().Msgf("%v : %v", ctxPeer, err)
		case <-done:
		}
	}()
	for {
		item, dup, err := c.queue.Next(ctx)
		if coalesce.IsClosedQueue(err) {
			c.errC <- nil
			return
		}
		if err != nil {
			c.errC <- err
			return
		}
		if _, ok := item.(syncMarker); ok {
			if err = c.stream.Send(subscribeSync); err != nil {
				c.errC <- err
				return
			}
			continue
		}
		n, ok := item.(*ctree.Leaf)
		if !ok || n == nil {
			c.errC <- status.Errorf(codes.Internal, "invalid cache node: %#v", item)
			return
		}
		if clusterMember {
			notification, ok := n.Value().(*pb.Notification)
			if !ok || notification == nil {
				c.errC <- status.Errorf(codes.Internal, "invalid notification type: %#v", item)
				return
			}
			target := notification.GetPrefix().GetTarget()
			if !connMgr.Forwardable(target) {
				return
			}
		}
		if err = s.sendSubscribeResponse(&resp{
			stream: c.stream,
			n:      n,
			dup:    dup,
			t:      t,
		}, c); err != nil {
			c.errC <- err
			return
		}
		if isTargetDelete(n) && c.target != "*" {
			s.config.Log.Info().Msgf("Target %q was deleted. Closing stream.", c.target)
			c.errC <- nil
			return
		}
	}
}
func MakeSubscribeResponse(n interface{}, dup uint32) (*pb.SubscribeResponse, error) {
	var notification *pb.Notification
	var ok bool
	notification, ok = n.(*pb.Notification)
	if !ok {
		return nil, status.Errorf(codes.Internal, "invalid notification type: %#v", n)
	}
	if dup > 0 && len(notification.Update) > 0 {
		notification = proto.Clone(notification).(*pb.Notification)
		notification.Update[0].Duplicates = dup
	}
	response := &pb.SubscribeResponse{
		Response: &pb.SubscribeResponse_Update{
			Update: notification,
		},
	}
	return response, nil
}