feat: add waitTime config field to azure queue storage event source

api/jsonschema/schema.json

@@ -978,6 +978,11 @@
         "storageAccountName": {
           "description": "StorageAccountName is the name of the storage account where the queue is. This field is necessary to access via Azure AD (managed identity) and it is ignored if ConnectionString is set.",
           "type": "string"
+        },
+        "waitTime": {
+          "description": "WaitTime is the duration (in seconds) for which the event source waits between empty results from the queue. The default value is 3 seconds.",
+          "format": "int32",
+          "type": "integer"
         }
       },
       "required": [

eventsources/sources/azurequeuestorage/start.go

@@ -90,6 +90,10 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 	}
 	var numMessages int32 = 10
 	var visibilityTimeout int32 = 120
+	var waitTime int32 = 3 // Defaults to 3 seconds
+	if el.AzureQueueStorageEventSource.WaitTime != nil {
+		waitTime = *el.AzureQueueStorageEventSource.WaitTime
+	}
 	log.Info("listening for messages on the queue...")
 	for {
 		select {
@@ -98,6 +102,7 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 			return nil
 		default:
 		}
+		log.Info("dequeing messages....")
 		messages, err := queueClient.DequeueMessages(ctx, &azqueue.DequeueMessagesOptions{
 			NumberOfMessages:  &numMessages,
 			VisibilityTimeout: &visibilityTimeout,
@@ -115,6 +120,9 @@ func (el *EventListener) StartListening(ctx context.Context, dispatch func([]byt
 				}
 			}, log)
 		}
+		if len(messages.Messages) == 0 {
+			time.Sleep(time.Second * time.Duration(waitTime))
+		}
 	}
 }
 

pkg/apis/eventsource/v1alpha1/types.go

@@ -1299,6 +1299,10 @@ type AzureQueueStorageEventSource struct {
 	// If set to true the decoding is done before the evaluation of JSONBody
 	// +optional
 	DecodeMessage bool `json:"decodeMessage,omitempty" protobuf:"bytes,8,opt,name=decodeMessage"`
+	// WaitTime is the duration (in seconds) for which the event source waits between empty results from the queue.
+	// The default value is 3 seconds.
+	// +optional
+	WaitTime *int32 `json:"waitTime,omitempty" protobuf:"varint,9,opt,name=waitTime"`
 }
 
 // StripeEventSource describes the event source for stripe webhook notifications