enable faster shutdown - useful for dgrep/dmap and dcat commands

3 files changed, 11 insertions, 13 deletions

diff --git a/internal/clients/handlers/basehandler.go b/internal/clients/handlers/basehandler.go
index 04124e7..b520c25 100644
--- a/internal/clients/handlers/basehandler.go
+++ b/internal/clients/handlers/basehandler.go
@@ -69,7 +69,7 @@ func (h *baseHandler) Write(p []byte) (n int, err error) {
 		*/
 		case '\n', protocol.MessageDelimiter:
 			message := h.receiveBuf.String()
-			h.handleMessageType(message)
+			h.handleMessage(message)
 			h.receiveBuf.Reset()
 		default:
 			h.receiveBuf.WriteByte(b)
@@ -90,9 +90,7 @@ func (h *baseHandler) Read(p []byte) (n int, err error) {
 	return
 }
 
-// Handle various message types.
-func (h *baseHandler) handleMessageType(message string) {
-	// Hidden server commands starti with a dot "."
+func (h *baseHandler) handleMessage(message string) {
 	if len(message) > 0 && message[0] == '.' {
 		h.handleHiddenMessage(message)
 		return
@@ -106,7 +104,7 @@ func (h *baseHandler) handleMessageType(message string) {
 func (h *baseHandler) handleHiddenMessage(message string) {
 	switch {
 	case strings.HasPrefix(message, ".syn close connection"):
-		h.SendMessage(".ack close connection")
+		go h.SendMessage(".ack close connection")
 		h.Shutdown()
 	}
 }
diff --git a/internal/clients/handlers/healthhandler.go b/internal/clients/handlers/healthhandler.go
index 4949985..4b16ce4 100644
--- a/internal/clients/handlers/healthhandler.go
+++ b/internal/clients/handlers/healthhandler.go
@@ -12,7 +12,6 @@ import (
 // HealthHandler is the handler used on the client side for running mapreduce aggregations.
 type HealthHandler struct {
 	baseHandler
-	HealthStatusCh chan<- int
 }
 
 // NewHealthHandler returns a new health client handler.
@@ -26,7 +25,6 @@ func NewHealthHandler(server string) *HealthHandler {
 			status:       -1,
 			done:         internal.NewDone(),
 		},
-		HealthStatusCh: make(chan int),
 	}
 }
 
@@ -34,12 +32,10 @@ func NewHealthHandler(server string) *HealthHandler {
 func (h *HealthHandler) Write(p []byte) (n int, err error) {
 	for _, b := range p {
 		switch b {
-		case '\n':
-			continue
-		case protocol.MessageDelimiter:
+		case '\n', protocol.MessageDelimiter:
 			message := h.baseHandler.receiveBuf.String()
 			dlog.Client.Debug(message)
-			h.handleHealthMessage(message)
+			h.handleMessage(message)
 			h.baseHandler.receiveBuf.Reset()
 		default:
 			h.baseHandler.receiveBuf.WriteByte(b)
@@ -49,7 +45,11 @@ func (h *HealthHandler) Write(p []byte) (n int, err error) {
 	return len(p), nil
 }
 
-func (h *HealthHandler) handleHealthMessage(message string) {
+func (h *HealthHandler) handleMessage(message string) {
+	if len(message) > 0 && message[0] == '.' {
+		h.baseHandler.handleHiddenMessage(message)
+		return
+	}
 	s := strings.Split(message, protocol.FieldDelimiter)
 	message = s[len(s)-1]
 	status := strings.Split(message, ":")
diff --git a/internal/clients/handlers/maprhandler.go b/internal/clients/handlers/maprhandler.go
index 848e7f0..d1acfbd 100644
--- a/internal/clients/handlers/maprhandler.go
+++ b/internal/clients/handlers/maprhandler.go
@@ -44,7 +44,7 @@ func (h *MaprHandler) Write(p []byte) (n int, err error) {
 			if message[0] == 'A' {
 				h.handleAggregateMessage(message)
 			} else {
-				h.baseHandler.handleMessageType(message)
+				h.baseHandler.handleMessage(message)
 			}
 			h.baseHandler.receiveBuf.Reset()
 		default: