Azure Storage Queues: Make polling interval configurable (#2781)

Signed-off-by: ItalyPaleAle <43508+ItalyPaleAle@users.noreply.github.com>

bindings/azure/storagequeues/metadata.yaml

@@ -44,6 +44,12 @@ metadata:
     example: |
       "http://127.0.0.1:10001"
       "https://accountName.queue.example.com"
+  - name: "pollingInterval"
+    type: duration
+    description: |
+      Set the interval to poll Azure Storage Queues for new messages
+    example: '"30s"'
+    default: '"10s"'
   - name: "ttlInSeconds"
     type: number
     description: |
@@ -74,10 +80,9 @@ metadata:
       output: true
       input: false
   - name: "visibilityTimeout"
-    type: number
+    type: duration
     description: |
       Allows setting a custom queue visibility timeout to avoid immediate retrying of recently-failed messages.
-      Value is formatted as a Go duration, with a 30s default.
     example: '1m'
     default: '30s'
     binding:

bindings/azure/storagequeues/storagequeues.go

@@ -35,7 +35,9 @@ import (
 )
 
 const (
-	defaultTTL = time.Minute * 10
+	defaultTTL               = 10 * time.Minute
+	defaultVisibilityTimeout = 30 * time.Second
+	defaultPollingInterval   = 10 * time.Second
 )
 
 type consumer struct {
@@ -56,6 +58,7 @@ type AzureQueueHelper struct {
 	logger            logger.Logger
 	decodeBase64      bool
 	encodeBase64      bool
+	pollingInterval   time.Duration
 	visibilityTimeout time.Duration
 }
 
@@ -105,6 +108,7 @@ func (d *AzureQueueHelper) Init(ctx context.Context, meta bindings.Metadata) (*s
 
 	d.decodeBase64 = m.DecodeBase64
 	d.encodeBase64 = m.EncodeBase64
+	d.pollingInterval = m.PollingInterval
 	d.visibilityTimeout = *m.VisibilityTimeout
 	d.queueClient = queueServiceClient.NewQueueClient(m.QueueName)
 
@@ -151,9 +155,9 @@ func (d *AzureQueueHelper) Read(ctx context.Context, consumer *consumer) error {
 		return err
 	}
 	if len(res.Messages) == 0 {
-		// Queue was empty so back off by 10 seconds before trying again
+		// Queue was empty so back off seconds before trying again
 		select {
-		case <-time.After(10 * time.Second):
+		case <-time.After(d.pollingInterval):
 		case <-ctx.Done():
 		}
 		return nil
@@ -222,6 +226,7 @@ type storageQueuesMetadata struct {
 	AccountKey        string
 	DecodeBase64      bool
 	EncodeBase64      bool
+	PollingInterval   time.Duration  `mapstructure:"pollingInterval"`
 	TTL               *time.Duration `mapstructure:"ttlInSeconds"`
 	VisibilityTimeout *time.Duration
 }
@@ -257,7 +262,8 @@ func (a *AzureStorageQueues) Init(ctx context.Context, metadata bindings.Metadat
 
 func parseMetadata(meta bindings.Metadata) (*storageQueuesMetadata, error) {
 	m := storageQueuesMetadata{
-		VisibilityTimeout: ptr.Of(time.Second * 30),
+		PollingInterval:   defaultPollingInterval,
+		VisibilityTimeout: ptr.Of(defaultVisibilityTimeout),
 	}
 	contribMetadata.DecodeMetadata(meta.Properties, &m)
 
@@ -281,6 +287,10 @@ func parseMetadata(meta bindings.Metadata) (*storageQueuesMetadata, error) {
 		m.AccountKey = val
 	}
 
+	if m.PollingInterval < (100 * time.Millisecond) {
+		return nil, errors.New("invalid value for 'pollingInterval': must be greater than 100ms")
+	}
+
 	ttl, ok, err := contribMetadata.TryGetTTL(meta.Properties)
 	if err != nil {
 		return nil, err

bindings/azure/storagequeues/storagequeues_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 
 	"github.com/dapr/components-contrib/bindings"
 	"github.com/dapr/components-contrib/metadata"
@@ -96,13 +97,13 @@ func TestWriteQueue(t *testing.T) {
 	m.Properties = map[string]string{"storageAccessKey": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "queue": "queue1", "storageAccount": "devstoreaccount1"}
 
 	err := a.Init(context.Background(), m)
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	r := bindings.InvokeRequest{Data: []byte("This is my message")}
 
 	_, err = a.Invoke(context.Background(), &r)
 
-	assert.Nil(t, err)
+	require.NoError(t, err)
 	assert.NoError(t, a.Close())
 }
 
@@ -118,13 +119,13 @@ func TestWriteWithTTLInQueue(t *testing.T) {
 	m.Properties = map[string]string{"storageAccessKey": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "queue": "queue1", "storageAccount": "devstoreaccount1", metadata.TTLMetadataKey: "1"}
 
 	err := a.Init(context.Background(), m)
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	r := bindings.InvokeRequest{Data: []byte("This is my message")}
 
 	_, err = a.Invoke(context.Background(), &r)
 
-	assert.Nil(t, err)
+	require.NoError(t, err)
 	assert.NoError(t, a.Close())
 }
 
@@ -140,7 +141,7 @@ func TestWriteWithTTLInWrite(t *testing.T) {
 	m.Properties = map[string]string{"storageAccessKey": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "queue": "queue1", "storageAccount": "devstoreaccount1", metadata.TTLMetadataKey: "1"}
 
 	err := a.Init(context.Background(), m)
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	r := bindings.InvokeRequest{
 		Data:     []byte("This is my message"),
@@ -149,7 +150,7 @@ func TestWriteWithTTLInWrite(t *testing.T) {
 
 	_, err = a.Invoke(context.Background(), &r)
 
-	assert.Nil(t, err)
+	require.NoError(t, err)
 	assert.NoError(t, a.Close())
 }
 
@@ -162,13 +163,13 @@ func TestWriteWithTTLInWrite(t *testing.T) {
 	m.Properties = map[string]string{"storageAccessKey": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "queue": "queue1", "storageAccount": "devstoreaccount1"}
 
 	err := a.Init(context.Background(), m)
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	r := bindings.InvokeRequest{Data: []byte("This is my message")}
 
 	err = a.Write(&r)
 
-	assert.Nil(t, err)
+	require.NoError(t, err)
 } */
 
 func TestReadQueue(t *testing.T) {
@@ -181,14 +182,14 @@ func TestReadQueue(t *testing.T) {
 	m.Properties = map[string]string{"storageAccessKey": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "queue": "queue1", "storageAccount": "devstoreaccount1"}
 
 	err := a.Init(context.Background(), m)
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	r := bindings.InvokeRequest{Data: []byte("This is my message")}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	_, err = a.Invoke(ctx, &r)
 
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	received := 0
 	handler := func(ctx context.Context, data *bindings.ReadResponse) ([]byte, error) {
@@ -223,14 +224,14 @@ func TestReadQueueDecode(t *testing.T) {
 	m.Properties = map[string]string{"storageAccessKey": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "queue": "queue1", "storageAccount": "devstoreaccount1", "decodeBase64": "true"}
 
 	err := a.Init(context.Background(), m)
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	r := bindings.InvokeRequest{Data: []byte("VGhpcyBpcyBteSBtZXNzYWdl")}
 
 	ctx, cancel := context.WithCancel(context.Background())
 	_, err = a.Invoke(ctx, &r)
 
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	received := 0
 	handler := func(ctx context.Context, data *bindings.ReadResponse) ([]byte, error) {
@@ -263,13 +264,13 @@ func TestReadQueueDecode(t *testing.T) {
 	m.Properties = map[string]string{"storageAccessKey": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "queue": "queue1", "storageAccount": "devstoreaccount1"}
 
 	err := a.Init(context.Background(), m)
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	r := bindings.InvokeRequest{Data: []byte("This is my message")}
 
 	err = a.Write(&r)
 
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	var handler = func(data *bindings.ReadResponse) ([]byte, error) {
 		s := string(data.Data)
@@ -294,7 +295,7 @@ func TestReadQueueNoMessage(t *testing.T) {
 	m.Properties = map[string]string{"storageAccessKey": "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==", "queue": "queue1", "storageAccount": "devstoreaccount1"}
 
 	err := a.Init(context.Background(), m)
-	assert.Nil(t, err)
+	require.NoError(t, err)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	received := 0
@@ -323,6 +324,7 @@ func TestParseMetadata(t *testing.T) {
 		// expectedAccountKey       string
 		expectedQueueName         string
 		expectedQueueEndpointURL  string
+		expectedPollingInterval   time.Duration
 		expectedTTL               *time.Duration
 		expectedVisibilityTimeout *time.Duration
 	}{
@@ -332,7 +334,8 @@ func TestParseMetadata(t *testing.T) {
 			// expectedAccountKey:       "myKey",
 			expectedQueueName:         "queue1",
 			expectedQueueEndpointURL:  "",
-			expectedVisibilityTimeout: ptr.Of(30 * time.Second),
+			expectedPollingInterval:   defaultPollingInterval,
+			expectedVisibilityTimeout: ptr.Of(defaultVisibilityTimeout),
 		},
 		{
 			name:       "Accout, key, and endpoint",
@@ -340,7 +343,8 @@ func TestParseMetadata(t *testing.T) {
 			// expectedAccountKey:       "myKey",
 			expectedQueueName:         "queue1",
 			expectedQueueEndpointURL:  "https://foo.example.com:10001",
-			expectedVisibilityTimeout: ptr.Of(30 * time.Second),
+			expectedPollingInterval:   defaultPollingInterval,
+			expectedVisibilityTimeout: ptr.Of(defaultVisibilityTimeout),
 		},
 		{
 			name:       "Empty TTL",
@@ -349,7 +353,8 @@ func TestParseMetadata(t *testing.T) {
 			expectedQueueName:         "queue1",
 			expectedQueueEndpointURL:  "",
 			expectedTTL:               ptr.Of(time.Duration(0)),
-			expectedVisibilityTimeout: ptr.Of(30 * time.Second),
+			expectedPollingInterval:   defaultPollingInterval,
+			expectedVisibilityTimeout: ptr.Of(defaultVisibilityTimeout),
 		},
 		{
 			name:       "With TTL",
@@ -358,14 +363,25 @@ func TestParseMetadata(t *testing.T) {
 			expectedQueueName:         "queue1",
 			expectedTTL:               &oneSecondDuration,
 			expectedQueueEndpointURL:  "",
-			expectedVisibilityTimeout: ptr.Of(30 * time.Second),
+			expectedPollingInterval:   defaultPollingInterval,
+			expectedVisibilityTimeout: ptr.Of(defaultVisibilityTimeout),
 		},
 		{
 			name:                      "With visibility timeout",
 			properties:                map[string]string{"accessKey": "myKey", "storageAccountQueue": "queue1", "storageAccount": "devstoreaccount1", "visibilityTimeout": "5s"},
 			expectedQueueName:         "queue1",
+			expectedPollingInterval:   defaultPollingInterval,
 			expectedVisibilityTimeout: ptr.Of(5 * time.Second),
 		},
+		{
+			name:       "With polling interval",
+			properties: map[string]string{"accessKey": "myKey", "storageAccountQueue": "queue1", "storageAccount": "devstoreaccount1", "pollingInterval": "2s"},
+			// expectedAccountKey:       "myKey",
+			expectedQueueName:         "queue1",
+			expectedQueueEndpointURL:  "",
+			expectedPollingInterval:   2 * time.Second,
+			expectedVisibilityTimeout: ptr.Of(defaultVisibilityTimeout),
+		},
 	}
 
 	for _, tt := range testCases {
@@ -375,7 +391,7 @@ func TestParseMetadata(t *testing.T) {
 
 			meta, err := parseMetadata(m)
 
-			assert.Nil(t, err)
+			require.NoError(t, err)
 			// assert.Equal(t, tt.expectedAccountKey, meta.AccountKey)
 			assert.Equal(t, tt.expectedQueueName, meta.QueueName)
 			assert.Equal(t, tt.expectedTTL, meta.TTL)
@@ -383,6 +399,20 @@ func TestParseMetadata(t *testing.T) {
 			assert.Equal(t, tt.expectedVisibilityTimeout, meta.VisibilityTimeout)
 		})
 	}
+
+	t.Run("invalid pollingInterval", func(t *testing.T) {
+		m := bindings.Metadata{Base: metadata.Base{
+			Properties: map[string]string{
+				"accessKey":           "myKey",
+				"storageAccountQueue": "queue1",
+				"storageAccount":      "devstoreaccount1",
+				"pollingInterval":     "-1s",
+			},
+		}}
+
+		_, err := parseMetadata(m)
+		require.Error(t, err)
+	})
 }
 
 func TestParseMetadataWithInvalidTTL(t *testing.T) {