kafka

Documentation

Index

• Constants
• type AuthMethod
• type Module
  • func NewModule(stateStorage *state.Storage, appLogger *logrus.Logger) (*Module, error)
  • func (m *Module) Brokers(projectID string) (*TabBrokersData, error)
  • func (m *Module) Close() error
  • func (m *Module) Connect(projectID string) (*State, error)
  • func (m *Module) Consumers(projectID string) (*TabConsumersData, error)
  • func (m *Module) CreateNewProject(projectID string) (*State, error)
  • func (m *Module) DeleteProject(projectID string) error
  • func (m *Module) ProjectState(projectID string) (*State, error)
  • func (m *Module) SaveState(projectID string, state *State) (*State, error)
  • func (m *Module) StartTopicConsuming(projectID string, consumingStrategy TopicConsumingStrategy, ...) (*TopicOutput, error)
  • func (m *Module) StopTopicConsuming(projectID string) error
  • func (m *Module) Topics(projectID string) (*TabTopicsData, error)
• type Project
  • func NewProject(projectID string, stateStorage *state.Storage, appLogger *logrus.Logger) (*Project, error)
  • func (p *Project) Brokers() (*TabBrokersData, error)
  • func (p *Project) Close() error
  • func (p *Project) Connect() error
  • func (p *Project) Consumers() (*TabConsumersData, error)
  • func (p *Project) SaveState(state *State) error
  • func (p *Project) StartTopicConsuming(ctx context.Context, consumingStrategy TopicConsumingStrategy, ...) (*TopicOutput, error)
  • func (p *Project) StopTopicConsuming() error
  • func (p *Project) Topics() (*TabTopicsData, error)
• type State
• type Tab
• type TabBrokersData
• type TabBrokersDataBroker
• type TabConsumersData
• type TabConsumersDataConsumer
• type TabTopicsData
• type TabTopicsDataTopic
• type TopicConsumingOutput
• type TopicConsumingStrategy
• type TopicMessage
• type TopicOutput
• type TopicPartition

Constants

const (
	AuthMethodPlaintext = "plaintext"
	AuthMethodSASLSSL   = "sasl_ssl"
)

const (
	TabOverview  = "overview"
	TabBrokers   = "brokers"
	TabTopics    = "topics"
	TabConsumers = "consumers"
)

const (
	TopicConsumingStrategyTime           = "time"
	TopicConsumingStrategyOffsetSpecific = "offset_specific"
	TopicConsumingStrategyOffsetNewest   = "offset_newest"
	TopicConsumingStrategyOffsetOldest   = "offset_oldest"
)

Types

type AuthMethod

type AuthMethod string

type Module

type Module struct {
	AppCtx context.Context
	// contains filtered or unexported fields
}

func NewModule

func NewModule(stateStorage *state.Storage, appLogger *logrus.Logger) (*Module, error)

func (*Module) Brokers

func (m *Module) Brokers(projectID string) (*TabBrokersData, error)

func (*Module) Close

func (m *Module) Close() error

func (*Module) Connect

func (m *Module) Connect(projectID string) (*State, error)

func (*Module) Consumers

func (m *Module) Consumers(projectID string) (*TabConsumersData, error)

func (*Module) CreateNewProject

func (m *Module) CreateNewProject(projectID string) (*State, error)

func (*Module) DeleteProject

func (m *Module) DeleteProject(projectID string) error

func (*Module) ProjectState

func (m *Module) ProjectState(projectID string) (*State, error)

func (*Module) SaveState

func (m *Module) SaveState(projectID string, state *State) (*State, error)

func (*Module) StartTopicConsuming

func (m *Module) StartTopicConsuming(
	projectID string,
	consumingStrategy TopicConsumingStrategy,
	topic,
	timeFrom string,
	offsetValue int64,
) (*TopicOutput, error)

func (*Module) StopTopicConsuming

func (m *Module) StopTopicConsuming(projectID string) error

func (*Module) Topics

func (m *Module) Topics(projectID string) (*TabTopicsData, error)

type Project

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

func NewProject

func NewProject(projectID string, stateStorage *state.Storage, appLogger *logrus.Logger) (*Project, error)

func (*Project) Brokers

func (p *Project) Brokers() (*TabBrokersData, error)

func (*Project) Close

func (p *Project) Close() error

func (*Project) Connect

func (p *Project) Connect() error

func (*Project) Consumers

func (p *Project) Consumers() (*TabConsumersData, error)

func (*Project) SaveState

func (p *Project) SaveState(state *State) error

func (*Project) StartTopicConsuming

func (p *Project) StartTopicConsuming(
	ctx context.Context,
	consumingStrategy TopicConsumingStrategy,
	topic,
	timeFrom string,
	offsetValue int64,
) (*TopicOutput, error)

nolint: funlen, cyclop, gocognit

func (*Project) StopTopicConsuming

func (p *Project) StopTopicConsuming() error

func (*Project) Topics

func (p *Project) Topics() (*TabTopicsData, error)

nolint: funlen

type State

type State struct {
	ID           string     `json:"id"`
	Address      string     `json:"address"`
	AuthMethod   AuthMethod `json:"authMethod"`
	AuthUsername string     `json:"authUsername"`
	AuthPassword string     `json:"authPassword"`
	IsConnected  bool       `json:"isConnected"`
	CurrentTab   Tab        `json:"currentTab"`
}

type Tab

type Tab string

type TabBrokersData

type TabBrokersData struct {
	IsConnected bool                    `json:"isConnected"`
	Count       int                     `json:"count"`
	List        []*TabBrokersDataBroker `json:"list"`
}

type TabBrokersDataBroker

type TabBrokersDataBroker struct {
	ID   int    `json:"id"`
	Rack string `json:"rack"`
	Host string `json:"host"`
	Port int    `json:"port"`
}

type TabConsumersData

type TabConsumersData struct {
	IsConnected bool                        `json:"isConnected"`
	Count       int                         `json:"count"`
	List        []*TabConsumersDataConsumer `json:"list"`
}

type TabConsumersDataConsumer

type TabConsumersDataConsumer struct {
	Name  string `json:"name"`
	State string `json:"state"`
}

type TabTopicsData

type TabTopicsData struct {
	IsConnected bool                  `json:"isConnected"`
	Count       int                   `json:"count"`
	List        []*TabTopicsDataTopic `json:"list"`
}

type TabTopicsDataTopic

type TabTopicsDataTopic struct {
	Name           string `json:"name"`
	PartitionCount int    `json:"partitionCount"`
	MessageCount   int64  `json:"messageCount"`
}

type TopicConsumingOutput

type TopicConsumingOutput struct {
	Messages []*TopicMessage `json:"messages"`
}

type TopicConsumingStrategy

type TopicConsumingStrategy string

type TopicMessage

type TopicMessage struct {
	TimestampUnix      int64  `json:"timestampUnix"`
	TimestampFormatted string `json:"timestampFormatted"`
	PartitionID        int    `json:"partitionID"`
	Offset             int64  `json:"offset"`
	Key                string `json:"key"`
	Data               string `json:"data"`
	Headers            string `json:"headers"`
}

type TopicOutput

type TopicOutput struct {
	TopicName     string            `json:"topicName"`
	StartFromTime string            `json:"startFromTime"`
	CountTotal    int64             `json:"countTotal"`
	CountCurrent  int64             `json:"countCurrent"`
	Partitions    []*TopicPartition `json:"partitions"`
}

type TopicPartition

type TopicPartition struct {
	ID                 int   `json:"id"`
	OffsetTotalStart   int64 `json:"offsetTotalStart"`
	OffsetTotalEnd     int64 `json:"offsetTotalEnd"`
	OffsetCurrentStart int64 `json:"offsetCurrentStart"`
	OffsetCurrentEnd   int64 `json:"offsetCurrentEnd"`
}