Timeout blocking read call.

emitter.go

@@ -8,11 +8,10 @@ import (
 	"time"
 )
 
-var wg *sync.WaitGroup
+var wg sync.WaitGroup
 
 // Start initialize loop for sending data from inputs to outputs
 func Start(plugins *InOutPlugins, stop chan int) {
-	wg = &sync.WaitGroup{}
 	if Settings.middleware != "" {
 		middleware := NewMiddleware(Settings.middleware)
 

test_input.go

@@ -21,19 +21,22 @@ func NewTestInput() (i *TestInput) {
 }
 
 func (i *TestInput) Read(data []byte) (int, error) {
-	buf := <-i.data
+	select {
+	case buf := <-i.data:
+		var header []byte
 
-	var header []byte
+		if !i.skipHeader {
+			header = payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1)
+			copy(data[0:len(header)], header)
+			copy(data[len(header):], buf)
+		} else {
+			copy(data, buf)
+		}
 
-	if !i.skipHeader {
-		header = payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1)
-		copy(data[0:len(header)], header)
-		copy(data[len(header):], buf)
-	} else {
-		copy(data, buf)
+		return len(buf) + len(header), nil
+	case <-time.After(10* time.Second):
+		return 0, nil
 	}
-
-	return len(buf) + len(header), nil
 }
 
 func (i *TestInput) EmitBytes(data []byte) {