types

Documentation

Index

• Constants
• Variables
• func NewProtoFromPChannelInfo(pchannel PChannelInfo) *streamingpb.PChannelInfo
• func NewProtoFromStreamingNodeInfo(info StreamingNodeInfo) *streamingpb.StreamingNodeInfo
• type AppendResponse
• type AppendResponses
  • func NewAppendResponseN(n int) AppendResponses
  • func (a *AppendResponses) FillAllError(err error)
  • func (a *AppendResponses) FillAllResponse(resp AppendResponse)
  • func (a *AppendResponses) FillResponseAtIdx(resp AppendResponse, idx int)
  • func (a AppendResponses) MaxTimeTick() uint64
  • func (a AppendResponses) UnwrapFirstError() error
• type AppendResult
  • func (r *AppendResult) GetExtra(m proto.Message) error
  • func (r *AppendResult) IntoProto() *streamingpb.ProduceMessageResponseResult
• type AssignmentDiscoverWatcher
• type AssignmentRebalanceTrigger
• type BroadcastAckRequest
• type BroadcastAppendResult
  • func (r *BroadcastAppendResult) GetAppendResult(channelName string) *AppendResult
• type PChannelInfo
  • func NewPChannelInfoFromProto(pchannel *streamingpb.PChannelInfo) PChannelInfo
  • func (c *PChannelInfo) String() string
• type PChannelInfoAssigned
• type StreamingNodeAssignment
• type StreamingNodeInfo
  • func NewStreamingNodeInfoFromProto(proto *streamingpb.StreamingNodeInfo) StreamingNodeInfo
• type StreamingNodeStatus
  • func (n *StreamingNodeStatus) ErrorOfNode() error
  • func (n *StreamingNodeStatus) IsHealthy() bool
• type VersionedStreamingNodeAssignments

Constants

const (
	InitialTerm int64 = -1
)

Variables

var (
	ErrStopping = errors.New("streaming node is stopping")
	ErrNotAlive = errors.New("streaming node is not alive")
)

Functions

func NewProtoFromPChannelInfo

func NewProtoFromPChannelInfo(pchannel PChannelInfo) *streamingpb.PChannelInfo

NewProtoFromPChannelInfo converts PChannelInfo to protobuf PChannelInfo

func NewProtoFromStreamingNodeInfo

func NewProtoFromStreamingNodeInfo(info StreamingNodeInfo) *streamingpb.StreamingNodeInfo

NewProtoFromStreamingNodeInfo creates a proto from StreamingNodeInfo.

Types

type AppendResponse

type AppendResponse struct {
	AppendResult *AppendResult
	Error        error
}

AppendResponse is the response of one append operation.

type AppendResponses

type AppendResponses struct {
	Responses []AppendResponse
}

AppendResponses is the response of append operation.

func NewAppendResponseN

func NewAppendResponseN(n int) AppendResponses

NewAppendResponseN creates a new append response.

func (*AppendResponses) FillAllError

func (a *AppendResponses) FillAllError(err error)

FillAllError fills all the responses with the same error.

func (*AppendResponses) FillAllResponse

func (a *AppendResponses) FillAllResponse(resp AppendResponse)

func (*AppendResponses) FillResponseAtIdx

func (a *AppendResponses) FillResponseAtIdx(resp AppendResponse, idx int)

FillResponseAtIdx fill the response at idx

func (AppendResponses) MaxTimeTick

func (a AppendResponses) MaxTimeTick() uint64

func (AppendResponses) UnwrapFirstError

func (a AppendResponses) UnwrapFirstError() error

UnwrapFirstError returns the first error in the responses.

type AppendResult

type AppendResult struct {
	// MessageID is generated by underlying walimpls.
	MessageID message.MessageID

	// TimeTick is the time tick of the message.
	// Set by timetick interceptor.
	TimeTick uint64

	// TxnCtx is the transaction context of the message.
	// If the message is not belong to a transaction, the TxnCtx will be nil.
	TxnCtx *message.TxnContext

	// Extra is the extra information of the append result.
	Extra *anypb.Any
}

AppendResult is the result of append operation.

func (*AppendResult) GetExtra

func (r *AppendResult) GetExtra(m proto.Message) error

GetExtra unmarshal the extra information to the given message.

func (*AppendResult) IntoProto

func (r *AppendResult) IntoProto() *streamingpb.ProduceMessageResponseResult

IntoProto converts the append result to proto.

type AssignmentDiscoverWatcher

type AssignmentDiscoverWatcher interface {
	// AssignmentDiscover watches the assignment discovery.
	// The callback will be called when the discovery is changed.
	// The final error will be returned when the watcher is closed or broken.
	AssignmentDiscover(ctx context.Context, cb func(*VersionedStreamingNodeAssignments) error) error

	AssignmentRebalanceTrigger
}

AssignmentDiscoverWatcher is the interface for watching the assignment discovery.

type AssignmentRebalanceTrigger

type AssignmentRebalanceTrigger interface {
	// ReportStreamingError is used to report the streaming error.
	// Trigger a re-balance of the pchannel.
	ReportAssignmentError(ctx context.Context, pchannel PChannelInfo, err error) error
}

AssignmentRebalanceTrigger is the interface for triggering the re-balance of the pchannel.

type BroadcastAckRequest

type BroadcastAckRequest struct {
	// BroadcastID is the broadcast id of the ack request.
	BroadcastID uint64
	VChannel    string
}

type BroadcastAppendResult

type BroadcastAppendResult struct {
	BroadcastID   uint64                   // the broadcast id of the append operation.
	AppendResults map[string]*AppendResult // make the channel name to the append result.
}

BroadcastAppendResult is the result of broadcast append operation.

func (*BroadcastAppendResult) GetAppendResult

func (r *BroadcastAppendResult) GetAppendResult(channelName string) *AppendResult

GetAppendResult returns the append result of the given channel.

type PChannelInfo

type PChannelInfo struct {
	Name string // name of pchannel.
	Term int64  // term of pchannel.
}

PChannelInfo is the struct for pchannel info.

func NewPChannelInfoFromProto

func NewPChannelInfoFromProto(pchannel *streamingpb.PChannelInfo) PChannelInfo

NewPChannelInfoFromProto converts protobuf PChannelInfo to PChannelInfo

func (*PChannelInfo) String

func (c *PChannelInfo) String() string

type PChannelInfoAssigned

type PChannelInfoAssigned struct {
	Channel PChannelInfo
	Node    StreamingNodeInfo
}

type StreamingNodeAssignment

type StreamingNodeAssignment struct {
	NodeInfo StreamingNodeInfo
	Channels map[string]PChannelInfo
}

StreamingNodeAssignment is the relation between server and channels.

type StreamingNodeInfo

type StreamingNodeInfo struct {
	ServerID int64
	Address  string
}

StreamingNodeInfo is the relation between server and channels.

func NewStreamingNodeInfoFromProto

func NewStreamingNodeInfoFromProto(proto *streamingpb.StreamingNodeInfo) StreamingNodeInfo

NewStreamingNodeInfoFromProto creates a StreamingNodeInfo from proto.

type StreamingNodeStatus

type StreamingNodeStatus struct {
	StreamingNodeInfo
	// TODO: balance attributes should added here in future.
	Err error
}

StreamingNodeStatus is the information of a streaming node.

func (*StreamingNodeStatus) ErrorOfNode

func (n *StreamingNodeStatus) ErrorOfNode() error

ErrorOfNode returns the error of the streaming node.

func (*StreamingNodeStatus) IsHealthy

func (n *StreamingNodeStatus) IsHealthy() bool

IsHealthy returns whether the streaming node is healthy.

type VersionedStreamingNodeAssignments

type VersionedStreamingNodeAssignments struct {
	Version     typeutil.VersionInt64Pair
	Assignments map[int64]StreamingNodeAssignment
}

VersionedStreamingNodeAssignments is the relation between server and channels with version.