kafka

Documentation

Index

• func RandStringBytes(n int) string
• type Client
  • func NewKafkaClient(cfg *common_utils.BaseConfig) *Client
  • func (k *Client) Close() error
  • func (k *Client) CreateTopic(topic string, numPart int) error
  • func (k *Client) IsReaderConnected() bool
  • func (k *Client) IsWriters() bool
  • func (k *Client) ListTopics() []kafka.Topic
  • func (k *Client) Listen(f HandlerFunc) error
  • func (k *Client) NewConsumer()
  • func (k *Client) NewPublisher() error
  • func (k *Client) Publish(ctx context.Context, topic string, event Event) error
  • func (k *Client) PublishWithTracer(ctx context.Context, topic string, event Event) error
• type Event
  • func NewEvent(eventType EventType, data []byte, metadata ...[]byte) *Event
  • func (e *Event) GetData() []byte
  • func (e *Event) GetEventID() string
  • func (e *Event) GetEventType() EventType
  • func (e *Event) GetJsonData(data any) error
  • func (e *Event) GetJsonMetadata(metaData any) error
  • func (e *Event) GetMetadata() []byte
  • func (e *Event) GetTimestamp() time.Time
  • func (e *Event) GetVersion() uint64
  • func (e *Event) SetMetadata(metaData any) error
  • func (e *Event) String() string
• type EventType
• type HandlerFunc
• type IClient
• type Message

Functions

func RandStringBytes

func RandStringBytes(n int) string

Types

type Client

type Client struct {
	Backoff backoff.BackOff
	// contains filtered or unexported fields
}

func NewKafkaClient

func NewKafkaClient(cfg *common_utils.BaseConfig) *Client

func (*Client) Close

func (k *Client) Close() error

func (*Client) CreateTopic

func (k *Client) CreateTopic(topic string, numPart int) error

func (*Client) IsReaderConnected

func (k *Client) IsReaderConnected() bool

func (*Client) IsWriters

func (k *Client) IsWriters() bool

func (*Client) ListTopics

func (k *Client) ListTopics() []kafka.Topic

func (*Client) Listen

func (k *Client) Listen(f HandlerFunc) error

Listen manual listen need call msg.Commit() when process done recommend for this process

func (*Client) NewConsumer

func (k *Client) NewConsumer()

func (*Client) NewPublisher

func (k *Client) NewPublisher() error

func (*Client) Publish

func (k *Client) Publish(ctx context.Context, topic string, event Event) error

func (*Client) PublishWithTracer

func (k *Client) PublishWithTracer(ctx context.Context, topic string, event Event) error

type Event

type Event struct {
	EventID   string
	EventType EventType
	Version   uint64
	Data      []byte
	Metadata  []byte
	Timestamp time.Time
}

Event is an internal representation of an event, returned when the Aggregate uses NewEvent to create a new event. The events loaded from the db is represented by each DBs internal event type, implementing Event.

func NewEvent

func NewEvent(eventType EventType, data []byte, metadata ...[]byte) *Event

NewEvent creates a new event, with the given aggregateID, eventType and data. The eventID is generated automatically, and the version is set to 0.

func (*Event) GetData

func (e *Event) GetData() []byte

func (*Event) GetEventID

func (e *Event) GetEventID() string

func (*Event) GetEventType

func (e *Event) GetEventType() EventType

func (*Event) GetJsonData

func (e *Event) GetJsonData(data any) error

GetJsonData json unmarshal data attached to the Event.

func (*Event) GetJsonMetadata

func (e *Event) GetJsonMetadata(metaData any) error

GetJsonMetadata unmarshal app-specific metadata serialized as json for the Event.

func (*Event) GetMetadata

func (e *Event) GetMetadata() []byte

func (*Event) GetTimestamp

func (e *Event) GetTimestamp() time.Time

func (*Event) GetVersion

func (e *Event) GetVersion() uint64

func (*Event) SetMetadata

func (e *Event) SetMetadata(metaData any) error

func (*Event) String

func (e *Event) String() string

type EventType

type EventType string

EventType is the type of any event, used as its unique identifier.

type HandlerFunc

type HandlerFunc func(context.Context, *Message) error

type IClient

type IClient interface {
	Listen(ctx context.Context, f HandlerFunc) error
	NewConsumer()
	IsWriters() bool
	Close() error

	NewPublisher() error
	Publish(ctx context.Context, topic string, msg any) error
	PublishWithTracer(ctx context.Context, topic string, msg any) error

	IsReaderConnected() bool
	// contains filtered or unexported methods
}

type Message

type Message struct {
	Offset        int64  `json:"offset,omitempty"`
	Partition     int    `json:"partition,omitempty"`
	Topic         string `json:"topic,omitempty"`
	Key           string `json:"key,omitempty"`
	Body          []byte `json:"body,omitempty"`
	Timestamp     int64  `json:"timestamp,omitempty"`
	ConsumerGroup string `json:"consumer_group,omitempty"`
	Retry         int    `json:"retry,omitempty"`
	Commit        func() error
	MoveToDLQ     func() error
	Headers       map[string]string
}

Message define message encode/decode sarama message