consumer

Documentation

Index

• Constants
• type AvroMessageDeserializer
  • func CreateAvroMessageDeserializer(topic string, avroSchemaRegistry string) (AvroMessageDeserializer, error)
  • func (deserializer AvroMessageDeserializer) CanDeserialize(topic string) (bool, error)
  • func (deserializer AvroMessageDeserializer) Deserialize(rawMsg *sarama.ConsumerMessage, flags ConsumerFlags) error
• type CachingSchemaRegistry
  • func CreateCachingSchemaRegistry(avroSchemaRegistry string) (*CachingSchemaRegistry, error)
  • func (registry *CachingSchemaRegistry) GetSchemaByID(id int) (string, error)
  • func (registry *CachingSchemaRegistry) Subjects() ([]string, error)
• type ConsumedMessage
• type ConsumerFlags
• type ConsumerOperation
  • func (operation *ConsumerOperation) Consume(topic string, flags ConsumerFlags) error
• type DefaultMessageDeserializer
  • func (deserializer DefaultMessageDeserializer) CanDeserialize(_ string) (bool, error)
  • func (deserializer DefaultMessageDeserializer) Deserialize(rawMsg *sarama.ConsumerMessage, flags ConsumerFlags) error
• type MessageDeserializer

Constants

const (
	HEX    = "hex"
	BASE64 = "base64"
	NONE   = "none"
)

Types

type AvroMessageDeserializer

type AvroMessageDeserializer struct {
	// contains filtered or unexported fields
}

func CreateAvroMessageDeserializer

func CreateAvroMessageDeserializer(topic string, avroSchemaRegistry string) (AvroMessageDeserializer, error)

func (AvroMessageDeserializer) CanDeserialize

func (deserializer AvroMessageDeserializer) CanDeserialize(topic string) (bool, error)

func (AvroMessageDeserializer) Deserialize

func (deserializer AvroMessageDeserializer) Deserialize(rawMsg *sarama.ConsumerMessage, flags ConsumerFlags) error

type CachingSchemaRegistry

type CachingSchemaRegistry struct {
	// contains filtered or unexported fields
}

func CreateCachingSchemaRegistry

func CreateCachingSchemaRegistry(avroSchemaRegistry string) (*CachingSchemaRegistry, error)

func (*CachingSchemaRegistry) GetSchemaByID

func (registry *CachingSchemaRegistry) GetSchemaByID(id int) (string, error)

func (*CachingSchemaRegistry) Subjects

func (registry *CachingSchemaRegistry) Subjects() ([]string, error)

type ConsumedMessage

type ConsumedMessage struct {
	Partition int32
	Offset    int64
	Key       []byte
	Value     []byte
	Timestamp *time.Time
}

type ConsumerFlags

type ConsumerFlags struct {
	PrintKeys       bool
	PrintTimestamps bool
	PrintAvroSchema bool
	PrintHeaders    bool
	OutputFormat    string
	Separator       string
	Partitions      []int
	Offsets         []string
	FromBeginning   bool
	Tail            int
	Exit            bool
	EncodeValue     string
	EncodeKey       string
}

type ConsumerOperation

type ConsumerOperation struct {
}

func (*ConsumerOperation) Consume

func (operation *ConsumerOperation) Consume(topic string, flags ConsumerFlags) error

type DefaultMessageDeserializer

type DefaultMessageDeserializer struct {
}

func (DefaultMessageDeserializer) CanDeserialize

func (deserializer DefaultMessageDeserializer) CanDeserialize(_ string) (bool, error)

func (DefaultMessageDeserializer) Deserialize

func (deserializer DefaultMessageDeserializer) Deserialize(rawMsg *sarama.ConsumerMessage, flags ConsumerFlags) error

type MessageDeserializer

type MessageDeserializer interface {
	CanDeserialize(topic string) (bool, error)
	Deserialize(msg *sarama.ConsumerMessage, flags ConsumerFlags) error
}