metadata

Documentation

Index

• Constants
• Variables
• func MessageFromMessage(consumed *sarama.ConsumerMessage) *commander.Message
• func MessageToMessage(produce *commander.Message) *sarama.ProducerMessage
• func NewKafkaContext(ctx context.Context, info Kafka) context.Context
• type Kafka
  • func KafkaFromContext(ctx context.Context) (info Kafka, ok bool)
• type Key
  • func (k Key) String() string

Constants

const (
	// CtxKafka represents the kafka context type
	CtxKafka = Key("kafka")
)

Variables

var (
	HeaderID              = "x-id"
	HeaderAction          = "x-action"
	HeaderStatusCode      = "x-status-code"
	HeaderVersion         = "x-version"
	HeaderParentID        = "x-parent-id"
	HeaderParentTimestamp = "x-parent-timestamp"
)

Kafka message headers

Functions

func MessageFromMessage

func MessageFromMessage(consumed *sarama.ConsumerMessage) *commander.Message

MessageFromMessage attempts to construct a commander message of the given sarama consumer message

func MessageToMessage

func MessageToMessage(produce *commander.Message) *sarama.ProducerMessage

MessageToMessage attempts to construct a sarama consumer message of the given commander message.

func NewKafkaContext

func NewKafkaContext(ctx context.Context, info Kafka) context.Context

NewKafkaContext creates a new context with Retries attached. If used NewKafkaContext will overwrite any previously-appended metadata.

Types

type Kafka

type Kafka struct {
	Offset    int64
	Partition int32
}

Kafka message metadata

func KafkaFromContext

func KafkaFromContext(ctx context.Context) (info Kafka, ok bool)

KafkaFromContext returns the Kafka in ctx if it exists. The returned Kafka should not be modified. Writing to it may cause races. Modification should be made to copies of the returned Kafka.

type Key

type Key string

Key typed context key

func (Key) String

func (k Key) String() string