shared

Documentation

Index

• Constants
• func AddSHeader(msg *KafkaMessage, key string, value string)
• func AddSXOrigin(msg *KafkaMessage)
• func AddSXTrace(msg *KafkaMessage) error
• func GetSHeader(msg *KafkaMessage, key string) (bool, string)
• func GetSXOrigin(msg *KafkaMessage) (bool, string)
• func IsInTrace(msg *KafkaMessage) bool
• func IsSameOrigin(msg *KafkaMessage) bool
• func ToConsumerMessage(message *KafkaMessage) *sarama.ConsumerMessage
• func ToProducerMessage(message *KafkaMessage) *sarama.ProducerMessage
• type KafkaMessage
  • func FromConsumerMessage(message *sarama.ConsumerMessage) *KafkaMessage
• type TraceValue
  • func GetSXTrace(msg *KafkaMessage) (bool, TraceValue)

Constants

const CycleTime = 100 * time.Millisecond

CycleTime is the default cycle time for loops. It is used in the consumer and producer.

Functions

func AddSHeader

func AddSHeader(msg *KafkaMessage, key string, value string)

func AddSXOrigin

func AddSXOrigin(msg *KafkaMessage)

func AddSXTrace

func AddSXTrace(msg *KafkaMessage) error

func GetSHeader

func GetSHeader(msg *KafkaMessage, key string) (bool, string)

func GetSXOrigin

func GetSXOrigin(msg *KafkaMessage) (bool, string)

func IsInTrace

func IsInTrace(msg *KafkaMessage) bool

func IsSameOrigin

func IsSameOrigin(msg *KafkaMessage) bool

func ToConsumerMessage

func ToConsumerMessage(message *KafkaMessage) *sarama.ConsumerMessage

ToConsumerMessage converts a KafkaMessage to a sarama.ConsumerMessage.

func ToProducerMessage

func ToProducerMessage(message *KafkaMessage) *sarama.ProducerMessage

ToProducerMessage converts a KafkaMessage to a sarama.ProducerMessage. It ignores the Partition and Offset fields and sets trace headers.

Types

type KafkaMessage

type KafkaMessage struct {
	Headers   map[string]string `json:"headers"`
	Topic     string            `json:"topic"`
	Key       []byte            `json:"key"`
	Value     []byte            `json:"value"`
	Offset    int64             `json:"offset"`
	Partition int32             `json:"partition"`
}

KafkaMessage represents a message in the Kafka queue.

func FromConsumerMessage

func FromConsumerMessage(message *sarama.ConsumerMessage) *KafkaMessage

FromConsumerMessage converts a sarama.ConsumerMessage to a KafkaMessage.

type TraceValue

type TraceValue struct {
	Traces map[int64]string `json:"trace"`
}

func GetSXTrace

func GetSXTrace(msg *KafkaMessage) (bool, TraceValue)