Class: Aws::SQS::ExtendedClient

Inherits:
SimpleDelegator
  • Object
show all
Includes:
Extended
Defined in:
lib/aws/sdk/extended/sqs.rb

Overview

Amazon SQS extended client, API-compatible with Aws::SQS::Client from the aws-sdk-sqs gem, invokes service methods on Amazon SQS with extended functionality to support large payloads.

Constant Summary

Constants included from Extended

Extended::DEFAULT_S3_MUTEX, Extended::EMPTY_ARRAY, Extended::EMPTY_HASH, Extended::EMPTY_STRING, Extended::LEGACY_RESERVED_ATTRIBUTE_NAME, Extended::MAX_ALLOWED_ATTRIBUTES, Extended::MESSAGE_SIZE_THRESHOLD, Extended::RECEIPT_HANDLE_FORMAT, Extended::RESERVED_ATTRIBUTE_NAME, Extended::S3_BUCKET_NAME_MARKER, Extended::S3_KEY_ATTRIBUTE_NAME, Extended::S3_KEY_MARKER

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Extended

default_s3_client, message_size, store_large_message

Constructor Details

#initialize(sqs_client, bucket: ENV.fetch("AWS_EXTENDED_CLIENT_S3_BUCKET"), s3_client: Extended.default_s3_client, always_through_s3: false, delete_payload_from_s3: false, payload_size_threshold: Extended::MESSAGE_SIZE_THRESHOLD) ⇒ ExtendedClient

Constructs a new SQS extended client, where all SQS service calls are forwarded to sqs_client, and calls to S3 to store and retrieve large payloads are forwarded to s3_client using the bucket specified by bucket, and not just message bodies which are bigger than the number of bytes specified by payload_size_threshold. This can be circumvented by always_through_s3 though, which will cause it to store all message bodies in the S3 bucket.

When the message is deleted, delete_payload_from_s3 determines whether the large payload stored in the S3 bucket should be deleted alongside.

: ( | SQS::Client, | ?bucket: String, | ?s3_client: S3::Client, | ?always_through_s3: bool, | delete_payload_from_s3: bool, | ?payload_size_threshold) -> void



42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/aws/sdk/extended/sqs.rb', line 42

def initialize(
  sqs_client,
  bucket: ENV.fetch("AWS_EXTENDED_CLIENT_S3_BUCKET"),
  s3_client: Extended.default_s3_client,
  always_through_s3: false,
  delete_payload_from_s3: false,
  payload_size_threshold: Extended::MESSAGE_SIZE_THRESHOLD
)
  @sqs_client = sqs_client
  @s3_client = s3_client
  @bucket = bucket
  @always_through_s3 = always_through_s3
  @delete_payload_from_s3 = delete_payload_from_s3
  @payload_size_threshold = payload_size_threshold
  super(sqs_client)
end

Instance Attribute Details

#always_through_s3 ⇒ Object (readonly)

: bool



20
21
22
 
# File 'lib/aws/sdk/extended/sqs.rb', line 20

def always_through_s3
  @always_through_s3
end

#bucket ⇒ Object (readonly)

: String



18
19
20
 
# File 'lib/aws/sdk/extended/sqs.rb', line 18

def bucket
  @bucket
end

#delete_payload_from_s3 ⇒ Object (readonly)

: bool



22
23
24
 
# File 'lib/aws/sdk/extended/sqs.rb', line 22

def delete_payload_from_s3
  @delete_payload_from_s3
end

#payload_size_threshold ⇒ Object (readonly)

: Integer



24
25
26
 
# File 'lib/aws/sdk/extended/sqs.rb', line 24

def payload_size_threshold
  @payload_size_threshold
end

#s3_client ⇒ Object (readonly)

: S3::Client



16
17
18
 
# File 'lib/aws/sdk/extended/sqs.rb', line 16

def s3_client
  @s3_client
end

#sqs_client ⇒ Object (readonly)

: SQS::Client



14
15
16
 
# File 'lib/aws/sdk/extended/sqs.rb', line 14

def sqs_client
  @sqs_client
end

Instance Method Details

#change_message_visibility(params = {}) ⇒ Object

Same functionality as Client#change_message_visibility

: (Hash[Symbol, untyped], *untyped) -> Seahorse::Client::_ResponseSuccess



188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/aws/sdk/extended/sqs.rb', line 188

def change_message_visibility(params = {}, *)
  request = Aws::SQS::Types::ChangeMessageVisibilityRequest.new(params)

  if request.queue_url && request.receipt_handle &&
     request.visibility_timeout && large_message_receipt_handle?(request.receipt_handle)

    *_, params[:receipt_handle] = bucket_key_original_receipt_handle(request.receipt_handle)
  end

  super
end

#delete_message(params = {}) ⇒ Object

Same functionality as Client#delete_message, will (if @delete_payload_froms3) delete the respective s3 objects from the S3 bucket. and store them as the message object in the returned object.

: (Hash[Symbol, untyped], *untyped) -> Seahorse::Client::_ResponseSuccess



146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/aws/sdk/extended/sqs.rb', line 146

def delete_message(params = {}, *)
  receipt_handle = params[:receipt_handle]

  return super unless @delete_payload_from_s3 && receipt_handle && large_message_receipt_handle?(receipt_handle)

  bucket, key, params[:receipt_handle] = bucket_key_original_receipt_handle(receipt_handle)

  # TODO: what to do with error?
  @s3_client.delete_object(bucket: bucket, key: key)

  super
end

#delete_message_batch(params = {}) ⇒ Object

Same functionality as Client#delete_message_batch, will (if @delete_payload_froms3) delete the respective s3 objects from the S3 bucket.

: (Hash[Symbol, untyped], *untyped) -> SQS::Client::_DeleteMessageBatchResponseSuccess



163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/aws/sdk/extended/sqs.rb', line 163

def delete_message_batch(params = {}, *)
  request = Aws::SQS::Types::DeleteMessageBatchRequest.new(params)

  if request.queue_url && request.entries
    s3_to_delete = request.entries.map do |entry|
      receipt_handle = entry[:receipt_handle]
      next unless @delete_payload_from_s3 && receipt_handle && large_message_receipt_handle?(receipt_handle)

      *bucket_key, entry[:receipt_handle] = bucket_key_original_receipt_handle(receipt_handle)

      bucket_key
    end

    s3_to_delete.each do |bucket, key|
      # TODO: what to do with error?
      @s3_client.delete_object(bucket: bucket, key: key)
    end
  end

  super
end

#freeze ⇒ Object 



59
60
61
62
63
64
 
# File 'lib/aws/sdk/extended/sqs.rb', line 59

def freeze
  @sqs_client.freeze
  @s3_client.freeze
  @bucket.freeze
  super
end

#receive_message(params = {}) ⇒ Object

Same functionality as Client#receive_message, will transparently retrieve large payloads and store them as the message object in the returned object.

: (Hash[Symbol, untyped], *untyped) -> SQS::Client::_ReceiveMessageResponseSuccess



108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/aws/sdk/extended/sqs.rb', line 108

def receive_message(params = {}, *)
  requested_attributes = params.fetch(:message_attribute_names, [])

  unless requested_attributes.include?("All") ||
         requested_attributes.include?("*")
    requested_attributes << RESERVED_ATTRIBUTE_NAME unless requested_attributes.include?(RESERVED_ATTRIBUTE_NAME)

    unless requested_attributes.include?(LEGACY_RESERVED_ATTRIBUTE_NAME)
      requested_attributes << LEGACY_RESERVED_ATTRIBUTE_NAME
    end
  end

  params[:message_attribute_names] = requested_attributes

  super.tap do |resp|
    messages = resp.messages || EMPTY_ARRAY #: Array[Aws::SQS::Types::Message]

    messages.each do |message|
      message_attributes = message.message_attributes

      next unless message_attributes && (
        message_attributes.include?(RESERVED_ATTRIBUTE_NAME) ||
        message_attributes.include?(LEGACY_RESERVED_ATTRIBUTE_NAME)
      )

      get_large_message(message)

      message_attributes.delete(RESERVED_ATTRIBUTE_NAME)
      message_attributes.delete(LEGACY_RESERVED_ATTRIBUTE_NAME)
    end
  end
end

#send_message(params = {}) ⇒ Object

Same functionality as Client#send_message, will store the message body in S3 if all conditions match.

: (Hash[Symbol, untyped], *untyped) -> SQS::Client::_SendMessageResponseSuccess



70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/aws/sdk/extended/sqs.rb', line 70

def send_message(params = {}, *)
  request = Aws::SQS::Types::SendMessageRequest.new(params)

  return super unless request.queue_url && request.message_body && (@always_through_s3 || large_message?(request))

  params[:message_attributes], params[:message_body] =
    Extended.store_large_message(
      request.message_attributes,
      request.message_body, @s3_client, @bucket
    )

  super
end

#send_message_batch(params = {}, options = {}) ⇒ Object

Same functionality as Client#send_message_batch, will store the message bodies in S3 if all conditions match, and store the reference in the SQS message.

: (Hash[Symbol, untyped], *untyped) -> SQS::Client::_SendMessageBatchResponseSuccess



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/aws/sdk/extended/sqs.rb', line 88

def send_message_batch(params = {}, options = {})
  request = Aws::SQS::Types::SendMessageBatchRequest.new(params)

  return super unless request.queue_url && request.entries

  request.entries.each do |entry|
    request = Aws::SQS::Types::SendMessageBatchRequestEntry.new(entry)

    next unless @always_through_s3 || large_message?(request)

    Extended.store_large_message(request.message_attributes, request.message_body, @s3_client, @bucket)
  end

  super
end