From f9a882f5e48daf494fdc207bbcb8e6e52ac2b4a8 Mon Sep 17 00:00:00 2001
From: Ben Grewell <BGrewell@gmail.com>
Date: Wed, 10 Mar 2021 14:48:33 -0800
Subject: [PATCH] fix windows tailer closing issues

---
 .idea/workspace.xml |  3 +-
 client.go           |  1 -
 reporter.go         |  2 -
 reporter_linux.go   |  5 +--
 reporter_windows.go | 96 ++++++++++++++++++++++++---------------------
 5 files changed, 56 insertions(+), 51 deletions(-)

diff --git a/.idea/workspace.xml b/.idea/workspace.xml
index 2048bf8..5b141f7 100644
--- a/.idea/workspace.xml
+++ b/.idea/workspace.xml
@@ -3,9 +3,10 @@
   <component name="ChangeListManager">
     <list default="true" id="fc2840de-29dc-4fca-8e0e-a283562f60ca" name="Default Changelist" comment="">
       <change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/client.go" beforeDir="false" afterPath="$PROJECT_DIR$/client.go" afterDir="false" />
       <change beforePath="$PROJECT_DIR$/reporter.go" beforeDir="false" afterPath="$PROJECT_DIR$/reporter.go" afterDir="false" />
       <change beforePath="$PROJECT_DIR$/reporter_linux.go" beforeDir="false" afterPath="$PROJECT_DIR$/reporter_linux.go" afterDir="false" />
-      <change beforePath="$PROJECT_DIR$/tests/client/client.go" beforeDir="false" afterPath="$PROJECT_DIR$/tests/client/client.go" afterDir="false" />
+      <change beforePath="$PROJECT_DIR$/reporter_windows.go" beforeDir="false" afterPath="$PROJECT_DIR$/reporter_windows.go" afterDir="false" />
     </list>
     <option name="SHOW_DIALOG" value="false" />
     <option name="HIGHLIGHT_CONFLICTS" value="true" />
diff --git a/client.go b/client.go
index 650ce5d..e7e27b9 100644
--- a/client.go
+++ b/client.go
@@ -492,7 +492,6 @@ func (c *Client) Start() (err error) {
 	if err != nil {
 		return err
 	}
-	fmt.Println(cmd)
 	var exit chan int
 	c.outputStream, c.errorStream, exit, c.cancel, err = ExecuteAsyncWithCancel(cmd)
 	if err != nil {
diff --git a/reporter.go b/reporter.go
index efae6d4..b677cdd 100644
--- a/reporter.go
+++ b/reporter.go
@@ -1,7 +1,6 @@
 package iperf
 
 import (
-	"fmt"
 	"github.com/BGrewell/tail"
 )
 
@@ -22,7 +21,6 @@ func (r *Reporter) Stop() {
 	r.tailer.Stop()
 	r.tailer.Cleanup()
 	close(r.ReportingChannel)
-	fmt.Println("reporter stopped")
 }
 
 // runLogProcessor is OS specific because of differences in iperf on Windows and Linux
diff --git a/reporter_linux.go b/reporter_linux.go
index 30b845a..e36c5c4 100644
--- a/reporter_linux.go
+++ b/reporter_linux.go
@@ -45,13 +45,12 @@ Connecting to host 10.254.100.100, port 5201
 iperf Done.
 
 */
-//TODO: NEED TO UPDATE WINDOWS IMPLEMENTATION ALSO!!!!!
 func (r *Reporter) runLogProcessor() {
 	var err error
 	r.tailer, err = tail.TailFile(r.LogFile, tail.Config{
 		Follow:    true,
 		ReOpen:    true,
-		Poll:      true, // on linux we don't need to poll as the fsnotify works properly
+		Poll:      false, // on linux we don't need to poll as the fsnotify works properly
 		MustExist: true,
 	})
 	if err != nil {
@@ -120,7 +119,7 @@ func (r *Reporter) runLogProcessor() {
 						r.ReportingChannel <- report
 					}
 				}
-			case <- time.After(1 * time.Second):
+			case <- time.After(100 * time.Millisecond):
 				if !r.running {
 					return
 				}
diff --git a/reporter_windows.go b/reporter_windows.go
index cb4c81d..684e776 100644
--- a/reporter_windows.go
+++ b/reporter_windows.go
@@ -7,10 +7,12 @@ import (
 	"log"
 	"strconv"
 	"strings"
+	"time"
 )
 
 func (r *Reporter) runLogProcessor() {
-	tailer, err := tail.TailFile(r.LogFile, tail.Config{
+	var err error
+	r.tailer, err = tail.TailFile(r.LogFile, tail.Config{
 		Follow:    true,
 		ReOpen:    true,
 		Poll:      true,
@@ -19,55 +21,61 @@ func (r *Reporter) runLogProcessor() {
 	if err != nil {
 		log.Fatalf("failed to tail log file: %v", err)
 	}
-	for line := range tailer.Lines {
-		// TODO: For now this only cares about individual streams it ignores the sum lines
-		if len(line.Text) > 5 {
-			id := line.Text[1:4]
-			stream, err := strconv.Atoi(strings.TrimSpace(id))
-			if err != nil {
+
+	for {
+		select {
+		case line := <- r.tailer.Lines:
+			if line == nil {
 				continue
 			}
-			fields := strings.Fields(line.Text[5:])
-			if len(fields) >= 6 {
-				if fields[0] == "local" {
-					continue
-				}
-				timeFields := strings.Split(fields[0], "-")
-				start, err := strconv.ParseFloat(timeFields[0], 32)
-				if err != nil {
-					log.Printf("failed to convert start time: %s\n", err)
-				}
-				end, err := strconv.ParseFloat(timeFields[1], 32)
-				transferedStr := fmt.Sprintf("%s%s", fields[2], fields[3])
-				transferedBytes, err := conversions.StringBitRateToInt(transferedStr)
-				if err != nil {
-					log.Printf("failed to convert units: %s\n", err)
-				}
-				transferedBytes = transferedBytes / 8
-				rateStr := fmt.Sprintf("%s%s", fields[4], fields[5])
-				rate, err := conversions.StringBitRateToInt(rateStr)
+			if len(line.Text) > 5 {
+				id := line.Text[1:4]
+				stream, err := strconv.Atoi(strings.TrimSpace(id))
 				if err != nil {
-					log.Printf("failed to convert units: %s\n", err)
-				}
-				omitted := false
-				if len(fields) >= 7 && fields[6] == "(omitted)" {
-					omitted = true
+					continue
 				}
-				report := &StreamIntervalReport{
-					Socket:        stream,
-					StartInterval: float32(start),
-					EndInterval:   float32(end),
-					Seconds:       float32(end - start),
-					Bytes:         int(transferedBytes),
-					BitsPerSecond: float64(rate),
-					Omitted:       omitted,
+				fields := strings.Fields(line.Text[5:])
+				if len(fields) >= 6 {
+					if fields[0] == "local" {
+						continue
+					}
+					timeFields := strings.Split(fields[0], "-")
+					start, err := strconv.ParseFloat(timeFields[0], 32)
+					if err != nil {
+						log.Printf("failed to convert start time: %s\n", err)
+					}
+					end, err := strconv.ParseFloat(timeFields[1], 32)
+					transferedStr := fmt.Sprintf("%s%s", fields[2], fields[3])
+					transferedBytes, err := conversions.StringBitRateToInt(transferedStr)
+					if err != nil {
+						log.Printf("failed to convert units: %s\n", err)
+					}
+					transferedBytes = transferedBytes / 8
+					rateStr := fmt.Sprintf("%s%s", fields[4], fields[5])
+					rate, err := conversions.StringBitRateToInt(rateStr)
+					if err != nil {
+						log.Printf("failed to convert units: %s\n", err)
+					}
+					omitted := false
+					if len(fields) >= 7 && fields[6] == "(omitted)" {
+						omitted = true
+					}
+					report := &StreamIntervalReport{
+						Socket:        stream,
+						StartInterval: float32(start),
+						EndInterval:   float32(end),
+						Seconds:       float32(end - start),
+						Bytes:         int(transferedBytes),
+						BitsPerSecond: float64(rate),
+						Omitted:       omitted,
+					}
+					r.ReportingChannel <- report
 				}
-				r.ReportingChannel <- report
 			}
-		}
-
-		if !r.running {
-			return
+		case <- time.After(100 * time.Millisecond):
+			if !r.running {
+				return
+			}
 		}
 	}
 }