Merge pull request #381 from wneessen/bug/380_use-of-closed-network-connection-error-when-using-concurrently

Add mutex for concurrent send protection

client.go

@@ -170,6 +170,9 @@ type (
 		// requestDSN indicates wether we want to request DSN (Delivery Status Notifications).
 		requestDSN bool
 
+		// sendMutex is used to synchronize access to shared resources during the dial and send methods.
+		sendMutex sync.Mutex
+
 		// smtpAuth is the authentication type that is used to authenticate the user with SMTP server. It
 		// satisfies the smtp.Auth interface.
 		//
@@ -1058,6 +1061,8 @@ func (c *Client) DialAndSend(messages ...*Msg) error {
 //   - An error if the connection fails, if sending the messages fails, or if closing the
 //     connection fails; otherwise, returns nil.
 func (c *Client) DialAndSendWithContext(ctx context.Context, messages ...*Msg) error {
+	c.sendMutex.Lock()
+	defer c.sendMutex.Unlock()
 	if err := c.DialWithContext(ctx); err != nil {
 		return fmt.Errorf("dial failed: %w", err)
 	}

client_test.go

@@ -2297,6 +2297,45 @@ func TestClient_DialAndSendWithContext(t *testing.T) {
 			t.Errorf("client was supposed to fail on dial")
 		}
 	})
+	t.Run("concurrent sending via DialAndSendWithContext", func(t *testing.T) {
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+		PortAdder.Add(1)
+		serverPort := int(TestServerPortBase + PortAdder.Load())
+		featureSet := "250-8BITMIME\r\n250-DSN\r\n250 SMTPUTF8"
+		go func() {
+			if err := simpleSMTPServer(ctx, t, &serverProps{
+				FeatureSet: featureSet,
+				ListenPort: serverPort,
+			}); err != nil {
+				t.Errorf("failed to start test server: %s", err)
+				return
+			}
+		}()
+		time.Sleep(time.Millisecond * 30)
+
+		client, err := NewClient(DefaultHost, WithPort(serverPort), WithTLSPolicy(NoTLS))
+		if err != nil {
+			t.Fatalf("failed to create new client: %s", err)
+		}
+
+		wg := sync.WaitGroup{}
+		for i := 0; i < 50; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				msg := testMessage(t)
+				msg.SetMessageIDWithValue("this.is.a.message.id")
+
+				ctxDial, cancelDial := context.WithTimeout(ctx, time.Minute)
+				defer cancelDial()
+				if goroutineErr := client.DialAndSendWithContext(ctxDial, msg); goroutineErr != nil {
+					t.Errorf("failed to dial and send message: %s", goroutineErr)
+				}
+			}()
+		}
+		wg.Wait()
+	})
 }
 
 func TestClient_auth(t *testing.T) {