kafkaframe

package module

v1.0.0 Latest Latest Go to latest Published: May 31, 2023 License: MIT Imports: 9 Imported by: 0

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/covrom/kafkaframe

Links

Open Source Insights

README ¶

kafkaframe

Go and Kafka pubsub templates

Documentation ¶

Index ¶

type Config
type Kafka
- func NewKafka(cfg Config) *Kafka
- func (k *Kafka) OpenSubscription(group string, topics ...string) (*Subscription, error)
- func (k *Kafka) OpenTopic(topicName string) (*Topic, error)
type MessageHeader
- func (m MessageHeader) CreatedAt() time.Time
- func (m MessageHeader) Xid() xid.ID
type ReceivedMessage
type Subscription
type Topic

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

This section is empty.

Types ¶

type Config ¶

type Config struct {
	ClientIDPrefix string   `yaml:"clientIdPrefix"`
	Brokers        []string `yaml:"brokers"`
}

type Kafka ¶

type Kafka struct {
	Config    Config
	SaramaCfg *sarama.Config
}

func NewKafka ¶

func NewKafka(cfg Config) *Kafka

func (*Kafka) OpenSubscription ¶

func (k *Kafka) OpenSubscription(group string, topics ...string) (*Subscription, error)

func (*Kafka) OpenTopic ¶

func (k *Kafka) OpenTopic(topicName string) (*Topic, error)

type MessageHeader ¶

type MessageHeader struct {
	ID       string `json:"id"`
	Topic    string `json:"topic"`
	Source   string `json:"source"`   // source system (clientIdPrefix prefix from config)
	Producer string `json:"producer"` // SaramaCfg.ClientID
}

func (MessageHeader) CreatedAt ¶

func (m MessageHeader) CreatedAt() time.Time

func (MessageHeader) Xid ¶

func (m MessageHeader) Xid() xid.ID

type ReceivedMessage ¶

type ReceivedMessage struct {
	MessageHeader
	*pubsub.Message
}

type Subscription ¶

type Subscription struct {
	PubsubSubscription *pubsub.Subscription
	// contains filtered or unexported fields
}

func (*Subscription) Close ¶

func (t *Subscription) Close() error

func (*Subscription) ReceiveJSON ¶

func (t *Subscription) ReceiveJSON(ctx context.Context, dest interface{}) (*ReceivedMessage, error)

ReceiveJSON (blocking) and unmarshal from JSON a next message from subscription into 'dest' that must be a pointer. The Ack method of the returned ReceivedMessage must be called once the message has been processed, to prevent it from being received again.

func (*Subscription) ReceiveProtobuf ¶

func (t *Subscription) ReceiveProtobuf(ctx context.Context, dest proto.Message) (*ReceivedMessage, error)

ReceiveProtobuf (blocking) and unmarshal from proto a next message from subscription into 'dest' that must be a pointer. The Ack method of the returned ReceivedMessage must be called once the message has been processed, to prevent it from being received again.

type Topic ¶

type Topic struct {
	Topic       string
	PubsubTopic *pubsub.Topic
	// contains filtered or unexported fields
}

func (*Topic) Close ¶

func (t *Topic) Close() error

func (*Topic) SendJSON ¶

func (t *Topic) SendJSON(ctx context.Context, v interface{}) (*MessageHeader, error)

func (*Topic) SendProtobuf ¶

func (t *Topic) SendProtobuf(ctx context.Context, v proto.Message) (*MessageHeader, error)

Source Files ¶

View all Source files

Directories ¶

Path	Synopsis
examples
consumer
consumer/kfkempl
producer
producer/api
producer/etl
producer/hr
producer/kafkaemployees

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL