heartbeat

Documentation

Index

• Constants
• type Client
• type Heartbeat
• type HeartbeatClient
  • func NewClient(conn *nats.Conn, nodeID string, publisher ncl.Publisher) (*HeartbeatClient, error)
  • func (h *HeartbeatClient) Close(ctx context.Context) error
  • func (h *HeartbeatClient) SendHeartbeat(ctx context.Context, sequence uint64) error
• type HeartbeatServer
  • func NewServer(params HeartbeatServerParams) (*HeartbeatServer, error)
  • func (h *HeartbeatServer) FilterNodeInfos(nodeInfos []*models.NodeInfo, state models.NodeConnectionState) []*models.NodeInfo
  • func (h *HeartbeatServer) Handle(ctx context.Context, heartbeat Heartbeat) error
  • func (h *HeartbeatServer) HandleMessage(ctx context.Context, message *ncl.Message) error
  • func (h *HeartbeatServer) RemoveNode(nodeID string)
  • func (h *HeartbeatServer) ShouldProcess(ctx context.Context, message *ncl.Message) bool
  • func (h *HeartbeatServer) Start(ctx context.Context) error
  • func (h *HeartbeatServer) UpdateNodeInfo(state *models.NodeState)
• type HeartbeatServerParams
• type TimestampedHeartbeat

Constants

const (
	// HeartbeatMessageType is the message type for heartbeats
	HeartbeatMessageType = "heartbeat"
)

Types

type Client

type Client interface {
	SendHeartbeat(ctx context.Context, sequence uint64) error
	Close(ctx context.Context) error
}

type Heartbeat

type Heartbeat struct {
	NodeID   string
	Sequence uint64
}

Heartbeat represents a heartbeat message from a specific node. It contains the node ID and the sequence number of the heartbeat which is monotonically increasing (reboots aside). We do not use timestamps on the client, we rely solely on the server-side time to avoid clock drift issues.

type HeartbeatClient

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

func NewClient

func NewClient(conn *nats.Conn, nodeID string, publisher ncl.Publisher) (*HeartbeatClient, error)

func (*HeartbeatClient) Close

func (h *HeartbeatClient) Close(ctx context.Context) error

func (*HeartbeatClient) SendHeartbeat

func (h *HeartbeatClient) SendHeartbeat(ctx context.Context, sequence uint64) error

type HeartbeatServer

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

func NewServer

func NewServer(params HeartbeatServerParams) (*HeartbeatServer, error)

func (*HeartbeatServer) FilterNodeInfos

func (h *HeartbeatServer) FilterNodeInfos(nodeInfos []*models.NodeInfo, state models.NodeConnectionState) []*models.NodeInfo

FilterNodeInfos will return only those NodeInfos that have the requested liveness

func (*HeartbeatServer) Handle

func (h *HeartbeatServer) Handle(ctx context.Context, heartbeat Heartbeat) error

Handle will handle a message received through the legacy heartbeat topic

func (*HeartbeatServer) HandleMessage

func (h *HeartbeatServer) HandleMessage(ctx context.Context, message *ncl.Message) error

HandleMessage will handle a message received through ncl and will call the Handle method

func (*HeartbeatServer) RemoveNode

func (h *HeartbeatServer) RemoveNode(nodeID string)

RemoveNode will handle removing the liveness for a specific node. This is useful when a node is removed from the cluster.

func (*HeartbeatServer) ShouldProcess

func (h *HeartbeatServer) ShouldProcess(ctx context.Context, message *ncl.Message) bool

func (*HeartbeatServer) Start

func (h *HeartbeatServer) Start(ctx context.Context) error

func (*HeartbeatServer) UpdateNodeInfo

func (h *HeartbeatServer) UpdateNodeInfo(state *models.NodeState)

UpdateNode will add the liveness for specific nodes to their NodeInfo

type HeartbeatServerParams

type HeartbeatServerParams struct {
	NodeID                string
	Client                *nats.Conn
	Clock                 clock.Clock
	NodeDisconnectedAfter time.Duration
}

type TimestampedHeartbeat

type TimestampedHeartbeat struct {
	Heartbeat
	Timestamp int64
}