inputs for multiple topics

graph.go

@@ -3,9 +3,11 @@ package goka
 import (
 	"errors"
 	"fmt"
+	"strings"
 )
 
 type Stream string
+type Streams []Stream
 type Table string
 type Group string
 
@@ -88,6 +90,13 @@ func DefineGroup(group Group, edges ...Edge) *GroupGraph {
 
 	for _, e := range edges {
 		switch e := e.(type) {
+		case inputStreams:
+			for _, input := range e {
+				inputStr := input.(*inputStream)
+				gg.codecs[input.Topic()] = input.Codec()
+				gg.callbacks[input.Topic()] = inputStr.cb
+				gg.inputStreams = append(gg.inputStreams, inputStr)
+			}
 		case *inputStream:
 			gg.codecs[e.Topic()] = e.Codec()
 			gg.callbacks[e.Topic()] = e.cb
@@ -182,6 +191,48 @@ 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 for 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 defines a consume callback on the loop topic

graph_test.go

@@ -2,6 +2,7 @@ package goka
 
 import (
 	"reflect"
+	"strings"
 	"testing"
 
 	"github.com/facebookgo/ensure"
@@ -75,18 +76,27 @@ func TestGroupGraph_Validate(t *testing.T) {
 
 func TestGroupGraph_codec(t *testing.T) {
 	g := DefineGroup("group",
-		Input("input-topic", c, cb))
+		Input("input-topic", c, cb),
+		Inputs(Streams{"input-topic2", "input-topic3"}, c, cb),
+	)
+
+	for _, topic := range []string{"input-topic", "input-topic2", "input-topic3"} {
+		codec := g.codec(topic)
+		ensure.DeepEqual(t, codec, c)
+	}
 
-	codec := g.codec("input-topic")
-	ensure.DeepEqual(t, codec, c)
 }
 
 func TestGroupGraph_callback(t *testing.T) {
 	g := DefineGroup("group",
-		Input("input-topic", c, cb))
+		Input("input-topic", c, cb),
+		Inputs(Streams{"input-topic2", "input-topic3"}, c, cb),
+	)
 
-	callback := g.callback("input-topic")
-	ensure.True(t, reflect.ValueOf(callback).Pointer() == reflect.ValueOf(cb).Pointer())
+	for _, topic := range []string{"input-topic", "input-topic2", "input-topic3"} {
+		callback := g.callback(topic)
+		ensure.True(t, reflect.ValueOf(callback).Pointer() == reflect.ValueOf(cb).Pointer())
+	}
 }
 
 func TestGroupGraph_getters(t *testing.T) {
@@ -96,9 +106,10 @@ func TestGroupGraph_getters(t *testing.T) {
 		Output("t3", c),
 		Output("t4", c),
 		Output("t5", c),
+		Inputs(Streams{"t6", "t7"}, c, cb),
 	)
 	ensure.True(t, g.Group() == "group")
-	ensure.True(t, len(g.InputStreams()) == 2)
+	ensure.True(t, len(g.InputStreams()) == 4)
 	ensure.True(t, len(g.OutputStreams()) == 3)
 	ensure.True(t, g.LoopStream() == nil)
 
@@ -136,3 +147,10 @@ func TestGroupGraph_getters(t *testing.T) {
 	ensure.True(t, len(g.LookupTables()) == 2)
 	ensure.DeepEqual(t, g.GroupTable().Topic(), tableName("group"))
 }
+
+func TestGroupGraph_Inputs(t *testing.T) {
+
+	topics := Inputs(Streams{"a", "b", "c"}, c, cb)
+	ensure.DeepEqual(t, topics.Topic(), "a,b,c")
+	ensure.True(t, strings.Contains(topics.String(), "a,b,c/*codec.String"))
+}