consistent_service

Documentation

Overview

Package consistent_service is consistent service for consensus engine

Index

• Variables
• type Broadcast
• type ConsistentEngine
• type Decoder
• type Logger
• type Message
• type Node
• type Status
• type StatusBroadcaster
• type StatusConsistentEngine
  • func NewConsistentService(local Node, msg Message, log Logger) *StatusConsistentEngine
  • func (e *StatusConsistentEngine) AddBroadcaster(id string, broadcaster StatusBroadcaster) error
  • func (e *StatusConsistentEngine) PutRemoter(id string, node Node) error
  • func (e *StatusConsistentEngine) RegisterStatusCoder(decoderType int8, decoder Decoder) error
  • func (e *StatusConsistentEngine) RegisterStatusInterceptor(interceptorType int8, interceptor StatusInterceptor) error
  • func (e *StatusConsistentEngine) RemoveRemoter(id string) error
  • func (e *StatusConsistentEngine) Start(ctx context.Context) error
  • func (e *StatusConsistentEngine) Stop(ctx context.Context) error
  • func (e *StatusConsistentEngine) UpdateNodeStatus(id string, node Node) error
• type StatusInterceptor

Variables

var (
	// ErrorInvalidParameter invalid parameter
	ErrorInvalidParameter = errors.New("invalid parameter")
	// ErrorBroadcasterExist broadcaster exist
	ErrorBroadcasterExist = errors.New("broadcaster exist")
	// ErrorDecoderExist decoder exist
	ErrorDecoderExist = errors.New("decoder exist")
	// ErrorInterceptorExist interceptor exist
	ErrorInterceptorExist = errors.New("interceptor exist")
	// ErrorRunRepeatedly run repeatedly
	ErrorRunRepeatedly = errors.New("run repeatedly")
	// ErrorNotRunning not running
	ErrorNotRunning = errors.New("not running")
	// ErrorRemoterExist remoter exist
	ErrorRemoterExist = errors.New("remoter exist")
	// ErrorRemoterNotExist remoter not exist
	ErrorRemoterNotExist = errors.New("remoter not exist")
	// ErrorRemoterEqualLocal remoter equal local
	ErrorRemoterEqualLocal = errors.New("remoter equal local")
)

Types

type Broadcast

type Broadcast func(Node, Node) (interface{}, error)

Broadcast 广播器

type ConsistentEngine

type ConsistentEngine interface {

	// Start 启动一致性引擎
	// ctx 后续扩展用，启动失败返回error
	Start(ctx context.Context) error

	// Stop 停止一致性引擎
	// ctx 后续扩展用，，停止失败返回error
	Stop(ctx context.Context) error

	// AddBroadcaster 添加状态广播器（如一个tbft状态广播器）
	// id 广播器标识，需要保证唯一性
	// broadcast 广播器，需要用户自己实现（如：tbft广播器、maxbft广播器）
	AddBroadcaster(id string, broadcast StatusBroadcaster) error

	// UpdateNodeStatus 更新本地状态
	// id 节点标识，需要保证唯一性
	// node 需要更新的节点信息（包含节点状态，节点状态可以有多种）
	UpdateNodeStatus(id string, node Node) error

	// PutRemoter 添加节点
	// id 节点标识，需要保证唯一性
	// node 需要添加的节点信息（包含节点状态，节点状态可以有多种）
	PutRemoter(id string, node Node) error

	// RemoveRemoter 删除节点
	// id 节点标识，当节点不存在时返回错误消息
	RemoveRemoter(id string) error

	// RegisterStatusCoder 注册状态解析器
	// decoderType 解析器标识，需要保证唯一性
	// decoder 需要添加的解析器，由用户实现（如：tbft解析器）
	RegisterStatusCoder(decoderType int8, decoder Decoder) error

	// RegisterStatusInterceptor 注册过滤器
	// interceptorType 过滤器标识，需要保证唯一性
	// interceptor 需要添加的过滤器，由用户实现（如：tbft过滤器）
	RegisterStatusInterceptor(interceptorType int8, interceptor StatusInterceptor) error
}

ConsistentEngine 一致性引擎，用于节点间（共识状态）信息同步

type Decoder

type Decoder interface {

	// MsgType 解析器处理的消息类型
	MsgType() int8

	// Decode 解析器解析对应类型的消息，返回解析后数据对象
	Decode(interface{}) interface{}
}

Decoder 解析器

type Logger

type Logger interface {
	Debug(args ...interface{})
	Debugf(format string, args ...interface{})
	Debugw(msg string, keysAndValues ...interface{})
	Error(args ...interface{})
	Errorf(format string, args ...interface{})
	Errorw(msg string, keysAndValues ...interface{})
	Warn(args ...interface{})
	Warnf(format string, args ...interface{})
	Warnw(msg string, keysAndValues ...interface{})
	Info(args ...interface{})
	Infof(format string, args ...interface{})
	Infow(msg string, keysAndValues ...interface{})
}

Logger is logger interface of consistentEngine.

type Message

type Message interface {

	// Send 一致性引擎对外发送数据
	// payload 需要发送的消息
	Send(payload interface{})

	// Receive 一致性引擎接收外部数据
	// 返回接收到的消息
	Receive() interface{}

	// Start 启动
	Start() error

	// Stop 关闭
	Stop() error
}

Message 用户一致性引擎于其他模块（如tbft/maxbft）数据交互

type Node

type Node interface {

	// ID 节点标识
	ID() string

	// Statuses 节点状态，可以有多种
	Statuses() map[int8]Status

	// UpdateStatus 更新节点状态
	// status 节点状态（如：tbft状态）
	UpdateStatus(status Status)
}

Node 节点信息（基本信息/共识状态）

type Status

type Status interface {

	// Type 状态类型
	Type() int8

	// Data 状态内容
	Data() interface{}

	// Update 更新状态
	Update(status Status)
}

Status 节点共识状态

type StatusBroadcaster

type StatusBroadcaster interface {

	// ID 广播器标识
	ID() string

	// TimePattern 状态广播触发模式
	// 返回触发间隔
	TimePattern() interface{}

	// PreBroadcaster 广播器，判断是否要发送消息
	// 返回广播器方法
	PreBroadcaster() Broadcast

	// Start 启动
	Start() error

	// Stop 停止
	Stop() error

	// IsRunning 获取运行状态
	IsRunning() bool
}

StatusBroadcaster 状态广播器 由内部定时器触发广播 根据LocalNodeStatus和RemoteNodeStatus当前状态，确认是否进行状态广播

type StatusConsistentEngine

type StatusConsistentEngine struct {
	// lock read write lock
	sync.RWMutex
	// contains filtered or unexported fields
}

StatusConsistentEngine 一致性引擎实现

func NewConsistentService

func NewConsistentService(local Node,
	msg Message,
	log Logger) *StatusConsistentEngine

NewConsistentService nolint: unused

func (*StatusConsistentEngine) AddBroadcaster

func (e *StatusConsistentEngine) AddBroadcaster(id string, broadcaster StatusBroadcaster) error

AddBroadcaster add a broadcaster

func (*StatusConsistentEngine) PutRemoter

func (e *StatusConsistentEngine) PutRemoter(id string, node Node) error

PutRemoter put a remoter

func (*StatusConsistentEngine) RegisterStatusCoder

func (e *StatusConsistentEngine) RegisterStatusCoder(decoderType int8, decoder Decoder) error

RegisterStatusCoder register status coder

func (*StatusConsistentEngine) RegisterStatusInterceptor

func (e *StatusConsistentEngine) RegisterStatusInterceptor(interceptorType int8, interceptor StatusInterceptor) error

RegisterStatusInterceptor register status interceptor

func (*StatusConsistentEngine) RemoveRemoter

func (e *StatusConsistentEngine) RemoveRemoter(id string) error

RemoveRemoter remove remoter

func (*StatusConsistentEngine) Start

func (e *StatusConsistentEngine) Start(ctx context.Context) error

Start engine

func (*StatusConsistentEngine) Stop

func (e *StatusConsistentEngine) Stop(ctx context.Context) error

Stop engine

func (*StatusConsistentEngine) UpdateNodeStatus

func (e *StatusConsistentEngine) UpdateNodeStatus(id string, node Node) error

UpdateNodeStatus 更新本地状态

type StatusInterceptor

type StatusInterceptor interface {

	// Handle 过滤处理方法
	// status 节点状态（如：tbft状态）
	Handle(status Status) error
}

StatusInterceptor 过滤器