kafkaframe

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
  • func (t *Subscription) Close() error
  • func (t *Subscription) ReceiveJSON(ctx context.Context, dest interface{}) (*ReceivedMessage, error)
  • func (t *Subscription) ReceiveProtobuf(ctx context.Context, dest proto.Message) (*ReceivedMessage, error)
• type Topic
  • func (t *Topic) Close() error
  • func (t *Topic) SendJSON(ctx context.Context, v interface{}) (*MessageHeader, error)
  • func (t *Topic) SendProtobuf(ctx context.Context, v proto.Message) (*MessageHeader, error)

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)