eventing

Documentation

Index

• Variables
• type CommitOverride
• type Consumer
• type ConsumerCallback
• type ConsumerCallbackAdapter
  • func (cb *ConsumerCallbackAdapter) Close() error
  • func (cb *ConsumerCallbackAdapter) DataReceived(msg Message) (err error)
  • func (cb *ConsumerCallbackAdapter) ErrorReceived(err error)
  • func (cb *ConsumerCallbackAdapter) ShouldFilter(m *Message) bool
  • func (cb *ConsumerCallbackAdapter) ShouldProcess(o interface{}) bool
  • func (cb *ConsumerCallbackAdapter) Stats(stats map[string]interface{})
• type ConsumerCallbackEventFilter
• type ConsumerCallbackMessageFilter
• type Message
  • func (m Message) Commit() error
  • func (m Message) IsAutoCommit() bool
• type Producer
• type ValueEncodingType

Variables

var ErrMessageNotAutoCommit = errors.New("message isn't auto commit")

ErrMessageNotAutoCommit is returned if you call commit on a message that isn't in auto commit mode

Types

type CommitOverride

type CommitOverride func(m Message) error

CommitOverride is called on msg.Commit

type Consumer

type Consumer interface {
	// Close will stop listening for events
	Close() error
	// Consume will start consuming from the consumer using the callback
	Consume(callback ConsumerCallback) error
}

Consumer will create a consumer for receiving events

type ConsumerCallback

type ConsumerCallback interface {
	// OnDataReceived is called when an event is received
	DataReceived(msg Message) error
	// OnErrorReceived is called when an error is received
	ErrorReceived(err error)
}

ConsumerCallback will receive events from producers

type ConsumerCallbackAdapter

type ConsumerCallbackAdapter struct {
	// OnDataReceived is called when an event is received
	OnDataReceived func(msg Message) error
	// OnErrorReceived is called when an error is received
	OnErrorReceived func(err error)
	// OnStats is called when topic stats is generated
	OnStats func(stats map[string]interface{})
	// OnShouldProcess is called before unmarshalling to allow the
	// consumer control over forwarding decisions
	OnShouldProcess func(o interface{}) bool
	// OnShouldFilter is called before forwarding to the consumer
	// to give the consumer control over filtering messages
	OnShouldFilter func(m *Message) bool
	// contains filtered or unexported fields
}

ConsumerCallbackAdapter will receive events from producers

func (*ConsumerCallbackAdapter) Close

func (cb *ConsumerCallbackAdapter) Close() error

func (*ConsumerCallbackAdapter) DataReceived

func (cb *ConsumerCallbackAdapter) DataReceived(msg Message) (err error)

func (*ConsumerCallbackAdapter) ErrorReceived

func (cb *ConsumerCallbackAdapter) ErrorReceived(err error)

func (*ConsumerCallbackAdapter) ShouldFilter

func (cb *ConsumerCallbackAdapter) ShouldFilter(m *Message) bool

func (*ConsumerCallbackAdapter) ShouldProcess

func (cb *ConsumerCallbackAdapter) ShouldProcess(o interface{}) bool

func (*ConsumerCallbackAdapter) Stats

func (cb *ConsumerCallbackAdapter) Stats(stats map[string]interface{})

type ConsumerCallbackEventFilter

type ConsumerCallbackEventFilter interface {
	ShouldFilter(m *Message) bool
}

ConsumerCallbackEventFilter is a filter for handling forwarding decisions after deserialization from kafka before deliver to the consumer callback. return true to forward or false to stop processing

type ConsumerCallbackMessageFilter

type ConsumerCallbackMessageFilter interface {
	ShouldProcess(o interface{}) bool
}

ConsumerCallbackMessageFilter is a filter for handling forwarding decisions after receiving from kafka before before unmarshalling NOTE: we explicitly use an object here which the implementer needs to cast as a *kafka.Message so as not to create an explicit dependency on the go confluent kafka library for this package where you don't need it (and that library is a bit of a pain because its Cgo and requires third-party library to compile)

type Message

type Message struct {
	Encoding  ValueEncodingType
	Key       string // Deprecated
	Value     []byte
	Headers   map[string]string
	Timestamp time.Time
	Extra     map[string]interface{} // Deprecated
	Topic     string
	Partition int32
	Offset    int64

	// internal, do not set
	CommitOverride CommitOverride `json:"-"`
	AutoCommit     bool           `json:"-"`
}

Message encapsulates data for the event system

func (Message) Commit

func (m Message) Commit() error

Commit is used to commit the processing of this event and store the offset

func (Message) IsAutoCommit

func (m Message) IsAutoCommit() bool

IsAutoCommit returns true if the message is automatically commit or false if you must call Commit when completed

type Producer

type Producer interface {
	// Send will send the event
	Send(ctx context.Context, msg Message) error
	// Close will close the producer
	Close() error
}

Producer will emit events to consumers

type ValueEncodingType

type ValueEncodingType string

ValueEncodingType is the type of encoding for the Value payload

const (
	// JSONEncoding is json encoding for the Value payload
	JSONEncoding ValueEncodingType = "json"
)