Avoid brokers when unnecessary

redpanda-data · Jan 17, 2018 · fe7ab2f · fe7ab2f
1 parent 27fde7d
commit fe7ab2f
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 0 deletions.
diff --git a/lib/input/fan_in.go b/lib/input/fan_in.go
@@ -238,6 +238,12 @@ func NewFanIn(
 		return nil, err
 	}
 
+	if len(inputConfs) == 0 {
+		return nil, ErrFanInNoInputs
+	} else if len(inputConfs) == 1 {
+		return New(inputConfs[0], log, stats, pipelines...)
+	}
+
 	inputs := make([]types.Producer, len(inputConfs))
 
 	for i, iConf := range inputConfs {

diff --git a/lib/output/fan_out.go b/lib/output/fan_out.go
@@ -85,6 +85,12 @@ func NewFanOut(conf Config, log log.Modular, stats metrics.Type) (Type, error) {
 		return nil, err
 	}
 
+	if len(outputConfs) == 0 {
+		return nil, ErrFanOutNoOutputs
+	} else if len(outputConfs) == 1 {
+		return New(outputConfs[0], log, stats)
+	}
+
 	outputs := make([]types.Consumer, len(outputConfs))
 
 	for i, oConf := range outputConfs {

diff --git a/lib/output/round_robin.go b/lib/output/round_robin.go
@@ -80,6 +80,12 @@ func NewRoundRobin(conf Config, log log.Modular, stats metrics.Type) (Type, erro
 		return nil, err
 	}
 
+	if len(outputConfs) == 0 {
+		return nil, ErrFanOutNoOutputs
+	} else if len(outputConfs) == 1 {
+		return New(outputConfs[0], log, stats)
+	}
+
 	outputs := make([]types.Consumer, len(outputConfs))
 
 	for i, oConf := range outputConfs {