From b3c161b7887d98ab7aba2fce90c9b5965991ab62 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/handlers')

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 1eed2b211044415f5f3da0f1b154a8ac9f1967c0 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/handlers')

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 d10426ac1f380e9cfc9571856c4c7f7cedd7307a 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/handlers')

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 9ccd8ae93db0d3c6ed204114e1f1ab79f91fadf7 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/handlers')

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 f98eab5780d8bc4454cc5293628bad118f424f68 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/handlers')

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 d8c204cd68f21f66ef29bcdeb7decc715af24b6e 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/handlers')

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 4bc9e30f9742c58bdb39adfcc5aa2dc55ecb86f3 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/handlers')

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 62bf0b0e9c2d5420349ad9c365d9484fc0a1c46b 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/handlers')

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.Line
-	regex              string
-	aggregate          *server.Aggregate
-	aggregatedMessages chan string
-	serverMessages     chan string
-	hostname           string
-	user               *user.User
-	catLimiter         chan struct{}
-	tailLimiter        chan struct{}
-	ackCloseReceived   chan struct{}
-	activeCommands     int32
-	activeReaders      int32
-	quiet              bool
-	readBuf            bytes.Buffer
-	writeBuf           bytes.Buffer
+	done             *internal.Done
+	lines            chan line.Line
+	regex            string
+	aggregate        *server.Aggregate
+	maprMessages     chan string
+	serverMessages   chan string
+	hostname         string
+	user             *user.User
+	catLimiter       chan struct{}
+	tailLimiter      chan struct{}
+	ackCloseReceived chan struct{}
+	activeCommands   int32
+	quiet            bool
+	readBuf          bytes.Buffer
+	writeBuf         bytes.Buffer
 }
 
 // NewServerHandler returns the server handler.
 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,
-		regex:              ".",
-		user:               user,
+		done:             internal.NewDone(),
+		lines:            make(chan line.Line, 100),
+		serverMessages:   make(chan string, 10),
+		maprMessages:     make(chan string, 10),
+		ackCloseReceived: make(chan struct{}),
+		catLimiter:       catLimiter,
+		tailLimiter:      tailLimiter,
+		regex:            ".",
+		user:             user,
 	}
 
 	fqdn, err := os.Hostname()
@@ -108,7 +107,7 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 		h.readBuf.WriteByte(protocol.MessageDelimiter)
 		n = copy(p, h.readBuf.Bytes())
 
-	case message := <-h.aggregatedMessages:
+	case message := <-h.maprMessages:
 		// Send mapreduce-aggregated data as a message.
 		h.readBuf.WriteString("AGGREGATE")
 		h.readBuf.WriteString(protocol.FieldDelimiter)
@@ -260,14 +259,6 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 			h.shutdown()
 		}
 	}
-	readerFinished := func() {
-		if h.decrementActiveReaders() == 0 {
-			if h.aggregate == nil {
-				return
-			}
-			h.aggregate.Shutdown()
-		}
-	}
 
 	splitted := strings.Split(args[0], ":")
 	commandName := splitted[0]
@@ -289,18 +280,14 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 	case "grep", "cat":
 		command := newReadCommand(h, omode.CatClient)
 		go func() {
-			h.incrementActiveReaders()
 			command.Start(ctx, argc, args, 1)
-			readerFinished()
 			commandFinished()
 		}()
 
 	case "tail":
 		command := newReadCommand(h, omode.TailClient)
 		go func() {
-			h.incrementActiveReaders()
 			command.Start(ctx, argc, args, 10)
-			readerFinished()
 			commandFinished()
 		}()
 
@@ -315,7 +302,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 
 		h.aggregate = aggregate
 		go func() {
-			command.Start(ctx, h.aggregatedMessages)
+			command.Start(ctx, h.maprMessages)
 			commandFinished()
 		}()
 
@@ -361,15 +348,11 @@ func (h *ServerHandler) serverMessageC() chan<- string {
 	return h.serverMessages
 }
 
-func (h *ServerHandler) flush() {
-	logger.Debug(h.user, "flush()")
-
-	if h.aggregate != nil {
-		h.aggregate.Flush()
-	}
+func (h *ServerHandler) flushMessages() {
+	logger.Debug(h.user, "flushMessages()")
 
 	unsentMessages := func() int {
-		return len(h.lines) + len(h.serverMessages) + len(h.aggregatedMessages)
+		return len(h.lines) + len(h.serverMessages) + len(h.maprMessages)
 	}
 	for i := 0; i < 3; i++ {
 		if unsentMessages() == 0 {
@@ -385,7 +368,7 @@ func (h *ServerHandler) flush() {
 
 func (h *ServerHandler) shutdown() {
 	logger.Debug(h.user, "shutdown()")
-	h.flush()
+	h.flushMessages()
 
 	go func() {
 		select {
@@ -413,15 +396,6 @@ func (h *ServerHandler) decrementActiveCommands() int32 {
 	return atomic.LoadInt32(&h.activeCommands)
 }
 
-func (h *ServerHandler) incrementActiveReaders() {
-	atomic.AddInt32(&h.activeReaders, 1)
-}
-
-func (h *ServerHandler) decrementActiveReaders() int32 {
-	atomic.AddInt32(&h.activeReaders, -1)
-	return atomic.LoadInt32(&h.activeReaders)
-}
-
 func readOptions(opts []string) (map[string]string, error) {
 	options := make(map[string]string, len(opts))
 
-- 
cgit v1.2.3


From 03df87ee9464f6d7a1ca50254dcc5b81d1cf2625 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sat, 18 Sep 2021 14:41:25 +0300
Subject: add spartan mode

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

(limited to 'internal/server/handlers')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index ed19412..2f3b73b 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -45,6 +45,7 @@ type ServerHandler struct {
 	ackCloseReceived chan struct{}
 	activeCommands   int32
 	quiet            bool
+	spartan          bool
 	readBuf          bytes.Buffer
 	writeBuf         bytes.Buffer
 }
@@ -118,16 +119,18 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 		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)
+		if !h.spartan {
+			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())
@@ -275,6 +278,12 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 			h.quiet = true
 		}
 	}
+	if spartan, ok := options["spartan"]; ok {
+		if spartan == "true" {
+			logger.Debug(h.user, "Enabling spartan mode")
+			h.spartan = true
+		}
+	}
 
 	switch commandName {
 	case "grep", "cat":
@@ -397,6 +406,7 @@ func (h *ServerHandler) decrementActiveCommands() int32 {
 }
 
 func readOptions(opts []string) (map[string]string, error) {
+	logger.Debug("Parsing options", opts)
 	options := make(map[string]string, len(opts))
 
 	for _, o := range opts {
@@ -416,6 +426,7 @@ func readOptions(opts []string) (map[string]string, error) {
 			val = string(decoded)
 		}
 
+		logger.Debug("Setting option", key, val)
 		options[key] = val
 	}
 
-- 
cgit v1.2.3


From 8ba431b0953c3b1458baefaeebf4ad92916315d9 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sat, 18 Sep 2021 19:27:50 +0300
Subject: fix auto reconnect

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

(limited to 'internal/server/handlers')

diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 2f3b73b..4820476 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -331,7 +331,11 @@ func (h *ServerHandler) handleAckCommand(argc int, args []string) {
 		return
 	}
 	if args[1] == "close" && args[2] == "connection" {
-		close(h.ackCloseReceived)
+		select {
+		case <-h.ackCloseReceived:
+		default:
+			close(h.ackCloseReceived)
+		}
 	}
 }
 
-- 
cgit v1.2.3


From 3223be4cf95d0b6828196ac7a84277c18f3f5655 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sun, 19 Sep 2021 13:22:59 +0300
Subject: move args to config package

logger package rewrite as dlog
---
 internal/server/handlers/controlhandler.go | 12 ++++----
 internal/server/handlers/readcommand.go    | 26 ++++++++---------
 internal/server/handlers/serverhandler.go  | 46 +++++++++++++++---------------
 3 files changed, 42 insertions(+), 42 deletions(-)

(limited to 'internal/server/handlers')

diff --git a/internal/server/handlers/controlhandler.go b/internal/server/handlers/controlhandler.go
index 1e17c78..ae70675 100644
--- a/internal/server/handlers/controlhandler.go
+++ b/internal/server/handlers/controlhandler.go
@@ -7,7 +7,7 @@ import (
 	"strings"
 
 	"github.com/mimecast/dtail/internal"
-	"github.com/mimecast/dtail/internal/io/logger"
+	"github.com/mimecast/dtail/internal/io/dlog"
 	user "github.com/mimecast/dtail/internal/user/server"
 )
 
@@ -22,7 +22,7 @@ type ControlHandler struct {
 
 // NewControlHandler returns a new control handler.
 func NewControlHandler(user *user.User) *ControlHandler {
-	logger.Debug(user, "Creating control handler")
+	dlog.Server.Debug(user, "Creating control handler")
 
 	h := ControlHandler{
 		done:           internal.NewDone(),
@@ -32,7 +32,7 @@ func NewControlHandler(user *user.User) *ControlHandler {
 
 	fqdn, err := os.Hostname()
 	if err != nil {
-		logger.FatalExit(err)
+		dlog.Server.FatalPanic(err)
 	}
 
 	s := strings.Split(fqdn, ".")
@@ -84,15 +84,15 @@ func (h *ControlHandler) Write(p []byte) (n int, err error) {
 }
 
 func (h *ControlHandler) handleCommand(command string) {
-	logger.Info(h.user, command)
+	dlog.Server.Info(h.user, command)
 	s := strings.Split(command, " ")
-	logger.Debug(h.user, "Receiving command", command, s)
+	dlog.Server.Debug(h.user, "Receiving command", command, s)
 
 	switch s[0] {
 	case "health":
 		h.serverMessages <- "OK: DTail SSH Server seems fine"
 		h.serverMessages <- "done;"
 	default:
-		h.serverMessages <- logger.Error(h.user, "Received unknown control command", command, s)
+		h.serverMessages <- dlog.Server.Error(h.user, "Received unknown control command", command, s)
 	}
 }
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 69dd4a5..60ad2a0 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -9,7 +9,7 @@ import (
 
 	"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/io/dlog"
 	"github.com/mimecast/dtail/internal/omode"
 	"github.com/mimecast/dtail/internal/regex"
 )
@@ -32,13 +32,13 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string, retrie
 	if argc >= 4 {
 		deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " "))
 		if err != nil {
-			r.server.sendServerMessage(logger.Error(r.server.user, commandParseWarning, err))
+			r.server.sendServerMessage(dlog.Server.Error(r.server.user, commandParseWarning, err))
 			return
 		}
 		re = deserializedRegex
 	}
 	if argc < 3 {
-		r.server.sendServerWarnMessage(logger.Warn(r.server.user, commandParseWarning, args, argc))
+		r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, commandParseWarning, args, argc))
 		return
 	}
 	r.readGlob(ctx, args[1], re, retries)
@@ -51,14 +51,14 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex,
 	for retryCount := 0; retryCount < retries; retryCount++ {
 		paths, err := filepath.Glob(glob)
 		if err != nil {
-			logger.Warn(r.server.user, glob, err)
+			dlog.Server.Warn(r.server.user, glob, err)
 			time.Sleep(retryInterval)
 			continue
 		}
 
 		if numPaths := len(paths); numPaths == 0 {
-			logger.Error(r.server.user, "No such file(s) to read", glob)
-			r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs"))
+			dlog.Server.Error(r.server.user, "No such file(s) to read", glob)
+			r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs"))
 			select {
 			case <-ctx.Done():
 				return
@@ -72,7 +72,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex,
 		return
 	}
 
-	r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Giving up to read file(s)"))
+	r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Giving up to read file(s)"))
 	return
 }
 
@@ -92,8 +92,8 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGr
 	globID := r.makeGlobID(path, glob)
 
 	if !r.server.user.HasFilePermission(path, "readfiles") {
-		logger.Error(r.server.user, "No permission to read file", path, globID)
-		r.server.sendServerWarnMessage(logger.Warn(r.server.user, "Unable to read file(s), check server logs"))
+		dlog.Server.Error(r.server.user, "No permission to read file", path, globID)
+		r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs"))
 		return
 	}
 
@@ -101,7 +101,7 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGr
 }
 
 func (r *readCommand) readFile(ctx context.Context, path, globID string, re regex.Regex) {
-	logger.Info(r.server.user, "Start reading file", path, globID)
+	dlog.Server.Info(r.server.user, "Start reading file", path, globID)
 
 	var reader fs.FileReader
 	switch r.mode {
@@ -122,7 +122,7 @@ func (r *readCommand) readFile(ctx context.Context, path, globID string, re rege
 			aggregate.NextLinesCh <- lines
 		}
 		if err := reader.Start(ctx, lines, re); err != nil {
-			logger.Error(r.server.user, path, globID, err)
+			dlog.Server.Error(r.server.user, path, globID, err)
 		}
 		if aggregate != nil {
 			// Also makes aggregate to Flush
@@ -139,7 +139,7 @@ func (r *readCommand) readFile(ctx context.Context, path, globID string, re rege
 		}
 
 		time.Sleep(time.Second * 2)
-		logger.Info(path, globID, "Reading file again")
+		dlog.Server.Info(path, globID, "Reading file again")
 	}
 }
 
@@ -161,6 +161,6 @@ func (r *readCommand) makeGlobID(path, glob string) string {
 		return pathParts[len(pathParts)-1]
 	}
 
-	r.server.sendServerWarnMessage(logger.Warn("Empty file path given?", path, glob))
+	r.server.sendServerWarnMessage(dlog.Server.Warn("Empty file path given?", path, glob))
 	return ""
 }
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index 4820476..b664566 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -16,7 +16,7 @@ import (
 	"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/io/dlog"
 	"github.com/mimecast/dtail/internal/io/pool"
 	"github.com/mimecast/dtail/internal/mapr/server"
 	"github.com/mimecast/dtail/internal/omode"
@@ -66,7 +66,7 @@ func NewServerHandler(user *user.User, catLimiter, tailLimiter chan struct{}) *S
 
 	fqdn, err := os.Hostname()
 	if err != nil {
-		logger.FatalExit(err)
+		dlog.Server.FatalPanic(err)
 	}
 
 	s := strings.Split(fqdn, ".")
@@ -165,18 +165,18 @@ func (h *ServerHandler) Write(p []byte) (n int, err error) {
 }
 
 func (h *ServerHandler) handleCommand(commandStr string) {
-	logger.Debug(h.user, commandStr)
+	dlog.Server.Debug(h.user, commandStr)
 	ctx := context.Background()
 
 	args, argc, add, err := h.handleProtocolVersion(strings.Split(commandStr, " "))
 	if err != nil {
-		h.send(h.serverMessages, logger.Error(h.user, err)+add)
+		h.send(h.serverMessages, dlog.Server.Error(h.user, err)+add)
 		return
 	}
 
 	args, argc, err = h.handleBase64(args, argc)
 	if err != nil {
-		h.send(h.serverMessages, logger.Error(h.user, err))
+		h.send(h.serverMessages, dlog.Server.Error(h.user, err))
 		return
 	}
 
@@ -239,7 +239,7 @@ func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, er
 
 	args = strings.Split(decodedStr, " ")
 	argc = len(decodedStr)
-	logger.Trace(h.user, "Base64 decoded received command", decodedStr, argc, args)
+	dlog.Server.Trace(h.user, "Base64 decoded received command", decodedStr, argc, args)
 
 	return args, argc, nil
 }
@@ -247,14 +247,14 @@ func (h *ServerHandler) handleBase64(args []string, argc int) ([]string, int, er
 func (h *ServerHandler) handleControlCommand(argc int, args []string) {
 	switch args[0] {
 	case "debug":
-		h.send(h.serverMessages, logger.Debug(h.user, "Receiving debug command", argc, args))
+		h.send(h.serverMessages, dlog.Server.Debug(h.user, "Receiving debug command", argc, args))
 	default:
-		logger.Warn(h.user, "Received unknown control command", argc, args)
+		dlog.Server.Warn(h.user, "Received unknown control command", argc, args)
 	}
 }
 
 func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []string) {
-	logger.Debug(h.user, "handleUserCommand", argc, args)
+	dlog.Server.Debug(h.user, "handleUserCommand", argc, args)
 
 	h.incrementActiveCommands()
 	commandFinished := func() {
@@ -268,19 +268,19 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 
 	options, err := readOptions(splitted[1:])
 	if err != nil {
-		h.sendServerMessage(logger.Error(h.user, err))
+		h.sendServerMessage(dlog.Server.Error(h.user, err))
 		commandFinished()
 		return
 	}
 	if quiet, ok := options["quiet"]; ok {
 		if quiet == "true" {
-			logger.Debug(h.user, "Enabling quiet mode")
+			dlog.Server.Debug(h.user, "Enabling quiet mode")
 			h.quiet = true
 		}
 	}
 	if spartan, ok := options["spartan"]; ok {
 		if spartan == "true" {
-			logger.Debug(h.user, "Enabling spartan mode")
+			dlog.Server.Debug(h.user, "Enabling spartan mode")
 			h.spartan = true
 		}
 	}
@@ -304,7 +304,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 		command, aggregate, err := newMapCommand(h, argc, args)
 		if err != nil {
 			h.sendServerMessage(err.Error())
-			logger.Error(h.user, err)
+			dlog.Server.Error(h.user, err)
 			commandFinished()
 			return
 		}
@@ -320,14 +320,14 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 		commandFinished()
 
 	default:
-		h.sendServerMessage(logger.Error(h.user, "Received unknown user command", commandName, argc, args, options))
+		h.sendServerMessage(dlog.Server.Error(h.user, "Received unknown user command", commandName, argc, args, options))
 		commandFinished()
 	}
 }
 
 func (h *ServerHandler) handleAckCommand(argc int, args []string) {
 	if argc < 3 {
-		h.sendServerWarnMessage(logger.Warn(h.user, commandParseWarning, args, argc))
+		h.sendServerWarnMessage(dlog.Server.Warn(h.user, commandParseWarning, args, argc))
 		return
 	}
 	if args[1] == "close" && args[2] == "connection" {
@@ -362,25 +362,25 @@ func (h *ServerHandler) serverMessageC() chan<- string {
 }
 
 func (h *ServerHandler) flushMessages() {
-	logger.Debug(h.user, "flushMessages()")
+	dlog.Server.Debug(h.user, "flushMessages()")
 
 	unsentMessages := func() int {
 		return len(h.lines) + len(h.serverMessages) + len(h.maprMessages)
 	}
 	for i := 0; i < 3; i++ {
 		if unsentMessages() == 0 {
-			logger.Debug(h.user, "All lines sent")
+			dlog.Server.Debug(h.user, "All lines sent")
 			return
 		}
-		logger.Debug(h.user, "Still lines to be sent")
+		dlog.Server.Debug(h.user, "Still lines to be sent")
 		time.Sleep(time.Second)
 	}
 
-	logger.Warn(h.user, "Some lines remain unsent", unsentMessages())
+	dlog.Server.Warn(h.user, "Some lines remain unsent", unsentMessages())
 }
 
 func (h *ServerHandler) shutdown() {
-	logger.Debug(h.user, "shutdown()")
+	dlog.Server.Debug(h.user, "shutdown()")
 	h.flushMessages()
 
 	go func() {
@@ -393,7 +393,7 @@ func (h *ServerHandler) shutdown() {
 	select {
 	case <-h.ackCloseReceived:
 	case <-time.After(time.Second * 5):
-		logger.Debug(h.user, "Shutdown timeout reached, enforcing shutdown")
+		dlog.Server.Debug(h.user, "Shutdown timeout reached, enforcing shutdown")
 	case <-h.done.Done():
 	}
 
@@ -410,7 +410,7 @@ func (h *ServerHandler) decrementActiveCommands() int32 {
 }
 
 func readOptions(opts []string) (map[string]string, error) {
-	logger.Debug("Parsing options", opts)
+	dlog.Server.Debug("Parsing options", opts)
 	options := make(map[string]string, len(opts))
 
 	for _, o := range opts {
@@ -430,7 +430,7 @@ func readOptions(opts []string) (map[string]string, error) {
 			val = string(decoded)
 		}
 
-		logger.Debug("Setting option", key, val)
+		dlog.Server.Debug("Setting option", key, val)
 		options[key] = val
 	}
 
-- 
cgit v1.2.3


From a4d25626414ee36f937badd13e164aaf271c65d5 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sun, 26 Sep 2021 16:42:47 +0300
Subject: refactor config reader - also looks in additional search paths for
 config file unless NONE is specified

---
 internal/server/handlers/readcommand.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'internal/server/handlers')

diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 60ad2a0..c76ae2a 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -7,9 +7,9 @@ import (
 	"sync"
 	"time"
 
+	"github.com/mimecast/dtail/internal/io/dlog"
 	"github.com/mimecast/dtail/internal/io/fs"
 	"github.com/mimecast/dtail/internal/io/line"
-	"github.com/mimecast/dtail/internal/io/dlog"
 	"github.com/mimecast/dtail/internal/omode"
 	"github.com/mimecast/dtail/internal/regex"
 )
-- 
cgit v1.2.3


From dde2a1702ea1283dc5b2754f751f9381e8d0f005 Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Sat, 2 Oct 2021 11:54:07 +0300
Subject: reduce logging in serverless mode

---
 internal/server/handlers/readcommand.go   | 12 ++--
 internal/server/handlers/serverhandler.go | 93 ++++++++++---------------------
 2 files changed, 36 insertions(+), 69 deletions(-)

(limited to 'internal/server/handlers')

diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index c76ae2a..6579018 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -32,13 +32,13 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string, retrie
 	if argc >= 4 {
 		deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " "))
 		if err != nil {
-			r.server.sendServerMessage(dlog.Server.Error(r.server.user, commandParseWarning, err))
+			r.server.send(r.server.serverMessages, dlog.Server.Error(r.server.user, commandParseWarning, err))
 			return
 		}
 		re = deserializedRegex
 	}
 	if argc < 3 {
-		r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, commandParseWarning, args, argc))
+		r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, commandParseWarning, args, argc))
 		return
 	}
 	r.readGlob(ctx, args[1], re, retries)
@@ -58,7 +58,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex,
 
 		if numPaths := len(paths); numPaths == 0 {
 			dlog.Server.Error(r.server.user, "No such file(s) to read", glob)
-			r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs"))
+			r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs"))
 			select {
 			case <-ctx.Done():
 				return
@@ -72,7 +72,7 @@ func (r *readCommand) readGlob(ctx context.Context, glob string, re regex.Regex,
 		return
 	}
 
-	r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Giving up to read file(s)"))
+	r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Giving up to read file(s)"))
 	return
 }
 
@@ -93,7 +93,7 @@ func (r *readCommand) readFileIfPermissions(ctx context.Context, wg *sync.WaitGr
 
 	if !r.server.user.HasFilePermission(path, "readfiles") {
 		dlog.Server.Error(r.server.user, "No permission to read file", path, globID)
-		r.server.sendServerWarnMessage(dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs"))
+		r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to read file(s), check server logs"))
 		return
 	}
 
@@ -161,6 +161,6 @@ func (r *readCommand) makeGlobID(path, glob string) string {
 		return pathParts[len(pathParts)-1]
 	}
 
-	r.server.sendServerWarnMessage(dlog.Server.Warn("Empty file path given?", path, glob))
+	r.server.send(r.server.serverMessages, dlog.Server.Warn("Empty file path given?", path, glob))
 	return ""
 }
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index b664566..ace2626 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -15,8 +15,8 @@ import (
 
 	"github.com/mimecast/dtail/internal"
 	"github.com/mimecast/dtail/internal/config"
-	"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/mapr/server"
 	"github.com/mimecast/dtail/internal/omode"
@@ -46,6 +46,7 @@ type ServerHandler struct {
 	activeCommands   int32
 	quiet            bool
 	spartan          bool
+	serverless       bool
 	readBuf          bytes.Buffer
 	writeBuf         bytes.Buffer
 }
@@ -99,6 +100,12 @@ func (h *ServerHandler) Read(p []byte) (n int, err error) {
 			return
 		}
 
+		if h.serverless {
+			// In serverless mode we have logged the server message already via the
+			// dlog logger, no need to send the message again to the client part.
+			return
+		}
+
 		// Handle normal server message (display to the user)
 		h.readBuf.WriteString("SERVER")
 		h.readBuf.WriteString(protocol.FieldDelimiter)
@@ -266,23 +273,24 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 	splitted := strings.Split(args[0], ":")
 	commandName := splitted[0]
 
-	options, err := readOptions(splitted[1:])
+	options, err := config.DeserializeOptions(splitted[1:])
 	if err != nil {
-		h.sendServerMessage(dlog.Server.Error(h.user, err))
+		h.send(h.serverMessages, dlog.Server.Error(h.user, err))
 		commandFinished()
 		return
 	}
-	if quiet, ok := options["quiet"]; ok {
-		if quiet == "true" {
-			dlog.Server.Debug(h.user, "Enabling quiet mode")
-			h.quiet = true
-		}
+
+	if quiet, _ := options["quiet"]; quiet == "true" {
+		dlog.Server.Debug(h.user, "Enabling quiet mode")
+		h.quiet = true
 	}
-	if spartan, ok := options["spartan"]; ok {
-		if spartan == "true" {
-			dlog.Server.Debug(h.user, "Enabling spartan mode")
-			h.spartan = true
-		}
+	if spartan, _ := options["spartan"]; spartan == "true" {
+		dlog.Server.Debug(h.user, "Enabling spartan mode")
+		h.spartan = true
+	}
+	if serverless, _ := options["serverless"]; serverless == "true" {
+		dlog.Server.Debug(h.user, "Enabling serverless mode")
+		h.serverless = true
 	}
 
 	switch commandName {
@@ -303,7 +311,7 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 	case "map":
 		command, aggregate, err := newMapCommand(h, argc, args)
 		if err != nil {
-			h.sendServerMessage(err.Error())
+			h.send(h.serverMessages, err.Error())
 			dlog.Server.Error(h.user, err)
 			commandFinished()
 			return
@@ -320,14 +328,16 @@ func (h *ServerHandler) handleUserCommand(ctx context.Context, argc int, args []
 		commandFinished()
 
 	default:
-		h.sendServerMessage(dlog.Server.Error(h.user, "Received unknown user command", commandName, argc, args, options))
+		h.send(h.serverMessages, dlog.Server.Error(h.user, "Received unknown user command", commandName, argc, args, options))
 		commandFinished()
 	}
 }
 
 func (h *ServerHandler) handleAckCommand(argc int, args []string) {
 	if argc < 3 {
-		h.sendServerWarnMessage(dlog.Server.Warn(h.user, commandParseWarning, args, argc))
+		if !h.quiet {
+			h.send(h.serverMessages, dlog.Server.Warn(h.user, commandParseWarning, args, argc))
+		}
 		return
 	}
 	if args[1] == "close" && args[2] == "connection" {
@@ -346,23 +356,8 @@ func (h *ServerHandler) send(ch chan<- string, message string) {
 	}
 }
 
-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
-}
-
-func (h *ServerHandler) flushMessages() {
-	dlog.Server.Debug(h.user, "flushMessages()")
+func (h *ServerHandler) flush() {
+	dlog.Server.Debug(h.user, "flush()")
 
 	unsentMessages := func() int {
 		return len(h.lines) + len(h.serverMessages) + len(h.maprMessages)
@@ -381,11 +376,11 @@ func (h *ServerHandler) flushMessages() {
 
 func (h *ServerHandler) shutdown() {
 	dlog.Server.Debug(h.user, "shutdown()")
-	h.flushMessages()
+	h.flush()
 
 	go func() {
 		select {
-		case h.serverMessageC() <- ".syn close connection":
+		case h.serverMessages <- ".syn close connection":
 		case <-h.done.Done():
 		}
 	}()
@@ -408,31 +403,3 @@ func (h *ServerHandler) decrementActiveCommands() int32 {
 	atomic.AddInt32(&h.activeCommands, -1)
 	return atomic.LoadInt32(&h.activeCommands)
 }
-
-func readOptions(opts []string) (map[string]string, error) {
-	dlog.Server.Debug("Parsing options", opts)
-	options := make(map[string]string, len(opts))
-
-	for _, o := range opts {
-		kv := strings.SplitN(o, "=", 2)
-		if len(kv) != 2 {
-			return options, fmt.Errorf("Unable to parse options: %v", kv)
-		}
-		key := kv[0]
-		val := kv[1]
-
-		if strings.HasPrefix(val, "base64%") {
-			s := strings.SplitN(val, "%", 2)
-			decoded, err := base64.StdEncoding.DecodeString(s[1])
-			if err != nil {
-				return options, err
-			}
-			val = string(decoded)
-		}
-
-		dlog.Server.Debug("Setting option", key, val)
-		options[key] = val
-	}
-
-	return options, nil
-}
-- 
cgit v1.2.3


From 2ff78a71087d4ba95121996338d0c418b907acfe Mon Sep 17 00:00:00 2001
From: Paul Buetow <paul@buetow.org>
Date: Tue, 5 Oct 2021 10:00:38 +0300
Subject: more on this

---
 internal/server/handlers/basehandler.go   | 283 +++++++++++++++++++++++++
 internal/server/handlers/readcommand.go   |   4 +-
 internal/server/handlers/serverhandler.go | 332 +++---------------------------
 3 files changed, 319 insertions(+), 300 deletions(-)
 create mode 100644 internal/server/handlers/basehandler.go

(limited to 'internal/server/handlers')

diff --git a/internal/server/handlers/basehandler.go b/internal/server/handlers/basehandler.go
new file mode 100644
index 0000000..12fb2b3
--- /dev/null
+++ b/internal/server/handlers/basehandler.go
@@ -0,0 +1,283 @@
+package handlers
+
+import (
+	"bytes"
+	"context"
+	"encoding/base64"
+	"errors"
+	"fmt"
+	"io"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/mimecast/dtail/internal"
+	"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/mapr/server"
+	"github.com/mimecast/dtail/internal/protocol"
+	user "github.com/mimecast/dtail/internal/user/server"
+)
+
+type handleCommandCb func(context.Context, int, []string)
+
+type baseHandler struct {
+	done             *internal.Done
+	handleCommandCb  handleCommandCb
+	lines            chan line.Line
+	aggregate        *server.Aggregate
+	maprMessages     chan string
+	serverMessages   chan string
+	hostname         string
+	user             *user.User
+	ackCloseReceived chan struct{}
+	activeCommands   int32
+	quiet            bool
+	spartan          bool
+	serverless       bool
+	readBuf          bytes.Buffer
+	writeBuf         bytes.Buffer
+}
+
+// Shutdown the handler.
+func (h *baseHandler) Shutdown() {
+	h.done.Shutdown()
+}
+
+// Done channel of the handler.
+func (h *baseHandler) Done() <-chan struct{} {
+	return h.done.Done()
+}
+
+// Read is to send data to the dtail client via Reader interface.
+func (h *baseHandler) Read(p []byte) (n int, err error) {
+	defer h.readBuf.Reset()
+
+	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
+		}
+
+		if h.serverless {
+			// In serverless mode we have logged the server message already via the
+			// dlog logger, no need to send the message again to the client part.
+			return
+		}
+
+		// 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.maprMessages:
+		// 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:
+		if !h.spartan {
+			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
+		default:
+		}
+	}
+	return
+}
+
+// Write is to receive data from the dtail client via Writer interface.
+func (h *baseHandler) Write(p []byte) (n int, err error) {
+	for _, b := range p {
+		switch b {
+		case ';':
+			h.handleCommand(string(h.writeBuf.Bytes()))
+			h.writeBuf.Reset()
+		default:
+			h.writeBuf.WriteByte(b)
+		}
+	}
+
+	n = len(p)
+	return
+}
+
+func (h *baseHandler) handleCommand(commandStr string) {
+	dlog.Server.Debug(h.user, commandStr)
+
+	args, argc, add, err := h.handleProtocolVersion(strings.Split(commandStr, " "))
+	if err != nil {
+		h.send(h.serverMessages, dlog.Server.Error(h.user, err)+add)
+		return
+	}
+
+	args, argc, err = h.handleBase64(args, argc)
+	if err != nil {
+		h.send(h.serverMessages, dlog.Server.Error(h.user, err))
+		return
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-h.done.Done()
+		cancel()
+	}()
+
+	h.handleCommandCb(ctx, argc, args)
+}
+
+func (h *baseHandler) handleProtocolVersion(args []string) ([]string, int, string, error) {
+	argc := len(args)
+	var add string
+
+	if argc <= 2 || args[0] != "protocol" {
+		return args, argc, add, errors.New("unable to determine protocol version")
+	}
+
+	if args[1] != protocol.ProtocolCompat {
+		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, add, nil
+}
+
+func (h *baseHandler) handleBase64(args []string, argc int) ([]string, int, error) {
+	err := errors.New("Unable to decode client message, DTail server and client versions may not be compatible")
+
+	if argc != 2 || args[0] != "base64" {
+		return args, argc, err
+	}
+
+	decoded, err := base64.StdEncoding.DecodeString(args[1])
+	if err != nil {
+		return args, argc, err
+	}
+	decodedStr := string(decoded)
+
+	args = strings.Split(decodedStr, " ")
+	argc = len(decodedStr)
+	dlog.Server.Trace(h.user, "Base64 decoded received command", decodedStr, argc, args)
+
+	return args, argc, nil
+}
+
+func (h *baseHandler) handleAckCommand(argc int, args []string) {
+	if argc < 3 {
+		if !h.quiet {
+			h.send(h.serverMessages, dlog.Server.Warn(h.user, "Unable to parse command", args, argc))
+		}
+		return
+	}
+	if args[1] == "close" && args[2] == "connection" {
+		select {
+		case <-h.ackCloseReceived:
+		default:
+			close(h.ackCloseReceived)
+		}
+	}
+}
+
+func (h *baseHandler) send(ch chan<- string, message string) {
+	select {
+	case ch <- message:
+	case <-h.done.Done():
+	}
+}
+
+func (h *baseHandler) flush() {
+	dlog.Server.Debug(h.user, "flush()")
+
+	numUnsentMessages := func() int {
+		return len(h.lines) + len(h.serverMessages) + len(h.maprMessages)
+	}
+
+	for i := 0; i < 3; i++ {
+		if numUnsentMessages() == 0 {
+			dlog.Server.Debug(h.user, "All lines sent")
+			return
+		}
+		dlog.Server.Debug(h.user, "Still lines to be sent")
+		time.Sleep(time.Second)
+	}
+
+	dlog.Server.Warn(h.user, "Some lines remain unsent", numUnsentMessages())
+}
+
+func (h *baseHandler) shutdown() {
+	dlog.Server.Debug(h.user, "shutdown()")
+	h.flush()
+
+	go func() {
+		select {
+		case h.serverMessages <- ".syn close connection":
+		case <-h.done.Done():
+		}
+	}()
+
+	select {
+	case <-h.ackCloseReceived:
+	case <-time.After(time.Second * 5):
+		dlog.Server.Debug(h.user, "Shutdown timeout reached, enforcing shutdown")
+	case <-h.done.Done():
+	}
+
+	h.done.Shutdown()
+}
+
+func (h *baseHandler) incrementActiveCommands() {
+	atomic.AddInt32(&h.activeCommands, 1)
+}
+
+func (h *baseHandler) decrementActiveCommands() int32 {
+	atomic.AddInt32(&h.activeCommands, -1)
+	return atomic.LoadInt32(&h.activeCommands)
+}
diff --git a/internal/server/handlers/readcommand.go b/internal/server/handlers/readcommand.go
index 6579018..abc44c7 100644
--- a/internal/server/handlers/readcommand.go
+++ b/internal/server/handlers/readcommand.go
@@ -32,13 +32,13 @@ func (r *readCommand) Start(ctx context.Context, argc int, args []string, retrie
 	if argc >= 4 {
 		deserializedRegex, err := regex.Deserialize(strings.Join(args[2:], " "))
 		if err != nil {
-			r.server.send(r.server.serverMessages, dlog.Server.Error(r.server.user, commandParseWarning, err))
+			r.server.send(r.server.serverMessages, dlog.Server.Error(r.server.user, "Unable to parse command", err))
 			return
 		}
 		re = deserializedRegex
 	}
 	if argc < 3 {
-		r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, commandParseWarning, args, argc))
+		r.server.send(r.server.serverMessages, dlog.Server.Warn(r.server.user, "Unable to parse command", args, argc))
 		return
 	}
 	r.readGlob(ctx, args[1], re, retries)
diff --git a/internal/server/handlers/serverhandler.go b/internal/server/handlers/serverhandler.go
index ace2626..2ec4fbf 100644
--- a/internal/server/handlers/serverhandler.go
+++ b/internal/server/handlers/serverhandler.go
@@ -1,69 +1,60 @@
 package handlers
 
 import (
-	"bytes"
 	"context"
-	"encoding/base64"
-	"errors"
-	"fmt"
-	"io"
 	"os"
-	"strconv"
 	"strings"
-	"sync/atomic"
-	"time"
 
 	"github.com/mimecast/dtail/internal"
 	"github.com/mimecast/dtail/internal/config"
 	"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/mapr/server"
 	"github.com/mimecast/dtail/internal/omode"
-	"github.com/mimecast/dtail/internal/protocol"
 	user "github.com/mimecast/dtail/internal/user/server"
 )
 
-const (
-	commandParseWarning string = "Unable to parse command"
-)
-
 // ServerHandler implements the Reader and Writer interfaces to handle
 // 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
-	maprMessages     chan string
-	serverMessages   chan string
-	hostname         string
-	user             *user.User
-	catLimiter       chan struct{}
-	tailLimiter      chan struct{}
-	ackCloseReceived chan struct{}
-	activeCommands   int32
-	quiet            bool
-	spartan          bool
-	serverless       bool
-	readBuf          bytes.Buffer
-	writeBuf         bytes.Buffer
+	baseHandler
+	catLimiter  chan struct{}
+	tailLimiter chan struct{}
+	regex       string
+	/*
+		done             *internal.Done
+		lines            chan line.Line
+		aggregate        *server.Aggregate
+		maprMessages     chan string
+		serverMessages   chan string
+		hostname         string
+		user             *user.User
+		ackCloseReceived chan struct{}
+		activeCommands   int32
+		quiet            bool
+		spartan          bool
+		serverless       bool
+		readBuf          byt