diff --git a/emitter.go b/emitter.go index 644b784..3db337c 100644 --- a/emitter.go +++ b/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 { diff --git a/gor.go b/gor.go index 7827bc0..07e5474 100644 --- a/gor.go +++ b/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) }) } diff --git a/output_http.go b/output_http.go index 2dde5fe..5f40e63 100644 --- a/output_http.go +++ b/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 { diff --git a/output_s3.go b/output_s3.go index b135d4d..5779a3f 100644 --- a/output_s3.go +++ b/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{}{} }