package storage // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. See License.txt in the project root for license information. import ( "encoding/xml" "fmt" "io" "net/http" "net/url" "strconv" "time" ) const ( // casing is per Golang's http.Header canonicalizing the header names. approximateMessagesCountHeader = "X-Ms-Approximate-Messages-Count" ) // QueueAccessPolicy represents each access policy in the queue ACL. type QueueAccessPolicy struct { ID string StartTime time.Time ExpiryTime time.Time CanRead bool CanAdd bool CanUpdate bool CanProcess bool } // QueuePermissions represents the queue ACLs. type QueuePermissions struct { AccessPolicies []QueueAccessPolicy } // SetQueuePermissionOptions includes options for a set queue permissions operation type SetQueuePermissionOptions struct { Timeout uint RequestID string `header:"x-ms-client-request-id"` } // Queue represents an Azure queue. type Queue struct { qsc *QueueServiceClient Name string Metadata map[string]string AproxMessageCount uint64 } func (q *Queue) buildPath() string { return fmt.Sprintf("/%s", q.Name) } func (q *Queue) buildPathMessages() string { return fmt.Sprintf("%s/messages", q.buildPath()) } // QueueServiceOptions includes options for some queue service operations type QueueServiceOptions struct { Timeout uint RequestID string `header:"x-ms-client-request-id"` } // Create operation creates a queue under the given account. // // See https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/Create-Queue4 func (q *Queue) Create(options *QueueServiceOptions) error { params := url.Values{} headers := q.qsc.client.getStandardHeaders() headers = q.qsc.client.addMetadataToHeaders(headers, q.Metadata) if options != nil { params = addTimeout(params, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPath(), params) resp, err := q.qsc.client.exec(http.MethodPut, uri, headers, nil, q.qsc.auth) if err != nil { return err } defer drainRespBody(resp) return checkRespCode(resp, []int{http.StatusCreated}) } // Delete operation permanently deletes the specified queue. // // See https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/Delete-Queue3 func (q *Queue) Delete(options *QueueServiceOptions) error { params := url.Values{} headers := q.qsc.client.getStandardHeaders() if options != nil { params = addTimeout(params, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPath(), params) resp, err := q.qsc.client.exec(http.MethodDelete, uri, headers, nil, q.qsc.auth) if err != nil { return err } defer drainRespBody(resp) return checkRespCode(resp, []int{http.StatusNoContent}) } // Exists returns true if a queue with given name exists. func (q *Queue) Exists() (bool, error) { uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPath(), url.Values{"comp": {"metadata"}}) resp, err := q.qsc.client.exec(http.MethodGet, uri, q.qsc.client.getStandardHeaders(), nil, q.qsc.auth) if resp != nil { defer drainRespBody(resp) if resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusNotFound { return resp.StatusCode == http.StatusOK, nil } err = getErrorFromResponse(resp) } return false, err } // SetMetadata operation sets user-defined metadata on the specified queue. // Metadata is associated with the queue as name-value pairs. // // See https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/Set-Queue-Metadata func (q *Queue) SetMetadata(options *QueueServiceOptions) error { params := url.Values{"comp": {"metadata"}} headers := q.qsc.client.getStandardHeaders() headers = q.qsc.client.addMetadataToHeaders(headers, q.Metadata) if options != nil { params = addTimeout(params, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPath(), params) resp, err := q.qsc.client.exec(http.MethodPut, uri, headers, nil, q.qsc.auth) if err != nil { return err } defer drainRespBody(resp) return checkRespCode(resp, []int{http.StatusNoContent}) } // GetMetadata operation retrieves user-defined metadata and queue // properties on the specified queue. Metadata is associated with // the queue as name-values pairs. // // See https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/Set-Queue-Metadata // // Because the way Golang's http client (and http.Header in particular) // canonicalize header names, the returned metadata names would always // be all lower case. func (q *Queue) GetMetadata(options *QueueServiceOptions) error { params := url.Values{"comp": {"metadata"}} headers := q.qsc.client.getStandardHeaders() if options != nil { params = addTimeout(params, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPath(), params) resp, err := q.qsc.client.exec(http.MethodGet, uri, headers, nil, q.qsc.auth) if err != nil { return err } defer drainRespBody(resp) if err := checkRespCode(resp, []int{http.StatusOK}); err != nil { return err } aproxMessagesStr := resp.Header.Get(http.CanonicalHeaderKey(approximateMessagesCountHeader)) if aproxMessagesStr != "" { aproxMessages, err := strconv.ParseUint(aproxMessagesStr, 10, 64) if err != nil { return err } q.AproxMessageCount = aproxMessages } q.Metadata = getMetadataFromHeaders(resp.Header) return nil } // GetMessageReference returns a message object with the specified text. func (q *Queue) GetMessageReference(text string) *Message { return &Message{ Queue: q, Text: text, } } // GetMessagesOptions is the set of options can be specified for Get // Messsages operation. A zero struct does not use any preferences for the // request. type GetMessagesOptions struct { Timeout uint NumOfMessages int VisibilityTimeout int RequestID string `header:"x-ms-client-request-id"` } type messages struct { XMLName xml.Name `xml:"QueueMessagesList"` Messages []Message `xml:"QueueMessage"` } // GetMessages operation retrieves one or more messages from the front of the // queue. // // See https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/Get-Messages func (q *Queue) GetMessages(options *GetMessagesOptions) ([]Message, error) { query := url.Values{} headers := q.qsc.client.getStandardHeaders() if options != nil { if options.NumOfMessages != 0 { query.Set("numofmessages", strconv.Itoa(options.NumOfMessages)) } if options.VisibilityTimeout != 0 { query.Set("visibilitytimeout", strconv.Itoa(options.VisibilityTimeout)) } query = addTimeout(query, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPathMessages(), query) resp, err := q.qsc.client.exec(http.MethodGet, uri, headers, nil, q.qsc.auth) if err != nil { return []Message{}, err } defer resp.Body.Close() var out messages err = xmlUnmarshal(resp.Body, &out) if err != nil { return []Message{}, err } for i := range out.Messages { out.Messages[i].Queue = q } return out.Messages, err } // PeekMessagesOptions is the set of options can be specified for Peek // Messsage operation. A zero struct does not use any preferences for the // request. type PeekMessagesOptions struct { Timeout uint NumOfMessages int RequestID string `header:"x-ms-client-request-id"` } // PeekMessages retrieves one or more messages from the front of the queue, but // does not alter the visibility of the message. // // See https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/Peek-Messages func (q *Queue) PeekMessages(options *PeekMessagesOptions) ([]Message, error) { query := url.Values{"peekonly": {"true"}} // Required for peek operation headers := q.qsc.client.getStandardHeaders() if options != nil { if options.NumOfMessages != 0 { query.Set("numofmessages", strconv.Itoa(options.NumOfMessages)) } query = addTimeout(query, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPathMessages(), query) resp, err := q.qsc.client.exec(http.MethodGet, uri, headers, nil, q.qsc.auth) if err != nil { return []Message{}, err } defer resp.Body.Close() var out messages err = xmlUnmarshal(resp.Body, &out) if err != nil { return []Message{}, err } for i := range out.Messages { out.Messages[i].Queue = q } return out.Messages, err } // ClearMessages operation deletes all messages from the specified queue. // // See https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/Clear-Messages func (q *Queue) ClearMessages(options *QueueServiceOptions) error { params := url.Values{} headers := q.qsc.client.getStandardHeaders() if options != nil { params = addTimeout(params, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPathMessages(), params) resp, err := q.qsc.client.exec(http.MethodDelete, uri, headers, nil, q.qsc.auth) if err != nil { return err } defer drainRespBody(resp) return checkRespCode(resp, []int{http.StatusNoContent}) } // SetPermissions sets up queue permissions // See https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/set-queue-acl func (q *Queue) SetPermissions(permissions QueuePermissions, options *SetQueuePermissionOptions) error { body, length, err := generateQueueACLpayload(permissions.AccessPolicies) if err != nil { return err } params := url.Values{ "comp": {"acl"}, } headers := q.qsc.client.getStandardHeaders() headers["Content-Length"] = strconv.Itoa(length) if options != nil { params = addTimeout(params, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPath(), params) resp, err := q.qsc.client.exec(http.MethodPut, uri, headers, body, q.qsc.auth) if err != nil { return err } defer drainRespBody(resp) return checkRespCode(resp, []int{http.StatusNoContent}) } func generateQueueACLpayload(policies []QueueAccessPolicy) (io.Reader, int, error) { sil := SignedIdentifiers{ SignedIdentifiers: []SignedIdentifier{}, } for _, qapd := range policies { permission := qapd.generateQueuePermissions() signedIdentifier := convertAccessPolicyToXMLStructs(qapd.ID, qapd.StartTime, qapd.ExpiryTime, permission) sil.SignedIdentifiers = append(sil.SignedIdentifiers, signedIdentifier) } return xmlMarshal(sil) } func (qapd *QueueAccessPolicy) generateQueuePermissions() (permissions string) { // generate the permissions string (raup). // still want the end user API to have bool flags. permissions = "" if qapd.CanRead { permissions += "r" } if qapd.CanAdd { permissions += "a" } if qapd.CanUpdate { permissions += "u" } if qapd.CanProcess { permissions += "p" } return permissions } // GetQueuePermissionOptions includes options for a get queue permissions operation type GetQueuePermissionOptions struct { Timeout uint RequestID string `header:"x-ms-client-request-id"` } // GetPermissions gets the queue permissions as per https://docs.microsoft.com/en-us/rest/api/storageservices/fileservices/get-queue-acl // If timeout is 0 then it will not be passed to Azure func (q *Queue) GetPermissions(options *GetQueuePermissionOptions) (*QueuePermissions, error) { params := url.Values{ "comp": {"acl"}, } headers := q.qsc.client.getStandardHeaders() if options != nil { params = addTimeout(params, options.Timeout) headers = mergeHeaders(headers, headersFromStruct(*options)) } uri := q.qsc.client.getEndpoint(queueServiceName, q.buildPath(), params) resp, err := q.qsc.client.exec(http.MethodGet, uri, headers, nil, q.qsc.auth) if err != nil { return nil, err } defer resp.Body.Close() var ap AccessPolicy err = xmlUnmarshal(resp.Body, &ap.SignedIdentifiersList) if err != nil { return nil, err } return buildQueueAccessPolicy(ap, &resp.Header), nil } func buildQueueAccessPolicy(ap AccessPolicy, headers *http.Header) *QueuePermissions { permissions := QueuePermissions{ AccessPolicies: []QueueAccessPolicy{}, } for _, policy := range ap.SignedIdentifiersList.SignedIdentifiers { qapd := QueueAccessPolicy{ ID: policy.ID, StartTime: policy.AccessPolicy.StartTime, ExpiryTime: policy.AccessPolicy.ExpiryTime, } qapd.CanRead = updatePermissions(policy.AccessPolicy.Permission, "r") qapd.CanAdd = updatePermissions(policy.AccessPolicy.Permission, "a") qapd.CanUpdate = updatePermissions(policy.AccessPolicy.Permission, "u") qapd.CanProcess = updatePermissions(policy.AccessPolicy.Permission, "p") permissions.AccessPolicies = append(permissions.AccessPolicies, qapd) } return &permissions }