support for realtime error aggegration

1 files changed, 16 insertions, 0 deletions

diff --git a/error_buffer.go b/error_buffer.go
index fcf566c8..27f35e78 100644
--- a/error_buffer.go
+++ b/error_buffer.go
@@ -5,10 +5,22 @@ import (
 	"sync"
 )
 
+// EventStderrOutput - is triggered when worker sends data into stderr. The context is output data in []bytes form.
+const EventStderrOutput = 1900
+
 // thread safe errBuffer
 type errBuffer struct {
 	mu     sync.Mutex
 	buffer *bytes.Buffer
+	lsn    func(event int, ctx interface{})
+}
+
+// Listen attaches error stream even listener.
+func (b *errBuffer) Listen(l func(event int, ctx interface{})) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	b.lsn = l
 }
 
 // Len returns the number of bytes of the unread portion of the errBuffer;
@@ -27,6 +39,10 @@ func (b *errBuffer) Write(p []byte) (n int, err error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	if b.lsn != nil {
+		b.lsn(EventStderrOutput, p)
+	}
+
 	return b.buffer.Write(p)
 }