client

Documentation

Index

• func AckMessages(client proto.MessageServiceClient, topicName string, id string, ...) error
• func ConsumeMessages(client proto.MessageServiceClient, topicName string, id string, ...) error
• func PublishMessages(client proto.MessageServiceClient, topicName string, partition uint64, ...) error
• type ConsumerGroup
• type JetClient
  • func New(address string) (*JetClient, error)
  • func (j *JetClient) ConsumeMessage(topicName string, id string) ([]*proto.Message, error)
  • func (j *JetClient) CreateConsumerGroup(topicName string) (*proto.CreateConsumerGroupResponse, error)
  • func (j *JetClient) CreateTopic(name string, partitions int) (*proto.CreateTopicResponse, error)
  • func (j *JetClient) PublishMessage(messages []*proto.KeyVal, topic string) (*proto.PublishMessageResponse, error)
• type KeyValuePair
• type MemberClient
• type Meta
• type PartitionMeta
  • func (p *PartitionMeta) String() string
• type ShardClient
  • func (s *ShardClient) GetLeader() *MemberClient
  • func (s *ShardClient) GetNextMember() *MemberClient
• type TopicMeta

Functions

func AckMessages

func AckMessages(client proto.MessageServiceClient, topicName string, id string, partitionMap map[uint64]uint64, channel chan *proto.AckConsumeResponse) error

func ConsumeMessages

func ConsumeMessages(client proto.MessageServiceClient, topicName string, id string, channel chan *proto.ConsumeResponse, offsets map[uint64]uint64) error

func PublishMessages

func PublishMessages(client proto.MessageServiceClient, topicName string, partition uint64, list []*proto.KeyVal, channel chan *proto.PublishMessageResponse) error

Types

type ConsumerGroup

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

type JetClient

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

func New

func New(address string) (*JetClient, error)

func (*JetClient) ConsumeMessage

func (j *JetClient) ConsumeMessage(topicName string, id string) ([]*proto.Message, error)

ConsumeMessage need to check if the consumer is created, if true then find

func (*JetClient) CreateConsumerGroup

func (j *JetClient) CreateConsumerGroup(topicName string) (*proto.CreateConsumerGroupResponse, error)

CreateConsumerGroup creates multiple consumers on each shard that contains the partitions of the topic, stores the id of each consumer

func (*JetClient) CreateTopic

func (j *JetClient) CreateTopic(name string, partitions int) (*proto.CreateTopicResponse, error)

func (*JetClient) PublishMessage

func (j *JetClient) PublishMessage(messages []*proto.KeyVal, topic string) (*proto.PublishMessageResponse, error)

PublishMessage need to get each partition's location within the cluster, then do consistent hashing with the key to get the partition needed

type KeyValuePair

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

type MemberClient

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

type Meta

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

type PartitionMeta

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

func (*PartitionMeta) String

func (p *PartitionMeta) String() string

type ShardClient

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

func (*ShardClient) GetLeader

func (s *ShardClient) GetLeader() *MemberClient

func (*ShardClient) GetNextMember

func (s *ShardClient) GetNextMember() *MemberClient

type TopicMeta

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