From 799b9b69ba08b898e13026b7ecab9f9f58580a82 Mon Sep 17 00:00:00 2001
From: Paul Buetow <pbuetow@mimecast.com>
Date: Tue, 8 Dec 2020 14:49:41 +0000
Subject: merge develop

---
 internal/server/continuous.go              |   2 +-
 internal/server/handlers/controlhandler.go |  28 +++--
 internal/server/handlers/handler.go        |   2 +
 internal/server/handlers/serverhandler.go  | 182 +++++++----------------------
 internal/server/scheduler.go               |   2 +-
 internal/server/server.go                  |  44 +++----
 6 files changed, 78 insertions(+), 182 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/continuous.go b/internal/server/continuous.go
index 583d136..f75c732 100644
--- a/internal/server/continuous.go
+++ b/internal/server/continuous.go
@@ -92,7 +92,7 @@ func (c *continuous) runJob(ctx context.Context, job config.Continuous) {
 	}
 
 	logger.Info(fmt.Sprintf("Starting job %s", job.Name))
-	status := client.Start(jobCtx, make(chan struct{}))
+	status := client.Start(jobCtx, make(chan string))
 	logMessage := fmt.Sprintf("Job exited with status %d", status)
 
 	if status != 0 {
diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go
index daa9835..8cc5a40 100644
--- a/internal/server/handlers/controlhandler.go
+++ b/internal/server/handlers/controlhandler.go
@@ -1,20 +1,19 @@
 package handlers
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"os"
 	"strings"
 
+	"github.com/mimecast/dtail/internal"
 	"github.com/mimecast/dtail/internal/io/logger"
 	user "github.com/mimecast/dtail/internal/user/server"
 )
 
 // ControlHandler is used for control functions and health monitoring.
 type ControlHandler struct {
-	ctx            context.Context
-	done           chan struct{}
+	done           *internal.Done
 	hostname       string
 	payload        []byte
 	serverMessages chan string
@@ -22,12 +21,11 @@ type ControlHandler struct {
 }
 
 // NewControlHandler returns a new control handler.
-func NewControlHandler(ctx context.Context, user *user.User) (*ControlHandler, <-chan struct{}) {
+func NewControlHandler(user *user.User) *ControlHandler {
 	logger.Debug(user, "Creating control handler")
 
 	h := ControlHandler{
-		ctx:            ctx,
-		done:           make(chan struct{}),
+		done:           internal.NewDone(),
 		serverMessages: make(chan string, 10),
 		user:           user,
 	}
@@ -40,7 +38,17 @@ func NewControlHandler(ctx context.Context, user *user.User) (*ControlHandler, <
 	s := strings.Split(fqdn, ".")
 	h.hostname = s[0]
 
-	return &h, h.done
+	return &h
+}
+
+// Shutdown the handler.
+func (h *ControlHandler) Shutdown() {
+	h.done.Shutdown()
+}
+
+// Done channel of the handler.
+func (h *ControlHandler) Done() <-chan struct{} {
+	return h.done.Done()
 }
 
 // Read is to send data to the client via the Reader interface.
@@ -51,7 +59,7 @@ func (h *ControlHandler) Read(p []byte) (n int, err error) {
 			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message))
 			n = copy(p, wholePayload)
 			return
-		case <-h.ctx.Done():
+		case <-h.done.Done():
 			return 0, io.EOF
 		}
 	}
@@ -63,7 +71,7 @@ func (h *ControlHandler) Write(p []byte) (n int, err error) {
 		switch c {
 		case ';':
 			wholePayload := strings.TrimSpace(string(h.payload))
-			h.handleCommand(h.ctx, wholePayload)
+			h.handleCommand(wholePayload)
 			h.payload = nil
 
 		default:
@@ -75,7 +83,7 @@ func (h *ControlHandler) Write(p []byte) (n int, err error) {
 	return
 }
 
-func (h *ControlHandler) handleCommand(ctx context.Context, command string) {
+func (h *ControlHandler) handleCommand(command string) {
 	logger.Info(h.user, command)
 	s := strings.Split(command, " ")
 	logger.Debug(h.user, "Receiving command", command, s)
diff --git a/internal/server/handlers/handler.go b/internal/server/handlers/handler.go
index c42ceb9..b04e854 100644
--- a/internal/server/handlers/handler.go
+++ b/internal/server/handlers/handler.go
@@ -5,4 +5,6 @@ import "io"
 // Handler interface for server side functionality.
 type Handler interface {
 	io.ReadWriter
+	Shutdown()
+	Done() <-chan struct{}
 }
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 7017f3e..5cf8041 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -7,18 +7,16 @@ import (
 	"fmt"
 	"io"
 	"os"
-	"strconv"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"time"
 
+	"github.com/mimecast/dtail/internal"
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/io/logger"
 	"github.com/mimecast/dtail/internal/mapr/server"
 	"github.com/mimecast/dtail/internal/omode"
-	"github.com/mimecast/dtail/internal/server/background"
 	user "github.com/mimecast/dtail/internal/user/server"
 	"github.com/mimecast/dtail/internal/version"
 )
@@ -31,33 +29,27 @@ const (
 // the Bi-directional communication between SSH client and server.
 // This handler implements the handler of the SSH server.
 type ServerHandler struct {
-	lines              chan line.Line
-	regex              string
-	aggregate          *server.Aggregate
-	aggregatedMessages chan string
-	serverMessages     chan string
-	payload            []byte
-	hostname           string
-	user               *user.User
-	// TODO: Move all these channels into a separate struct for readability!
+	done                *internal.Done
+	lines               chan line.Line
+	regex               string
+	aggregate           *server.Aggregate
+	aggregatedMessages  chan string
+	serverMessages      chan string
+	payload             []byte
+	hostname            string
+	user                *user.User
 	catLimiter          chan struct{}
 	tailLimiter         chan struct{}
 	globalServerWaitFor chan struct{}
 	ackCloseReceived    chan struct{}
-	serverCtx           context.Context
-	handlerCtx          context.Context
-	done                chan struct{}
 	activeCommands      int32
 	activeReaders       int32
-	background          background.Background
 }
 
 // NewServerHandler returns the server handler.
-func NewServerHandler(handlerCtx, serverCtx context.Context, user *user.User, catLimiter, tailLimiter, globalServerWaitFor chan struct{}, background background.Background) (*ServerHandler, <-chan struct{}) {
+func NewServerHandler(user *user.User, catLimiter, tailLimiter, globalServerWaitFor chan struct{}) *ServerHandler {
 	h := ServerHandler{
-		serverCtx:           serverCtx,
-		handlerCtx:          handlerCtx,
-		done:                make(chan struct{}),
+		done:                internal.NewDone(),
 		lines:               make(chan line.Line, 100),
 		serverMessages:      make(chan string, 10),
 		aggregatedMessages:  make(chan string, 10),
@@ -67,7 +59,6 @@ func NewServerHandler(handlerCtx, serverCtx context.Context, user *user.User, ca
 		globalServerWaitFor: globalServerWaitFor,
 		regex:               ".",
 		user:                user,
-		background:          background,
 	}
 
 	fqdn, err := os.Hostname()
@@ -78,7 +69,17 @@ func NewServerHandler(handlerCtx, serverCtx context.Context, user *user.User, ca
 	s := strings.Split(fqdn, ".")
 	h.hostname = s[0]
 
-	return &h, h.done
+	return &h
+}
+
+// Shutdown the handler.
+func (h *ServerHandler) Shutdown() {
+	h.done.Shutdown()
+}
+
+// Done channel of the handler.
+func (h *ServerHandler) Done() <-chan struct{} {
+	return h.done.Done()
 }
 
 // Read is to send data to the dtail client via Reader interface.
@@ -120,7 +121,7 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 		case <-time.After(time.Second):
 			// Once in a while check whether we are done.
 			select {
-			case <-h.handlerCtx.Done():
+			case <-h.done.Done():
 				return 0, io.EOF
 			default:
 			}
@@ -134,7 +135,7 @@ func (h *ServerHandler) Write(p []byte) (n int, err error) {
 		switch c {
 		case ';':
 			commandStr := strings.TrimSpace(string(h.payload))
-			h.handleCommand(h.handlerCtx, commandStr)
+			h.handleCommand(commandStr)
 			h.payload = nil
 		default:
 			h.payload = append(h.payload, c)
@@ -145,9 +146,9 @@ func (h *ServerHandler) Write(p []byte) (n int, err error) {
 	return
 }
 
-func (h *ServerHandler) handleCommand(ctx context.Context, commandStr string) {
+func (h *ServerHandler) handleCommand(commandStr string) {
 	logger.Debug(h.user, commandStr)
-	var timeout time.Duration
+	ctx := context.Background()
 
 	args, argc, err := h.handleProtocolVersion(strings.Split(commandStr, " "))
 	if err != nil {
@@ -161,30 +162,18 @@ func (h *ServerHandler) handleCommand(ctx context.Context, commandStr string) {
 		return
 	}
 
-	args, argc, timeout, err = h.handleTimeout(args, argc)
-	if err != nil {
-		h.send(h.serverMessages, logger.Error(h.user, err))
-		return
-	}
-
 	if h.user.Name == config.ControlUser {
 		h.handleControlCommand(argc, args)
 		return
 	}
 
-	if timeout > 0 {
-		logger.Info(h.user, "Command with timeout context", argc, args, timeout)
-		commandCtx, cancel := context.WithTimeout(ctx, timeout)
-		go func() {
-			<-commandCtx.Done()
-			logger.Info(h.user, "Command timed out, canceling it", args, args, timeout)
-			cancel()
-		}()
-		h.handleUserCommand(commandCtx, argc, args, timeout)
-		return
-	}
+	ctx, cancel := context.WithCancel(ctx)
+	go func() {
+		<-h.done.Done()
+		cancel()
+	}()
 
-	h.handleUserCommand(ctx, argc, args, timeout)
+	h.handleUserCommand(ctx, argc, args)
 }
 
 func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, error) {
@@ -222,16 +211,6 @@ func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, er
 	return args, argc, nil
 }
 
-func (h *ServerHandler) handleTimeout(args []string, argc int) ([]string, int, time.Duration, error) {
-	if argc <= 2 || args[0] != "timeout" {
-		// No timeout specified
-		return args, argc, time.Duration(0) * time.Second, nil
-	}
-
-	timeout, err := strconv.Atoi(args[1])
-	return args[2:], argc - 2, time.Duration(timeout) * time.Second, err
-}
-
 func (h *ServerHandler) handleControlCommand(argc int, args []string) {
 	switch args[0] {
 	case "debug":
@@ -241,7 +220,7 @@ func (h *ServerHandler) handleControlCommand(argc int, args []string) {
 	}
 }
 
-func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string, timeout time.Duration) {
+func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string) {
 	logger.Debug(h.user, "handleUserCommand", argc, args)
 
 	h.incrementActiveCommands()
@@ -255,7 +234,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 			if h.aggregate == nil {
 				return
 			}
-			h.aggregate.Cancel()
+			h.aggregate.Shutdown()
 		}
 	}
 
@@ -303,86 +282,6 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 			commandFinished()
 		}()
 
-	case "run":
-		// TODO: Refactor this "run" case, move code to runcommand.go
-		command := newRunCommand(h)
-
-		jobName, _ := options["jobName"]
-		logger.Debug(h.user, "run", options)
-
-		if val, ok := options["background"]; ok && (val == "cancel" || val == "stop") {
-			if err := h.background.Cancel(h.user.Name, jobName); err != nil {
-				h.sendServerMessage(logger.Error(h.user, err, jobName, args))
-			} else {
-				h.sendServerMessage(logger.Info(h.user, "job cancelled", jobName))
-			}
-			commandFinished()
-			return
-		}
-
-		if val, ok := options["background"]; ok && val == "list" {
-			h.sendServerMessage("Listing jobs")
-			count := 0
-			for jobName := range h.background.ListJobsC(h.user.Name) {
-				h.sendServerMessage(jobName)
-				count++
-			}
-			h.sendServerMessage(fmt.Sprintf("Found %d jobs", count))
-			commandFinished()
-			return
-		}
-
-		str, _ := options["outerArgs"]
-		outerArgs := strings.Split(str, " ")
-
-		var background bool
-		if val, ok := options["background"]; ok && val == "start" {
-			background = true
-		}
-
-		var wg sync.WaitGroup
-		wg.Add(1)
-
-		if background {
-			if timeout == 0 {
-				// Set default background timeout.
-				timeout = time.Hour * 1
-			}
-			// Use a new context based on the server context, so that background job does not get
-			// terminated when handler/SSH connection terminates.
-			commandCtx, cancel := context.WithTimeout(h.serverCtx, timeout)
-
-			if err := h.background.Add(h.user.Name, jobName, cancel, &wg); err != nil {
-				h.sendServerMessage(logger.Error(h.user, err, jobName, args))
-				commandFinished()
-				return
-			}
-			ctx = commandCtx
-		}
-
-		if err := command.StartBackground(ctx, &wg, argc, args, outerArgs); err != nil {
-			h.sendServerMessage(logger.Error(h.user, "Unable to execute command", argc, args, err))
-			commandFinished()
-			return
-		}
-
-		// Make sure that server waits for all sub-processes to finish on shutdown
-		go func() { h.globalServerWaitFor <- struct{}{} }()
-		go func() {
-			wg.Wait()
-			<-h.globalServerWaitFor
-		}()
-
-		if background {
-			h.sendServerMessage(logger.Info(h.user, jobName, "job started in background"))
-			commandFinished()
-			return
-		}
-
-		// Command run in foreground, wait for it to complete before finishing the connection.
-		wg.Wait()
-		commandFinished()
-
 	case "ack", ".ack":
 		h.handleAckCommand(argc, args)
 		commandFinished()
@@ -406,7 +305,7 @@ func (h *ServerHandler) handleAckCommand(argc int, args []string) {
 func (h *ServerHandler) send(ch chan<- string, message string) {
 	select {
 	case ch <- message:
-	case <-h.handlerCtx.Done():
+	case <-h.done.Done():
 	}
 }
 
@@ -447,7 +346,7 @@ func (h *ServerHandler) shutdown() {
 	go func() {
 		select {
 		case h.serverMessageC() <- ".syn close connection":
-		case <-h.handlerCtx.Done():
+		case <-h.done.Done():
 		}
 	}()
 
@@ -455,13 +354,10 @@ func (h *ServerHandler) shutdown() {
 	case <-h.ackCloseReceived:
 	case <-time.After(time.Second * 5):
 		logger.Debug(h.user, "Shutdown timeout reached, enforcing shutdown")
-	case <-h.handlerCtx.Done():
+	case <-h.done.Done():
 	}
 
-	select {
-	case h.done <- struct{}{}:
-	default:
-	}
+	h.done.Shutdown()
 }
 
 func (h *ServerHandler) incrementActiveCommands() {
diff --git a/internal/server/scheduler.go b/internal/server/scheduler.go
index 9d76a3b..a1e9e36 100644
--- a/internal/server/scheduler.go
+++ b/internal/server/scheduler.go
@@ -93,7 +93,7 @@ func (s *scheduler) runJobs(ctx context.Context) {
 		defer cancel()
 
 		logger.Info(fmt.Sprintf("Starting job %s", job.Name))
-		status := client.Start(jobCtx, make(chan struct{}))
+		status := client.Start(jobCtx, make(chan string))
 		logMessage := fmt.Sprintf("Job exited with status %d", status)
 
 		if status != 0 {
diff --git a/internal/server/server.go b/internal/server/server.go
index a446738..31fa85d 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -11,7 +11,6 @@ import (
 
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/logger"
-	"github.com/mimecast/dtail/internal/server/background"
 	"github.com/mimecast/dtail/internal/server/handlers"
 	"github.com/mimecast/dtail/internal/ssh/server"
 	user "github.com/mimecast/dtail/internal/user/server"
@@ -35,9 +34,8 @@ type Server struct {
 	// Mointor log files for pattern (if configured)
 	cont *continuous
 	// Wait counter, e.g. there might be still subprocesses (forked by drun) to be killed.
+	// TODO: Remove this counter.
 	shutdownWaitFor chan struct{}
-	// Background jobs
-	background background.Background
 }
 
 // New returns a new server.
@@ -51,7 +49,6 @@ func New() *Server {
 		shutdownWaitFor: make(chan struct{}, 1000),
 		sched:           newScheduler(),
 		cont:            newContinuous(),
-		background:      background.New(),
 	}
 
 	s.sshServerConfig.PasswordCallback = s.Callback
@@ -178,53 +175,46 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch
 
 		switch req.Type {
 		case "shell":
-			handlerCtx, cancel := context.WithCancel(ctx)
-
 			var handler handlers.Handler
-			var done <-chan struct{}
-
 			switch user.Name {
 			case config.ControlUser:
-				handler, done = handlers.NewControlHandler(handlerCtx, user)
+				handler = handlers.NewControlHandler(user)
 			default:
-				handler, done = handlers.NewServerHandler(handlerCtx, ctx, user, s.catLimiter, s.tailLimiter, s.shutdownWaitFor, s.background)
+				handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter, s.shutdownWaitFor)
 			}
 
-			go func() {
-				// Handler finished work, cancel all remaining routines
-				defer cancel()
-
-				<-done
-			}()
+			terminate := func() {
+				handler.Shutdown()
+				sshConn.Close()
+			}
 
 			go func() {
 				// Broken pipe, cancel
-				defer cancel()
-
 				io.Copy(channel, handler)
+				terminate()
 			}()
 
 			go func() {
 				// Broken pipe, cancel
-				defer cancel()
-
 				io.Copy(handler, channel)
+				terminate()
 			}()
 
 			go func() {
-				defer cancel()
+				select {
+				case <-ctx.Done():
+				case <-handler.Done():
+				}
+				terminate()
+			}()
 
+			go func() {
 				if err := sshConn.Wait(); err != nil && err != io.EOF {
 					logger.Error(user, err)
 				}
 				s.stats.decrementConnections()
 				logger.Info(user, "Good bye Mister!")
-			}()
-
-			go func() {
-				<-handlerCtx.Done()
-				sshConn.Close()
-				logger.Info(user, "Closed SSH connection")
+				terminate()
 			}()
 
 			// Only serving shell type
-- 
cgit v1.2.3


From ab676c2b484225ed22765b23d8f0545088ecd610 Mon Sep 17 00:00:00 2001
From: Paul Buetow <git@mx.buetow.org>
Date: Sat, 26 Dec 2020 10:48:51 +0000
Subject: code cleanup and minor refactorings

---
 internal/server/handlers/runcommand.go    | 111 ------------------------------
 internal/server/handlers/serverhandler.go |  50 +++++++-------
 internal/server/server.go                 |  24 +------
 3 files changed, 26 insertions(+), 159 deletions(-)
 delete mode 100644 internal/server/handlers/runcommand.go

(limited to 'internal/server')

diff --git a/internal/server/handlers/runcommand.go b/internal/server/handlers/runcommand.go
deleted file mode 100644
index 8e5895b..0000000
--- a/internal/server/handlers/runcommand.go
+++ /dev/null
@@ -1,111 +0,0 @@
-package handlers
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"io/ioutil"
-	"os"
-	"os/exec"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/mimecast/dtail/internal/config"
-	"github.com/mimecast/dtail/internal/io/logger"
-	"github.com/mimecast/dtail/internal/io/run"
-)
-
-type runCommand struct {
-	server *ServerHandler
-	run    run.Run
-}
-
-func newRunCommand(server *ServerHandler) runCommand {
-	return runCommand{
-		server: server,
-	}
-}
-
-func (r runCommand) StartBackground(ctx context.Context, wg *sync.WaitGroup, argc int, args, outerArgs []string) error {
-	if argc < 2 {
-		return fmt.Errorf("%s: args:%v argc:%d", commandParseWarning, args, argc)
-	}
-
-	ec := make(chan int, 1)
-	var pid int
-	var err error
-
-	command := strings.Join(args[1:], " ")
-	if strings.Contains(command, ";") || strings.Contains(command, "\n") {
-		if pid, err = r.startScript(ctx, wg, ec, command, outerArgs); err != nil {
-			r.server.sendServerMessage(".run exitstatus 255")
-			return err
-		}
-		return nil
-	}
-
-	if pid, err = r.start(ctx, wg, ec, strings.TrimSpace(command), outerArgs); err != nil {
-		r.server.sendServerMessage(".run exitstatus 255")
-		return err
-	}
-
-	exitCode := <-ec
-	r.server.sendServerMessage(fmt.Sprintf(".run exitstatus %d", exitCode))
-	r.server.sendServerMessage(logger.Info(fmt.Sprintf("Process %d exited with status %d", pid, exitCode)))
-
-	return nil
-}
-
-func (r runCommand) startScript(ctx context.Context, wg *sync.WaitGroup, ec chan<- int, script string, outerArgs []string) (int, error) {
-	if _, err := os.Stat(config.Common.TmpDir); os.IsNotExist(err) {
-		return -1, err
-	}
-
-	timestamp := time.Now().UnixNano()
-	scriptPath := fmt.Sprintf("%s/%s_%v.sh", config.Common.TmpDir, r.server.user.Name, timestamp)
-
-	// TODO: On dserver startup delete all previously written scripts (there might be left overs due to a crash or so)
-	logger.Debug(r.server.user, "Writing temp script", scriptPath)
-
-	script = fmt.Sprintf("#!/bin/sh\n%s", script)
-	if err := ioutil.WriteFile(scriptPath, []byte(script), 0700); err != nil {
-		return -1, err
-	}
-
-	pid, err := r.start(ctx, wg, ec, scriptPath, outerArgs)
-	go func() {
-		wg.Wait()
-		logger.Debug("Deleting script", scriptPath)
-		os.Remove(scriptPath)
-	}()
-
-	return pid, err
-}
-
-func (r runCommand) start(ctx context.Context, wg *sync.WaitGroup, ec chan<- int, command string, outerArgs []string) (int, error) {
-	if len(command) == 0 {
-		return -1, errors.New("Empty command provided")
-	}
-
-	splitted := strings.Split(command, " ")
-	path := splitted[0]
-	args := splitted[1:]
-	args = append(args, outerArgs...)
-
-	qualifiedPath, err := exec.LookPath(path)
-	if err != nil {
-		return -1, err
-	}
-
-	if !r.server.user.HasFilePermission(qualifiedPath, "runcommands") {
-		return -1, fmt.Errorf("No permission to execute path: %s", qualifiedPath)
-	}
-
-	r.run = run.New(qualifiedPath, args)
-	pid, err := r.run.StartBackground(ctx, wg, ec, r.server.lines)
-	if err != nil {
-		return pid, err
-	}
-	return pid, nil
-}
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 5cf8041..3d1a53d 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -29,36 +29,34 @@ const (
 // the Bi-directional communication between SSH client and server.
 // This handler implements the handler of the SSH server.
 type ServerHandler struct {
-	done                *internal.Done
-	lines               chan line.Line
-	regex               string
-	aggregate           *server.Aggregate
-	aggregatedMessages  chan string
-	serverMessages      chan string
-	payload             []byte
-	hostname            string
-	user                *user.User
-	catLimiter          chan struct{}
-	tailLimiter         chan struct{}
-	globalServerWaitFor chan struct{}
-	ackCloseReceived    chan struct{}
-	activeCommands      int32
-	activeReaders       int32
+	done               *internal.Done
+	lines              chan line.Line
+	regex              string
+	aggregate          *server.Aggregate
+	aggregatedMessages chan string
+	serverMessages     chan string
+	payload            []byte
+	hostname           string
+	user               *user.User
+	catLimiter         chan struct{}
+	tailLimiter        chan struct{}
+	ackCloseReceived   chan struct{}
+	activeCommands     int32
+	activeReaders      int32
 }
 
 // NewServerHandler returns the server handler.
-func NewServerHandler(user *user.User, catLimiter, tailLimiter, globalServerWaitFor chan struct{}) *ServerHandler {
+func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *ServerHandler {
 	h := ServerHandler{
-		done:                internal.NewDone(),
-		lines:               make(chan line.Line, 100),
-		serverMessages:      make(chan string, 10),
-		aggregatedMessages:  make(chan string, 10),
-		ackCloseReceived:    make(chan struct{}),
-		catLimiter:          catLimiter,
-		tailLimiter:         tailLimiter,
-		globalServerWaitFor: globalServerWaitFor,
-		regex:               ".",
-		user:                user,
+		done:               internal.NewDone(),
+		lines:              make(chan line.Line, 100),
+		serverMessages:     make(chan string, 10),
+		aggregatedMessages: make(chan string, 10),
+		ackCloseReceived:   make(chan struct{}),
+		catLimiter:         catLimiter,
+		tailLimiter:        tailLimiter,
+		regex:              ".",
+		user:               user,
 	}
 
 	fqdn, err := os.Hostname()
diff --git a/internal/server/server.go b/internal/server/server.go
index 31fa85d..a20737e 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -7,7 +7,6 @@ import (
 	"io"
 	"net"
 	"strings"
-	"time"
 
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/logger"
@@ -33,9 +32,6 @@ type Server struct {
 	sched *scheduler
 	// Mointor log files for pattern (if configured)
 	cont *continuous
-	// Wait counter, e.g. there might be still subprocesses (forked by drun) to be killed.
-	// TODO: Remove this counter.
-	shutdownWaitFor chan struct{}
 }
 
 // New returns a new server.
@@ -46,7 +42,6 @@ func New() *Server {
 		sshServerConfig: &gossh.ServerConfig{},
 		catLimiter:      make(chan struct{}, config.Server.MaxConcurrentCats),
 		tailLimiter:     make(chan struct{}, config.Server.MaxConcurrentTails),
-		shutdownWaitFor: make(chan struct{}, 1000),
 		sched:           newScheduler(),
 		cont:            newContinuous(),
 	}
@@ -80,27 +75,12 @@ func (s *Server) Start(ctx context.Context) int {
 	go s.cont.start(ctx)
 	go s.listenerLoop(ctx, listener)
 
-	select {
-	case <-ctx.Done():
-		// Wait until all commands/jobs/children are no more!
-		s.wait()
-	}
+	<-ctx.Done()
 
 	// For future use.
 	return 0
 }
 
-func (s *Server) wait() {
-	for {
-		num := len(s.shutdownWaitFor)
-		logger.Debug("Waiting for stuff to finish", num)
-		if num <= 0 {
-			return
-		}
-		time.Sleep(time.Second)
-	}
-}
-
 func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) {
 	logger.Debug("Starting listener loop")
 
@@ -180,7 +160,7 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch
 			case config.ControlUser:
 				handler = handlers.NewControlHandler(user)
 			default:
-				handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter, s.shutdownWaitFor)
+				handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter)
 			}
 
 			terminate := func() {
-- 
cgit v1.2.3


From 60fa324cd5296b088c24d8db1d334a25ca955788 Mon Sep 17 00:00:00 2001
From: Paul Buetow <git@mx.buetow.org>
Date: Sat, 26 Dec 2020 11:44:31 +0000
Subject: initial spartan mode support

---
 internal/server/handlers/serverhandler.go | 7 +++++++
 1 file changed, 7 insertions(+)

(limited to 'internal/server')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 3d1a53d..5b948b3 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -43,6 +43,7 @@ type ServerHandler struct {
 	ackCloseReceived   chan struct{}
 	activeCommands     int32
 	activeReaders      int32
+	spartan            bool
 }
 
 // NewServerHandler returns the server handler.
@@ -245,6 +246,12 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 		commandFinished()
 		return
 	}
+	if spartan, ok := options["spartan"]; ok {
+		if spartan == "true" {
+			logger.Debug(h.user, "Enabling spartan mode")
+			h.spartan = true
+		}
+	}
 
 	switch commandName {
 	case "grep", "cat":
-- 
cgit v1.2.3


From b05ae938aa6ada831d19de076861e246a03a9d7d Mon Sep 17 00:00:00 2001
From: Paul Buetow <git@mx.buetow.org>
Date: Sat, 26 Dec 2020 14:00:41 +0000
Subject: rename spartan to quiet

---
 internal/server/handlers/serverhandler.go | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 5b948b3..3212ee1 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -43,7 +43,7 @@ type ServerHandler struct {
 	ackCloseReceived   chan struct{}
 	activeCommands     int32
 	activeReaders      int32
-	spartan            bool
+	quiet            bool
 }
 
 // NewServerHandler returns the server handler.
@@ -246,10 +246,10 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 		commandFinished()
 		return
 	}
-	if spartan, ok := options["spartan"]; ok {
-		if spartan == "true" {
-			logger.Debug(h.user, "Enabling spartan mode")
-			h.spartan = true
+	if quiet, ok := options["quiet"]; ok {
+		if quiet == "true" {
+			logger.Debug(h.user, "Enabling quiet mode")
+			h.quiet = true
 		}
 	}
 
-- 
cgit v1.2.3


From f9415259bc5d74a89801be75587da87dd37518af Mon Sep 17 00:00:00 2001
From: Paul Buetow <git@mx.buetow.org>
Date: Sat, 26 Dec 2020 14:28:31 +0000
Subject: initial quiet switch

---
 internal/server/handlers/controlhandler.go |  4 +---
 internal/server/handlers/readcommand.go    | 11 +++++------
 internal/server/handlers/serverhandler.go  | 11 +++++++++--
 3 files changed, 15 insertions(+), 11 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go
index 8cc5a40..1e17c78 100644
--- a/internal/server/handlers/controlhandler.go
+++ b/internal/server/handlers/controlhandler.go
@@ -92,9 +92,7 @@ func (h *ControlHandler) handleCommand(command string) {
 	case "health":
 		h.serverMessages <- "OK: DTail SSH Server seems fine"
 		h.serverMessages <- "done;"
-	case "debug":
-		h.serverMessages <- logger.Debug(h.user, "Receiving debug command", command, s)
 	default:
-		h.serverMessages <- logger.Warn(h.user, "Received unknown control command", command, s)
+		h.serverMessages <- logger.Error(h.user, "Received unknown control command", command, s)
 	}
 }
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 0f9207d..5b8ce3a 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -31,14 +31,13 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string) {
 	if argc >= 4 {
 		deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " "))
 		if err != nil {
-			logger.Error(err)
 			r.server.sendServerMessage(logger.Error(r.server.user, commandParseWarning, err))
 			return
 		}
 		re = deserializedRegex
 	}
 	if argc < 3 {
-		r.server.sendServerMessage(logger.Warn(r.server.user, commandParseWarning, args, argc))
+		r.server.sendServerWarnMessage(logger.Warn(r.server.user, commandParseWarning, args, argc))
 		return
 	}
 	r.readGlob(ctx, args[1], re)
@@ -52,7 +51,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex)
 	for {
 		maxRetries--
 		if maxRetries < 0 {
-			r.server.sendServerMessage(logger.Warn(r.server.user, "Giving up to read file(s)"))
+			r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Giving up to read file(s)"))
 			return
 		}
 
@@ -65,7 +64,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex)
 
 		if numPaths := len(paths); numPaths == 0 {
 			logger.Error(r.server.user, "No such file(s) to read", glob)
-			r.server.sendServerMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs"))
+			r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs"))
 			select {
 			case <-ctx.Done():
 				return
@@ -97,7 +96,7 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGr
 
 	if !r.server.user.HasFilePermission(path, "readfiles") {
 		logger.Error(r.server.user, "No permission to read file", path, globID)
-		r.server.sendServerMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs"))
+		r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs"))
 		return
 	}
 
@@ -161,6 +160,6 @@ func (r *readCommand) makeGlobID(path, glob string) string {
 		return pathParts[len(pathParts)-1]
 	}
 
-	r.server.sendServerMessage(logger.Error("Empty file path given?", path, glob))
+	r.server.sendServerWarnMessage(logger.Warn("Empty file path given?", path, glob))
 	return ""
 }
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 3212ee1..681598c 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -43,7 +43,7 @@ type ServerHandler struct {
 	ackCloseReceived   chan struct{}
 	activeCommands     int32
 	activeReaders      int32
-	quiet            bool
+	quiet              bool
 }
 
 // NewServerHandler returns the server handler.
@@ -299,7 +299,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 
 func (h *ServerHandler) handleAckCommand(argc int, args []string) {
 	if argc < 3 {
-		h.sendServerMessage(logger.Warn(h.user, commandParseWarning, args, argc))
+		h.sendServerWarnMessage(logger.Warn(h.user, commandParseWarning, args, argc))
 		return
 	}
 	if args[1] == "close" && args[2] == "connection" {
@@ -318,6 +318,13 @@ func (h *ServerHandler) sendServerMessage(message string) {
 	h.send(h.serverMessageC(), message)
 }
 
+func (h *ServerHandler) sendServerWarnMessage(message string) {
+	if h.quiet {
+		return
+	}
+	h.send(h.serverMessageC(), message)
+}
+
 func (h *ServerHandler) serverMessageC() chan<- string {
 	return h.serverMessages
 }
-- 
cgit v1.2.3


From 94e37105c5a8c0ce22104add751e9938f239261e Mon Sep 17 00:00:00 2001
From: Paul Buetow <git@mx.buetow.org>
Date: Sun, 27 Dec 2020 16:31:03 +0000
Subject: only try to read a file once in cat and grep mode but 10 times in
 tail mode

---
 internal/server/handlers/readcommand.go   | 20 ++++++++------------
 internal/server/handlers/serverhandler.go |  4 ++--
 2 files changed, 10 insertions(+), 14 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 5b8ce3a..5bab26f 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -25,7 +25,7 @@ func newReadCommand(server *ServerHandler, mode omode.Mode) *readCommand {
 	}
 }
 
-func (r *readCommand) Start(ctx context.Context, argc int, args []string) {
+func (r *readCommand) Start(ctx context.Context, argc int, args []string, retries int) {
 	re := regex.NewNoop()
 
 	if argc >= 4 {
@@ -40,21 +40,14 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string) {
 		r.server.sendServerWarnMessage(logger.Warn(r.server.user, commandParseWarning, args, argc))
 		return
 	}
-	r.readGlob(ctx, args[1], re)
+	r.readGlob(ctx, args[1], re, retries)
 }
 
-func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex) {
+func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex, retries int) {
 	retryInterval := time.Second * 5
 	glob = filepath.Clean(glob)
 
-	maxRetries := 10
-	for {
-		maxRetries--
-		if maxRetries < 0 {
-			r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Giving up to read file(s)"))
-			return
-		}
-
+	for retryCount := 0; retryCount < retries; retryCount++ {
 		paths, err := filepath.Glob(glob)
 		if err != nil {
 			logger.Warn(r.server.user, glob, err)
@@ -75,8 +68,11 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex)
 		}
 
 		r.readFiles(ctx, paths, glob, re, retryInterval)
-		break
+		return
 	}
+
+	r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Giving up to read file(s)"))
+	return
 }
 
 func (r *readCommand) readFiles(ctx context.Context, paths []string, glob string, re regex.Regex, retryInterval time.Duration) {
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 681598c..185e7c2 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -258,7 +258,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 		command := newReadCommand(h, omode.CatClient)
 		go func() {
 			h.incrementActiveReaders()
-			command.Start(ctx, argc, args)
+			command.Start(ctx, argc, args, 1)
 			readerFinished()
 			commandFinished()
 		}()
@@ -267,7 +267,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 		command := newReadCommand(h, omode.TailClient)
 		go func() {
 			h.incrementActiveReaders()
-			command.Start(ctx, argc, args)
+			command.Start(ctx, argc, args, 10)
 			readerFinished()
 			commandFinished()
 		}()
-- 
cgit v1.2.3


From ae8ffc84331ca72f0d33fff69edd85d6e03d29ae Mon Sep 17 00:00:00 2001
From: Paul Buetow <git@mx.buetow.org>
Date: Mon, 28 Dec 2020 09:40:38 +0000
Subject: refactor

---
 internal/server/handlers/serverhandler.go | 50 +++++++++++++++----------------
 internal/server/server.go                 | 20 +------------
 2 files changed, 25 insertions(+), 45 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 843eabc..db917bd 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -29,36 +29,34 @@ const (
 // the Bi-directional communication between SSH client and server.
 // This handler implements the handler of the SSH server.
 type ServerHandler struct {
-	done                *internal.Done
-	lines               chan line.Line
-	regex               string
-	aggregate           *server.Aggregate
-	aggregatedMessages  chan string
-	serverMessages      chan string
-	payload             []byte
-	hostname            string
-	user                *user.User
-	catLimiter          chan struct{}
-	tailLimiter         chan struct{}
-	globalServerWaitFor chan struct{}
-	ackCloseReceived    chan struct{}
-	activeCommands      int32
-	activeReaders       int32
+	done               *internal.Done
+	lines              chan line.Line
+	regex              string
+	aggregate          *server.Aggregate
+	aggregatedMessages chan string
+	serverMessages     chan string
+	payload            []byte
+	hostname           string
+	user               *user.User
+	catLimiter         chan struct{}
+	tailLimiter        chan struct{}
+	ackCloseReceived   chan struct{}
+	activeCommands     int32
+	activeReaders      int32
 }
 
 // NewServerHandler returns the server handler.
-func NewServerHandler(user *user.User, catLimiter, tailLimiter, globalServerWaitFor chan struct{}) *ServerHandler {
+func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *ServerHandler {
 	h := ServerHandler{
-		done:                internal.NewDone(),
-		lines:               make(chan line.Line, 100),
-		serverMessages:      make(chan string, 10),
-		aggregatedMessages:  make(chan string, 10),
-		ackCloseReceived:    make(chan struct{}),
-		catLimiter:          catLimiter,
-		tailLimiter:         tailLimiter,
-		globalServerWaitFor: globalServerWaitFor,
-		regex:               ".",
-		user:                user,
+		done:               internal.NewDone(),
+		lines:              make(chan line.Line, 100),
+		serverMessages:     make(chan string, 10),
+		aggregatedMessages: make(chan string, 10),
+		ackCloseReceived:   make(chan struct{}),
+		catLimiter:         catLimiter,
+		tailLimiter:        tailLimiter,
+		regex:              ".",
+		user:               user,
 	}
 
 	fqdn, err := os.Hostname()
diff --git a/internal/server/server.go b/internal/server/server.go
index 31fa85d..d8d43c9 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -7,7 +7,6 @@ import (
 	"io"
 	"net"
 	"strings"
-	"time"
 
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/logger"
@@ -33,9 +32,6 @@ type Server struct {
 	sched *scheduler
 	// Mointor log files for pattern (if configured)
 	cont *continuous
-	// Wait counter, e.g. there might be still subprocesses (forked by drun) to be killed.
-	// TODO: Remove this counter.
-	shutdownWaitFor chan struct{}
 }
 
 // New returns a new server.
@@ -46,7 +42,6 @@ func New() *Server {
 		sshServerConfig: &gossh.ServerConfig{},
 		catLimiter:      make(chan struct{}, config.Server.MaxConcurrentCats),
 		tailLimiter:     make(chan struct{}, config.Server.MaxConcurrentTails),
-		shutdownWaitFor: make(chan struct{}, 1000),
 		sched:           newScheduler(),
 		cont:            newContinuous(),
 	}
@@ -82,25 +77,12 @@ func (s *Server) Start(ctx context.Context) int {
 
 	select {
 	case <-ctx.Done():
-		// Wait until all commands/jobs/children are no more!
-		s.wait()
 	}
 
 	// For future use.
 	return 0
 }
 
-func (s *Server) wait() {
-	for {
-		num := len(s.shutdownWaitFor)
-		logger.Debug("Waiting for stuff to finish", num)
-		if num <= 0 {
-			return
-		}
-		time.Sleep(time.Second)
-	}
-}
-
 func (s *Server) listenerLoop(ctx context.Context, listener net.Listener) {
 	logger.Debug("Starting listener loop")
 
@@ -180,7 +162,7 @@ func (s *Server) handleRequests(ctx context.Context, sshConn gossh.Conn, in <-ch
 			case config.ControlUser:
 				handler = handlers.NewControlHandler(user)
 			default:
-				handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter, s.shutdownWaitFor)
+				handler = handlers.NewServerHandler(user, s.catLimiter, s.tailLimiter)
 			}
 
 			terminate := func() {
-- 
cgit v1.2.3


From c2522ffb59514443816a96386c16bb7527cbe57c Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sat, 21 Aug 2021 14:54:24 +0300
Subject: read files bytewise for more control of whats happening - change
 transport protocol for more control over newlines

---
 internal/server/handlers/serverhandler.go | 15 +++++++++------
 1 file changed, 9 insertions(+), 6 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 185e7c2..23e3aeb 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -17,8 +17,8 @@ import (
 	"github.com/mimecast/dtail/internal/io/logger"
 	"github.com/mimecast/dtail/internal/mapr/server"
 	"github.com/mimecast/dtail/internal/omode"
+	"github.com/mimecast/dtail/internal/protocol"
 	user "github.com/mimecast/dtail/internal/user/server"
-	"github.com/mimecast/dtail/internal/version"
 )
 
 const (
@@ -92,24 +92,27 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 			}
 			if message[0] == '.' {
 				// Handle hidden message (don't display to the user, interpreted by dtail client)
-				wholePayload := []byte(fmt.Sprintf("%s\n", message))
+				wholePayload := []byte(fmt.Sprintf("%s%b", message, protocol.MessageDelimiter))
 				n = copy(p, wholePayload)
 				return
 			}
 
 			// Handle normal server message (display to the user)
-			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message))
+			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%b", h.hostname, message, protocol.MessageDelimiter))
 			n = copy(p, wholePayload)
 			return
 
 		case message := <-h.aggregatedMessages:
 			// Send mapreduce-aggregated data as a message.
-			data := fmt.Sprintf("AGGREGATE➔%s➔%s\n", h.hostname, message)
+			data := fmt.Sprintf("AGGREGATE%s%s%s%s%b",
+				protocol.AggregateDelimiter, h.hostname,
+				protocol.AggregateDelimiter, message, protocol.MessageDelimiter)
 			wholePayload := []byte(data)
 			n = copy(p, wholePayload)
 			return
 
 		case line := <-h.lines:
+			//fmt.Printf("<<<%d,%s>>>\n", len(line.Content), line.Content)
 			// Send normal file content data as a message.
 			serverInfo := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|",
 				h.hostname, line.TransmittedPerc, line.Count, line.SourceID))
@@ -182,8 +185,8 @@ func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, err
 		return args, argc, errors.New("unable to determine protocol version")
 	}
 
-	if args[1] != version.ProtocolCompat {
-		err := fmt.Errorf("server with protocol version '%s' but client with '%s', please update DTail", version.ProtocolCompat, args[1])
+	if args[1] != protocol.ProtocolCompat {
+		err := fmt.Errorf("server with protocol version '%s' but client with '%s', please update DTail", protocol.ProtocolCompat, args[1])
 		return args, argc, err
 	}
 
-- 
cgit v1.2.3


From 9883a190109623b64e6d311dc2b462a6eae68003 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sun, 22 Aug 2021 10:07:00 +0300
Subject: introduces the protocol package

---
 internal/server/handlers/controlhandler.go | 3 ++-
 internal/server/handlers/serverhandler.go  | 6 +++---
 2 files changed, 5 insertions(+), 4 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go
index 1e17c78..a217b40 100644
--- a/internal/server/handlers/controlhandler.go
+++ b/internal/server/handlers/controlhandler.go
@@ -8,6 +8,7 @@ import (
 
 	"github.com/mimecast/dtail/internal"
 	"github.com/mimecast/dtail/internal/io/logger"
+	"github.com/mimecast/dtail/internal/protocol"
 	user "github.com/mimecast/dtail/internal/user/server"
 )
 
@@ -56,7 +57,7 @@ func (h *ControlHandler) Read(p []byte) (n int, err error) {
 	for {
 		select {
 		case message := <-h.serverMessages:
-			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message))
+			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%b", h.hostname, message, protocol.MessageDelimiter))
 			n = copy(p, wholePayload)
 			return
 		case <-h.done.Done():
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 23e3aeb..9541a34 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -92,13 +92,13 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 			}
 			if message[0] == '.' {
 				// Handle hidden message (don't display to the user, interpreted by dtail client)
-				wholePayload := []byte(fmt.Sprintf("%s%b", message, protocol.MessageDelimiter))
+				wholePayload := []byte(fmt.Sprintf("%s%s", message, string(protocol.MessageDelimiter)))
 				n = copy(p, wholePayload)
 				return
 			}
 
 			// Handle normal server message (display to the user)
-			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%b", h.hostname, message, protocol.MessageDelimiter))
+			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%s", h.hostname, message, string(protocol.MessageDelimiter)))
 			n = copy(p, wholePayload)
 			return
 
@@ -112,7 +112,7 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 			return
 
 		case line := <-h.lines:
-			//fmt.Printf("<<<%d,%s>>>\n", len(line.Content), line.Content)
+			//fmt.Printf("\t<<<%d,%s>>>\n", len(line.Content), line.Content)
 			// Send normal file content data as a message.
 			serverInfo := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|",
 				h.hostname, line.TransmittedPerc, line.Count, line.SourceID))
-- 
cgit v1.2.3


From 6d727b9bdbc387c8a5c34406a2c4de9140face38 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sat, 28 Aug 2021 19:36:46 +0100
Subject: use a byte.Buffer in the file reader

---
 internal/server/handlers/serverhandler.go | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 9541a34..62f3c2b 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -15,6 +15,7 @@ import (
 	"github.com/mimecast/dtail/internal/config"
 	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/io/logger"
+	"github.com/mimecast/dtail/internal/io/pool"
 	"github.com/mimecast/dtail/internal/mapr/server"
 	"github.com/mimecast/dtail/internal/omode"
 	"github.com/mimecast/dtail/internal/protocol"
@@ -114,10 +115,10 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 		case line := <-h.lines:
 			//fmt.Printf("\t<<<%d,%s>>>\n", len(line.Content), line.Content)
 			// Send normal file content data as a message.
-			serverInfo := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|",
-				h.hostname, line.TransmittedPerc, line.Count, line.SourceID))
-			wholePayload := append(serverInfo, line.Content[:]...)
-			n = copy(p, wholePayload)
+			payload := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|%s",
+				h.hostname, line.TransmittedPerc, line.Count, line.SourceID, line.Content.String()))
+			n = copy(p, payload)
+			pool.RecycleBytesBuffer(line.Content)
 			return
 
 		case <-time.After(time.Second):
-- 
cgit v1.2.3


From 8c2e94030d0e31289c35fcfb56499707fd4a7ccd Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sat, 28 Aug 2021 20:10:54 +0100
Subject: make use of more buffers on server side

---
 internal/server/handlers/serverhandler.go | 40 +++++++++++++++++++++----------
 1 file changed, 27 insertions(+), 13 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 62f3c2b..f5aefa2 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -1,6 +1,7 @@
 package handlers
 
 import (
+	"bytes"
 	"context"
 	"encoding/base64"
 	"errors"
@@ -93,31 +94,44 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 			}
 			if message[0] == '.' {
 				// Handle hidden message (don't display to the user, interpreted by dtail client)
-				wholePayload := []byte(fmt.Sprintf("%s%s", message, string(protocol.MessageDelimiter)))
-				n = copy(p, wholePayload)
+				payload := []byte(fmt.Sprintf("%s%s", message, string(protocol.MessageDelimiter)))
+				n = copy(p, payload)
 				return
 			}
 
 			// Handle normal server message (display to the user)
-			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%s", h.hostname, message, string(protocol.MessageDelimiter)))
-			n = copy(p, wholePayload)
+			payload := []byte(fmt.Sprintf("SERVER|%s|%s%s", h.hostname, message, string(protocol.MessageDelimiter)))
+			n = copy(p, payload)
 			return
 
 		case message := <-h.aggregatedMessages:
 			// Send mapreduce-aggregated data as a message.
-			data := fmt.Sprintf("AGGREGATE%s%s%s%s%b",
-				protocol.AggregateDelimiter, h.hostname,
-				protocol.AggregateDelimiter, message, protocol.MessageDelimiter)
-			wholePayload := []byte(data)
-			n = copy(p, wholePayload)
+			buf := pool.BytesBuffer.Get().(*bytes.Buffer)
+			buf.WriteString("AGGREGATE")
+			buf.WriteString(protocol.AggregateDelimiter)
+			buf.WriteString(h.hostname)
+			buf.WriteString(protocol.AggregateDelimiter)
+			buf.WriteString(message)
+			buf.WriteByte(protocol.MessageDelimiter)
+			n = copy(p, buf.Bytes())
+			pool.RecycleBytesBuffer(buf)
 			return
 
 		case line := <-h.lines:
-			//fmt.Printf("\t<<<%d,%s>>>\n", len(line.Content), line.Content)
-			// Send normal file content data as a message.
-			payload := []byte(fmt.Sprintf("REMOTE|%s|%3d|%v|%s|%s",
-				h.hostname, line.TransmittedPerc, line.Count, line.SourceID, line.Content.String()))
+			buf := pool.BytesBuffer.Get().(*bytes.Buffer)
+			buf.WriteString("REMOTE")
+			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(h.hostname)
+			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc))
+			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(fmt.Sprintf("%v", line.Count))
+			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(line.SourceID)
+			buf.WriteByte(protocol.FieldDelimiter)
+			payload := append(buf.Bytes(), line.Content.Bytes()...)
 			n = copy(p, payload)
+			pool.RecycleBytesBuffer(buf)
 			pool.RecycleBytesBuffer(line.Content)
 			return
 
-- 
cgit v1.2.3


From 23982f331c2154a66b86d596226c24454fd06be5 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sat, 28 Aug 2021 20:26:32 +0100
Subject: 1. Major performance gain by not checking for file truncation aftter 
   each bytes read. 2. Introduce field separator to the protocol package.

---
 internal/server/handlers/serverhandler.go | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index f5aefa2..14fc5d0 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -120,15 +120,15 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 		case line := <-h.lines:
 			buf := pool.BytesBuffer.Get().(*bytes.Buffer)
 			buf.WriteString("REMOTE")
-			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(protocol.FieldDelimiter)
 			buf.WriteString(h.hostname)
-			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(protocol.FieldDelimiter)
 			buf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc))
-			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(protocol.FieldDelimiter)
 			buf.WriteString(fmt.Sprintf("%v", line.Count))
-			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(protocol.FieldDelimiter)
 			buf.WriteString(line.SourceID)
-			buf.WriteByte(protocol.FieldDelimiter)
+			buf.WriteString(protocol.FieldDelimiter)
 			payload := append(buf.Bytes(), line.Content.Bytes()...)
 			n = copy(p, payload)
 			pool.RecycleBytesBuffer(buf)
-- 
cgit v1.2.3


From cc89d3fb8be2465af276d7ef03ea2a8affd87b2e Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Mon, 6 Sep 2021 13:48:55 +0300
Subject: Print out client/server update notice even from dtail server 4 to
 dtail client 3.

---
 internal/server/handlers/serverhandler.go | 30 +++++++++++++++++++++++-------
 1 file changed, 23 insertions(+), 7 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 14fc5d0..14f46a3 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -167,9 +168,9 @@ func (h *ServerHandler) handleCommand(commandStr string) {
 	logger.Debug(h.user, commandStr)
 	ctx := context.Background()
 
-	args, argc, err := h.handleProtocolVersion(strings.Split(commandStr, " "))
+	args, argc, add, err := h.handleProtocolVersion(strings.Split(commandStr, " "))
 	if err != nil {
-		h.send(h.serverMessages, logger.Error(h.user, err))
+		h.send(h.serverMessages, logger.Error(h.user, err)+add)
 		return
 	}
 
@@ -193,19 +194,34 @@ func (h *ServerHandler) handleCommand(commandStr string) {
 	h.handleUserCommand(ctx, argc, args)
 }
 
-func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, error) {
+func (h *ServerHandler) handleProtocolVersion(args []string) ([]string, int, string, error) {
 	argc := len(args)
+	var add string
 
 	if argc <= 2 || args[0] != "protocol" {
-		return args, argc, errors.New("unable to determine protocol version")
+		return args, argc, add, errors.New("unable to determine protocol version")
 	}
 
 	if args[1] != protocol.ProtocolCompat {
-		err := fmt.Errorf("server with protocol version '%s' but client with '%s', please update DTail", protocol.ProtocolCompat, args[1])
-		return args, argc, err
+		clientCompat, _ := strconv.Atoi(args[1])
+		serverCompat, _ := strconv.Atoi(protocol.ProtocolCompat)
+		if clientCompat <= 3 {
+			// Protocol version 3 or lower expect a newline as message separator
+			// One day (after 2 major versions) this exception may be removed!
+			add = "\n"
+		}
+
+		toUpdate := "client"
+		if clientCompat > serverCompat {
+			toUpdate = "server"
+		}
+
+		err := fmt.Errorf("DTail server protocol version '%s' does not match client protocol version '%s', please update DTail %s!",
+			protocol.ProtocolCompat, args[1], toUpdate)
+		return args, argc, add, err
 	}
 
-	return args[2:], argc - 2, nil
+	return args[2:], argc - 2, add, nil
 }
 
 func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, error) {
-- 
cgit v1.2.3


From f74a9e4b35feb8c07d8a70b5a581088a0a59889d Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Tue, 7 Sep 2021 10:01:32 +0300
Subject: Produce MAPREDUCE lines, can aggregate these via default log format

---
 internal/server/stats.go | 12 ++++++++----
 1 file changed, 8 insertions(+), 4 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/stats.go b/internal/server/stats.go
index ac579ad..3e8c71d 100644
--- a/internal/server/stats.go
+++ b/internal/server/stats.go
@@ -50,10 +50,14 @@ func (s *stats) logServerStats() {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
-	currentConnections := fmt.Sprintf("currentConnections=%d", s.currentConnections)
-	lifetimeConnections := fmt.Sprintf("lifetimeConnections=%d", s.lifetimeConnections)
-	goroutines := fmt.Sprintf("goroutines=%d", runtime.NumGoroutine())
-	logger.Info("stats", currentConnections, lifetimeConnections, goroutines)
+	data := make(map[string]interface{})
+	data["currentConnections"] = s.currentConnections
+	data["lifetimeConnections"] = s.lifetimeConnections
+	data["goroutines"] = runtime.NumGoroutine()
+	data["cgocalls"] = runtime.NumCgoCall()
+	data["cpu"] = runtime.NumCPU()
+
+	logger.Mapreduce("STATS", data)
 }
 
 func (s *stats) serverLimitExceeded() error {
-- 
cgit v1.2.3


From 16dc57e1e1c28e9d762424e596223a980770e059 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Wed, 8 Sep 2021 19:10:50 +0300
Subject: mapreduce tables are in colors now too

---
 internal/server/handlers/controlhandler.go |   3 +-
 internal/server/handlers/serverhandler.go  | 118 ++++++++++++++---------------
 2 files changed, 59 insertions(+), 62 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go
index a217b40..1e17c78 100644
--- a/internal/server/handlers/controlhandler.go
+++ b/internal/server/handlers/controlhandler.go
@@ -8,7 +8,6 @@ import (
 
 	"github.com/mimecast/dtail/internal"
 	"github.com/mimecast/dtail/internal/io/logger"
-	"github.com/mimecast/dtail/internal/protocol"
 	user "github.com/mimecast/dtail/internal/user/server"
 )
 
@@ -57,7 +56,7 @@ func (h *ControlHandler) Read(p []byte) (n int, err error) {
 	for {
 		select {
 		case message := <-h.serverMessages:
-			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s%b", h.hostname, message, protocol.MessageDelimiter))
+			wholePayload := []byte(fmt.Sprintf("SERVER|%s|%s\n", h.hostname, message))
 			n = copy(p, wholePayload)
 			return
 		case <-h.done.Done():
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 14f46a3..e74e686 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -38,7 +38,6 @@ type ServerHandler struct {
 	aggregate          *server.Aggregate
 	aggregatedMessages chan string
 	serverMessages     chan string
-	payload            []byte
 	hostname           string
 	user               *user.User
 	catLimiter         chan struct{}
@@ -47,6 +46,8 @@ type ServerHandler struct {
 	activeCommands     int32
 	activeReaders      int32
 	quiet              bool
+	readBuf            bytes.Buffer
+	writeBuf           bytes.Buffer
 }
 
 // NewServerHandler returns the server handler.
@@ -86,77 +87,74 @@ func (h *ServerHandler) Done() <-chan struct{} {
 
 // Read is to send data to the dtail client via Reader interface.
 func (h *ServerHandler) Read(p []byte) (n int, err error) {
-	for {
-		select {
-		case message := <-h.serverMessages:
-			if len(message) == 0 {
-				logger.Warn(h.user, "Empty message recieved")
-				return
-			}
-			if message[0] == '.' {
-				// Handle hidden message (don't display to the user, interpreted by dtail client)
-				payload := []byte(fmt.Sprintf("%s%s", message, string(protocol.MessageDelimiter)))
-				n = copy(p, payload)
-				return
-			}
-
-			// Handle normal server message (display to the user)
-			payload := []byte(fmt.Sprintf("SERVER|%s|%s%s", h.hostname, message, string(protocol.MessageDelimiter)))
-			n = copy(p, payload)
-			return
+	defer h.readBuf.Reset()
 
-		case message := <-h.aggregatedMessages:
-			// Send mapreduce-aggregated data as a message.
-			buf := pool.BytesBuffer.Get().(*bytes.Buffer)
-			buf.WriteString("AGGREGATE")
-			buf.WriteString(protocol.AggregateDelimiter)
-			buf.WriteString(h.hostname)
-			buf.WriteString(protocol.AggregateDelimiter)
-			buf.WriteString(message)
-			buf.WriteByte(protocol.MessageDelimiter)
-			n = copy(p, buf.Bytes())
-			pool.RecycleBytesBuffer(buf)
+	select {
+	case message := <-h.serverMessages:
+		if message[0] == '.' {
+			// Handle hidden message (don't display to the user, interpreted by dtail client)
+			h.readBuf.WriteString(message)
+			h.readBuf.WriteByte(protocol.MessageDelimiter)
+			n = copy(p, h.readBuf.Bytes())
 			return
+		}
 
-		case line := <-h.lines:
-			buf := pool.BytesBuffer.Get().(*bytes.Buffer)
-			buf.WriteString("REMOTE")
-			buf.WriteString(protocol.FieldDelimiter)
-			buf.WriteString(h.hostname)
-			buf.WriteString(protocol.FieldDelimiter)
-			buf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc))
-			buf.WriteString(protocol.FieldDelimiter)
-			buf.WriteString(fmt.Sprintf("%v", line.Count))
-			buf.WriteString(protocol.FieldDelimiter)
-			buf.WriteString(line.SourceID)
-			buf.WriteString(protocol.FieldDelimiter)
-			payload := append(buf.Bytes(), line.Content.Bytes()...)
-			n = copy(p, payload)
-			pool.RecycleBytesBuffer(buf)
-			pool.RecycleBytesBuffer(line.Content)
+		// Handle normal server message (display to the user)
+		h.readBuf.WriteString("SERVER")
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(h.hostname)
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(message)
+		h.readBuf.WriteByte(protocol.MessageDelimiter)
+		n = copy(p, h.readBuf.Bytes())
+
+	case message := <-h.aggregatedMessages:
+		// Send mapreduce-aggregated data as a message.
+		h.readBuf.WriteString("AGGREGATE")
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(h.hostname)
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(message)
+		h.readBuf.WriteByte(protocol.MessageDelimiter)
+		n = copy(p, h.readBuf.Bytes())
+
+	case line := <-h.lines:
+		h.readBuf.WriteString("REMOTE")
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(h.hostname)
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(fmt.Sprintf("%3d", line.TransmittedPerc))
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(fmt.Sprintf("%v", line.Count))
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(line.SourceID)
+		h.readBuf.WriteString(protocol.FieldDelimiter)
+		h.readBuf.WriteString(line.Content.String())
+		h.readBuf.WriteByte(protocol.MessageDelimiter)
+		n = copy(p, h.readBuf.Bytes())
+		pool.RecycleBytesBuffer(line.Content)
+
+	case <-time.After(time.Second):
+		// Once in a while check whether we are done.
+		select {
+		case <-h.done.Done():
+			err = io.EOF
 			return
-
-		case <-time.After(time.Second):
-			// Once in a while check whether we are done.
-			select {
-			case <-h.done.Done():
-				return 0, io.EOF
-			default:
-			}
+		default:
 		}
 	}
+	return
 }
 
 // Write is to receive data from the dtail client via Writer interface.
 func (h *ServerHandler) Write(p []byte) (n int, err error) {
-	for _, c := range p {
-		switch c {
+	for _, b := range p {
+		switch b {
 		case ';':
-			commandStr := strings.TrimSpace(string(h.payload))
-			h.handleCommand(commandStr)
-			h.payload = nil
+			h.handleCommand(string(h.writeBuf.Bytes()))
+			h.writeBuf.Reset()
 		default:
-			h.payload = append(h.payload, c)
+			h.writeBuf.WriteByte(b)
 		}
 	}
 
-- 
cgit v1.2.3


From 2ebe7e9d63ba62c6f19749c39fe0a577d86ca775 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sun, 12 Sep 2021 19:04:42 +0300
Subject: bugfix: dmap skipped the last couple of mapreduce lines

---
 internal/server/handlers/readcommand.go   | 15 ++++--
 internal/server/handlers/serverhandler.go | 86 +++++++++++--------------------
 2 files changed, 40 insertions(+), 61 deletions(-)

(limited to 'internal/server')

diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 5bab26f..69dd4a5 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/mimecast/dtail/internal/io/fs"
+	"github.com/mimecast/dtail/internal/io/line"
 	"github.com/mimecast/dtail/internal/io/logger"
 	"github.com/mimecast/dtail/internal/omode"
 	"github.com/mimecast/dtail/internal/regex"
@@ -113,16 +114,20 @@ func (r *readCommand) readFile(ctx context.Context, path, globID string, re rege
 	}
 
 	lines := r.server.lines
-
-	// Plug in mappreduce engine
-	if r.server.aggregate != nil {
-		lines = r.server.aggregate.Lines
-	}
+	aggregate := r.server.aggregate
 
 	for {
+		if aggregate != nil {
+			lines = make(chan line.Line, 100)
+			aggregate.NextLinesCh <- lines
+		}
 		if err := reader.Start(ctx, lines, re); err != nil {
 			logger.Error(r.server.user, path, globID, err)
 		}
+		if aggregate != nil {
+			// Also makes aggregate to Flush
+			close(lines)
+		}
 
 		select {
 		case <-ctx.Done():
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index e74e686..ed19412 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -32,36 +32,35 @@ const (
 // the Bi-directional communication between SSH client and server.
 // This handler implements the handler of the SSH server.
 type ServerHandler struct {
-	done               *internal.Done
-	lines              chan line.L