Module: Aws::Extended

Included in:
SNS::ExtendedClient, SQS::ExtendedClient
Defined in:
lib/aws/sdk/extended.rb

Defined Under Namespace

Classes: Error

Constant Summary collapse

EMPTY_HASH =

@rbs! interface _WithSize def size: () -> Integer end

{}.freeze
EMPTY_ARRAY =

: Hash[untyped, untyped]

[].freeze
EMPTY_STRING =

: Array

""
MESSAGE_SIZE_THRESHOLD = 
262_144
MAX_ALLOWED_ATTRIBUTES = 
10
LEGACY_RESERVED_ATTRIBUTE_NAME = 
"SQSLargePayloadSize"
RESERVED_ATTRIBUTE_NAME = 
"ExtendedPayloadSize"
S3_KEY_ATTRIBUTE_NAME = 
"S3Key"
S3_BUCKET_NAME_MARKER = 
"-..s3BucketName..-"
S3_KEY_MARKER = 
"-..s3Key..-"
RECEIPT_HANDLE_FORMAT = 
"#{S3_BUCKET_NAME_MARKER}%s" \
"#{S3_BUCKET_NAME_MARKER}#{S3_KEY_MARKER}%s" \
"#{S3_KEY_MARKER}%s".freeze
DEFAULT_S3_MUTEX = 
Mutex.new

Class Method Summary collapse

Class Method Details

.default_s3_client ⇒ Object 



40
41
42
43
44
 
# File 'lib/aws/sdk/extended.rb', line 40

def default_s3_client #: Aws::S3::Client
  @default_s3_client || DEFAULT_S3_MUTEX.synchronize do
    @default_s3_client ||= Aws::S3::Client.new
  end
end

.message_size(message_attributes, body) ⇒ Object

: (Hash[String, untyped], _WithSize) -> Integer



47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/aws/sdk/extended.rb', line 47

def message_size(message_attributes, body)
  total_size = body.size

  if message_attributes
    total_size = message_attributes.sum(total_size) do |key, value|
      key.size + value.fetch(:data_type, EMPTY_STRING).size +
        value.fetch(:binary_value, EMPTY_STRING).size +
        value.fetch(:string_value, EMPTY_STRING).size
    end
  end

  total_size
end

.store_large_message(message_attributes, body, s3_client, bucket) ⇒ Object

: (Hash[String, untyped], _WithSize, S3::Client, String) -> [Hash[Symbol, untyped], String]



62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/aws/sdk/extended.rb', line 62

def store_large_message(message_attributes, body, s3_client, bucket)
  message_attributes ||= Extended::EMPTY_HASH

  if message_attributes.size > Extended::MAX_ALLOWED_ATTRIBUTES - 1
    raise Extended::Error,
          "Number of message attributes exceeds the maximum allowed for large payload messages"
  end

  if message_attributes.key?(Extended::RESERVED_ATTRIBUTE_NAME) ||
     message_attributes.key?(Extended::LEGACY_RESERVED_ATTRIBUTE_NAME)
    raise Extended::Error,
          "Use of an attribute reserved by SQS extended client."
  end

  s3_key = message_attributes.fetch(Extended::S3_KEY_ATTRIBUTE_NAME, SecureRandom.urlsafe_base64)

  message_attributes = message_attributes.merge(
    # TODO: optionally use legacy attribute
    Extended::RESERVED_ATTRIBUTE_NAME => {
      data_type: "Number",
      string_value: body.size.to_s
    },
    Extended::S3_KEY_ATTRIBUTE_NAME => {
      data_type: "String",
      string_value: s3_key
    }
  )

  s3_client.put_object(bucket: bucket, key: s3_key, body: body)

  message_body = JSON.generate(
    [
      Extended::RESERVED_ATTRIBUTE_NAME,
      { s3BucketName: bucket, s3Key: s3_key }
    ]
  )

  [message_attributes, message_body]
end