message

func Copy ¶

func Copy(src *kafka.Message) func(*kafka.Message) error

Copy copies the Value and Key from a source message.

Headers are NOT copied.

func Header ¶

func Header(k, v string) func(*kafka.Message) error

Header adds or replaces a message header.

func Headers ¶

func Headers(h map[string]string) func(*kafka.Message) error

Headers adds or replaces message headers.

func JSON ¶

func JSON(v any) func(*kafka.Message) error

JSON sets the message value to a JSON representation of the given value.

func Key ¶

func Key(k string) func(*kafka.Message) error

Key sets the message key.

func Partition ¶

func Partition(p int32) func(*kafka.Message) error

Partition sets the message partition.

func String ¶

func String(v string) func(*kafka.Message) error

String sets the message value to the specified string.

func Topic ¶

func Topic[T any](t T) func(*kafka.Message) error

Topic sets the message topic.

func Value ¶

func Value(v []byte) func(*kafka.Message) error

Value sets the message value to a copy of the specified byte slice.