flow

package

v0.3.4 Latest Latest Go to latest Published: Jan 21, 2024 License: GPL-3.0 Imports: 8 Imported by: 5

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/hjwalt/flows

Links

Open Source Insights

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
type MessageFormat
type Producer
type Topic

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func Format ¶

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

func Produce ¶ added in v0.2.11

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 ¶ added in v0.2.8

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 ¶ added in v0.2.1

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

func GenericTopic ¶ added in v0.2.1

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

func JsonTopic ¶ added in v0.2.2

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

func ProtobufTopic ¶ added in v0.2.2

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

func StringTopic ¶ added in v0.2.2

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

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL