appropriate emitter close

emitter.go

@@ -13,7 +13,6 @@ import (
 
 // Emitter represents an abject to manage plugins communication
 type Emitter struct {
-	sync.Mutex
 	sync.WaitGroup
 	plugins *InOutPlugins
 }
@@ -25,7 +24,6 @@ func NewEmitter() *Emitter {
 
 // Start initialize loop for sending data from inputs to outputs
 func (e *Emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
-	defer e.Wait()
 	if Settings.CopyBufferSize < 1 {
 		Settings.CopyBufferSize = 5 << 20
 	}
@@ -67,7 +65,11 @@ func (e *Emitter) Close() {
 			cp.Close()
 		}
 	}
-	e.plugins.All = nil // avoid further accidental close
+	if len(e.plugins.All) > 0 {
+		// wait for everything to stop
+		e.Wait()
+	}
+	e.plugins.All = nil // avoid Close to make changes again
 }
 
 // CopyMulty copies from 1 reader to multiple writers
@@ -146,11 +148,7 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error {
 						return err
 					}
 
-					wIndex++
-
-					if wIndex >= len(writers) {
-						wIndex = 0
-					}
+					wIndex = (wIndex + 1) % len(writers)
 				}
 			} else {
 				for _, dst := range writers {

gor.go

@@ -4,7 +4,6 @@ package main
 
 import (
 	"flag"
-	"fmt"
 	"log"
 	"net/http"
 	"net/http/httputil"
@@ -80,7 +79,7 @@ func main() {
 		log.Printf("Running gor for a duration of %s\n", Settings.ExitAfter)
 
 		time.AfterFunc(Settings.ExitAfter, func() {
-			fmt.Printf("gor run timeout %s\n", Settings.ExitAfter)
+			log.Printf("gor run timeout %s\n", Settings.ExitAfter)
 			close(closeCh)
 		})
 	}

output_http.go

@@ -59,7 +59,7 @@ type HTTPOutput struct {
 	client        *HTTPClient
 	stopWorker    chan struct{}
 	queue         chan *Message
-	responses     chan response
+	responses     chan *response
 	stop          chan bool // Channel used only to indicate goroutine should shutdown
 }
 
@@ -110,7 +110,9 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) PluginReadWriter {
 	}
 
 	o.queue = make(chan *Message, o.config.QueueLen)
-	o.responses = make(chan response, o.config.QueueLen)
+	if o.config.TrackResponses {
+		o.responses = make(chan *response, o.config.QueueLen)
+	}
 	// it should not be buffered to avoid races
 	o.stopWorker = make(chan struct{})
 
@@ -192,7 +194,10 @@ func (o *HTTPOutput) PluginWrite(msg *Message) (n int, err error) {
 
 // PluginRead reads message from this plugin
 func (o *HTTPOutput) PluginRead() (*Message, error) {
-	var resp response
+	if !o.config.TrackResponses {
+		return nil, ErrorStopped
+	}
+	var resp *response
 	var msg Message
 	select {
 	case <-o.stop:
@@ -224,7 +229,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, msg *Message) {
 	}
 
 	if o.config.TrackResponses {
-		o.responses <- response{resp, uuid, start.UnixNano(), stop.UnixNano() - start.UnixNano()}
+		o.responses <- &response{resp, uuid, start.UnixNano(), stop.UnixNano() - start.UnixNano()}
 	}
 
 	if o.elasticSearch != nil {

output_s3.go

@@ -108,22 +108,23 @@ func (o *S3Output) onBufferUpdate(path string) {
 	idx := getFileIndex(path)
 	bucket, key := o.keyPath(idx)
 
-	file, _ := os.Open(path)
-	// reader := bufio.NewReader(file)
+	file, err := os.Open(path)
+	if err != nil {
+		Debug(0, fmt.Sprintf("[S3 Output] Failed to open file %q. err: %q", path, err))
+		return
+	}
+	defer os.Remove(path)
 
-	_, err := svc.PutObject(&s3.PutObjectInput{
+	_, err = svc.PutObject(&s3.PutObjectInput{
 		Body:   file,
 		Bucket: aws.String(bucket),
 		Key:    aws.String(key),
 	})
 	if err != nil {
-		log.Printf("[S3 Output] Failed to upload data to %s/%s, %s\n", bucket, key, err)
-		os.Remove(path)
+		Debug(0, fmt.Sprintf("[S3 Output] Failed to upload data to %q/%q, %q", bucket, key, err))
 		return
 	}
 
-	os.Remove(path)
-
 	if o.closeCh != nil {
 		o.closeCh <- struct{}{}
 	}