Merge pull request #25 from bitflow-stream/anton-fix-code-smells

Fix some code smells

Author: mwallschlaeger

bitflow/endpoints.go

@@ -296,52 +296,10 @@ func (f *EndpointFactory) CreateInput(inputs ...string) (SampleSource, error) {
 			return nil, fmt.Errorf("Format cannot be specified for data input: %v", input)
 		}
 		if result == nil {
-			reader := f.Reader(nil) // nil as Unmarshaller makes the SampleSource auto-detect the format
-			if f.FlagSourceTag != "" {
-				reader.Handler = sourceTagger(f.FlagSourceTag)
-			}
 			inputType = endpoint.Type
-			switch endpoint.Type {
-			case StdEndpoint:
-				source := NewConsoleSource()
-				source.Reader = reader
-				result = source
-			case TcpEndpoint, HttpEndpoint:
-				source := &TCPSource{
-					RemoteAddrs:   []string{endpoint.Target},
-					PrintErrors:   !f.FlagTcpSourceDropErrors,
-					RetryInterval: tcp_download_retry_interval,
-					DialTimeout:   tcp_dial_timeout,
-					UseHTTP:       endpoint.Type == HttpEndpoint,
-				}
-				source.TcpConnLimit = f.FlagTcpConnectionLimit
-				source.Reader = reader
-				result = source
-			case TcpListenEndpoint:
-				source := NewTcpListenerSource(endpoint.Target)
-				source.SimultaneousConnections = f.FlagInputTcpAcceptLimit
-				source.TcpConnLimit = f.FlagTcpConnectionLimit
-				source.Reader = reader
-				result = source
-			case FileEndpoint:
-				source := &FileSource{
-					FileNames: []string{endpoint.Target},
-					IoBuffer:  f.FlagIoBuffer,
-					Robust:    f.FlagInputFilesRobust,
-					KeepAlive: f.FlagFilesKeepAlive,
-				}
-				source.Reader = reader
-				result = source
-			default:
-				if factory, ok := f.CustomDataSources[endpoint.Type]; ok && endpoint.IsCustomType {
-					var factoryErr error
-					result, factoryErr = factory(endpoint.Target)
-					if factoryErr != nil {
-						return nil, fmt.Errorf("Error creating '%v' input: %v", endpoint.Type, factoryErr)
-					}
-				} else {
-					return nil, errors.New("Unknown input endpoint type: " + string(endpoint.Type))
-				}
+			result, err = f.createInput(endpoint)
+			if err != nil {
+				return nil, err
 			}
 		} else {
 			if inputType != endpoint.Type {
@@ -369,6 +327,55 @@ func (f *EndpointFactory) CreateInput(inputs ...string) (SampleSource, error) {
 	return result, nil
 }
 
+func (f *EndpointFactory) createInput(endpoint EndpointDescription) (SampleSource, error) {
+	reader := f.Reader(nil) // nil as Unmarshaller makes the SampleSource auto-detect the format
+	if f.FlagSourceTag != "" {
+		reader.Handler = sourceTagger(f.FlagSourceTag)
+	}
+	switch endpoint.Type {
+	case StdEndpoint:
+		source := NewConsoleSource()
+		source.Reader = reader
+		return source, nil
+	case TcpEndpoint, HttpEndpoint:
+		source := &TCPSource{
+			RemoteAddrs:   []string{endpoint.Target},
+			PrintErrors:   !f.FlagTcpSourceDropErrors,
+			RetryInterval: tcp_download_retry_interval,
+			DialTimeout:   tcp_dial_timeout,
+			UseHTTP:       endpoint.Type == HttpEndpoint,
+		}
+		source.TcpConnLimit = f.FlagTcpConnectionLimit
+		source.Reader = reader
+		return source, nil
+	case TcpListenEndpoint:
+		source := NewTcpListenerSource(endpoint.Target)
+		source.SimultaneousConnections = f.FlagInputTcpAcceptLimit
+		source.TcpConnLimit = f.FlagTcpConnectionLimit
+		source.Reader = reader
+		return source, nil
+	case FileEndpoint:
+		source := &FileSource{
+			FileNames: []string{endpoint.Target},
+			IoBuffer:  f.FlagIoBuffer,
+			Robust:    f.FlagInputFilesRobust,
+			KeepAlive: f.FlagFilesKeepAlive,
+		}
+		source.Reader = reader
+		return source, nil
+	default:
+		if factory, ok := f.CustomDataSources[endpoint.Type]; ok && endpoint.IsCustomType {
+			result, err := factory(endpoint.Target)
+			if err != nil {
+				err = fmt.Errorf("Error creating '%v' input: %v", endpoint.Type, err)
+			}
+			return result, err
+		} else {
+			return nil, errors.New("Unknown input endpoint type: " + string(endpoint.Type))
+		}
+	}
+}
+
 // Writer returns an instance of SampleWriter, configured by the values stored in the EndpointFactory.
 func (f *EndpointFactory) Writer() SampleWriter {
 	return SampleWriter{f.FlagParallelHandler}
@@ -377,7 +384,6 @@ func (f *EndpointFactory) Writer() SampleWriter {
 // CreateInput creates a SampleSink object based on the given output endpoint description
 // and the configuration flags in the EndpointFactory.
 func (f *EndpointFactory) CreateOutput(output string) (SampleProcessor, error) {
-	var resultSink SampleProcessor
 	endpoint, err := f.ParseEndpointDescription(output, true)
 	if err != nil {
 		return nil, err
@@ -389,18 +395,28 @@ func (f *EndpointFactory) CreateOutput(output string) (SampleProcessor, error) {
 			return nil, err
 		}
 	}
-	var marshallingSink *AbstractMarshallingSampleOutput
+	resultSink, marshallingSink, err := f.createOutput(endpoint, &marshaller)
+	if err != nil {
+		return nil, err
+	}
+	if marshallingSink != nil {
+		marshallingSink.SetMarshaller(marshaller)
+		marshallingSink.Writer = f.Writer()
+	}
+	return resultSink, nil
+}
+
+func (f *EndpointFactory) createOutput(endpoint EndpointDescription, marshaller *Marshaller) (SampleProcessor, *AbstractMarshallingSampleOutput, error) {
 	switch endpoint.Type {
 	case StdEndpoint:
 		sink := NewConsoleSink()
-		marshallingSink = &sink.AbstractMarshallingSampleOutput
-		if txt, ok := marshaller.(TextMarshaller); ok {
-			txt.AssumeStdout = true
-			marshaller = txt
-		} else if txt, ok := marshaller.(*TextMarshaller); ok {
+		if txt, ok := (*marshaller).(TextMarshaller); ok {
 			txt.AssumeStdout = true
+			*marshaller = txt
+		} else if txt2, ok2 := (*marshaller).(*TextMarshaller); ok2 {
+			txt2.AssumeStdout = true
 		}
-		resultSink = sink
+		return sink, &sink.AbstractMarshallingSampleOutput, nil
 	case FileEndpoint:
 		sink := &FileSink{
 			Filename:          endpoint.Target,
@@ -409,8 +425,7 @@ func (f *EndpointFactory) CreateOutput(output string) (SampleProcessor, error) {
 			Append:            f.FlagFilesAppend,
 			VanishedFileCheck: f.FlagFileVanishedCheck,
 		}
-		marshallingSink = &sink.AbstractMarshallingSampleOutput
-		resultSink = sink
+		return sink, &sink.AbstractMarshallingSampleOutput, nil
 	case TcpEndpoint:
 		sink := &TCPSink{
 			Endpoint:    endpoint.Target,
@@ -420,8 +435,7 @@ func (f *EndpointFactory) CreateOutput(output string) (SampleProcessor, error) {
 		if f.FlagTcpLogReceivedData {
 			sink.LogReceivedTraffic = log.ErrorLevel
 		}
-		marshallingSink = &sink.AbstractMarshallingSampleOutput
-		resultSink = sink
+		return sink, &sink.AbstractMarshallingSampleOutput, nil
 	case TcpListenEndpoint:
 		sink := &TCPListenerSink{
 			Endpoint:        endpoint.Target,
@@ -431,12 +445,11 @@ func (f *EndpointFactory) CreateOutput(output string) (SampleProcessor, error) {
 		if f.FlagTcpLogReceivedData {
 			sink.LogReceivedTraffic = log.ErrorLevel
 		}
-		marshallingSink = &sink.AbstractMarshallingSampleOutput
-		resultSink = sink
+		return sink, &sink.AbstractMarshallingSampleOutput, nil
 	case HttpEndpoint:
 		theUrl, err := url.Parse("http://" + endpoint.Target)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		sink := &HttpServerSink{
 			Endpoint:        theUrl.Host,
@@ -448,24 +461,18 @@ func (f *EndpointFactory) CreateOutput(output string) (SampleProcessor, error) {
 		if f.FlagTcpLogReceivedData {
 			sink.LogReceivedTraffic = log.ErrorLevel
 		}
-		marshallingSink = &sink.AbstractMarshallingSampleOutput
-		resultSink = sink
+		return sink, &sink.AbstractMarshallingSampleOutput, nil
 	default:
 		if factory, ok := f.CustomDataSinks[endpoint.Type]; ok && endpoint.IsCustomType {
-			var factoryErr error
-			resultSink, factoryErr = factory(endpoint.Target)
-			if factoryErr != nil {
-				return nil, fmt.Errorf("Error creating '%v' output: %v", endpoint.Type, factoryErr)
+			sink, err := factory(endpoint.Target)
+			if err != nil {
+				err = fmt.Errorf("Error creating '%v' output: %v", endpoint.Type, err)
 			}
+			return sink, nil, err
 		} else {
-			return nil, errors.New("Unknown output endpoint type: " + string(endpoint.Type))
+			return nil, nil, errors.New("Unknown output endpoint type: " + string(endpoint.Type))
 		}
 	}
-	if marshallingSink != nil {
-		marshallingSink.SetMarshaller(marshaller)
-		marshallingSink.Writer = f.Writer()
-	}
-	return resultSink, nil
 }
 
 func (f *EndpointFactory) CreateMarshaller(format MarshallingFormat) (Marshaller, error) {
@@ -562,31 +569,8 @@ func (f *EndpointFactory) ParseUrlEndpointDescription(endpoint string) (res Endp
 	target := urlParts[1]
 	res.Target = target
 	for _, part := range strings.Split(urlParts[0], "+") {
-		// TODO unclean: this parsing method is used for both marshalling/unmarshalling endpoints
-		if f.isMarshallingFormat(part) {
-			if res.Format != "" {
-				err = fmt.Errorf("Multiple formats defined in: %v", endpoint)
-				return
-			}
-			res.Format = MarshallingFormat(part)
-		} else {
-			if res.Type != UndefinedEndpoint {
-				err = fmt.Errorf("Multiple transport types defined: %v", endpoint)
-				return
-			}
-			switch EndpointType(part) {
-			case TcpEndpoint, TcpListenEndpoint, FileEndpoint, HttpEndpoint:
-				res.Type = EndpointType(part)
-			case StdEndpoint:
-				if target != stdTransportTarget {
-					err = fmt.Errorf("Transport '%v' can only be defined with target '%v'", part, stdTransportTarget)
-					return
-				}
-				res.Type = StdEndpoint
-			default:
-				res.IsCustomType = true
-				res.Type = EndpointType(part)
-			}
+		if err = f.parseEndpointUrlPart(part, &res, endpoint, target); err != nil {
+			return
 		}
 	}
 	if res.Type == UndefinedEndpoint {
@@ -602,6 +586,33 @@ func (f *EndpointFactory) ParseUrlEndpointDescription(endpoint string) (res Endp
 	return
 }
 
+func (f *EndpointFactory) parseEndpointUrlPart(part string, res *EndpointDescription, endpoint, target string) error {
+	// TODO unclean: this parsing method is used for both marshalling/unmarshalling endpoints
+	if f.isMarshallingFormat(part) {
+		if res.Format != "" {
+			return fmt.Errorf("Multiple formats defined in: %v", endpoint)
+		}
+		res.Format = MarshallingFormat(part)
+	} else {
+		if res.Type != UndefinedEndpoint {
+			return fmt.Errorf("Multiple transport types defined: %v", endpoint)
+		}
+		switch EndpointType(part) {
+		case TcpEndpoint, TcpListenEndpoint, FileEndpoint, HttpEndpoint:
+			res.Type = EndpointType(part)
+		case StdEndpoint:
+			if target != stdTransportTarget {
+				return fmt.Errorf("Transport '%v' can only be defined with target '%v'", part, stdTransportTarget)
+			}
+			res.Type = StdEndpoint
+		default:
+			res.IsCustomType = true
+			res.Type = EndpointType(part)
+		}
+	}
+	return nil
+}
+
 func (f *EndpointFactory) isMarshallingFormat(formatName string) bool {
 	_, ok := f.Marshallers[MarshallingFormat(formatName)]
 	return ok

bitflow/fork/fork_distributors.go

@@ -130,6 +130,16 @@ func (d *PipelineCache) getPipelines(key string, build PipelineBuildFunc) ([]Sub
 		}
 		d.pipelines[key] = pipes
 	}
+	d.updateSortedPipelineKeys(key, pipes)
+	result := make([]Subpipeline, len(pipes))
+	for i, pipe := range pipes {
+		result[i].Key = key
+		result[i].Pipe = pipe
+	}
+	return result, nil
+}
+
+func (d *PipelineCache) updateSortedPipelineKeys(key string, pipes []*bitflow.SamplePipeline) {
 	// Maintain a sorted list of keys that lead to each pipeline
 	for _, pipe := range pipes {
 		if keys, ok := d.keys[pipe]; ok {
@@ -146,12 +156,6 @@ func (d *PipelineCache) getPipelines(key string, build PipelineBuildFunc) ([]Sub
 			d.keys[pipe] = []string{key}
 		}
 	}
-	result := make([]Subpipeline, len(pipes))
-	for i, pipe := range pipes {
-		result[i].Key = key
-		result[i].Pipe = pipe
-	}
-	return result, nil
 }
 
 func (d *PipelineCache) ContainedStringers() []fmt.Stringer {

bitflow/marshall_text.go

@@ -29,7 +29,6 @@ const (
 // in a key = value format. The width of the header line, the number of columns
 // in the table, and the spacing between the columns in the table can be configured.
 type TextMarshaller struct {
-
 	// TextWidths sets the width of the header line and value table.
 	// If Columns > 0, this value is ignored as the width is determined by the
 	// number of columns. If this is 0, the width will be determined automatically:
@@ -174,23 +173,29 @@ func (m TextMarshaller) writeHeader(header string, textWidth int, writer io.Writ
 }
 
 func (m TextMarshaller) writeLines(lines []string, widths []int, writer io.Writer) error {
-	columns := len(widths)
 	for i, line := range lines {
-		if _, err := writer.Write([]byte(line)); err != nil {
+		if err := m.writeLine(line, i, lines, widths, writer); err != nil {
 			return err
 		}
-		col := i % columns
-		if col >= columns-1 || i == len(lines)-1 {
-			if _, err := writer.Write([]byte("\n")); err != nil {
+	}
+	return nil
+}
+
+func (m TextMarshaller) writeLine(line string, lineIndex int, lines []string, widths []int, writer io.Writer) error {
+	if _, err := writer.Write([]byte(line)); err != nil {
+		return err
+	}
+	col := lineIndex % len(widths)
+	if col >= len(widths)-1 || lineIndex == len(lines)-1 {
+		if _, err := writer.Write([]byte("\n")); err != nil {
+			return err
+		}
+	} else {
+		extraSpace := widths[col] - len(line)
+		for j := 0; j < extraSpace; j++ {
+			if _, err := writer.Write([]byte(" ")); err != nil {
 				return err
 			}
-		} else {
-			extraSpace := widths[col] - len(line)
-			for j := 0; j < extraSpace; j++ {
-				if _, err := writer.Write([]byte(" ")); err != nil {
-					return err
-				}
-			}
 		}
 	}
 	return nil

bitflow/pipeline.go

@@ -64,7 +64,7 @@ func (p *SamplePipeline) Construct(tasks *golib.TaskGroup) {
 			tasks.Add(&ProcessorTaskWrapper{proc})
 		}
 	}
-    tasks.Add(&SourceTaskWrapper{firstSource})
+	tasks.Add(&SourceTaskWrapper{firstSource})
 }
 
 // Add adds the SampleProcessor parameter to the list of SampleProcessors in the
@@ -74,11 +74,10 @@ func (p *SamplePipeline) Construct(tasks *golib.TaskGroup) {
 //   pipeline.Add(processor1).Add(processor2)
 func (p *SamplePipeline) Add(processor SampleProcessor) *SamplePipeline {
 	if p.lastProcessor != nil {
-		if merger, ok := p.lastProcessor.(MergeableProcessor); ok {
-			if merger.MergeProcessor(processor) {
-				// Merge successful: drop the incoming step
-				return p
-			}
+		merger, ok := p.lastProcessor.(MergeableProcessor)
+		if ok && merger.MergeProcessor(processor) {
+			// Merge successful: drop the incoming step
+			return p
 		}
 	}
 	p.lastProcessor = processor