message

Documentation

Index

• func Format() format.Format[Message[[]byte, []byte]]
• type Bytes
• type Message
  • func Convert[K1 any, V1 any, K2 any, V2 any](source Message[K1, V1], k1 format.Format[K1], v1 format.Format[V1], ...) (Message[K2, V2], error)
  • func EmptySlice() []Message[Bytes, Bytes]
• type MessageFormat
  • func (helper MessageFormat) Default() Message[[]byte, []byte]
  • func (helper MessageFormat) Marshal(value Message[[]byte, []byte]) ([]byte, error)
  • func (helper MessageFormat) Unmarshal(value []byte) (Message[[]byte, []byte], error)
• type Producer

Functions

func Format

func Format() format.Format[Message[[]byte, []byte]]

Types

type Bytes

type Bytes = []byte

type Message

type Message[K any, V any] struct {
	Topic     string
	Partition int32
	Offset    int64
	Key       K
	Value     V
	Headers   map[string][]Bytes
	Timestamp time.Time
}

func Convert

func Convert[K1 any, V1 any, K2 any, V2 any](
	source Message[K1, V1],
	k1 format.Format[K1],
	v1 format.Format[V1],
	k2 format.Format[K2],
	v2 format.Format[V2],
) (Message[K2, V2], error)

assuming byte compatibility, i.e. bytes <-> proto, string <-> json

func EmptySlice

func EmptySlice() []Message[Bytes, Bytes]

type MessageFormat

type MessageFormat struct{}

for byte based storage of a message

func (MessageFormat) Default

func (helper MessageFormat) Default() Message[[]byte, []byte]

func (MessageFormat) Marshal

func (helper MessageFormat) Marshal(value Message[[]byte, []byte]) ([]byte, error)

func (MessageFormat) Unmarshal

func (helper MessageFormat) Unmarshal(value []byte) (Message[[]byte, []byte], error)

type Producer

type Producer interface {
	Produce(context.Context, []Message[Bytes, Bytes]) error
	Start() error
	Stop()
}