mirror of
https://github.com/fog/fog-aws.git
synced 2022-11-09 13:50:52 -05:00
87 lines
3.5 KiB
Ruby
87 lines
3.5 KiB
Ruby
module Fog
|
|
module AWS
|
|
class Kinesis
|
|
class Real
|
|
# Merges two adjacent shards in a stream and combines them into a single shard to reduce the stream's capacity to ingest and transport data.
|
|
#
|
|
# ==== Options
|
|
# * AdjacentShardToMerge<~String>: The shard ID of the adjacent shard for the merge.
|
|
# * ShardToMerge<~String>: The shard ID of the shard to combine with the adjacent shard for the merge.
|
|
# * StreamName<~String>: The name of the stream for the merge.
|
|
# ==== Returns
|
|
# * response<~Excon::Response>:
|
|
#
|
|
# ==== See Also
|
|
# https://docs.aws.amazon.com/kinesis/latest/APIReference/API_MergeShards.html
|
|
#
|
|
def merge_shards(options={})
|
|
body = {
|
|
"AdjacentShardToMerge" => options.delete("AdjacentShardToMerge"),
|
|
"ShardToMerge" => options.delete("ShardToMerge"),
|
|
"StreamName" => options.delete("StreamName")
|
|
}.reject{ |_,v| v.nil? }
|
|
|
|
request({
|
|
'X-Amz-Target' => "Kinesis_#{@version}.MergeShards",
|
|
:body => body,
|
|
}.merge(options))
|
|
end
|
|
end
|
|
|
|
class Mock
|
|
def merge_shards(options={})
|
|
stream_name = options.delete("StreamName")
|
|
shard_to_merge_id = options.delete("ShardToMerge")
|
|
adjacent_shard_to_merge_id = options.delete("AdjacentShardToMerge")
|
|
|
|
unless stream = data[:kinesis_streams].detect{ |s| s["StreamName"] == stream_name }
|
|
raise Fog::AWS::Kinesis::ResourceNotFound.new("Stream #{stream_name} under account #{@account_id} not found.")
|
|
end
|
|
|
|
unless shard_to_merge = stream["Shards"].detect{ |shard| shard["ShardId"] == shard_to_merge_id }
|
|
raise Fog::AWS::Kinesis::ResourceNotFound.new("Could not find shard foo in stream #{stream_name} under account #{@account_id}.")
|
|
|
|
end
|
|
|
|
unless adjacent_shard_to_merge = stream["Shards"].detect{ |shard| shard["ShardId"] == adjacent_shard_to_merge_id }
|
|
raise Fog::AWS::Kinesis::ResourceNotFound.new("Could not find shard foo in stream #{stream_name} under account #{@account_id}.")
|
|
|
|
end
|
|
|
|
# Close shards (set an EndingSequenceNumber on them)
|
|
shard_to_merge["SequenceNumberRange"]["EndingSequenceNumber"] = next_sequence_number
|
|
adjacent_shard_to_merge["SequenceNumberRange"]["EndingSequenceNumber"] = next_sequence_number
|
|
|
|
new_starting_hash_key = [
|
|
shard_to_merge["HashKeyRange"]["StartingHashKey"].to_i,
|
|
adjacent_shard_to_merge["HashKeyRange"]["StartingHashKey"].to_i
|
|
].min.to_s
|
|
|
|
new_ending_hash_key = [
|
|
shard_to_merge["HashKeyRange"]["EndingHashKey"].to_i,
|
|
adjacent_shard_to_merge["HashKeyRange"]["EndingHashKey"].to_i
|
|
].max.to_s
|
|
|
|
# create a new shard with ParentShardId and AdjacentParentShardID
|
|
stream["Shards"] << {
|
|
"HashKeyRange"=> {
|
|
"EndingHashKey" => new_ending_hash_key,
|
|
"StartingHashKey" => new_starting_hash_key
|
|
},
|
|
"SequenceNumberRange" => {
|
|
"StartingSequenceNumber" => next_sequence_number
|
|
},
|
|
"ShardId" => next_shard_id,
|
|
"ParentShardId" => shard_to_merge_id,
|
|
"AdjacentParentShardId" => adjacent_shard_to_merge_id
|
|
}
|
|
|
|
response = Excon::Response.new
|
|
response.status = 200
|
|
response.body = ""
|
|
response
|
|
end
|
|
end
|
|
end
|
|
end
|
|
end
|