func (n *Node) SetIdentity(identity string) error {
	return n.memberlist.SetLocalLabel(membership.IdentityLabelKey, identity)
}
SetIdentity changes the identity of the local node. This will change the
state of the local node and will be gossiped around in the network.
package swim
import (
	"errors"
	"sync"
	"time"
	"github.com/uber/ringpop-go/membership"
	"github.com/uber/ringpop-go/discovery"
	"github.com/uber/ringpop-go/events"
	"github.com/benbjohnson/clock"
	"github.com/rcrowley/go-metrics"
	log "github.com/uber-common/bark"
	"github.com/uber/ringpop-go/logging"
	"github.com/uber/ringpop-go/shared"
	"github.com/uber/ringpop-go/util"
)
var (
	ErrNodeNotReady = errors.New("node is not ready to handle requests")
)
type Options struct {
	StateTimeouts     StateTimeouts
	MinProtocolPeriod time.Duration
	JoinTimeout, PingTimeout, PingRequestTimeout time.Duration
	PingRequestSize int
	RollupFlushInterval time.Duration
	RollupMaxUpdates    int
	MaxReverseFullSyncJobs int
	PartitionHealPeriod           time.Duration
	PartitionHealBaseProbabillity float64
	LabelLimits   LabelOptions
	InitialLabels LabelMap
	Clock clock.Clock
	SelfEvict SelfEvictOptions
	RequiresAppInPing bool
}
func defaultOptions() *Options {
	opts := &Options{
		StateTimeouts: StateTimeouts{
			Suspect:   5 * time.Second,
			Faulty:    24 * time.Hour,
			Tombstone: 1 * time.Minute,
		},
		MinProtocolPeriod: 200 * time.Millisecond,
		JoinTimeout:        1000 * time.Millisecond,
		PingTimeout:        1500 * time.Millisecond,
		PingRequestTimeout: 5000 * time.Millisecond,
		PingRequestSize: 3,
		RollupFlushInterval: 5000 * time.Millisecond,
		RollupMaxUpdates:    250,
		PartitionHealPeriod:           30 * time.Second,
		PartitionHealBaseProbabillity: 3,
		LabelLimits: DefaultLabelOptions,
		Clock: clock.New(),
		MaxReverseFullSyncJobs: 5,
		SelfEvict: SelfEvictOptions{
			PingRatio: .4,
		},
	}
	return opts
}
func mergeDefaultOptions(opts *Options) *Options {
	def := defaultOptions()
	if opts == nil {
		return def
	}
	opts.StateTimeouts = mergeStateTimeouts(opts.StateTimeouts, def.StateTimeouts)
	opts.LabelLimits = mergeLabelOptions(opts.LabelLimits, def.LabelLimits)
	opts.MinProtocolPeriod = util.SelectDuration(opts.MinProtocolPeriod, def.MinProtocolPeriod)
	opts.RollupMaxUpdates = util.SelectInt(opts.RollupMaxUpdates, def.RollupMaxUpdates)
	opts.RollupFlushInterval = util.SelectDuration(opts.RollupFlushInterval, def.RollupFlushInterval)
	opts.PartitionHealPeriod = util.SelectDuration(opts.PartitionHealPeriod, def.PartitionHealPeriod)
	opts.PartitionHealBaseProbabillity = util.SelectFloat(opts.PartitionHealBaseProbabillity, def.PartitionHealBaseProbabillity)
	opts.JoinTimeout = util.SelectDuration(opts.JoinTimeout, def.JoinTimeout)
	opts.PingTimeout = util.SelectDuration(opts.PingTimeout, def.PingTimeout)
	opts.PingRequestTimeout = util.SelectDuration(opts.PingRequestTimeout, def.PingRequestTimeout)
	opts.PingRequestSize = util.SelectInt(opts.PingRequestSize, def.PingRequestSize)
	opts.MaxReverseFullSyncJobs = util.SelectInt(opts.MaxReverseFullSyncJobs, def.MaxReverseFullSyncJobs)
	opts.SelfEvict = mergeSelfEvictOptions(opts.SelfEvict, def.SelfEvict)
	if opts.Clock == nil {
		opts.Clock = def.Clock
	}
	return opts
}
type NodeInterface interface {
	Bootstrap(opts *BootstrapOptions) ([]string, error)
	CountReachableMembers(predicates ...MemberPredicate) int
	Destroy()
	GetChecksum() uint32
	GetReachableMembers(predicates ...MemberPredicate) []Member
	Labels() *NodeLabels
	MemberStats() MemberStats
	ProtocolStats() ProtocolStats
	Ready() bool
	AddListener(events.EventListener) bool
	RemoveListener(events.EventListener) bool
	RegisterSelfEvictHook(hooks SelfEvictHook) error
	SelfEvict() error
	SetIdentity(identity string) error
}
type Node struct {
	events.SyncEventEmitter
	app     string
	service string
	address string
	state struct {
		stopped, destroyed, pinging, ready bool
		sync.RWMutex
	}
	channel          shared.SubChannel
	discoverProvider discovery.DiscoverProvider
	memberlist       *memberlist
	memberiter       memberIter
	disseminator     *disseminator
	stateTransitions *stateTransitions
	gossip           *gossip
	healer *discoverProviderHealer
	joinTimeout, pingTimeout, pingRequestTimeout time.Duration
	pingRequestSize int
	maxReverseFullSyncJobs int
	clientRate metrics.Meter
	serverRate metrics.Meter
	totalRate  metrics.Meter
	startTime time.Time
	logger log.Logger
	labelLimits LabelOptions
	clock clock.Clock
	selfEvict         *selfEvict
	selfEvictOptions  SelfEvictOptions
	requiresAppInPing bool
}
func NewNode(app, address string, channel shared.SubChannel, opts *Options) *Node {
	opts = mergeDefaultOptions(opts)
	node := &Node{
		address: address,
		app:     app,
		channel: channel,
		logger:  logging.Logger("node").WithField("local", address),
		joinTimeout:        opts.JoinTimeout,
		pingTimeout:        opts.PingTimeout,
		pingRequestTimeout: opts.PingRequestTimeout,
		pingRequestSize: opts.PingRequestSize,
		maxReverseFullSyncJobs: opts.MaxReverseFullSyncJobs,
		clientRate: metrics.NewMeter(),
		serverRate: metrics.NewMeter(),
		totalRate:  metrics.NewMeter(),
		clock:      opts.Clock,
	}
	node.selfEvict = newSelfEvict(node, opts.SelfEvict)
	node.requiresAppInPing = opts.RequiresAppInPing
	node.labelLimits = opts.LabelLimits
	node.memberlist = newMemberlist(node, opts.InitialLabels)
	node.memberiter = newMemberlistIter(node.memberlist)
	node.stateTransitions = newStateTransitions(node, opts.StateTimeouts)
	node.healer = newDiscoverProviderHealer(
		node,
		opts.PartitionHealBaseProbabillity,
		opts.PartitionHealPeriod,
	)
	node.gossip = newGossip(node, opts.MinProtocolPeriod)
	node.disseminator = newDisseminator(node)
	for _, member := range node.memberlist.GetMembers() {
		change := Change{}
		change.populateSubject(&member)
		node.disseminator.RecordChange(change)
	}
	if node.channel != nil {
		node.registerHandlers()
		node.service = node.channel.ServiceName()
	}
	return node
}
func (n *Node) Address() string {
	return n.address
}
func (n *Node) App() string {
	return n.app
}
func (n *Node) HasChanges() bool {
	return n.disseminator.HasChanges()
}
func (n *Node) Incarnation() int64 {
	if n.memberlist != nil && n.memberlist.local != nil {
		n.memberlist.members.RLock()
		incarnation := n.memberlist.local.Incarnation
		n.memberlist.members.RUnlock()
		return incarnation
	}
	return -1
}
func (n *Node) Start() {
	n.gossip.Start()
	n.stateTransitions.Enable()
	n.healer.Start()
	n.state.Lock()
	n.state.stopped = false
	n.state.Unlock()
}
func (n *Node) Stop() {
	n.gossip.Stop()
	n.stateTransitions.Disable()
	n.healer.Stop()
	n.state.Lock()
	n.state.stopped = true
	n.state.Unlock()
}
func (n *Node) Stopped() bool {
	n.state.RLock()
	stopped := n.state.stopped
	n.state.RUnlock()
	return stopped
}
func (n *Node) Destroy() {
	n.state.Lock()
	if n.state.destroyed {
		n.state.Unlock()
		return
	}
	n.state.destroyed = true
	n.state.Unlock()
	n.Stop()
}
func (n *Node) Destroyed() bool {
	n.state.RLock()
	destroyed := n.state.destroyed
	n.state.RUnlock()
	return destroyed
}
func (n *Node) Ready() bool {
	n.state.RLock()
	ready := n.state.ready
	n.state.RUnlock()
	return ready
}
func (n *Node) RegisterSelfEvictHook(hooks SelfEvictHook) error {
	return n.selfEvict.RegisterSelfEvictHook(hooks)
}
func (n *Node) SelfEvict() error {
	return n.selfEvict.SelfEvict()
}
type BootstrapOptions struct {
	DiscoverProvider discovery.DiscoverProvider
	Stopped bool
	JoinTimeout time.Duration
	JoinSize int
	MaxJoinDuration time.Duration
	ParallelismFactor int
}
func (n *Node) Bootstrap(opts *BootstrapOptions) ([]string, error) {
	if n.channel == nil {
		return nil, errors.New("channel required")
	}
	if opts == nil {
		opts = &BootstrapOptions{}
	}
	n.discoverProvider = opts.DiscoverProvider
	joinOpts := &joinOpts{
		timeout:           opts.JoinTimeout,
		size:              opts.JoinSize,
		maxJoinDuration:   opts.MaxJoinDuration,
		parallelismFactor: opts.ParallelismFactor,
	}
	joined, err := sendJoin(n, joinOpts)
	if err != nil {
		n.logger.WithFields(log.Fields{
			"err": err.Error(),
		}).Error("bootstrap failed")
		return nil, err
	}
	if !opts.Stopped {
		n.gossip.Start()
		n.healer.Start()
	}
	n.state.Lock()
	n.state.ready = true
	n.state.Unlock()
	n.startTime = time.Now()
	return joined, nil
}
func (n *Node) handleChanges(changes []Change) {
	n.disseminator.AdjustMaxPropagations()
	for _, change := range changes {
		n.disseminator.RecordChange(change)
		switch change.Status {
		case Alive:
			n.stateTransitions.Cancel(change)
		case Suspect:
			n.stateTransitions.ScheduleSuspectToFaulty(change)
		case Faulty:
			n.stateTransitions.ScheduleFaultyToTombstone(change)
		case Leave:
			n.stateTransitions.Cancel(change)
		case Tombstone:
			n.stateTransitions.ScheduleTombstoneToEvict(change)
		}
	}
}
func (n *Node) pinging() bool {
	n.state.RLock()
	pinging := n.state.pinging
	n.state.RUnlock()
	return pinging
}
func (n *Node) setPinging(pinging bool) {
	n.state.Lock()
	n.state.pinging = pinging
	n.state.Unlock()
}
func (n *Node) pingNextMember() {
	member, ok := n.memberiter.Next()
	if !ok {
		n.logger.Warn("no pingable members")
		return
	}
	if n.pinging() {
		n.logger.Warn("node already pinging")
		return
	}
	n.setPinging(true)
	defer n.setPinging(false)
	res, err := sendPing(n, member.Address, n.pingTimeout)
	if err == nil {
		n.memberlist.Update(res.Changes)
		return
	}
	target := member.Address
	targetReached, errs := indirectPing(n, target, n.pingRequestSize, n.pingRequestTimeout)
	if len(errs) == n.pingRequestSize {
		n.logger.WithFields(log.Fields{
			"target":    target,
			"errors":    errs,
			"numErrors": len(errs),
		}).Warn("ping request inconclusive due to errors")
		return
	}
	if !targetReached {
		n.logger.WithField("target", target).Info("ping request target unreachable")
		n.memberlist.MakeSuspect(member.Address, member.Incarnation)
		return
	}
	n.logger.WithField("target", target).Info("ping request target reachable")
}
func (n *Node) GetReachableMembers(predicates ...MemberPredicate) []Member {
	predicates = append(predicates, memberIsReachable)
	return n.memberlist.GetMembers(predicates...)
}
func (n *Node) CountReachableMembers(predicates ...MemberPredicate) int {
	predicates = append(predicates, memberIsReachable)
	return n.memberlist.CountMembers(predicates...)
}
func (n *Node) Labels() *NodeLabels {
	return &NodeLabels{n}
}