rpc

Documentation

Index

• Constants
• func NewRpcCtx() *poleRpcContext
• type ChannelHandler
• type ProtobufMessageRegistry
  • func (pmr *ProtobufMessageRegistry) FindProtoMessageSupplier(key string) func() proto.Message
  • func (pmr *ProtobufMessageRegistry) RegistryProtoMessageSupplier(key string, supplier func() proto.Message) bool
• type RaftClient
  • func NewRaftClient(openTSL bool) (*RaftClient, error)
  • func (c *RaftClient) CheckConnection(endpoint entity.Endpoint) (bool, error)
  • func (c *RaftClient) RegisterConnectEventWatcher(watcher func(event api.ConnectEventType, con net.Conn))
  • func (c *RaftClient) SendRequest(endpoint entity.Endpoint, req *api.ServerRequest) (*api.ServerResponse, error)
• type RaftRPCServer
  • func NewRaftRPCServer(label string, port int32, openTSL bool) (*RaftRPCServer, error)
  • func (rpcServer *RaftRPCServer) GetRealServer() api.TransportServer
  • func (rpcServer *RaftRPCServer) RegisterChannelHandler(funName string, handler ChannelHandler)
  • func (rpcServer *RaftRPCServer) RegisterSimpleHandler(funName string, handler SimpleHandler)
• type RpcContext
• type RpcServerContext
  • func (rpc *RpcServerContext) Close() error
  • func (rpc *RpcServerContext) GetReq() (proto.Message, error)
  • func (rpc *RpcServerContext) SendResp(msg proto.Message)
• type SimpleHandler

Constants

const (
	// cli command
	// 添加一个学习者的命令
	CliAddLearnerRequest string = "CliAddLearnerCommand"
	// 添加一个成员的命令
	CliAddPeerRequest string = "CliAddPeerCommand"
	// 更改成员的命令
	CliChangePeersRequest string = "CliChangePeersCommand"
	// 获取 Leader 的命令
	CliGetLeaderRequest string = "CliGetLeaderCommand"
	// 获取所有的成员
	CliGetPeersRequest string = "CliGetPeersCommand"
	// 批量移除学习者
	CliRemoveLearnersRequest string = "CliRemoveLearnersCommand"
	CliResetLearnersRequest  string = "CliResetLearnersCommand"
	CliResetPeersRequest     string = "CliResetPeersCommand"
	CliSnapshotRequest       string = "CliSnapshotCommand"
	CliTransferLeaderRequest string = "CliTransferLeaderCommand"

	// proto command
	CoreAppendEntriesRequest   string = "CoreAppendEntriesCommand"
	CoreGetFileRequest         string = "CoreGetFileCommand"
	CoreInstallSnapshotRequest string = "CoreInstallSnapshotCommand"
	CoreNodeRequest            string = "CoreNodeCommand"
	CoreReadIndexRequest       string = "CoreReadIndexCommand"
	CoreRequestPreVoteRequest  string = "CoreRequestPreVoteRequest"
	CoreRequestVoteRequest     string = "CoreRequestVoteCommand"
	CoreTimeoutNowRequest      string = "CoreTimeoutNowCommand"

	//
	CommonRpcErrorCommand string = "CommonRpcErrorCommand"
)

const (
	//RequestIDKey 每次请求的唯一标识
	RequestIDKey    string = "RequestID"
	RequestFuncName string = "RequestFuncName"
)

Functions

func NewRpcCtx

func NewRpcCtx() *poleRpcContext

NewRPCCtx 创建一个新的 RPCContext

Types

type ChannelHandler

type ChannelHandler func(ctx context.Context, rpcCtx *RpcServerContext)

type ProtobufMessageRegistry

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

ProtobufMessageRegistry 用于自动化的将字节数组转换为业务结构体

var GlobalProtoRegistry *ProtobufMessageRegistry

GlobalProtoRegistry 全局的 proto.Message supplier 的仓库

func (*ProtobufMessageRegistry) FindProtoMessageSupplier

func (pmr *ProtobufMessageRegistry) FindProtoMessageSupplier(key string) func() proto.Message

FindProtoMessageSupplier 根据名称找到 proto.Message 的对象构造者，每次调用对象构造者，会创建出一个新的 proto.Message

func (*ProtobufMessageRegistry) RegistryProtoMessageSupplier

func (pmr *ProtobufMessageRegistry) RegistryProtoMessageSupplier(key string, supplier func() proto.Message) bool

RegistryProtoMessageSupplier 注册一个 proto.Message 对象构造者

type RaftClient

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

RaftClient raft rpc 客户端，负责进行远程方法调用

func NewRaftClient

func NewRaftClient(openTSL bool) (*RaftClient, error)

NewRaftClient 创建一个 Raft 的 RPC 客户端

func (*RaftClient) CheckConnection

func (c *RaftClient) CheckConnection(endpoint entity.Endpoint) (bool, error)

CheckConnection 检查链接是否存在，如果不存在则会自动去创建一个链接

func (*RaftClient) RegisterConnectEventWatcher

func (c *RaftClient) RegisterConnectEventWatcher(watcher func(event api.ConnectEventType, con net.Conn))

RegisterConnectEventWatcher 注册链接事件监听器

func (*RaftClient) SendRequest

func (c *RaftClient) SendRequest(endpoint entity.Endpoint, req *api.ServerRequest) (*api.ServerResponse, error)

SendRequest 发送一个 RPC 请求，基于 request-response 模型

type RaftRPCServer

type RaftRPCServer struct {
	IsReady chan struct{}

	Ctx context.Context
	// contains filtered or unexported fields
}

RaftRPCServer raft 的 rpc-server，封装 pole-group/pole-rpc

func NewRaftRPCServer

func NewRaftRPCServer(label string, port int32, openTSL bool) (*RaftRPCServer, error)

NewRaftRPCServer 创建 RpcServer

func (*RaftRPCServer) GetRealServer

func (rpcServer *RaftRPCServer) GetRealServer() api.TransportServer

GetRealServer 获取真正的 RpcServer

func (*RaftRPCServer) RegisterChannelHandler

func (rpcServer *RaftRPCServer) RegisterChannelHandler(funName string, handler ChannelHandler)

RegisterChannelRequestHandler 注册一个 Request-Channel的Server端处理者，名称为name

func (*RaftRPCServer) RegisterSimpleHandler

func (rpcServer *RaftRPCServer) RegisterSimpleHandler(funName string, handler SimpleHandler)

RegisterRequestHandler 注册一个 Request-Response的Server端处理者，名称为name

type RpcContext

type RpcContext interface {
	io.Closer

	SendResp(resp *api.ServerResponse)

	GetRemoteAddr() net.Addr

	Write(key, value interface{})
}

type RpcServerContext

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

func (*RpcServerContext) Close

func (rpc *RpcServerContext) Close() error

func (*RpcServerContext) GetReq

func (rpc *RpcServerContext) GetReq() (proto.Message, error)

func (*RpcServerContext) SendResp

func (rpc *RpcServerContext) SendResp(msg proto.Message)

type SimpleHandler

type SimpleHandler func(ctx context.Context, rpcCtx *RpcServerContext)