func (n *Node) Watch(service string) error {
	resp, err := n.ndc.Get(service, map[string]interface{}{})
	if err != nil {
		log.Errorf("Watch service %v error %v", service, err)
		return err
	}
	for _, node := range resp.Nodes {
		n.handleNeighborNodes(discovery.NodeUp, &node)
	}
	return n.ndc.Watch(context.Background(), service, n.handleNeighborNodes)
}
Watch the neighbor nodes
package ion
import (
	"context"
	"fmt"
	"sync"
	ndc "github.com/cloudwebrtc/nats-discovery/pkg/client"
	"github.com/cloudwebrtc/nats-discovery/pkg/discovery"
	nrpc "github.com/cloudwebrtc/nats-grpc/pkg/rpc"
	"github.com/nats-io/nats.go"
	log "github.com/pion/ion-log"
	"github.com/pion/ion/pkg/util"
	"google.golang.org/grpc"
)
type Node struct {
	NID string
	nc *nats.Conn
	nrpc *nrpc.Server
	ndc *ndc.Client
	nodeLock sync.RWMutex
	neighborNodes map[string]discovery.Node
	cliLock sync.RWMutex
	clis    map[string]*nrpc.Client
}
func NewNode(nid string) Node {
	return Node{
		NID:           nid,
		neighborNodes: make(map[string]discovery.Node),
		clis:          make(map[string]*nrpc.Client),
	}
}
func (n *Node) Start(natURL string) error {
	var err error
	n.nc, err = util.NewNatsConn(natURL)
	if err != nil {
		log.Errorf("new nats conn error %v", err)
		n.Close()
		return err
	}
	n.ndc, err = ndc.NewClient(n.nc)
	if err != nil {
		log.Errorf("new discovery client error %v", err)
		n.Close()
		return err
	}
	n.nrpc = nrpc.NewServer(n.nc, n.NID)
	return nil
}
func (n *Node) NatsConn() *nats.Conn {
	return n.nc
}
func (n *Node) KeepAlive(node discovery.Node) error {
	return n.ndc.KeepAlive(node)
}
func (n *Node) NewNatsRPCClient(service, peerNID string, parameters map[string]interface{}) (*nrpc.Client, error) {
	var cli *nrpc.Client = nil
	selfNID := n.NID
	for id, node := range n.neighborNodes {
		if node.Service == service && (id == peerNID || peerNID == "*") {
			cli = nrpc.NewClient(n.nc, id, selfNID)
		}
	}
	if cli == nil {
		resp, err := n.ndc.Get(service, parameters)
		if err != nil {
			log.Errorf("failed to Get service [%v]: %v", service, err)
			return nil, err
		}
		if len(resp.Nodes) == 0 {
			err := fmt.Errorf("get service [%v], node cnt == 0", service)
			return nil, err
		}
		cli = nrpc.NewClient(n.nc, resp.Nodes[0].NID, selfNID)
	}
	n.cliLock.Lock()
	defer n.cliLock.Unlock()
	n.clis[util.RandomString(12)] = cli
	return cli, nil
}