ethreceipts: port #188

Author: xiam

ethreceipts/ethreceipts_test.go

@@ -756,6 +756,8 @@ func TestFiltersAddDeadlock(t *testing.T) {
 }
 
 func TestFlakyProvider(t *testing.T) {
+	const subscribers = 20
+
 	t.Run("Wait for txn receipts with a healthy monitor and a healthy provider", func(t *testing.T) {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
@@ -800,7 +802,7 @@ func TestFlakyProvider(t *testing.T) {
 
 		var wg sync.WaitGroup
 
-		for i := 0; i < 10; i++ {
+		for i := 0; i < subscribers; i++ {
 			txr := &ethtxn.TransactionRequest{
 				To:       &walletBAddress,
 				ETHValue: ethtest.ETHValue(0.01),
@@ -917,7 +919,7 @@ func TestFlakyProvider(t *testing.T) {
 		var wg sync.WaitGroup
 
 		// Add a bunch of subscribers that will each send a txn and wait for it
-		for i := 0; i < 10; i++ {
+		for i := 0; i < subscribers; i++ {
 			txr := &ethtxn.TransactionRequest{
 				To:       &walletBAddress,
 				ETHValue: ethtest.ETHValue(0.01),
@@ -1034,7 +1036,7 @@ func TestFlakyProvider(t *testing.T) {
 		var wg sync.WaitGroup
 
 		// Add a bunch of subscribers that will each send a txn and wait for it
-		for i := 0; i < 10; i++ {
+		for i := 0; i < subscribers; i++ {
 			txr := &ethtxn.TransactionRequest{
 				To:       &walletBAddress,
 				ETHValue: ethtest.ETHValue(0.01),
@@ -1151,7 +1153,7 @@ func TestFlakyProvider(t *testing.T) {
 		var wg sync.WaitGroup
 
 		// Add a bunch of subscribers that will each send a txn and wait for it
-		for i := 0; i < 10; i++ {
+		for i := 0; i < subscribers; i++ {
 			txr := &ethtxn.TransactionRequest{
 				To:       &walletBAddress,
 				ETHValue: ethtest.ETHValue(0.01),

ethrpc/ethrpc.go

@@ -28,7 +28,7 @@ type Provider struct {
 	log                 logger.Logger
 	nodeURL             string
 	nodeWSURL           string
-	httpClient          httpClient
+	httpClient          atomic.Value
 	br                  breaker.Breaker
 	jwtToken            string // optional
 	streamClosers       []StreamCloser
@@ -47,17 +47,20 @@ type Provider struct {
 func NewProvider(nodeURL string, options ...Option) (*Provider, error) {
 	p := &Provider{
 		nodeURL: nodeURL,
-		httpClient: &http.Client{
-			// default timeout of 60 seconds
-			Timeout: 60 * time.Second,
-		},
 	}
 	for _, opt := range options {
 		if opt == nil {
 			continue
 		}
 		opt(p)
 	}
+	if p.httpClient.Load() == nil {
+		httpClient := &http.Client{
+			// default timeout of 60 seconds
+			Timeout: 60 * time.Second,
+		}
+		p.httpClient.Store(httpClient)
+	}
 	return p, nil
 }
 
@@ -85,8 +88,12 @@ type StreamUnsubscriber interface {
 	Unsubscribe()
 }
 
-func (s *Provider) SetHTTPClient(httpClient *http.Client) {
-	s.httpClient = httpClient
+func (s *Provider) SetHTTPClient(client httpClient) {
+	s.httpClient.Store(client)
+}
+
+func (p *Provider) getHTTPClient() httpClient {
+	return p.httpClient.Load().(httpClient)
 }
 
 func (p *Provider) StrictnessLevel() StrictnessLevel {
@@ -126,7 +133,8 @@ func (p *Provider) Do(ctx context.Context, calls ...Call) ([]byte, error) {
 		req.Header.Set("Authorization", fmt.Sprintf("BEARER %s", p.jwtToken))
 	}
 
-	res, err := p.httpClient.Do(req)
+	httpClient := p.getHTTPClient()
+	res, err := httpClient.Do(req)
 	if err != nil {
 		return nil, superr.Wrap(ErrRequestFail, fmt.Errorf("failed to send request: %w", err))
 	}

ethrpc/option.go

@@ -25,7 +25,7 @@ func WithStreaming(nodeWebsocketURL string) Option {
 
 func WithHTTPClient(c httpClient) Option {
 	return func(p *Provider) {
-		p.httpClient = c
+		p.httpClient.Store(c)
 	}
 }