nextroll_erlcloud

%% Amazon Simple Queue Service (SQS)

-module(erlcloud_sqs).

-export([configure/2, configure/3, new/2, new/3]).

-export([
         add_permission/3, add_permission/4,
         change_message_visibility/3, change_message_visibility/4,
         create_queue/1, create_queue/2, create_queue/3,
         create_fifo_queue/1, create_fifo_queue/2, create_fifo_queue/3,
         delete_message/2, delete_message/3,
         delete_queue/1, delete_queue/2,
         purge_queue/1, purge_queue/2,
         get_queue_url/1, get_queue_url/2,
         get_queue_attributes/1, get_queue_attributes/2, get_queue_attributes/3,
         list_queues/0, list_queues/1, list_queues/2,
         receive_message/1, receive_message/2, receive_message/3, receive_message/4,
         receive_message/5, receive_message/6, receive_message/7,
         remove_permission/2, remove_permission/3,
         send_message/2, send_message/3, send_message/4, send_message/5,
         set_queue_attributes/2, set_queue_attributes/3,
         send_message_batch/2, send_message_batch/3, send_message_batch/4,
         delete_message_batch/2, delete_message_batch/3,
         change_message_visibility_batch/3, change_message_visibility_batch/4
        ]).

-include("erlcloud.hrl").
-include("erlcloud_aws.hrl").

-define(API_VERSION, "2012-11-05").

-define(SEND_BATCH_FIELD(N, F, V), {"SendMessageBatchRequestEntry." ++ N ++ F, V}).
-define(DELETE_BATCH_FIELD(N, F, V), {"DeleteMessageBatchRequestEntry." ++ N ++ F, V}).
-define(CHANGE_VISIBILITY_BATCH_FIELD(N, F, V), {"ChangeMessageVisibilityBatchRequestEntry." ++ N ++ F, V}).

-type(sqs_permission() :: all | send_message | receive_message | delete_message |
                          change_message_visibility | get_queue_attributes).
-type(sqs_acl() :: [{string(), sqs_permission()}]).
-type(sqs_msg_attribute_name() :: all | sender_id | sent_timestamp |
                                  approximate_receive_count |
                                  approximate_first_receive_timestamp |
                                  wait_time_seconds |
                                  receive_message_wait_time_seconds).
-type(sqs_queue_attribute_name() :: all | approximate_number_of_messages |
                                    kms_master_key_id | kms_data_key_reuse_period_seconds |
                                    approximate_number_of_messages_not_visible | visibility_timeout |
                                    created_timestamp | last_modified_timestamp | policy |
                                    queue_arn).

-type(batch_entry() ::   {string(), string()}
                       | {string(), string(), [message_attribute()]}
                       | {string(), string(), [message_attribute()], proplists:proplist()}).
-type(message_attribute() :: {string(), string() | integer() | float() | binary()}).

-spec new(string(), string()) -> aws_config().
new(AccessKeyID, SecretAccessKey) ->
    #aws_config{access_key_id=AccessKeyID,
                secret_access_key=SecretAccessKey}.

-spec new(string(), string(), string()) -> aws_config().
new(AccessKeyID, SecretAccessKey, Host) ->
    #aws_config{access_key_id=AccessKeyID,
                secret_access_key=SecretAccessKey,
                sqs_host=Host}.

-spec configure(string(), string()) -> ok.
configure(AccessKeyID, SecretAccessKey) ->
    put(aws_config, new(AccessKeyID, SecretAccessKey)),
    ok.

-spec configure(string(), string(), string()) -> ok.
configure(AccessKeyID, SecretAccessKey, Host) ->
    put(aws_config, new(AccessKeyID, SecretAccessKey, Host)),
    ok.


-spec add_permission(string(), string(), sqs_acl()) -> ok | no_return().
add_permission(QueueName, Label, Permissions) ->
    add_permission(QueueName, Label, Permissions, default_config()).

-spec add_permission(string(), string(), sqs_acl(), aws_config()) -> ok | no_return().
add_permission(QueueName, Label, Permissions, Config)
  when is_list(QueueName),
       is_list(Label), length(Label) =< 80,
       is_list(Permissions) ->
    sqs_simple_request(Config, QueueName, "AddPermission",
                       [{"Label", Label}|erlcloud_aws:param_list(encode_permissions(Permissions), {"AWSAccountId", "ActionName"})]).

encode_permissions(Permissions) ->
    [encode_permission(P) || P <- Permissions].

encode_permission({AccountId, Permission}) ->
    {AccountId,
     case Permission of
         all -> "*";
         send_message -> "SendMessage";
         receive_message -> "ReceiveMessage";
         delete_message -> "DeleteMessage";
         change_message_visibility -> "ChangeMessageVisibility";
         get_queue_attributes -> "GetQueueAttributes"
     end}.

-spec change_message_visibility(string(), string(), 0..43200) -> ok | no_return().
change_message_visibility(QueueName, ReceiptHandle, VisibilityTimeout) ->
    change_message_visibility(QueueName, ReceiptHandle, VisibilityTimeout,
                              default_config()).

-spec change_message_visibility(string(), string(), 0..43200, aws_config()) -> ok | no_return().
change_message_visibility(QueueName, ReceiptHandle, VisibilityTimeout, Config) ->
    sqs_simple_request(Config, QueueName, "ChangeMessageVisibility",
                       [{"ReceiptHandle", ReceiptHandle}, {"VisibilityTimeout", VisibilityTimeout}]).

-spec create_queue(string()) -> proplist() | no_return().
create_queue(QueueName) ->
    create_queue(QueueName, default_config()).

-spec create_queue(string(), 0..43200 | none | aws_config()) -> proplist() | no_return().
create_queue(QueueName, Config)
  when is_record(Config, aws_config) ->
    create_queue(QueueName, none, Config);
create_queue(QueueName, DefaultVisibilityTimeout) ->
    create_queue(QueueName, DefaultVisibilityTimeout, default_config()).

-spec create_queue(string(), 0..43200 | none, aws_config()) -> proplist() | no_return().
create_queue(QueueName, DefaultVisibilityTimeout, Config) ->
    create_queue_impl(QueueName, DefaultVisibilityTimeout, Config, []).

-spec create_fifo_queue(string()) -> proplist() | no_return().
create_fifo_queue(QueueName) ->
    create_fifo_queue(QueueName, default_config()).

-spec create_fifo_queue(string(), 0..43200 | none | aws_config()) -> proplist() | no_return().
create_fifo_queue(QueueName, Config)
  when is_record(Config, aws_config) ->
    create_fifo_queue(QueueName, none, Config);
create_fifo_queue(QueueName, DefaultVisibilityTimeout) ->
    create_fifo_queue(QueueName, DefaultVisibilityTimeout, default_config()).

-spec create_fifo_queue(string(), 0..43200 | none, aws_config()) -> proplist() | no_return().
create_fifo_queue(QueueName, DefaultVisibilityTimeout, Config) ->
    Attributes = erlcloud_aws:param_list([[{"Name", "FifoQueue"}, {"Value", true}]], "Attribute"),
    create_queue_impl(QueueName, DefaultVisibilityTimeout, Config, Attributes).

-spec create_queue_impl(string(), 0..43200 | none, aws_config(), proplists:proplist()) -> proplist() | no_return().
create_queue_impl(QueueName, DefaultVisibilityTimeout, Config, Attributes)
  when is_list(QueueName),
       (is_integer(DefaultVisibilityTimeout) andalso
        DefaultVisibilityTimeout >= 0 andalso
        DefaultVisibilityTimeout =< 43200) orelse
       DefaultVisibilityTimeout =:= none ->
    Doc = sqs_xml_request(Config, "/", "CreateQueue",
                          [{"QueueName", QueueName},
                           {"DefaultVisibilityTimeout", DefaultVisibilityTimeout} | Attributes]),
    erlcloud_xml:decode(
      [
       {queue_url, "CreateQueueResult/QueueUrl", text}
      ],
      Doc
     ).

-spec delete_message(string(), string()) -> ok | no_return().
delete_message(QueueName, ReceiptHandle) ->
    delete_message(QueueName, ReceiptHandle, default_config()).

-spec delete_message(string(), string(), aws_config()) -> ok | no_return().
delete_message(QueueName, ReceiptHandle, Config)
  when is_list(QueueName), is_list(ReceiptHandle), is_record(Config, aws_config) ->
    sqs_simple_request(Config, QueueName, "DeleteMessage",
                       [{"ReceiptHandle", ReceiptHandle}]).

-spec delete_queue(string()) -> ok | no_return().
delete_queue(QueueName) ->
    delete_queue(QueueName, default_config()).

-spec delete_queue(string(), aws_config()) -> ok | no_return().
delete_queue(QueueName, Config)
  when is_list(QueueName), is_record(Config, aws_config) ->
    sqs_simple_request(Config, QueueName, "DeleteQueue", []).

-spec purge_queue(string()) -> ok | no_return().
purge_queue(QueueName) ->
    purge_queue(QueueName, default_config()).

-spec purge_queue(string(), aws_config()) -> ok | no_return().
purge_queue(QueueName, Config)
  when is_list(QueueName), is_record(Config, aws_config) ->
    sqs_simple_request(Config, QueueName, "PurgeQueue", []).

-spec get_queue_url(string()) -> proplist() | no_return().
get_queue_url(QueueName) ->
    get_queue_url(QueueName, default_config()).

-spec get_queue_url(string(), aws_config()) -> proplist() | no_return().
get_queue_url(QueueName, Config) ->
    Doc = sqs_xml_request(Config, "/", "GetQueueUrl", [{"QueueName", QueueName}]),
    erlcloud_xml:decode(
        [
            {queue_url, "GetQueueUrlResult/QueueUrl", text}
        ],
        Doc
    ).

-spec get_queue_attributes(string()) -> proplist() | no_return().
get_queue_attributes(QueueName) ->
    get_queue_attributes(QueueName, all).

-spec get_queue_attributes(string(), all | [sqs_queue_attribute_name()] | aws_config()) -> proplist() | no_return().
get_queue_attributes(QueueName, Config)
  when is_record(Config, aws_config) ->
    get_queue_attributes(QueueName, all, Config);
get_queue_attributes(QueueName, AttributeNames) ->
    get_queue_attributes(QueueName, AttributeNames, default_config()).

-spec get_queue_attributes(string(), all | [sqs_queue_attribute_name()], aws_config()) -> proplist() | no_return().
get_queue_attributes(QueueName, all, Config) when is_record(Config, aws_config) ->
    get_queue_attributes(QueueName, [all], Config);
get_queue_attributes(QueueName, AttributeNames, Config)
  when is_list(QueueName), is_list(AttributeNames), is_record(Config, aws_config) ->
    Doc = sqs_xml_request(Config, QueueName, "GetQueueAttributes",
                          erlcloud_aws:param_list([encode_attribute_name(N) || N <- AttributeNames], "AttributeName")),
    Attrs = decode_attributes(xmerl_xpath:string("GetQueueAttributesResult/Attribute", Doc)),
    [{decode_attribute_name(Name), decode_attribute_value(Name, Value)} || {Name, Value} <- Attrs].


encode_attribute_name(message_retention_period) -> "MessageRetentionPeriod";
encode_attribute_name(queue_arn) -> "QueueArn";
encode_attribute_name(maximum_message_size) -> "MaximumMessageSize";
encode_attribute_name(visibility_timeout) -> "VisibilityTimeout";
encode_attribute_name(approximate_number_of_messages) -> "ApproximateNumberOfMessages";
encode_attribute_name(approximate_number_of_messages_not_visible) -> "ApproximateNumberOfMessagesNotVisible";
encode_attribute_name(approximate_number_of_messages_delayed) -> "ApproximateNumberOfMessagesDelayed";
encode_attribute_name(last_modified_timestamp) -> "LastModifiedTimestamp";
encode_attribute_name(created_timestamp) -> "CreatedTimestamp";
encode_attribute_name(delay_seconds) -> "DelaySeconds";
encode_attribute_name(receive_message_wait_time_seconds) -> "ReceiveMessageWaitTimeSeconds";
encode_attribute_name(policy) -> "Policy";
encode_attribute_name(redrive_policy) -> "RedrivePolicy";
encode_attribute_name(kms_master_key_id) -> "KmsMasterKeyId";
encode_attribute_name(kms_data_key_reuse_period_seconds) -> "KmsDataKeyReusePeriodSeconds";
encode_attribute_name(all) -> "All".


decode_attribute_name("MessageRetentionPeriod") -> message_retention_period;
decode_attribute_name("QueueArn") -> queue_arn;
decode_attribute_name("MaximumMessageSize") -> maximum_message_size;
decode_attribute_name("VisibilityTimeout") -> visibility_timeout;
decode_attribute_name("ApproximateNumberOfMessages") -> approximate_number_of_messages;
decode_attribute_name("ApproximateNumberOfMessagesNotVisible") -> approximate_number_of_messages_not_visible;
decode_attribute_name("ApproximateNumberOfMessagesDelayed") -> approximate_number_of_messages_delayed;
decode_attribute_name("LastModifiedTimestamp") -> last_modified_timestamp;
decode_attribute_name("CreatedTimestamp") -> created_timestamp;
decode_attribute_name("DelaySeconds") -> delay_seconds;
decode_attribute_name("ReceiveMessageWaitTimeSeconds") -> receive_message_wait_time_seconds;
decode_attribute_name("Policy") -> policy;
decode_attribute_name("RedrivePolicy") -> redrive_policy;
decode_attribute_name("ContentBasedDeduplication") -> content_based_deduplication;
decode_attribute_name("KmsMasterKeyId") -> kms_master_key_id;
decode_attribute_name("KmsDataKeyReusePeriodSeconds") -> kms_data_key_reuse_period_seconds;
decode_attribute_name("FifoQueue") -> fifo_queue.


decode_attribute_value("Policy", Value) -> Value;
decode_attribute_value("QueueArn", Value) -> Value;
decode_attribute_value("RedrivePolicy", Value) -> Value;
decode_attribute_value("KmsMasterKeyId", Value) -> Value;
decode_attribute_value(_, "true") -> true;
decode_attribute_value(_, "false") -> false;
decode_attribute_value(_, Value) -> list_to_integer(Value).


-spec list_queues() -> [string()] | no_return().
list_queues() ->
    list_queues("").

-spec list_queues(string() | aws_config()) -> [string()] | no_return().
list_queues(Config)
  when is_record(Config, aws_config) ->
    list_queues("", Config);
list_queues(QueueNamePrefix) ->
    list_queues(QueueNamePrefix, default_config()).

-spec list_queues(string(), aws_config()) -> [string()] | no_return().
list_queues(QueueNamePrefix, Config)
  when is_list(QueueNamePrefix), is_record(Config, aws_config) ->
    Doc = sqs_xml_request(Config, "/", "ListQueues",
                          [{"QueueNamePrefix", QueueNamePrefix}]),
    erlcloud_xml:get_list("ListQueuesResult/QueueUrl", Doc).

-spec receive_message(string()) -> proplist() | no_return().
receive_message(QueueName) ->
    receive_message(QueueName, default_config()).

-spec receive_message(string(), [sqs_msg_attribute_name()] | all | aws_config()) -> proplist() | no_return().
receive_message(QueueName, Config)
  when is_record(Config, aws_config) ->
    receive_message(QueueName, [], Config);
receive_message(QueueName, AttributeNames) ->
    receive_message(QueueName, AttributeNames, default_config()).

-spec receive_message(string(), [sqs_msg_attribute_name()] | all, 1..10 | aws_config()) -> proplist() | no_return().
receive_message(QueueName, AttributeNames, Config)
  when is_record(Config, aws_config) ->
    receive_message(QueueName, AttributeNames, 1, Config);
receive_message(QueueName, AttributeNames, MaxNumberOfMessages) ->
    receive_message(QueueName, AttributeNames, MaxNumberOfMessages, default_config()).

-spec receive_message(string(), [sqs_msg_attribute_name()] | all, 1..10, 0..43200 | none | aws_config()) -> proplist() | no_return().
receive_message(QueueName, AttributeNames, MaxNumberOfMessages, Config)
  when is_record(Config, aws_config) ->
    receive_message(QueueName, AttributeNames, MaxNumberOfMessages, none, Config);
receive_message(QueueName, AttributeNames, MaxNumberOfMessages, VisibilityTimeout) ->
    receive_message(QueueName, AttributeNames, MaxNumberOfMessages,
                    VisibilityTimeout, default_config()).

-spec receive_message(string(), [sqs_msg_attribute_name()] | all, 1..10,
                      0..43200 | none, 0..20 | none | aws_config()) -> proplist() | no_return().
receive_message(QueueName, AttributeNames, MaxNumberOfMessages,
                VisibilityTimeout, Config)
  when is_record(Config, aws_config) ->
    receive_message(QueueName, AttributeNames, MaxNumberOfMessages,
                    VisibilityTimeout, none, Config);
receive_message(QueueName, AttributeNames, MaxNumberOfMessages,
                VisibilityTimeout, WaitTimeSeconds) ->
    receive_message(QueueName, AttributeNames, MaxNumberOfMessages,
                    VisibilityTimeout, WaitTimeSeconds, default_config()).

-spec receive_message(string(), [sqs_msg_attribute_name()] | all, 1..10,
                      0..43200 | none, 0..20 | none, aws_config()) -> proplist() | no_return().
receive_message(QueueName, all, MaxNumberOfMessages, VisibilityTimeout,
                WaitTimeoutSeconds, Config)  when is_record(Config, aws_config) ->
    receive_message(QueueName, [all], MaxNumberOfMessages,
                    VisibilityTimeout, WaitTimeoutSeconds, Config);
receive_message(QueueName, AttributeNames, MaxNumberOfMessages,
                VisibilityTimeout, WaitTimeSeconds, Config) ->
    receive_message(QueueName, AttributeNames, MaxNumberOfMessages, VisibilityTimeout,
                    WaitTimeSeconds, [all], Config).

receive_message(QueueName, AttributeNames, MaxNumberOfMessages,
                VisibilityTimeout, WaitTimeSeconds, MessageAttributeNames, Config)
  when is_record(Config, aws_config), (is_list(AttributeNames) orelse AttributeNames =:= all),
       MaxNumberOfMessages >= 1, MaxNumberOfMessages =< 10,
       (VisibilityTimeout >= 0 andalso VisibilityTimeout =< 43200) orelse
       VisibilityTimeout =:= none,
       (WaitTimeSeconds >= 0 andalso WaitTimeSeconds =< 20) orelse
       WaitTimeSeconds =:= none,
       (is_list(MessageAttributeNames) orelse MessageAttributeNames =:= all) ->
    InitialTimeout = erlcloud_aws:get_timeout(Config),
    TotalTimeout = if
       (WaitTimeSeconds =/= none andalso WaitTimeSeconds >= 0 andalso InitialTimeout =/= infinity) ->
           InitialTimeout + (WaitTimeSeconds * 1000) ;
       true -> InitialTimeout
    end,
    Doc = sqs_xml_request(Config#aws_config{timeout=TotalTimeout}, QueueName, "ReceiveMessage",
                          [
                           {"MaxNumberOfMessages", MaxNumberOfMessages},
                           {"VisibilityTimeout", VisibilityTimeout},
                           {"WaitTimeSeconds", WaitTimeSeconds} |
                           erlcloud_aws:param_list(encode_msg_attribute_names(MessageAttributeNames), "MessageAttributeName") ++
                               erlcloud_aws:param_list(encode_msg_attribute_names(AttributeNames), "AttributeName")
                          ]
                         ),
    erlcloud_xml:decode(
      [
       {messages, "ReceiveMessageResult/Message", fun decode_messages/1}
      ],
      Doc
     ).


encode_msg_attribute_names(all) ->
    [encode_msg_attribute_name(all)];
encode_msg_attribute_names(Names) when is_list(Names) ->
    [encode_msg_attribute_name(Name) || Name <- Names].

encode_msg_attribute_name(all) -> "All";
encode_msg_attribute_name(sender_id) -> "SenderId";
encode_msg_attribute_name(sent_timestamp) -> "SentTimestamp";
encode_msg_attribute_name(message_group_id) -> "MessageGroupId";
encode_msg_attribute_name(message_deduplication_id) -> "MessageDeduplicationId";
encode_msg_attribute_name(approximate_receive_count) -> "ApproximateReceiveCount";
encode_msg_attribute_name(approximate_first_receive_timestamp) -> "ApproximateFirstReceiveTimestamp";
encode_msg_attribute_name(Name) when is_list(Name) -> Name.

decode_msg_attribute_name("SenderId") -> sender_id;
decode_msg_attribute_name("SentTimestamp") -> sent_timestamp;
decode_msg_attribute_name("MessageGroupId") -> message_group_id;
decode_msg_attribute_name("MessageDeduplicationId") -> message_deduplication_id;
decode_msg_attribute_name("ApproximateReceiveCount") -> approximate_receive_count;
decode_msg_attribute_name("ApproximateFirstReceiveTimestamp") -> approximate_first_receive_timestamp;
decode_msg_attribute_name(Name) when is_list(Name) -> Name.

decode_msg_attribute_value("SenderId", Value) -> Value;
decode_msg_attribute_value("MessageGroupId", Value) -> Value;
decode_msg_attribute_value("MessageDeduplicationId", Value) -> Value;
decode_msg_attribute_value(_Name, Value) -> list_to_integer(Value).

decode_messages(Messages) ->
    [decode_message(Message) || Message <- Messages].

decode_message(Message) ->
    erlcloud_xml:decode(
      [
       {body, "Body", text},
       {md5_of_body, "MD5OfBody", text},
       {message_id, "MessageId", text},
       {receipt_handle, "ReceiptHandle", text},
       {attributes, "Attribute", fun decode_msg_attributes/1},
       {message_attributes, "MessageAttribute", fun decode_message_attributes/1}
      ],
      Message
     ).

decode_message_attributes(Attributes) ->
    [{decode_msg_attribute_name(Name), decode_message_attribute_value(DataType, StringValue)} ||
        {Name, DataType, StringValue} <- decode_message_attribute(Attributes)].

decode_message_attribute(Attributes) ->
    F = fun(Attr) ->
                Name = erlcloud_xml:get_text("Name", Attr),
                DataType = erlcloud_xml:get_text("Value/DataType", Attr),
                Value = case string:rstr(DataType, "Binary") of
                            1 ->
                                erlcloud_xml:get_text("Value/BinaryValue", Attr);
                            _ ->
                                erlcloud_xml:get_text("Value/StringValue", Attr)
                        end,
                {Name, DataType, Value}
        end,
    [F(Attr) || Attr <- Attributes].

decode_message_attribute_value("Number", Value) ->
    list_to_integer(Value);
decode_message_attribute_value(["Number", "int"], Value) ->
    list_to_integer(Value);
decode_message_attribute_value(["Number", "int", CustomType], Value) ->
    {CustomType, list_to_integer(Value)};
decode_message_attribute_value(["Number", "float"], Value) ->
    list_to_float(Value);
decode_message_attribute_value(["Number", "float", CustomType], Value) ->
    {CustomType, list_to_float(Value)};
decode_message_attribute_value("String", Value) ->
    Value;
decode_message_attribute_value(["String", CustomType], Value) ->
    {CustomType, Value};
decode_message_attribute_value("Binary", Value) ->
    list_to_binary(Value);
decode_message_attribute_value(["Binary", CustomType], Value) ->
    {CustomType, list_to_binary(Value)};
decode_message_attribute_value(DataType, Value) ->
    case string:tokens(DataType, ".") of
        [_Other] -> % check if datatype is something not handled above by Number/String/Binary
            erlang:error(decode_message_attribute_value_error, [DataType, Value]);
        Parsed ->
            decode_message_attribute_value(Parsed, Value)
    end.

decode_msg_attributes(Attrs)  ->
    [{decode_msg_attribute_name(Name),
      decode_msg_attribute_value(Name, Value)} || {Name, Value} <- decode_attributes(Attrs)].

decode_attributes(Attrs) ->
    [{erlcloud_xml:get_text("Name", Attr), erlcloud_xml:get_text("Value", Attr)} ||
        Attr <- Attrs].

-spec remove_permission(string(), string()) -> ok | no_return().
remove_permission(QueueName, Label) ->
    remove_permission(QueueName, Label, default_config()).

-spec remove_permission(string(), string(), aws_config()) -> ok | no_return().
remove_permission(QueueName, Label, Config)
  when is_list(QueueName), is_list(Label), is_record(Config, aws_config) ->
    sqs_simple_request(Config, QueueName, "RemovePermission",
                       [{"Label", Label}]).

-spec send_message(string(), string()) -> proplist() | no_return().
send_message(QueueName, MessageBody) ->
    send_message(QueueName, MessageBody, default_config()).

-spec send_message(string(), string(), proplists:proplist() | 0..900 | none | aws_config()) -> proplist() | no_return().
send_message(QueueName, MessageBody, #aws_config{} = Config) ->
    send_message(QueueName, MessageBody, none, Config);
send_message(QueueName, MessageBody, DelaySeconds) 
  when ((DelaySeconds >= 0 andalso DelaySeconds =< 900) orelse DelaySeconds =:= none) ->
    send_message(QueueName, MessageBody, DelaySeconds, default_config());
send_message(QueueName, MessageBody, Opts) when is_list(Opts) ->
    send_message(QueueName, MessageBody, Opts, default_config()).

-spec send_message(string(), string(), proplists:proplist() | 0..900 | none, aws_config()) -> proplist() | no_return().
send_message(QueueName, MessageBody, DelaySeconds, Config)
  when ((DelaySeconds >= 0 andalso DelaySeconds =< 900) orelse DelaySeconds =:= none) ->
    send_message(QueueName, MessageBody, [{delay_seconds, DelaySeconds}], [], Config);
send_message(QueueName, MessageBody, Opts, Config) when is_list(Opts) ->
    send_message(QueueName, MessageBody, Opts, [], Config).

-spec send_message(string(), string(), proplists:proplist() | 0..900 | none, [message_attribute()], aws_config()) -> proplist() | no_return().
send_message(QueueName, MessageBody, DelaySeconds, MessageAttributes, #aws_config{}=Config)
  when ((DelaySeconds >= 0 andalso DelaySeconds =< 900) orelse DelaySeconds =:= none) ->
    send_message(QueueName, MessageBody, [{delay_seconds, DelaySeconds}], MessageAttributes, Config);
send_message(QueueName, MessageBody, Opts, MessageAttributes, #aws_config{}=Config)
  when is_list(Opts) andalso is_list(MessageAttributes) ->
    EncodedMessageAttributes = encode_message_attributes(MessageAttributes),
    EncodedMessageOpts = [{encode_send_msg_attribute_name(N), V} || {N, V} <- Opts],
    Doc = sqs_xml_request(Config, QueueName, "SendMessage",
                          [{"MessageBody", MessageBody} | EncodedMessageOpts] ++ EncodedMessageAttributes),
    erlcloud_xml:decode(
      [
       {message_id, "SendMessageResult/MessageId", text},
       {md5_of_message_body, "SendMessageResult/MD5OfMessageBody", text}
      ],
      Doc
     ).

-spec set_queue_attributes(string(), proplists:proplist()) -> ok | no_return().
set_queue_attributes(QueueName, Attributes) ->
    set_queue_attributes(QueueName, Attributes, default_config()).

-spec set_queue_attributes(string(), proplists:proplist(), aws_config()) -> ok | no_return().
set_queue_attributes(QueueName, Attributes, Config)
  when is_list(QueueName), is_list(Attributes), is_record(Config, aws_config) ->
    Params = erlcloud_aws:param_list([
        [{"Name", encode_attribute_name(Name)},
         {"Value", Value}] || {Name, Value} <- Attributes], "Attribute"),
    sqs_simple_request(Config, QueueName, "SetQueueAttributes", Params).

-spec send_message_batch(string(), [batch_entry()]) -> proplist() | no_return().
send_message_batch(QueueName, BatchMessages) ->
    send_message_batch(QueueName, BatchMessages, default_config()).

-spec send_message_batch(string(), [batch_entry()], 0..900 | none | aws_config()) -> proplist() | no_return().
send_message_batch(QueueName, BatchMessages, Config)
    when is_record(Config, aws_config) ->
    send_message_batch(QueueName, BatchMessages, none, Config);
send_message_batch(QueueName, BatchMessages, DelaySeconds) ->
    send_message_batch(QueueName, BatchMessages, DelaySeconds, default_config()).

-spec send_message_batch(string(), [batch_entry()], 0..900 | none, aws_config()) -> proplist() | no_return().
send_message_batch(QueueName, BatchMessages, DelaySeconds, Config)
    when is_list(QueueName), is_record(Config, aws_config),
         (DelaySeconds >= 0 andalso DelaySeconds =< 900) orelse
         DelaySeconds =:= none ->
    {_, BatchRequestEntries} =
    lists:foldr(fun(Message, {N, Acc}) ->
                    {N + 1, [mk_send_batch_entry(N, Message, DelaySeconds)| Acc]}
                end, {1, []}, BatchMessages),

    Doc = sqs_xml_request(Config, QueueName, "SendMessageBatch",
                          lists:flatten(BatchRequestEntries)),

    BatchResponse =
    [{successful, "SendMessageBatchResult/SendMessageBatchResultEntry", fun decode_send_batch_successful/1},
     {failed, "SendMessageBatchResult/BatchResultErrorEntry", fun decode_batch_result_error/1}],

    erlcloud_xml:decode(BatchResponse, Doc).

-spec delete_message_batch(string(), [batch_entry()]) -> proplists:proplist() | no_return().
delete_message_batch(QueueName, BatchReceiptHandles) ->
    delete_message_batch(QueueName, BatchReceiptHandles, default_config()).

-spec delete_message_batch(string(), [batch_entry()], aws_config()) -> proplists:proplist() | no_return().
delete_message_batch(QueueName, [{Id, Handle}|_]=BatchReceiptHandles, Config)
    when is_list(QueueName), is_list(Id), is_list(Handle),
         is_record(Config, aws_config) ->

    {_, BatchRequestEntries} =
    lists:foldr(fun({BatchId, MessageBody}, {N, Acc}) ->
                    {N + 1, [mk_delete_batch_entry(N, BatchId, MessageBody)| Acc]}
                end, {1, []}, BatchReceiptHandles),

    Doc = sqs_xml_request(Config, QueueName, "DeleteMessageBatch",
                          lists:flatten(BatchRequestEntries)),

    BatchResponse =
    [{successful, "DeleteMessageBatchResult/DeleteMessageBatchResultEntry", fun decode_id_batch_successful/1},
     {failed, "DeleteMessageBatchResult/BatchResultErrorEntry", fun decode_batch_result_error/1}],
    erlcloud_xml:decode(BatchResponse, Doc).

-spec change_message_visibility_batch(string(), [batch_entry()], 0..43200) -> proplists:proplist() | no_return().
change_message_visibility_batch(QueueName, BatchReceiptHandles, VisibilityTimeout) ->
    change_message_visibility_batch(QueueName, BatchReceiptHandles, VisibilityTimeout,
                              default_config()).

-spec change_message_visibility_batch(string(), [batch_entry()], 0..43200, aws_config()) -> proplists:proplist() | no_return().
change_message_visibility_batch(QueueName, [{Id, Handle}|_]=BatchReceiptHandles, VisibilityTimeout, Config)
    when is_list(QueueName), is_list(Id), is_list(Handle), is_record(Config, aws_config) ->

    {_, BatchRequestEntries} =
    lists:foldr(fun({BatchId, MessageBody}, {N, Acc}) ->
                    {N + 1, [mk_ch_visibility_batch_entry(N, BatchId, MessageBody, VisibilityTimeout)| Acc]}
                end, {1, []}, BatchReceiptHandles),

    Doc = sqs_xml_request(Config, QueueName, "ChangeMessageVisibilityBatch",
                          lists:flatten(BatchRequestEntries)),

    BatchResponse =
    [{successful, "ChangeMessageVisibilityBatchResult/ChangeMessageVisibilityBatchResultEntry", fun decode_id_batch_successful/1},
     {failed, "ChangeMessageVisibilityBatchResult/BatchResultErrorEntry", fun decode_batch_result_error/1}],
    erlcloud_xml:decode(BatchResponse, Doc).

-spec mk_send_batch_entry(integer(), batch_entry(), 0..900 | none) -> [{string(), integer() | string()}].
mk_send_batch_entry(N, {MessageId, MessageBody}, DelaySeconds) ->
    mk_send_batch_entry(N, {MessageId, MessageBody, [], []}, DelaySeconds);
mk_send_batch_entry(N, {MessageId, MessageBody, MessageAttributes}, DelaySeconds) ->
    mk_send_batch_entry(N, {MessageId, MessageBody, MessageAttributes, []}, DelaySeconds);
mk_send_batch_entry(N, {MessageId, MessageBody, MessageAttributes, Opts}, DelaySeconds)
  when is_list(MessageId), is_list(MessageBody), is_list(MessageAttributes), is_list(Opts) ->
    N0 = integer_to_list(N),
    EncodedOpts = [
        ?SEND_BATCH_FIELD(N0, [$., encode_send_msg_attribute_name(Field)], Value)
        || {Field, Value} <- Opts],
    Base = [?SEND_BATCH_FIELD(N0, ".Id", MessageId),
            ?SEND_BATCH_FIELD(N0, ".MessageBody", MessageBody),
            ?SEND_BATCH_FIELD(N0, ".DelaySeconds", DelaySeconds) | EncodedOpts],
    lists:foldl(fun({Field, Value}, Acc) ->
                        [?SEND_BATCH_FIELD(N0, [$., Field], Value) | Acc]
                end, Base, encode_message_attributes(MessageAttributes)).

-spec mk_delete_batch_entry(integer(), string(), string()) -> [{string(), string()}].
mk_delete_batch_entry(N, MessageId, ReceiptHandle) ->
    N0 = integer_to_list(N),
    [
        ?DELETE_BATCH_FIELD(N0, ".Id", MessageId),
        ?DELETE_BATCH_FIELD(N0, ".ReceiptHandle", ReceiptHandle)
    ].

-spec mk_ch_visibility_batch_entry(integer(), string(), string(), 0..43200) -> [{string(), integer() | string()}].
mk_ch_visibility_batch_entry(N, MessageId, ReceiptHandle, VisibilityTimeout) ->
    N0 = integer_to_list(N),
    [
        ?CHANGE_VISIBILITY_BATCH_FIELD(N0, ".Id", MessageId),
        ?CHANGE_VISIBILITY_BATCH_FIELD(N0, ".ReceiptHandle", ReceiptHandle),
        ?CHANGE_VISIBILITY_BATCH_FIELD(N0, ".VisibilityTimeout", VisibilityTimeout)
    ].

decode_send_batch_successful(Nodes) ->
    [erlcloud_xml:decode([
                             {id, "Id", text},
                             {message_id, "MessageId", text},
                             {md5_of_message_body, "MD5OfMessageBody", text}
                         ], Node) || Node <- Nodes].

decode_id_batch_successful(Nodes) ->
    [erlcloud_xml:decode([{id, "Id", text}], Node) || Node <- Nodes].

decode_batch_result_error(Nodes) ->
    [erlcloud_xml:decode([
                             {id, "Id", text},
                             {sender_fault, "SenderFault", boolean},
                             {code, "Code", text},
                             {message, "Message", text}
                         ], Node) || Node <- Nodes].

default_config() -> erlcloud_aws:default_config().

sqs_simple_request(Config, QueueName, Action, Params) ->
    sqs_request(Config, QueueName, Action, Params),
    ok.

sqs_xml_request(Config, QueueName, Action, Params) ->
    case erlcloud_aws:aws_request_xml4(post, Config#aws_config.sqs_protocol,
                                  Config#aws_config.sqs_host, Config#aws_config.sqs_port,
                                  queue_path(QueueName),
                                  [{"Action", Action}, {"Version", ?API_VERSION}|Params],
                                  "sqs",Config)
    of
        {ok, Body} ->
            Body;
        {error, Reason} ->
            erlang:error({aws_error, Reason})
    end.

sqs_request(Config, QueueName, Action, Params) ->
    case erlcloud_aws:aws_request4(post, Config#aws_config.sqs_protocol,
                              Config#aws_config.sqs_host, Config#aws_config.sqs_port,
                              queue_path(QueueName),
                              [{"Action", Action}, {"Version", ?API_VERSION}|Params],
                              "sqs", Config)
    of
        {ok, Body} ->
            Body;
        {error, Reason} ->
            erlang:error({aws_error, Reason})
    end.

queue_path([$/|QueueName]) -> [$/ |erlcloud_http:url_encode(QueueName)];
queue_path([$h,$t,$t,$p|_] = URL) ->
    re:replace(URL, "^https?://[^/]*", "", [{return, list}]);
queue_path(QueueName) -> [$/ | erlcloud_http:url_encode(QueueName)].

encode_send_msg_attribute_name(delay_seconds) -> "DelaySeconds";
encode_send_msg_attribute_name(message_group_id) -> "MessageGroupId";
encode_send_msg_attribute_name(message_deduplication_id) -> "MessageDeduplicationId".

encode_message_attributes(Attributes) ->
    Map = fun(Attribute, Acc) ->
                  [encode_message_attribute(Attribute) | Acc]
          end,
    erlcloud_aws:param_list(lists:reverse(lists:foldl(Map, [], Attributes)), "MessageAttribute").

encode_message_attribute({Key, Value}) ->
    [
      {"Value.DataType", encode_message_attribute_type(Value)},
     encode_message_attribute_value(Value),
      {"Name", Key}
    ].

encode_message_attribute_value({_CustomType, Value}) ->
    encode_message_attribute_value(Value);
encode_message_attribute_value(Value) when is_integer(Value) ->
    {"Value.StringValue", integer_to_list(Value)};
encode_message_attribute_value(Value) when is_float(Value) ->
    {"Value.StringValue", float_to_list(Value, [{decimals, 12}, compact])};
encode_message_attribute_value(Value) when is_list(Value) ->
    {"Value.StringValue", Value};
encode_message_attribute_value(Value) when is_binary(Value) ->
    {"Value.BinaryValue", binary_to_list(Value)}.

encode_message_attribute_type({CustomType, Value})
  when is_list(CustomType) andalso is_integer(Value) ->
    ["Number.int.", CustomType];
encode_message_attribute_type({CustomType, Value})
  when is_list(CustomType) andalso is_float(Value) ->
    ["Number.float.", CustomType];
encode_message_attribute_type({CustomType, Value})
  when is_list(CustomType) andalso is_list(Value) ->
    ["String.", CustomType];
encode_message_attribute_type({CustomType, Value})
  when is_list(CustomType) andalso is_binary(Value) ->
    ["Binary.", CustomType];
encode_message_attribute_type(Value) when is_integer(Value) ->
    "Number.int";
encode_message_attribute_type(Value) when is_float(Value) ->
    "Number.float";
encode_message_attribute_type(Value) when is_list(Value) ->
    "String";
encode_message_attribute_type(Value) when is_binary(Value) ->
    "Binary".