flow

Documentation

Index

• func Format() format.Format[Message[structure.Bytes, structure.Bytes]]
• func Produce[K any, V any](c context.Context, flowProducer Producer, topic Topic[K, V], m Message[K, V]) error
• type Message
  • func Append[K any, V any](outputs []Message[structure.Bytes, structure.Bytes], source *Message[K, V], ...) ([]Message[structure.Bytes, structure.Bytes], error)
  • 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[structure.Bytes, structure.Bytes]
• type MessageFormat
  • func (helper MessageFormat) Default() Message[structure.Bytes, structure.Bytes]
  • func (helper MessageFormat) Marshal(value Message[structure.Bytes, structure.Bytes]) (structure.Bytes, error)
  • func (helper MessageFormat) Unmarshal(value structure.Bytes) (Message[structure.Bytes, structure.Bytes], error)
• type Producer
• type Topic
  • func GenericTopic[K any, V any](topic string, keyFormat format.Format[K], valueFormat format.Format[V]) Topic[K, V]
  • func JsonTopic[K any, V any](topic string) Topic[K, V]
  • func ProtobufTopic[K proto.Message, V proto.Message](topic string) Topic[K, V]
  • func StringTopic(topic string) Topic[string, string]

Functions

func Format

func Format() format.Format[Message[structure.Bytes, structure.Bytes]]

func Produce

func Produce[K any, V any](
	c context.Context,
	flowProducer Producer,
	topic Topic[K, V],
	m Message[K, V],
) error

Types

type Message

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

func Append

func Append[K any, V any](
	outputs []Message[structure.Bytes, structure.Bytes],
	source *Message[K, V],
	outputTopic Topic[K, V],
) (
	[]Message[structure.Bytes, structure.Bytes],
	error,
)

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[structure.Bytes, structure.Bytes]

type MessageFormat

type MessageFormat struct{}

for byte based storage of a message

func (MessageFormat) Default

func (helper MessageFormat) Default() Message[structure.Bytes, structure.Bytes]

func (MessageFormat) Marshal

func (helper MessageFormat) Marshal(value Message[structure.Bytes, structure.Bytes]) (structure.Bytes, error)

func (MessageFormat) Unmarshal

func (helper MessageFormat) Unmarshal(value structure.Bytes) (Message[structure.Bytes, structure.Bytes], error)

type Producer

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

type Topic

type Topic[K any, V any] interface {
	Name() string
	KeyFormat() format.Format[K]
	ValueFormat() format.Format[V]
}

func GenericTopic

func GenericTopic[K any, V any](topic string, keyFormat format.Format[K], valueFormat format.Format[V]) Topic[K, V]

func JsonTopic

func JsonTopic[K any, V any](topic string) Topic[K, V]

func ProtobufTopic

func ProtobufTopic[K proto.Message, V proto.Message](topic string) Topic[K, V]

func StringTopic

func StringTopic(topic string) Topic[string, string]