graph.go

package goka

import (
	"errors"
	"fmt"
	"strings"
)

var (
	tableSuffix = "-table"
	loopSuffix  = "-loop"
)

// Stream is the name of an event stream topic in Kafka, ie, a topic with
// cleanup.policy=delete
type Stream string

// Streams is a slice of Stream names.
type Streams []Stream

// Table is the name of a table topic in Kafka, ie, a topic with
// cleanup.policy=compact
type Table string

// Group is the name of a consumer group in Kafka and represents a processor
// group in Goka. A processor group may have a group table and a group loopback
// stream. By default, the group table is named <group>-table and the loopback
// stream <group>-loop.
type Group string

// GroupGraph is the specification of a processor group. It contains all input,
// output, and any other topic from which and into which the processor group
// may consume or produce events. Each of these links to Kafka is called Edge.
type GroupGraph struct {
	group         string
	inputTables   []Edge
	crossTables   []Edge
	inputStreams  []Edge
	outputStreams []Edge
	loopStream    []Edge
	groupTable    []Edge

	codecs    map[string]Codec
	callbacks map[string]ProcessCallback

	joinCheck map[string]bool
}

// Group returns the group name.
func (gg *GroupGraph) Group() Group {
	return Group(gg.group)
}

// InputStreams returns all input stream edges of the group.
func (gg *GroupGraph) InputStreams() Edges {
	return gg.inputStreams
}

// JointTables retuns all joint table edges of the group.
func (gg *GroupGraph) JointTables() Edges {
	return gg.inputTables
}

// LookupTables retuns all lookup table edges  of the group.
func (gg *GroupGraph) LookupTables() Edges {
	return gg.crossTables
}

// LoopStream returns the loopback edge of the group.
func (gg *GroupGraph) LoopStream() Edge {
	// only 1 loop stream is valid
	if len(gg.loopStream) > 0 {
		return gg.loopStream[0]
	}
	return nil
}

// GroupTable returns the group table edge of the group.
func (gg *GroupGraph) GroupTable() Edge {
	// only 1 group table is valid
	if len(gg.groupTable) > 0 {
		return gg.groupTable[0]
	}
	return nil
}

// OutputStreams returns the output stream edges of the group.
func (gg *GroupGraph) OutputStreams() Edges {
	return gg.outputStreams
}

// inputs returns all input topics (tables and streams)
func (gg *GroupGraph) inputs() Edges {
	return append(append(gg.inputStreams, gg.inputTables...), gg.crossTables...)
}

// copartitioned returns all copartitioned topics (joint tables and input streams)
func (gg *GroupGraph) copartitioned() Edges {
	return append(gg.inputStreams, gg.inputTables...)
}

func (gg *GroupGraph) codec(topic string) Codec {
	return gg.codecs[topic]
}

func (gg *GroupGraph) callback(topic string) ProcessCallback {
	return gg.callbacks[topic]
}

func (gg *GroupGraph) joint(topic string) bool {
	return gg.joinCheck[topic]
}

// DefineGroup creates a group graph with a given group name and a list of
// edges.
func DefineGroup(group Group, edges ...Edge) *GroupGraph {
	gg := GroupGraph{group: string(group),
		codecs:    make(map[string]Codec),
		callbacks: make(map[string]ProcessCallback),
		joinCheck: make(map[string]bool),
	}

	for _, e := range edges {
		switch e := e.(type) {
		case inputStreams:
			for _, input := range e {
				gg.validateInputTopic(input.Topic())
				inputStr := input.(*inputStream)
				gg.codecs[input.Topic()] = input.Codec()
				gg.callbacks[input.Topic()] = inputStr.cb
				gg.inputStreams = append(gg.inputStreams, inputStr)
			}
		case *inputStream:
			gg.validateInputTopic(e.Topic())
			gg.codecs[e.Topic()] = e.Codec()
			gg.callbacks[e.Topic()] = e.cb
			gg.inputStreams = append(gg.inputStreams, e)
		case *loopStream:
			e.setGroup(group)
			gg.codecs[e.Topic()] = e.Codec()
			gg.callbacks[e.Topic()] = e.cb
			gg.loopStream = append(gg.loopStream, e)
		case *outputStream:
			gg.codecs[e.Topic()] = e.Codec()
			gg.outputStreams = append(gg.outputStreams, e)
		case *inputTable:
			gg.codecs[e.Topic()] = e.Codec()
			gg.inputTables = append(gg.inputTables, e)
			gg.joinCheck[e.Topic()] = true
		case *crossTable:
			gg.codecs[e.Topic()] = e.Codec()
			gg.crossTables = append(gg.crossTables, e)
		case *groupTable:
			e.setGroup(group)
			gg.codecs[e.Topic()] = e.Codec()
			gg.groupTable = append(gg.groupTable, e)
		}
	}
	return &gg
}

func (gg *GroupGraph) validateInputTopic(topic string) {
	if topic == "" {
		panic("Input topic cannot be empty. This will not work.")
	}

	if _, exists := gg.callbacks[topic]; exists {
		panic(fmt.Errorf("Callback for topic %s already exists. It is illegal to consume a topic twice", topic))
	}
}

// Validate validates the group graph and returns an error if invalid.
// Main validation checks are:
// - at most one loopback stream edge is allowed
// - at most one group table edge is allowed
// - at least one input stream is required
// - table and loopback topics cannot be used in any other edge.
func (gg *GroupGraph) Validate() error {
	if len(gg.loopStream) > 1 {
		return errors.New("more than one loop stream in group graph")
	}
	if len(gg.groupTable) > 1 {
		return errors.New("more than one group table in group graph")
	}
	if len(gg.inputStreams) == 0 {
		return errors.New("no input stream in group graph")
	}
	for _, t := range append(gg.outputStreams,
		append(gg.inputStreams, append(gg.inputTables, gg.crossTables...)...)...) {
		if t.Topic() == loopName(gg.Group()) {
			return errors.New("should not directly use loop stream")
		}
		if t.Topic() == tableName(gg.Group()) {
			return errors.New("should not directly use group table")
		}
	}
	return nil
}

// Edge represents a topic in Kafka and the corresponding codec to encode and
// decode the messages of that topic.
type Edge interface {
	String() string
	Topic() string
	Codec() Codec
}

// Edges is a slice of edge objects.
type Edges []Edge

// Topics returns the names of the topics of the edges.
func (e Edges) Topics() []string {
	var t []string
	for _, i := range e {
		t = append(t, i.Topic())
	}
	return t
}

type topicDef struct {
	name  string
	codec Codec
}

func (t *topicDef) Topic() string {
	return t.name
}

func (t *topicDef) String() string {
	return fmt.Sprintf("%s/%T", t.name, t.codec)
}

func (t *topicDef) Codec() Codec {
	return t.codec
}

type inputStream struct {
	*topicDef
	cb ProcessCallback
}

// Input represents an edge of an input stream topic. The edge
// specifies the topic name, its codec and the ProcessorCallback used to
// process it. The topic has to be copartitioned with any other input stream of
// the group and with the group table.
// The group starts reading the topic from the newest offset.
func Input(topic Stream, c Codec, cb ProcessCallback) Edge {
	return &inputStream{&topicDef{string(topic), c}, cb}
}

type inputStreams Edges

func (is inputStreams) String() string {
	if is == nil {
		return "empty input streams"
	}

	return fmt.Sprintf("input streams: %s/%T", is.Topic(), is.Codec())
}

func (is inputStreams) Topic() string {
	if is == nil {
		return ""
	}
	var topics []string

	for _, stream := range is {
		topics = append(topics, stream.Topic())
	}
	return strings.Join(topics, ",")
}

func (is inputStreams) Codec() Codec {
	if is == nil {
		return nil
	}
	return is[0].Codec()
}

// Inputs creates edges of multiple input streams sharing the same
// codec and callback.
func Inputs(topics Streams, c Codec, cb ProcessCallback) Edge {
	if len(topics) == 0 {
		return nil
	}
	var edges Edges
	for _, topic := range topics {
		edges = append(edges, Input(topic, c, cb))
	}
	return inputStreams(edges)
}

type loopStream inputStream

// Loop represents the edge of the loopback topic of the group. The edge
// specifies the codec of the messages in the topic and ProcesCallback to
// process the messages of the topic. Context.Loopback() is used to write
// messages into this topic from any callback of the group.
func Loop(c Codec, cb ProcessCallback) Edge {
	return &loopStream{&topicDef{codec: c}, cb}
}

func (s *loopStream) setGroup(group Group) {
	s.topicDef.name = loopName(group)
}

type inputTable struct {
	*topicDef
}

// Join represents an edge of a copartitioned, log-compacted table topic. The
// edge specifies the topic name and the codec of the messages of the topic.
// The group starts reading the topic from the oldest offset.
// The processing of input streams is blocked until all partitions of the table
// are recovered.
func Join(topic Table, c Codec) Edge {
	return &inputTable{&topicDef{string(topic), c}}
}

type crossTable struct {
	*topicDef
}

// Lookup represents an edge of a non-copartitioned, log-compacted table
// topic. The edge specifies the topic name and the codec of the messages of
// the topic.  The group starts reading the topic from the oldest offset.
// The processing of input streams is blocked until the table is fully
// recovered.
func Lookup(topic Table, c Codec) Edge {
	return &crossTable{&topicDef{string(topic), c}}
}

type groupTable struct {
	*topicDef
}

// Persist represents the edge of the group table, which is log-compacted and
// copartitioned with the input streams. This edge specifies the codec of the
// messages in the topic, ie, the codec of the values of the table.
// The processing of input streams is blocked until all partitions of the group
// table are recovered.
func Persist(c Codec) Edge {
	return &groupTable{&topicDef{codec: c}}
}

func (t *groupTable) setGroup(group Group) {
	t.topicDef.name = string(GroupTable(group))
}

type outputStream struct {
	*topicDef
}

// Output represents an edge of an output stream topic. The edge
// specifies the topic name and the codec of the messages of the topic.
// Context.Emit() only emits messages into Output edges defined in the group
// graph.
// The topic does not have to be copartitioned with the input streams.
func Output(topic Stream, c Codec) Edge {
	return &outputStream{&topicDef{string(topic), c}}
}

// GroupTable returns the name of the group table of group.
func GroupTable(group Group) Table {
	return Table(tableName(group))
}

func tableName(group Group) string {
	return string(group) + tableSuffix
}

// loopName returns the name of the loop topic of group.
func loopName(group Group) string {
	return string(group) + loopSuffix
}