fix: singleflight needs check for existing #171

Author: emiago

client.go

@@ -174,6 +174,28 @@ func (c *Client) requestTransaction(ctx context.Context, req *sip.Request) (sip.
 	return c.tx.Request(ctx, req)
 }
 
+func (c *Client) newTransaction(ctx context.Context, req *sip.Request, onConnection func(conn sip.Connection) error) (sip.ClientTransaction, error) {
+	if c.TxRequester != nil {
+		return c.TxRequester.Request(ctx, req)
+	}
+
+	tx, err := c.tx.NewClientTransaction(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := onConnection(tx.Connection()); err != nil {
+		tx.Terminate()
+		return nil, err
+	}
+
+	err = tx.Init()
+	if err != nil {
+		tx.Terminate()
+	}
+	return tx, err
+}
+
 // Do request is HTTP client like Do request/response.
 // It returns on final response.
 // NOTE: Canceling ctx WILL not send Cancel Request which is needed for INVITE. Use dialog API for dealing with dialogs

client_test.go

@@ -2,9 +2,13 @@ package sipgo
 
 import (
 	"context"
+	"io"
+	"net"
 	"os"
+	"runtime"
 	"sort"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/emiago/sipgo/sip"
@@ -291,6 +295,40 @@ func TestDigestAuthLowerCase(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestClientParalelDialing(t *testing.T) {
+	if os.Getenv("TEST_INTEGRATION") == "" {
+		t.Skip("Use TEST_INTEGRATION env value to run this test")
+		return
+	}
+
+	ua, err := NewUA()
+	require.NoError(t, err)
+
+	l, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 15090})
+	require.NoError(t, err)
+	go func() {
+		io.ReadAll(l)
+	}()
+
+	c, err := NewClient(ua,
+		WithClientHostname("10.0.0.0"),
+		WithClientConnectionAddr("127.0.0.1:15080"),
+	)
+	require.NoError(t, err)
+	wg := sync.WaitGroup{}
+	for i := 0; i < 2*runtime.NumCPU(); i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			req := sip.NewRequest(sip.INVITE, sip.Uri{Host: "127.0.0.1", Port: 15090})
+			err := c.WriteRequest(req)
+			require.NoError(t, err)
+		}()
+	}
+
+	wg.Wait()
+}
+
 func BenchmarkClientTransactionRequestBuild(t *testing.B) {
 	ua, err := NewUA()
 	require.Nil(t, err)

dialog_client.go

@@ -90,6 +90,17 @@ func (s *DialogClientSession) Do(ctx context.Context, req *sip.Request) (*sip.Re
 // This ensures that you have proper request done within dialog. You should avoid setting any Dialog header (cseq, from, to, callid)
 func (s *DialogClientSession) TransactionRequest(ctx context.Context, req *sip.Request) (sip.ClientTransaction, error) {
 	s.buildReq(req)
+
+	// Overrides contact header with local connection addr
+	if cont := req.Contact(); cont.Address.Host == "" {
+		tx, err := s.UA.Client.newTransaction(ctx, req, func(conn sip.Connection) error {
+			var err error
+			cont.Address.Host, cont.Address.Port, err = sip.ParseAddr(conn.LocalAddr().String())
+			return err
+		})
+		return tx, err
+	}
+
 	// Passing option to avoid CSEQ apply
 	return s.UA.Client.TransactionRequest(ctx, req, s.requestValidate)
 }

sip/transaction_layer.go

@@ -193,6 +193,19 @@ func (txl *TransactionLayer) handleResponse(res *Response) error {
 }
 
 func (txl *TransactionLayer) Request(ctx context.Context, req *Request) (*ClientTx, error) {
+	tx, err := txl.NewClientTransaction(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := tx.Init(); err != nil {
+		tx.Terminate()
+		return nil, err
+	}
+	return tx, nil
+}
+
+func (txl *TransactionLayer) NewClientTransaction(ctx context.Context, req *Request) (*ClientTx, error) {
 	if req.IsAck() {
 		return nil, fmt.Errorf("ACK request must be sent directly through transport")
 	}
@@ -202,7 +215,11 @@ func (txl *TransactionLayer) Request(ctx context.Context, req *Request) (*Client
 		return nil, err
 	}
 
-	return txl.clientTxRequest(ctx, req, key)
+	tx, err := txl.clientTxRequest(ctx, req, key)
+	if err != nil {
+		return nil, err
+	}
+	return tx, nil
 }
 
 func (txl *TransactionLayer) clientTxRequest(ctx context.Context, req *Request, key string) (*ClientTx, error) {
@@ -232,11 +249,6 @@ func (txl *TransactionLayer) clientTxCreate(ctx context.Context, req *Request, k
 	}
 
 	tx := NewClientTx(key, req, conn, txl.log)
-	if err := tx.Init(); err != nil {
-		tx.Terminate()
-		return nil, err
-	}
-
 	return tx, nil
 }
 

sip/transport_connection_pool.go

@@ -55,23 +55,34 @@ func (p *ConnectionPool) addSingleflight(raddr Addr, laddr Addr, reuse bool, do
 
 	if laddr.Port > 0 || reuse {
 		// TODO: remplement this here to avoid type conversion
-		conn, err, shared := p.sf.Do(laddr.String()+raddr.String(), func() (any, error) {
-			return do()
-		})
-		if err != nil {
-			return nil, err
+		if c := p.Get(a); c != nil {
+			return c, nil
 		}
-		c := conn.(Connection)
 
-		if shared {
-			return c, nil
+		laddrStr := laddr.String()
+		if laddr.Port > 0 {
+			if c := p.Get(laddrStr); c != nil {
+				return c, nil
+			}
 		}
 
-		p.Lock()
-		defer p.Unlock()
+		conn, err, _ := p.sf.Do(laddrStr+a, func() (any, error) {
+			c, err := do()
+			if err != nil {
+				return nil, err
+			}
+
+			p.Lock()
+			defer p.Unlock()
 
-		p.m[a] = c
-		p.m[c.LocalAddr().String()] = c
+			p.m[a] = c
+			p.m[c.LocalAddr().String()] = c
+			return c, nil
+		})
+		if err != nil {
+			return nil, err
+		}
+		c := conn.(Connection)
 		return c, nil
 	}