controller

Documentation

Index

• type Queue
  • func NewQueue(pCtx context.Context, log logger.Logger) (*Queue, error)
  • func (q *Queue) AckMessage(pCtx context.Context, consumerID string, msg *model.Message) error
  • func (q *Queue) Close(ctx context.Context) error
  • func (q *Queue) Connect(pCtx context.Context, consumerID, consumerGroupID string, topics []string) (*pbTypes.Consumer, *pbTypes.ConsumerGroup, error)
  • func (q *Queue) CreateTopic(pCtx context.Context, name string, numberOfPartitions uint64, ...) (*pbTypes.Topic, error)
  • func (q *Queue) HealthCheck(pCtx context.Context, consumerID string, healthCheckAt time.Time) (*model.Consumer, error)
  • func (q *Queue) LeaderUpdated(info raftio.LeaderInfo)
  • func (q *Queue) ReceiveMessageForPartition(pCtx context.Context, consumerID string, partitionId string) (*model.Message, error)
  • func (q *Queue) RegisterNewNode(pCtx context.Context, newNodeID uint64, targetNodeAddr string) error
  • func (q *Queue) SendMessage(pCtx context.Context, msg *model.Message) (*model.Message, error)
  • func (q *Queue) ShardsInfo(pCtx context.Context, topics []string) (map[string]*model.ShardInfo, []*model.Broker, *model.Broker, error)

Types

type Queue

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

func NewQueue

func NewQueue(
	pCtx context.Context,
	log logger.Logger,
) (*Queue, error)

func (*Queue) AckMessage

func (q *Queue) AckMessage(
	pCtx context.Context,
	consumerID string,
	msg *model.Message,
) error

func (*Queue) Close

func (q *Queue) Close(ctx context.Context) error

func (*Queue) Connect

func (q *Queue) Connect(
	pCtx context.Context,
	consumerID, consumerGroupID string,
	topics []string,
) (*pbTypes.Consumer, *pbTypes.ConsumerGroup, error)

func (*Queue) CreateTopic

func (q *Queue) CreateTopic(
	pCtx context.Context,
	name string,
	numberOfPartitions uint64,
	replicationFactor uint64,
) (*pbTypes.Topic, error)

func (*Queue) HealthCheck

func (q *Queue) HealthCheck(
	pCtx context.Context,
	consumerID string,
	healthCheckAt time.Time,
) (*model.Consumer, error)

func (*Queue) LeaderUpdated

func (q *Queue) LeaderUpdated(info raftio.LeaderInfo)

func (*Queue) ReceiveMessageForPartition

func (q *Queue) ReceiveMessageForPartition(
	pCtx context.Context,
	consumerID string,
	partitionId string,
) (*model.Message, error)

func (*Queue) RegisterNewNode

func (q *Queue) RegisterNewNode(
	pCtx context.Context,
	newNodeID uint64,
	targetNodeAddr string,
) error

func (*Queue) SendMessage

func (q *Queue) SendMessage(
	pCtx context.Context,
	msg *model.Message,
) (*model.Message, error)

func (*Queue) ShardsInfo

func (q *Queue) ShardsInfo(
	pCtx context.Context,
	topics []string,
) (map[string]*model.ShardInfo, []*model.Broker, *model.Broker, error)