task: replace looped time.After with tickers (task 378)

2 files changed, 27 insertions, 3 deletions

diff --git a/internal/clients/maprclient.go b/internal/clients/maprclient.go
index 2af038f..95b3a9c 100644
--- a/internal/clients/maprclient.go
+++ b/internal/clients/maprclient.go
@@ -145,11 +145,32 @@ func (c MaprClient) makeCommands() (commands []string) {
 func (c *MaprClient) periodicReportResults(ctx context.Context) {
 	rampUpSleep := c.query.Interval / 2
 	dlog.Client.Debug("Ramp up sleeping before processing mapreduce results", rampUpSleep)
-	time.Sleep(rampUpSleep)
+
+	if rampUpSleep > 0 {
+		rampUpTimer := time.NewTimer(rampUpSleep)
+		select {
+		case <-rampUpTimer.C:
+		case <-ctx.Done():
+			if !rampUpTimer.Stop() {
+				select {
+				case <-rampUpTimer.C:
+				default:
+				}
+			}
+			return
+		}
+	}
+
+	interval := c.query.Interval
+	if interval <= 0 {
+		interval = time.Second
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
 
 	for {
 		select {
-		case <-time.After(c.query.Interval):
+		case <-ticker.C:
 			dlog.Client.Debug("Gathering interim mapreduce result")
 			if err := c.reportResults(false); err != nil {
 				dlog.Client.Error("Unable to gather mapreduce result", err)
diff --git a/internal/clients/stats.go b/internal/clients/stats.go
index 7a6643b..5880fd1 100644
--- a/internal/clients/stats.go
+++ b/internal/clients/stats.go
@@ -39,6 +39,9 @@ func newTailStats(servers int) *stats {
 func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{},
 	statsCh <-chan string, quiet bool) {
 
+	ticker := time.NewTicker(3 * time.Second)
+	defer ticker.Stop()
+
 	var connectedLast int
 	for {
 		var force bool
@@ -48,7 +51,7 @@ func (s *stats) Start(ctx context.Context, throttleCh <-chan struct{},
 		case message := <-statsCh:
 			messages = append(messages, message)
 			force = true
-		case <-time.After(time.Second * 3):
+		case <-ticker.C:
 		case <-ctx.Done():
 			return
 		}