event

Documentation

Index

• func KVFromProto(msg proto.Message) (string, []byte, error)
• type Event
  • func NewMessage(key string, value []byte) Event
  • func NewMessageFromProto(msg proto.Message) (Event, error)
• type Handler
  • func ChainHandler(h ...Handler) Handler
  • func FilterKeyHandler(key string, handler Handler) Handler
  • func ProtoHandler[T proto.Message](msg T, next HandlerOf[T]) Handler
  • func RecoverHandler(l klog.Logger, next Handler) Handler
  • func TransformHandler[T any](transformer TransformerOf[T], next HandlerOf[T]) Handler
  • func UowHandler(uowMgr uow.Manager, handler Handler) Handler
• type HandlerOf
• type Message
  • func (m *Message) Key() string
  • func (m *Message) Value() []byte
• type Receiver
• type Sender
• type TransformerOf

Functions

func KVFromProto

func KVFromProto(msg proto.Message) (string, []byte, error)

Types

type Event

type Event interface {
	Key() string
	Value() []byte
}

func NewMessage

func NewMessage(key string, value []byte) Event

func NewMessageFromProto

func NewMessageFromProto(msg proto.Message) (Event, error)

type Handler

type Handler func(context.Context, Event) error

func ChainHandler

func ChainHandler(h ...Handler) Handler

ChainHandler cmobine multiple handler one by one

func FilterKeyHandler

func FilterKeyHandler(key string, handler Handler) Handler

FilterKeyHandler filter event by key compare

func ProtoHandler

func ProtoHandler[T proto.Message](msg T, next HandlerOf[T]) Handler

func RecoverHandler

func RecoverHandler(l klog.Logger, next Handler) Handler

RecoverHandler wrap next with recover. prevent consumer panic

func TransformHandler

func TransformHandler[T any](transformer TransformerOf[T], next HandlerOf[T]) Handler

TransformHandler transform Event into type generic T

func UowHandler

func UowHandler(uowMgr uow.Manager, handler Handler) Handler

UowHandler wrap handler into a unit of work

type HandlerOf

type HandlerOf[T any] func(context.Context, T) error

type Message

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

func (*Message) Key

func (m *Message) Key() string

func (*Message) Value

func (m *Message) Value() []byte

type Receiver

type Receiver interface {
	Receive(ctx context.Context, handler Handler) error
	Close() error
}

type Sender

type Sender interface {
	Send(ctx context.Context, msg Event) error
	Close() error
}

type TransformerOf

type TransformerOf[T any] func(e Event) (T, error)