func (c *Cluster) Start() {
	l, err := listener.New(c.listenOn)
	if err != nil {
		panic(err)
	}
	l.SetReadTimeout(120 * time.Second)
	for _, n := range c.nodes {
		go n.reconnect()
	}
	if c.fo != nil {
		go c.run()
	}
	err = rpc.Register(c)
	if err != nil {
		log.Fatal("cluster.Start", "error registering rpc server", err)
	}
	go rpc.Accept(l)
	log.ConnLogger.Info().Str("context", "cluster.Start").Msgf("Cluster of %d nodes initialized, node '%s' listening on [%s]", len(Globals.Cluster.nodes)+1,
		Globals.Cluster.thisNodeName, c.listenOn)
}
Start accepting connections.
package broker
import (
	"encoding/gob"
	"encoding/json"
	"errors"
	"net"
	"net/rpc"
	"sort"
	"sync"
	"time"
	lp "github.com/unit-io/unitd/lineprotocol"
	"github.com/unit-io/unitd/message"
	"github.com/unit-io/unitd/message/security"
	"github.com/unit-io/unitd/net/listener"
	rh "github.com/unit-io/unitd/pkg/hash"
	"github.com/unit-io/unitd/pkg/log"
	"github.com/unit-io/unitd/pkg/uid"
)
const (
	defaultClusterReconnect = 200 * time.Millisecond
	clusterHashReplicas = 20
)
type clusterNodeConfig struct {
	Name string `json:"name"`
	Addr string `json:"addr"`
}
type clusterConfig struct {
	Nodes []clusterNodeConfig `json:"nodes"`
	ThisName string `json:"self"`
	Failover *clusterFailoverConfig
}
type ClusterNode struct {
	lock sync.Mutex
	endpoint *rpc.Client
	connected bool
	reconnecting bool
	address string
	name string
	failCount int
	done chan bool
}
type ClusterSess struct {
	RemoteAddr string
	ConnID uid.LID
	ClientID uid.ID
}
type ClusterReq struct {
	Node string
	Signature string
	MsgSub   *lp.Subscribe
	MsgPub   *lp.Publish
	MsgUnsub *lp.Unsubscribe
	Topic    *security.Topic
	Type     uint8
	Message  *message.Message
	Conn *ClusterSess
	ConnGone bool
}
type ClusterResp struct {
	Type     uint8
	MsgSub   *lp.Subscribe
	MsgPub   *lp.Publish
	MsgUnsub *lp.Unsubscribe
	Msg      []byte
	Topic    *security.Topic
	Message  *message.Message
	FromConnID uid.LID
}
func (n *ClusterNode) reconnect() {
	var reconnTicker *time.Ticker
	n.lock.Lock()
	if n.reconnecting {
		n.lock.Unlock()
		return
	}
	n.reconnecting = true
	n.lock.Unlock()
	var count = 0
	var err error
	for {
		if n.endpoint, err = rpc.Dial("tcp", n.address); err == nil {
			if reconnTicker != nil {
				reconnTicker.Stop()
			}
			n.lock.Lock()
			n.connected = true
			n.reconnecting = false
			n.lock.Unlock()
			log.Info("cluster.reconnect", "connection established "+n.name)
			return
		} else if count == 0 {
			reconnTicker = time.NewTicker(defaultClusterReconnect)
		}
		count++
		select {
		case <-reconnTicker.C:
		case <-n.done:
			log.Info("cluster.reconnect", "node shutdown started "+n.name)
			reconnTicker.Stop()
			if n.endpoint != nil {
				n.endpoint.Close()
			}
			n.lock.Lock()
			n.connected = false
			n.reconnecting = false
			n.lock.Unlock()
			log.Info("cluster", "node shut down completed "+n.name)
			return
		}
	}
}
func (n *ClusterNode) call(proc string, msg, resp interface{}) error {
	if !n.connected {
		return errors.New("cluster.call: node '" + n.name + "' not connected")
	}
	if err := n.endpoint.Call(proc, msg, resp); err != nil {
		log.Fatal("cluster.call", "call failed to "+n.name, err)
		n.lock.Lock()
		if n.connected {
			n.endpoint.Close()
			n.connected = false
			go n.reconnect()
		}
		n.lock.Unlock()
		return err
	}
	return nil
}
func (n *ClusterNode) callAsync(proc string, msg, resp interface{}, done chan *rpc.Call) *rpc.Call {
	if done != nil && cap(done) == 0 {
		log.Fatal("cluster.callAsync", "RPC done channel is unbuffered", nil)
	}
	if !n.connected {
		call := &rpc.Call{
			ServiceMethod: proc,
			Args:          msg,
			Reply:         resp,
			Error:         errors.New("cluster.callAsync: node '" + n.name + "' not connected"),
			Done:          done,
		}
		if done != nil {
			done <- call
		}
		return call
	}
	myDone := make(chan *rpc.Call, 1)
	go func() {
		select {
		case call := <-myDone:
			if call.Error != nil {
				n.lock.Lock()
				if n.connected {
					n.endpoint.Close()
					n.connected = false
					go n.reconnect()
				}
				n.lock.Unlock()
			}
			if done != nil {
				done <- call
			}
		}
	}()
	call := n.endpoint.Go(proc, msg, resp, myDone)
	call.Done = done
	return call
}
func (n *ClusterNode) forward(msg *ClusterReq) error {
	log.Info("cluster.forward", "forwarding request to node "+n.name)
	msg.Node = Globals.Cluster.thisNodeName
	rejected := false
	err := n.call("Cluster.Master", msg, &rejected)
	if err == nil && rejected {
		err = errors.New("cluster.forward: master node out of sync")
	}
	return err
}
type Cluster struct {
	nodes map[string]*ClusterNode
	thisNodeName string
	listenOn string
	inbound *net.TCPListener
	ring *rh.Ring
	fo *clusterFailover
}
func (c *Cluster) Master(msg *ClusterReq, rejected *bool) error {
	log.Info("cluster.Master", "master request received from node "+msg.Node)
	conn := Globals.ConnCache.Get(msg.Conn.ConnID)
	if msg.ConnGone {
		if conn != nil {
			conn.stop <- nil
		}
	} else if msg.Signature == c.ring.Signature() {
		if conn == nil {
			node := Globals.Cluster.nodes[msg.Node]
			if node == nil {
				log.Error("cluster.Master", "request from an unknown node "+msg.Node)
				return nil
			}
			log.Info("cluster.Master", "new connection request"+string(msg.Conn.ConnID))
			conn = Globals.Service.newRpcConn(node, msg.Conn.ConnID, msg.Conn.ClientID)
			go conn.rpcWriteLoop()
		}
		conn.connid = msg.Conn.ConnID
		switch msg.Type {
		case message.SUBSCRIBE:
			conn.handler(msg.MsgSub)
		case message.UNSUBSCRIBE:
			conn.handler(msg.MsgUnsub)
		case message.PUBLISH:
			conn.handler(msg.MsgPub)
		}
	} else {
		*rejected = true
	}
	return nil
}
func (Cluster) Proxy(resp *ClusterResp, unused *bool) error {
	log.Info("cluster.Proxy", "response from Master for connection "+string(resp.FromConnID))
	if conn := Globals.ConnCache.Get(resp.FromConnID); conn != nil {
		if !conn.SendRawBytes(resp.Msg) {
			log.Error("cluster.Proxy", "Proxy: timeout")
		}
	} else {
		log.ErrLogger.Error().Str("context", "cluster.Proxy").Uint64("connid", uint64(resp.FromConnID)).Msg("master response for unknown session")
	}
	return nil
}
func (c *Cluster) nodeForContract(contract string) *ClusterNode {
	key := c.ring.Get(contract)
	if key == c.thisNodeName {
		log.Error("cluster", "request to route to self")
		return nil
	}
	node := Globals.Cluster.nodes[key]
	if node == nil {
		log.Error("cluster", "no node for contract "+contract+key)
	}
	return node
}
func (c *Cluster) isRemoteContract(contract string) bool {
	if c == nil {
		return false
	}
	return c.ring.Get(contract) != c.thisNodeName
}
func (c *Cluster) routeToContract(msg lp.Packet, topic *security.Topic, msgType uint8, m *message.Message, conn *Conn) error {
	n := c.nodeForContract(string(conn.clientid.Contract()))
	if n == nil {
		return errors.New("cluster.routeToContract: attempt to route to non-existent node")
	}
	if conn.nodes == nil {
		conn.nodes = make(map[string]bool)
	}
	conn.nodes[n.name] = true
	var msgSub *lp.Subscribe
	var msgPub *lp.Publish
	var msgUnsub *lp.Unsubscribe
	switch msgType {
	case message.SUBSCRIBE:
		msgSub = msg.(*lp.Subscribe)
		msgSub.IsForwarded = true
	case message.UNSUBSCRIBE:
		msgUnsub = msg.(*lp.Unsubscribe)
		msgUnsub.IsForwarded = true
	case message.PUBLISH:
		msgPub = msg.(*lp.Publish)
		msgPub.IsForwarded = true
	}
	return n.forward(
		&ClusterReq{
			Node:      c.thisNodeName,
			Signature: c.ring.Signature(),
			MsgSub:    msgSub,
			MsgUnsub:  msgUnsub,
			MsgPub:    msgPub,
			Topic:     topic,
			Type:      msgType,
			Message:   m,
			Conn: &ClusterSess{
				ConnID:   conn.connid,
				ClientID: conn.clientid}})
}
func (c *Cluster) connGone(conn *Conn) error {
	if c == nil {
		return nil
	}
	for name := range conn.nodes {
		n := c.nodes[name]
		if n != nil {
			return n.forward(
				&ClusterReq{
					Node:     c.thisNodeName,
					ConnGone: true,
					Conn: &ClusterSess{
						ConnID: conn.connid}})
		}
	}
	return nil
}
func ClusterInit(configString json.RawMessage, self *string) int {
	if Globals.Cluster != nil {
		log.Fatal("cluster.ClusterInit", "Cluster already initialized.", nil)
	}
	if len(configString) == 0 {
		log.Info("cluster.ClusterInit", "Running as a standalone server.")
		return 1
	}
	var config clusterConfig
	if err := json.Unmarshal(configString, &config); err != nil {
		log.Fatal("cluster.ClusterInit", "error parsing cluster config", err)
	}
	thisName := *self
	if thisName == "" {
		thisName = config.ThisName
	}
	if thisName == "" {
		log.Info("cluster.ClusterInit", "Running as a standalone server.")
		return 1
	}
	gob.Register([]interface{}{})
	gob.Register(map[string]interface{}{})
	gob.Register(lp.Publish{})
	gob.Register(lp.Subscribe{})
	gob.Register(lp.Unsubscribe{})
	Globals.Cluster = &Cluster{
		thisNodeName: thisName,
		nodes:        make(map[string]*ClusterNode)}
	var nodeNames []string
	for _, host := range config.Nodes {
		nodeNames = append(nodeNames, host.Name)
		if host.Name == thisName {
			Globals.Cluster.listenOn = host.Addr
			continue
		}
		n := ClusterNode{
			address: host.Addr,
			name:    host.Name,
			done:    make(chan bool, 1)}
		Globals.Cluster.nodes[host.Name] = &n
	}
	if len(Globals.Cluster.nodes) == 0 {
		log.Info("cluster.ClusterInit", "Invalid cluster size: 1")
	}
	if !Globals.Cluster.failoverInit(config.Failover) {
		Globals.Cluster.rehash(nil)
	}
	sort.Strings(nodeNames)
	workerId := sort.SearchStrings(nodeNames, thisName) + 1
	return workerId
}
func (c *Conn) rpcWriteLoop() {
	defer func() {
		c.closeRPC()
		Globals.ConnCache.Delete(c.connid)
		c.unsubAll()
	}()
	var unused bool
	for {
		select {
		case msg, ok := <-c.send:
			if !ok || c.clnode.endpoint == nil {
				return
			}
			m, err := lp.Encode(c.proto, msg)
			if err != nil {
				log.Error("conn.writeRpc", err.Error())
				return
			}
			if err := c.clnode.call("Cluster.Proxy", &ClusterResp{Msg: m.Bytes(), FromConnID: c.connid}, &unused); err != nil {
				log.Error("conn.writeRPC", err.Error())
				return
			}
		case msg := <-c.stop:
			if msg != nil {
				c.clnode.call("Cluster.Proxy", &ClusterResp{Msg: msg.([]byte), FromConnID: c.connid}, &unused)
			}
			return
		}
	}
}
func (c *Conn) closeRPC() {
	log.Info("cluster.closeRPC", "session closed at master")
}