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 Flags) 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 DefaultMessageDeserializer
  • func (deserializer DefaultMessageDeserializer) CanDeserialize(_ string) (bool, error)
  • func (deserializer DefaultMessageDeserializer) Deserialize(rawMsg *sarama.ConsumerMessage, flags Flags) error
• type Flags
• type MessageDeserializer
• type MessageDeserializerChain
  • func (deserializer MessageDeserializerChain) CanDeserialize(topic string) (bool, error)
  • func (deserializer MessageDeserializerChain) Deserialize(msg *sarama.ConsumerMessage, flags Flags) error
• type Operation
  • func (operation *Operation) Consume(topic string, flags Flags) error
• type ProtobufMessageDeserializer
  • func CreateProtobufMessageDeserializer(context protobuf.SearchContext, keyType, valueType string) (*ProtobufMessageDeserializer, error)
  • func (deserializer ProtobufMessageDeserializer) CanDeserialize(_ string) (bool, error)
  • func (deserializer ProtobufMessageDeserializer) Deserialize(rawMsg *sarama.ConsumerMessage, flags Flags) error

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 Flags) 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 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 Flags) error

type Flags

type Flags 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
	ProtoFiles       []string
	ProtoImportPaths []string
	ProtosetFiles    []string
	KeyProtoType     string
	ValueProtoType   string
}

type MessageDeserializer

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

type MessageDeserializerChain

type MessageDeserializerChain []MessageDeserializer

func (MessageDeserializerChain) CanDeserialize

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

func (MessageDeserializerChain) Deserialize

func (deserializer MessageDeserializerChain) Deserialize(msg *sarama.ConsumerMessage, flags Flags) error

type Operation

type Operation struct {
}

func (*Operation) Consume

func (operation *Operation) Consume(topic string, flags Flags) error

type ProtobufMessageDeserializer

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

func CreateProtobufMessageDeserializer

func CreateProtobufMessageDeserializer(context protobuf.SearchContext, keyType, valueType string) (*ProtobufMessageDeserializer, error)

func (ProtobufMessageDeserializer) CanDeserialize

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

func (ProtobufMessageDeserializer) Deserialize

func (deserializer ProtobufMessageDeserializer) Deserialize(rawMsg *sarama.ConsumerMessage, flags Flags) error