add dcat test

2 files changed, 21 insertions, 48 deletions

diff --git a/internal/io/dlog/loggers/stdout.go b/internal/io/dlog/loggers/stdout.go
index 9738323..05485c6 100644
--- a/internal/io/dlog/loggers/stdout.go
+++ b/internal/io/dlog/loggers/stdout.go
@@ -8,66 +8,47 @@ import (
 )
 
 type stdout struct {
-	bufferCh chan string
 	pauseCh  chan struct{}
 	resumeCh chan struct{}
+	mutex    sync.Mutex
 }
 
 func newStdout() *stdout {
 	return &stdout{
-		bufferCh: make(chan string, 100),
 		pauseCh:  make(chan struct{}),
 		resumeCh: make(chan struct{}),
 	}
 }
 
 func (s *stdout) Start(ctx context.Context, wg *sync.WaitGroup) {
-	pause := func(ctx context.Context) {
-		select {
-		case <-s.resumeCh:
-			return
-		case <-ctx.Done():
-			return
-		}
-	}
-
-	go func() {
-		defer wg.Done()
-
-		for {
-			select {
-			case message := <-s.bufferCh:
-				fmt.Println(message)
-			case <-s.pauseCh:
-				pause(ctx)
-			case <-ctx.Done():
-				s.Flush()
-				return
-			}
-		}
-	}()
+	wg.Done()
 }
 
 func (s *stdout) Log(now time.Time, message string) {
-	s.bufferCh <- message
+	s.log(message)
 }
 
 func (s *stdout) LogWithColors(now time.Time, message, coloredMessage string) {
-	s.bufferCh <- coloredMessage
+	s.log(coloredMessage)
 }
 
-func (s *stdout) Flush() {
-	for {
-		select {
-		case message := <-s.bufferCh:
-			fmt.Println(message)
-		default:
-			return
-		}
+func (s *stdout) log(message string) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	select {
+	case <-s.pauseCh:
+		// Pause until resumed.
+		<-s.resumeCh
+	default:
 	}
+
+	fmt.Println(message)
 }
 
-func (s *stdout) Pause()            { s.pauseCh <- struct{}{} }
-func (s *stdout) Resume()           { s.resumeCh <- struct{}{} }
-func (s *stdout) Rotate()           {}
+func (s *stdout) Pause()  { s.pauseCh <- struct{}{} }
+func (s *stdout) Resume() { s.resumeCh <- struct{}{} }
+func (s *stdout) Flush()  {}
+func (s *stdout) Rotate() {}
+
 func (stdout) SupportsColors() bool { return true }
diff --git a/internal/io/fs/readfile.go b/internal/io/fs/readfile.go
index 07486a1..f128c07 100644
--- a/internal/io/fs/readfile.go
+++ b/internal/io/fs/readfile.go
@@ -13,8 +13,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/io/dlog"
+	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/io/pool"
 	"github.com/mimecast/dtail/internal/regex"
 
@@ -182,14 +182,6 @@ func (f readFile) read(ctx context.Context, fd *os.File, rawLines chan *bytes.Bu
 
 		switch b {
 		case '\n':
-			/*
-				// dcat/dgrep should actually transfer empty lines
-					if message.Len() == 0 {
-						time.Sleep(time.Millisecond * 100)
-						continue
-					}
-			*/
-			//message.WriteString("\n")
 			select {
 			case rawLines <- message:
 				message = pool.BytesBuffer.Get().(*bytes.Buffer)