kafka

Documentation

Index

• type ConsumerArgs
• type ConsumerResult
• type CreatePartitionArgs
• type CreatePartitionResult
• type CreateTopicArgs
• type CreateTopicResult
• type DeleteMessageArgs
• type DeleteMessageResult
• type DeleteTopicArgs
• type DeleteTopicResult
• type DescribeConsumerGroupArgs
• type DescribeConsumerGroupResult
• type DescribeTopicArgs
• type DescribeTopicResult
• type ListConsumerGroupsArgs
• type ListConsumerGroupsResult
• type ListTopicsArgs
• type ListTopicsResult
• type Message
• type ProducerArgs
• type ProducerResult
• type Server
  • func GetServer() *Server
  • func (s *Server) Consumer(ctx *gin.Context, args *ConsumerArgs) (*ConsumerResult, *kerr.KafkaError)
  • func (s *Server) CreatePartition(ctx *gin.Context, args *CreatePartitionArgs) (*CreatePartitionResult, *kerr.KafkaError)
  • func (s *Server) CreateTopic(ctx *gin.Context, args *CreateTopicArgs) (*CreateTopicResult, *kerr.KafkaError)
  • func (s *Server) DeleteMessage(ctx *gin.Context, args *DeleteMessageArgs) (*DeleteMessageResult, *kerr.KafkaError)
  • func (s *Server) DeleteTopic(ctx *gin.Context, args *DeleteTopicArgs) (*DeleteTopicResult, *kerr.KafkaError)
  • func (s *Server) DescribeConsumerGroup(ctx *gin.Context, args *DescribeConsumerGroupArgs) (*DescribeConsumerGroupResult, *kerr.KafkaError)
  • func (s *Server) DescribeTopic(ctx *gin.Context, args *DescribeTopicArgs) (*DescribeTopicResult, *kerr.KafkaError)
  • func (s *Server) ListConsumerGroups(ctx *gin.Context, args *ListConsumerGroupsArgs) (*ListConsumerGroupsResult, *kerr.KafkaError)
  • func (s *Server) ListTopics(ctx *gin.Context, args *ListTopicsArgs) (*ListTopicsResult, *kerr.KafkaError)
  • func (s *Server) Producer(ctx *gin.Context, args *ProducerArgs) (*ProducerResult, *kerr.KafkaError)

Types

type ConsumerArgs

type ConsumerArgs struct {
	Topic         string
	Brokers       string
	ConsumerGroup string
	Count         int
	TimeWait      int
}

ConsumerArgs consumer args

type ConsumerResult

type ConsumerResult struct {
	Messages []*sarama.ConsumerMessage `json:"messages"`
}

ConsumerResult consumer result

type CreatePartitionArgs

type CreatePartitionArgs struct {
	Topic          string
	Brokers        string
	TotalPartition int
}

CreatePartitionArgs create partition args

type CreatePartitionResult

type CreatePartitionResult struct {
	Error error
}

CreatePartitionResult create partition result

type CreateTopicArgs

type CreateTopicArgs struct {
	Topic          string
	Brokers        string
	PartitionCount int
	ReplicaCount   int
	RetentionTime  int
}

CreateTopicArgs create topic args

type CreateTopicResult

type CreateTopicResult struct {
	Error error
}

CreateTopicResult create topic result

type DeleteMessageArgs

type DeleteMessageArgs struct {
	Topic       string
	Brokers     string
	EndOffset   int64
	PartitionID int
}

DeleteMessageArgs delete topic args

type DeleteMessageResult

type DeleteMessageResult struct {
	Error error
}

DeleteMessageResult delete topic result

type DeleteTopicArgs

type DeleteTopicArgs struct {
	Topic   string
	Brokers string
}

DeleteTopicArgs delete topic args

type DeleteTopicResult

type DeleteTopicResult struct {
	Error error
}

DeleteTopicResult delete topic result

type DescribeConsumerGroupArgs

type DescribeConsumerGroupArgs struct {
	ConsumerGroup string
	Brokers       string
}

DescribeConsumerGroupArgs args

type DescribeConsumerGroupResult

type DescribeConsumerGroupResult struct {
	*sarama.GroupDescription
}

DescribeConsumerGroupResult result

type DescribeTopicArgs

type DescribeTopicArgs struct {
	Topic   string
	Brokers string
}

DescribeTopicArgs describe an topic args

type DescribeTopicResult

type DescribeTopicResult struct {
	TotalPartition int              `json:"total_partition"`
	Partitions     []*partitionInfo `json:"partitions"`
}

DescribeTopicResult describe an topic result

type ListConsumerGroupsArgs

type ListConsumerGroupsArgs struct {
	Brokers string
}

ListConsumerGroupsArgs args

type ListConsumerGroupsResult

type ListConsumerGroupsResult struct {
	Groups []string `json:"groups"`
}

ListConsumerGroupsResult result

type ListTopicsArgs

type ListTopicsArgs struct {
	Brokers string
}

ListTopicsArgs args

type ListTopicsResult

type ListTopicsResult struct {
	Topics []string `json:"topics"`
}

ListTopicsResult result

type Message

type Message struct {
	Key   string `json:"key"`
	Value string `json:"value"`
}

type ProducerArgs

type ProducerArgs struct {
	Topic    string
	Brokers  string
	Messages []*Message
}

ProducerArgs producer args

type ProducerResult

type ProducerResult struct {
	Error error
}

ProducerResult producer result

type Server

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

Server for kafka

func GetServer

func GetServer() *Server

GetServer get kafka server

func (*Server) Consumer

func (s *Server) Consumer(ctx *gin.Context, args *ConsumerArgs) (*ConsumerResult, *kerr.KafkaError)

Consumer kafka topic message

func (*Server) CreatePartition

func (s *Server) CreatePartition(ctx *gin.Context, args *CreatePartitionArgs) (*CreatePartitionResult, *kerr.KafkaError)

CreatePartition create partition

func (*Server) CreateTopic

func (s *Server) CreateTopic(ctx *gin.Context, args *CreateTopicArgs) (*CreateTopicResult, *kerr.KafkaError)

CreateTopic create topic

func (*Server) DeleteMessage

func (s *Server) DeleteMessage(ctx *gin.Context, args *DeleteMessageArgs) (*DeleteMessageResult, *kerr.KafkaError)

DeleteMessage delete message

func (*Server) DeleteTopic

func (s *Server) DeleteTopic(ctx *gin.Context, args *DeleteTopicArgs) (*DeleteTopicResult, *kerr.KafkaError)

DeleteTopic delete topic

func (*Server) DescribeConsumerGroup

func (s *Server) DescribeConsumerGroup(ctx *gin.Context, args *DescribeConsumerGroupArgs) (*DescribeConsumerGroupResult, *kerr.KafkaError)

DescribeConsumerGroup describe consumer group

func (*Server) DescribeTopic

func (s *Server) DescribeTopic(ctx *gin.Context, args *DescribeTopicArgs) (*DescribeTopicResult, *kerr.KafkaError)

DescribeTopic describe topic

func (*Server) ListConsumerGroups

func (s *Server) ListConsumerGroups(ctx *gin.Context, args *ListConsumerGroupsArgs) (*ListConsumerGroupsResult, *kerr.KafkaError)

ListConsumerGroups list consumer groups

func (*Server) ListTopics

func (s *Server) ListTopics(ctx *gin.Context, args *ListTopicsArgs) (*ListTopicsResult, *kerr.KafkaError)

ListTopics list kafka all topics

func (*Server) Producer

func (s *Server) Producer(ctx *gin.Context, args *ProducerArgs) (*ProducerResult, *kerr.KafkaError)

Producer write 2 kafka