From e9bbc252e0501c12fbbf4d4e0d2c901df3b30cbc Mon Sep 17 00:00:00 2001
From: Arijit Das <arijit@dgraph.io>
Date: Thu, 11 Jun 2020 20:34:09 +0530
Subject: [PATCH] Fix input raw test and go fmt.

---
 emitter.go     |  2 +-
 input_http.go  |  2 +-
 input_tcp.go   |  2 +-
 output_http.go | 24 ++++++++++++++++++++++--
 4 files changed, 25 insertions(+), 5 deletions(-)

diff --git a/emitter.go b/emitter.go
index 5ba6854..8836104 100644
--- a/emitter.go
+++ b/emitter.go
@@ -84,7 +84,7 @@ func (e *emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
 
 func (e *emitter) close() {
 	select {
-	case <- e.quit:
+	case <-e.quit:
 	default:
 		close(e.quit)
 	}
diff --git a/input_http.go b/input_http.go
index d3032a3..6789330 100644
--- a/input_http.go
+++ b/input_http.go
@@ -13,7 +13,7 @@ type HTTPInput struct {
 	data     chan []byte
 	address  string
 	listener net.Listener
-	stop       chan bool // Channel used only to indicate goroutine should shutdown
+	stop     chan bool // Channel used only to indicate goroutine should shutdown
 }
 
 // NewHTTPInput constructor for HTTPInput. Accepts address with port which he will listen on.
diff --git a/input_tcp.go b/input_tcp.go
index e9a87c6..5486a75 100644
--- a/input_tcp.go
+++ b/input_tcp.go
@@ -17,7 +17,7 @@ type TCPInput struct {
 	listener net.Listener
 	address  string
 	config   *TCPInputConfig
-	stop       chan bool // Channel used only to indicate goroutine should shutdown
+	stop     chan bool // Channel used only to indicate goroutine should shutdown
 }
 
 type TCPInputConfig struct {
diff --git a/output_http.go b/output_http.go
index b8f06df..538a913 100644
--- a/output_http.go
+++ b/output_http.go
@@ -64,6 +64,8 @@ type HTTPOutput struct {
 	queueStats *GorStat
 
 	elasticSearch *ESPlugin
+
+	stop chan bool // Channel used only to indicate goroutine should shutdown
 }
 
 // NewHTTPOutput constructor for HTTPOutput
@@ -73,6 +75,7 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer {
 
 	o.address = address
 	o.config = config
+	o.stop = make(chan bool)
 
 	if o.config.stats {
 		o.queueStats = NewGorStat("output_http", o.config.statsMs)
@@ -124,6 +127,8 @@ func (o *HTTPOutput) startWorker() {
 
 	for {
 		select {
+		case <-o.stop:
+			return
 		case data := <-o.queue:
 			o.sendRequest(client, data)
 			deathCount = 0
@@ -155,7 +160,11 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) {
 	buf := make([]byte, len(data))
 	copy(buf, data)
 
-	o.queue <- buf
+	select {
+	case <-o.stop:
+		return 0, ErrorStopped
+	case o.queue <- buf:
+	}
 
 	if o.config.stats {
 		o.queueStats.Write(len(o.queue))
@@ -180,7 +189,12 @@ func (o *HTTPOutput) Write(data []byte) (n int, err error) {
 }
 
 func (o *HTTPOutput) Read(data []byte) (int, error) {
-	resp := <-o.responses
+	var resp response
+	select {
+	case <-o.stop:
+		return 0, ErrorStopped
+	case resp = <-o.responses:
+	}
 
 	if Settings.debug {
 		Debug("[OUTPUT-HTTP] Received response:", string(resp.payload))
@@ -231,3 +245,9 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) {
 func (o *HTTPOutput) String() string {
 	return "HTTP output: " + o.address
 }
+
+// Close closes the data channel so that data
+func (o *HTTPOutput) Close() error {
+	close(o.stop)
+	return nil
+}