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

…2996)

Signed-off-by: eduardodbr <eduardodbr@hotmail.com>

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"
+ },
+ "waitTimeInSeconds": {
+ "description": "WaitTimeInSeconds 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.WaitTimeInSeconds != nil {
+ waitTime = *el.AzureQueueStorageEventSource.WaitTimeInSeconds
+ }
  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))
+ }
  }
 }
 

examples/event-sources/azure-queue-storage.yaml

@@ -12,6 +12,8 @@ spec:
  jsonBody: true
  # DecodeMessage specifies if all the messages from AQS should be base64 decoded
  decodeMessage: false
+ # waitTimeInSeconds defines the wait time between empty reads from the queue
+ waitTimeInSeconds: 2
  # connection string contains information about K8s secret that stores the connection string
  connectionString:
  # Key within the K8s secret whose corresponding value (must be base64 encoded) is access key

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"`
+ // WaitTimeInSeconds is the duration (in seconds) for which the event source waits between empty results from the queue.
+ // The default value is 3 seconds.
+ // +optional
+ WaitTimeInSeconds *int32 `json:"waitTimeInSeconds,omitempty" protobuf:"varint,9,opt,name=waitTimeInSeconds"`
 }
 
 // StripeEventSource describes the event source for stripe webhook notifications