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
	Topic     string
	Key       []byte
	Value     []byte
	Offset    int64
	Partition int32
}

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)