Reconnect on TCP disconnection from sender side

pkg/sender/tcp.go

@@ -38,6 +38,8 @@ func (s *TCPSender) Send(data []byte) error {
 	_, err := s.conn.Write(data)
 
 	if err != nil {
+		_ = s.conn.Close()
+		s.conn = nil
 		return fmt.Errorf("s.conn.Write: %w", err)
 	}
 

tests/captured/cmd.go

@@ -27,6 +27,13 @@ func (c *capturedOutput) Write(data []byte) (n int, err error) {
 	return len(data), nil
 }
 
+func (c *capturedOutput) reset() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.data = make([]byte, 0, 1024)
+}
+
 type RunningCmd struct {
 	cmd    *exec.Cmd
 	stdout *capturedOutput
@@ -54,6 +61,20 @@ func StartInBackground(ctx context.Context, command string, args ...string) (*Ru
 	return r, nil
 }
 
+func (r *RunningCmd) Stop() error {
+	if r == nil || r.cmd == nil {
+		return fmt.Errorf("cmd is nil")
+	}
+
+	err := r.cmd.Process.Kill()
+
+	if err != nil {
+		return fmt.Errorf("process.Kill: %w", err)
+	}
+
+	return nil
+}
+
 func (r *RunningCmd) Stdout() string {
 	r.stdout.mu.Lock()
 	defer r.stdout.mu.Unlock()
@@ -71,3 +92,8 @@ func (r *RunningCmd) Stderr() string {
 
 	return output
 }
+
+func (r *RunningCmd) ResetOutput() {
+	r.stdout.reset()
+	r.stderr.reset()
+}

tests/features/tcp.feature

@@ -14,6 +14,26 @@ Feature: send and receive TCP
     Then the stdout contains "connected"
     And the stdout contains "hi"
 
+  Scenario: the connection is closed
+    Given I run cyn -t 127.0.0.1:15236
+    And I run cyn -T 127.0.0.1:15236 -i 10ms
+    When I wait a moment
+    And I stop process #1
+    And I wait a moment
+    Then the stdout contains "broken pipe"
+
+  Scenario: the connection is reset
+    Given I run cyn -t 127.0.0.1:15236
+    And I run cyn -T 127.0.0.1:15236 -i 10ms
+    When I wait a moment
+    And I stop process #1
+    And I wait a moment
+    And I run cyn -t 127.0.0.1:15236
+    And I wait a moment
+    And I reset the output
+    And I wait 1 second
+    Then the stdout does not contain "broken pipe"
+
   Scenario: an instance is set to call itself via config file
     Given a configuration file that contains:
       """

tests/output_test.go

@@ -66,3 +66,15 @@ func (t *testContext) someStderrContains(output string) error {
 
 	return fmt.Errorf("failed to find %q in any stderr output", output)
 }
+
+func (t *testContext) noStdoutContains(output string) error {
+	for i, cmd := range t.cmds {
+		stdout := cmd.Stdout()
+
+		if strings.Contains(stdout, output) {
+			return fmt.Errorf("found %q in process #%d", output, i+1)
+		}
+	}
+
+	return nil
+}

tests/run_test.go

@@ -54,3 +54,30 @@ func (t *testContext) startCynInBackground(args ...string) error {
 
 	return nil
 }
+
+func (t *testContext) iStopProcess(index int) error {
+	// 1-indexed
+	index--
+
+	if len(t.cmds) <= index {
+		return fmt.Errorf("bad test: only have %d commands run but wanted to stop #%d", len(t.cmds), index+1)
+	}
+
+	cmd := t.cmds[index]
+
+	err := cmd.Stop()
+
+	if err != nil {
+		return fmt.Errorf("cmd.Stop: %w", err)
+	}
+
+	return nil
+}
+
+func (t *testContext) iResetTheOutput() error {
+	for _, cmd := range t.cmds {
+		cmd.ResetOutput()
+	}
+
+	return nil
+}

tests/steps_test.go

@@ -62,4 +62,7 @@ func InitializeScenario(sc *godog.ScenarioContext) {
 	sc.Step(`^I disconnect my TCP connection$`, t.iDisconnectMyTCPConnection)
 	sc.Step(`^some|the stdout contains "(.*)"$`, t.someStdoutContains)
 	sc.Step(`^some|the stderr contains "(.*)"$`, t.someStderrContains)
+	sc.Step(`^the stdout does not contain "(.*)"$`, t.noStdoutContains)
+	sc.Step(`^I reset the output$`, t.iResetTheOutput)
+	sc.Step(`^I stop process #(\d+)$`, t.iStopProcess)
 }