refactor: add context to pipe break detection; use timeout in pipe br…

…eak unit test

Signed-off-by: thediveo <thediveo@gmx.eu>

capturestream.go

@@ -13,6 +13,7 @@
 package cshargextcap
 
 import (
+	"context"
 	"os"
 	"strings"
 
@@ -70,7 +71,7 @@ func Capture(st csharg.SharkTank) int {
 	// might be idle for long times and thus we would otherwise not notice that
 	// Wireshark has already stopped capturing.
 	go func() {
-		pipe.WaitTillBreak(fifo)
+		pipe.WaitTillBreak(context.Background(), fifo)
 		cs.Stop()
 	}()
 	// ...and finally wait for the packet capture to terminate (or getting

pipe/checker_notwin.go

@@ -7,31 +7,39 @@
 package pipe
 
 import (
+	"context"
 	"os"
 
 	"golang.org/x/sys/unix"
 
 	log "github.com/sirupsen/logrus"
 )
 
-// WaitTillBreak continuously checks a fifo/pipe to see when it breaks. When
-// called, WaitTillBreak blocks until the fifo/pipe finally has broken.
+// WaitTillBreak continuously checks a fifo/pipe's producer end (writing end) to
+// see when it breaks. When called, WaitTillBreak blocks until the fifo/pipe
+// finally has broken. It also returns when the passed context is done.
 //
-// This implementation leverages [syscall.Select].
-func WaitTillBreak(fifo *os.File) {
+// This implementation leverages [unix.Poll].
+func WaitTillBreak(ctx context.Context, fifo *os.File) {
 	log.Debug("constantly monitoring packet capture fifo status...")
 	fds := []unix.PollFd{
 		{
 			Fd:     int32(fifo.Fd()),
-			Events: unix.POLLIN + unix.POLLERR,
+			Events: 0, // we're interested only in POLLERR and that is ignored here anyway.
 		},
 	}
 	for {
 		// Check the fifo becomming readable, which signals that it has been
 		// closed. In this case, ex-termi-nate ;) Oh, and remember to correctly
 		// initialize the fdset each time before calling select() ... well, just
 		// because that's a good idea to do. :(
-		n, err := unix.Poll(fds, 1000 /*ms*/)
+		n, err := unix.Poll(fds, 100 /* ms */)
+		select {
+		case <-ctx.Done():
+			log.Debug("context done while monitoring packet capture fifo")
+			return
+		default:
+		}
 		if err != nil {
 			if err == unix.EINTR {
 				continue
@@ -42,7 +50,6 @@ func WaitTillBreak(fifo *os.File) {
 		if n <= 0 {
 			continue
 		}
-		log.Debugf("poll: %+v", fds)
 		if fds[0].Revents&unix.POLLERR != 0 {
 			// Either the pipe was broken by Wireshark, or we did break it on
 			// purpose in the piping process. Anyway, we're done.

pipe/checker_notwin_test.go

@@ -5,6 +5,7 @@
 package pipe
 
 import (
+	"context"
 	"io"
 	"os"
 	"time"
@@ -13,12 +14,21 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	. "github.com/onsi/gomega/gleak"
 	. "github.com/thediveo/success"
 )
 
 var _ = Describe("pipes", func() {
 
-	It("detects on the write end when a pipe breaks", func() {
+	BeforeEach(func() {
+		goodgos := Goroutines()
+		DeferCleanup(func() {
+			Eventually(Goroutines).Within(2 * time.Second).ProbeEvery(100 * time.Millisecond).
+				ShouldNot(HaveLeaked(goodgos))
+		})
+	})
+
+	It("detects on the write end when a pipe breaks", func(ctx context.Context) {
 		// As Wireshark uses a named pipe it passes an extcap its name (path)
 		// and then expects the extcap to open this named pipe for writing
 		// packet capture data into it. For this test we simulate Wireshark
@@ -71,8 +81,11 @@ var _ = Describe("pipes", func() {
 		}()
 
 		By("waiting for pipe to break")
+		ctx, cancel := context.WithTimeout(ctx, 4*time.Second)
+		defer cancel()
 		start := time.Now()
-		WaitTillBreak(w)
+		WaitTillBreak(ctx, w)
+		Expect(ctx.Err()).To(BeNil(), "break detection failed")
 		Expect(time.Since(start).Milliseconds()).To(
 			BeNumerically(">", 1900), "false positive: pipe wasn't broken yet")
 	})