Skip to content

Commit

Permalink
Merge pull request #381 from wneessen/bug/380_use-of-closed-network-c…
Browse files Browse the repository at this point in the history
…onnection-error-when-using-concurrently

Add mutex for concurrent send protection
  • Loading branch information
wneessen authored Nov 21, 2024
2 parents 181539f + 6e9df0b commit d009698
Show file tree
Hide file tree
Showing 2 changed files with 44 additions and 0 deletions.
5 changes: 5 additions & 0 deletions client.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
//
Expand Down Expand Up @@ -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)
}
Expand Down
39 changes: 39 additions & 0 deletions client_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down

0 comments on commit d009698

Please sign in to comment.