cluster

Documentation

Index

• Constants
• func OnApplyLog(leaderId, nodeId string, tp byte, filter []byte, prompt string, err error)
• func OnConnectPacketLog(direction byte, node, clientId string)
• func OnJoinLog(nodeId, addr, prompt string, err error)
• func OnPublishPacketLog(direction byte, nodeId, cid, topic string, pid uint16)
• type Agent
  • func NewAgent(conf *config.Cluster) *Agent
  • func (a *Agent) AddRaftPeer(id, addr string)
  • func (a *Agent) BindMqttServer(server *mqtt.Server)
  • func (a *Agent) GetLocalName() string
  • func (a *Agent) GetMemberList() []discovery.Member
  • func (a *Agent) GetValue(key string) []string
  • func (a *Agent) RemoveRaftPeer(id string)
  • func (a *Agent) Start() (err error)
  • func (a *Agent) Stat() map[string]int64
  • func (a *Agent) Stop()
  • func (a *Agent) SubmitOutTask(pk *packets.Packet)
  • func (a *Agent) SubmitRaftTask(msg *message.Message)
• type ClientManager
  • func NewClientManager(a *Agent) *ClientManager
  • func (c *ClientManager) ConnectNotifyToNode(nodeId, clientId string)
  • func (c *ClientManager) ConnectNotifyToOthers(msg *message.Message)
  • func (c *ClientManager) RaftApplyToOthers(msg *message.Message)
  • func (c *ClientManager) RaftJoinToOthers()
  • func (c *ClientManager) RelayPublishPacket(nodeId string, msg *message.Message)
  • func (c *ClientManager) RelayRaftApply(nodeId string, msg *message.Message)
  • func (c *ClientManager) RelayRaftJoin(nodeId string)
  • func (c *ClientManager) RemoveGrpcClient(nodeId string)
• type MqttEventHook
  • func (h *MqttEventHook) ID() string
  • func (h *MqttEventHook) Init(config any) error
  • func (h *MqttEventHook) OnPublished(cl *mqtt.Client, pk packets.Packet)
  • func (h *MqttEventHook) OnSessionEstablished(cl *mqtt.Client, pk packets.Packet)
  • func (h *MqttEventHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int)
  • func (h *MqttEventHook) OnUnsubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int)
  • func (h *MqttEventHook) OnWillSent(cl *mqtt.Client, pk packets.Packet)
  • func (h *MqttEventHook) Provides(b byte) bool
• type RpcService
  • func NewRpcService(a *Agent) *RpcService
  • func (s *RpcService) ConnectNotify(ctx context.Context, req *crpc.ConnectRequest) (*crpc.Response, error)
  • func (s *RpcService) PublishPacket(ctx context.Context, req *crpc.PublishRequest) (*crpc.Response, error)
  • func (s *RpcService) RaftApply(ctx context.Context, req *crpc.ApplyRequest) (*crpc.Response, error)
  • func (s *RpcService) RaftJoin(ctx context.Context, req *crpc.JoinRequest) (*crpc.Response, error)
  • func (s *RpcService) StartRpcServer() error
  • func (s *RpcService) StopRpcServer()

Constants

const (
	DirectionInbound byte = iota
	DirectionOutbound
)

const (
	NodesFile = "nodes.json"
	PeersFIle = "peers.json"
)

const (
	ReqTimeout = 500 * time.Millisecond
)

Functions

func OnApplyLog

func OnApplyLog(leaderId, nodeId string, tp byte, filter []byte, prompt string, err error)

func OnConnectPacketLog

func OnConnectPacketLog(direction byte, node, clientId string)

func OnJoinLog

func OnJoinLog(nodeId, addr, prompt string, err error)

func OnPublishPacketLog

func OnPublishPacketLog(direction byte, nodeId, cid, topic string, pid uint16)

Types

type Agent

type Agent struct {
	Config *config.Cluster

	OutPool *ants.Pool
	// contains filtered or unexported fields
}

func NewAgent

func NewAgent(conf *config.Cluster) *Agent

func (*Agent) AddRaftPeer

func (a *Agent) AddRaftPeer(id, addr string)

func (*Agent) BindMqttServer

func (a *Agent) BindMqttServer(server *mqtt.Server)

func (*Agent) GetLocalName

func (a *Agent) GetLocalName() string

func (*Agent) GetMemberList

func (a *Agent) GetMemberList() []discovery.Member

func (*Agent) GetValue

func (a *Agent) GetValue(key string) []string

func (*Agent) RemoveRaftPeer

func (a *Agent) RemoveRaftPeer(id string)

func (*Agent) Start

func (a *Agent) Start() (err error)

func (*Agent) Stat

func (a *Agent) Stat() map[string]int64

func (*Agent) Stop

func (a *Agent) Stop()

func (*Agent) SubmitOutTask

func (a *Agent) SubmitOutTask(pk *packets.Packet)

func (*Agent) SubmitRaftTask

func (a *Agent) SubmitRaftTask(msg *message.Message)

type ClientManager

type ClientManager struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewClientManager

func NewClientManager(a *Agent) *ClientManager

func (*ClientManager) ConnectNotifyToNode

func (c *ClientManager) ConnectNotifyToNode(nodeId, clientId string)

func (*ClientManager) ConnectNotifyToOthers

func (c *ClientManager) ConnectNotifyToOthers(msg *message.Message)

func (*ClientManager) RaftApplyToOthers

func (c *ClientManager) RaftApplyToOthers(msg *message.Message)

func (*ClientManager) RaftJoinToOthers

func (c *ClientManager) RaftJoinToOthers()

func (*ClientManager) RelayPublishPacket

func (c *ClientManager) RelayPublishPacket(nodeId string, msg *message.Message)

func (*ClientManager) RelayRaftApply

func (c *ClientManager) RelayRaftApply(nodeId string, msg *message.Message)

func (*ClientManager) RelayRaftJoin

func (c *ClientManager) RelayRaftJoin(nodeId string)

func (*ClientManager) RemoveGrpcClient

func (c *ClientManager) RemoveGrpcClient(nodeId string)

type MqttEventHook

type MqttEventHook struct {
	mqtt.HookBase
	// contains filtered or unexported fields
}

MqttEventHook is a mqtt event hook that callback when events such as connect, publish, subscribe, etc. occur.

func (*MqttEventHook) ID

func (h *MqttEventHook) ID() string

ID returns the id of the hook.

func (*MqttEventHook) Init

func (h *MqttEventHook) Init(config any) error

Init initializes

func (*MqttEventHook) OnPublished

func (h *MqttEventHook) OnPublished(cl *mqtt.Client, pk packets.Packet)

OnPublished is called when a client has published a message to subscribers.

func (*MqttEventHook) OnSessionEstablished

func (h *MqttEventHook) OnSessionEstablished(cl *mqtt.Client, pk packets.Packet)

OnSessionEstablished notifies other nodes to perform local subscription cleanup when their session is established.

func (*MqttEventHook) OnSubscribed

func (h *MqttEventHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int)

OnSubscribed is called when a client subscribes to one or more filters.

func (*MqttEventHook) OnUnsubscribed

func (h *MqttEventHook) OnUnsubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int)

func (*MqttEventHook) OnWillSent

func (h *MqttEventHook) OnWillSent(cl *mqtt.Client, pk packets.Packet)

OnWillSent is called when an LWT message has been issued from a disconnecting client.

func (*MqttEventHook) Provides

func (h *MqttEventHook) Provides(b byte) bool

Provides indicates which hook methods this hook provides.

type RpcService

type RpcService struct {
	// contains filtered or unexported fields
}

func NewRpcService

func NewRpcService(a *Agent) *RpcService

func (*RpcService) ConnectNotify

func (s *RpcService) ConnectNotify(ctx context.Context, req *crpc.ConnectRequest) (*crpc.Response, error)

func (*RpcService) PublishPacket

func (s *RpcService) PublishPacket(ctx context.Context, req *crpc.PublishRequest) (*crpc.Response, error)

func (*RpcService) RaftApply

func (s *RpcService) RaftApply(ctx context.Context, req *crpc.ApplyRequest) (*crpc.Response, error)

func (*RpcService) RaftJoin

func (s *RpcService) RaftJoin(ctx context.Context, req *crpc.JoinRequest) (*crpc.Response, error)

func (*RpcService) StartRpcServer

func (s *RpcService) StartRpcServer() error

func (*RpcService) StopRpcServer

func (s *RpcService) StopRpcServer()