Class: Fog::AWS::Kinesis::Real

Inherits:
Object
  • Object
show all
Includes:
CredentialFetcher::ConnectionMethods
Defined in:
lib/fog/aws/kinesis.rb,
lib/fog/aws/requests/kinesis/put_record.rb,
lib/fog/aws/requests/kinesis/get_records.rb,
lib/fog/aws/requests/kinesis/put_records.rb,
lib/fog/aws/requests/kinesis/split_shard.rb,
lib/fog/aws/requests/kinesis/list_streams.rb,
lib/fog/aws/requests/kinesis/merge_shards.rb,
lib/fog/aws/requests/kinesis/create_stream.rb,
lib/fog/aws/requests/kinesis/delete_stream.rb,
lib/fog/aws/requests/kinesis/describe_stream.rb,
lib/fog/aws/requests/kinesis/add_tags_to_stream.rb,
lib/fog/aws/requests/kinesis/get_shard_iterator.rb,
lib/fog/aws/requests/kinesis/list_tags_for_stream.rb,
lib/fog/aws/requests/kinesis/remove_tags_from_stream.rb

Instance Method Summary collapse

Methods included from CredentialFetcher::ConnectionMethods

#refresh_credentials_if_expired

Constructor Details

#initialize(options = {}) ⇒ Real

Returns a new instance of Real.



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/fog/aws/kinesis.rb', line 36

def initialize(options={})
  @use_iam_profile = options[:use_iam_profile]

  @connection_options = options[:connection_options] || {}

  @instrumentor           = options[:instrumentor]
  @instrumentor_name      = options[:instrumentor_name] || 'fog.aws.kinesis'

  options[:region] ||= 'us-east-1'
  @region     = options[:region]
  @host       = options[:host] || "kinesis.#{options[:region]}.amazonaws.com"
  @path       = options[:path]        || '/'
  @persistent = options[:persistent]  || true
  @port       = options[:port]        || 443
  @scheme     = options[:scheme]      || 'https'
  @connection = Fog::XML::Connection.new("#{@scheme}://#{@host}:#{@port}#{@path}", @persistent, @connection_options)
  @version    = "20131202"

  setup_credentials(options)
end

Instance Method Details

#add_tags_to_stream(options = {}) ⇒ Object

Adds or updates tags for the specified Amazon Kinesis stream.

Options

  • StreamName<~String>: The name of the stream.

  • Tags<~Hash>: The set of key-value pairs to use to create the tags.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_AddTagsToStream.html



16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/fog/aws/requests/kinesis/add_tags_to_stream.rb', line 16

def add_tags_to_stream(options={})
  body = {
    "StreamName" => options.delete("StreamName"),
    "Tags" => options.delete("Tags")
  }.reject{ |_,v| v.nil? }

  request({
            'X-Amz-Target' => "Kinesis_#{@version}.AddTagsToStream",
            :body          => body,
          }.merge(options))
end

#create_stream(options = {}) ⇒ Object

Creates a Amazon Kinesis stream.

Options

  • ShardCount<~Number>: The number of shards that the stream will use.

  • StreamName<~String>: A name to identify the stream.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_CreateStream.html



16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/fog/aws/requests/kinesis/create_stream.rb', line 16

def create_stream(options={})
  body = {
    "ShardCount" => options.delete("ShardCount") || 1,
    "StreamName" => options.delete("StreamName")
  }.reject{ |_,v| v.nil? }

  request({
            'X-Amz-Target' => "Kinesis_#{@version}.CreateStream",
            :body          => body,
          }.merge(options))
end

#delete_stream(options = {}) ⇒ Object

Deletes a stream and all its shards and data.

Options

  • StreamName<~String>: A name to identify the stream.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_DeleteStream.html



15
16
17
18
19
20
21
22
23
24
 
# File 'lib/fog/aws/requests/kinesis/delete_stream.rb', line 15

def delete_stream(options={})
  body = {
    "StreamName" => options.delete("StreamName")
  }.reject{ |_,v| v.nil? }

  request({
            'X-Amz-Target' => "Kinesis_#{@version}.DeleteStream",
            :body          => body,
          }.merge(options))
end

#describe_stream(options = {}) ⇒ Object

Describes the specified stream.

Options

  • ExclusiveStartShardId<~String>: The shard ID of the shard to start with.

  • Limit<~Number>: The maximum number of shards to return.

  • StreamName<~String>: The name of the stream to describe.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStream.html



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/fog/aws/requests/kinesis/describe_stream.rb', line 17

def describe_stream(options={})
  body = {
    "ExclusiveStartShardId" => options.delete("ExclusiveStartShardId"),
    "Limit" => options.delete("Limit"),
    "StreamName" => options.delete("StreamName")
  }.reject{ |_,v| v.nil? }

  response = request({
                       :idempotent    => true,
                       'X-Amz-Target' => "Kinesis_#{@version}.DescribeStream",
                       :body          => body,
                     }.merge(options))
  response.body = Fog::JSON.decode(response.body) unless response.body.nil?
  response.body
  response
end

#get_records(options = {}) ⇒ Object

Gets data records from a shard.

Options

  • Limit<~Number>: The maximum number of records to return.

  • ShardIterator<~String>: The position in the shard from which you want to start sequentially reading data records.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html



16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/fog/aws/requests/kinesis/get_records.rb', line 16

def get_records(options={})
  body = {
    "Limit" => options.delete("Limit"),
    "ShardIterator" => options.delete("ShardIterator")
  }.reject{ |_,v| v.nil? }

  response = request({
                       'X-Amz-Target' => "Kinesis_#{@version}.GetRecords",
                       :body          => body,
                     }.merge(options))
  response.body = Fog::JSON.decode(response.body) unless response.body.nil?
  response
end

#get_shard_iterator(options = {}) ⇒ Object

Gets a shard iterator.

Options

  • ShardId<~String>: The shard ID of the shard to get the iterator for.

  • ShardIteratorType<~String>: Determines how the shard iterator is used to start reading data records from the shard.

  • StartingSequenceNumber<~String>: The sequence number of the data record in the shard from which to start reading from.

  • StreamName<~String>: A name to identify the stream.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html



18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/fog/aws/requests/kinesis/get_shard_iterator.rb', line 18

def get_shard_iterator(options={})
  body = {
    "ShardId" => options.delete("ShardId"),
    "ShardIteratorType" => options.delete("ShardIteratorType"),
    "StartingSequenceNumber" => options.delete("StartingSequenceNumber"),
    "StreamName" => options.delete("StreamName")
  }.reject{ |_,v| v.nil? }

  response = request({
                       'X-Amz-Target' => "Kinesis_#{@version}.GetShardIterator",
                       :body          => body,
                     }.merge(options))
  response.body = Fog::JSON.decode(response.body) unless response.body.nil?
  response
end

#list_streams(options = {}) ⇒ Object

List availabe streams

Options

  • ExclusiveStartStreamName<~String>: The name of the stream to start the list with.

  • Limit<~Number>: The maximum number of streams to list.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_ListStreams.html



16
17
18
19
20
21
22
23
24
 
# File 'lib/fog/aws/requests/kinesis/list_streams.rb', line 16

def list_streams(options={})
  response = request({
                       :idempotent    => true,
                       'X-Amz-Target' => "Kinesis_#{@version}.ListStreams",
                       :body          => {},
                     }.merge(options))
  response.body = Fog::JSON.decode(response.body) unless response.body.nil?
  response
end

#list_tags_for_stream(options = {}) ⇒ Object

Lists the tags for the specified Amazon Kinesis stream.

Options

  • ExclusiveStartTagKey<~String>: The key to use as the starting point for the list of tags.

  • Limit<~Number>: The number of tags to return.

  • StreamName<~String>: The name of the stream.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_ListTagsForStream.html



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/fog/aws/requests/kinesis/list_tags_for_stream.rb', line 17

def list_tags_for_stream(options={})
  body = {
    "ExclusiveStartTagKey" => options.delete("ExclusiveStartTagKey"),
    "Limit" => options.delete("Limit"),
    "StreamName" => options.delete("StreamName")
  }.reject{ |_,v| v.nil? }

  response = request({
                       :idempotent    => true,
                       'X-Amz-Target' => "Kinesis_#{@version}.ListTagsForStream",
                       :body          => body,
                     }.merge(options))
  response.body = Fog::JSON.decode(response.body) unless response.body.nil?
  response.body
  response
end

#merge_shards(options = {}) ⇒ Object

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

docs.aws.amazon.com/kinesis/latest/APIReference/API_MergeShards.html



17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/fog/aws/requests/kinesis/merge_shards.rb', line 17

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

#put_record(options = {}) ⇒ Object

Writes a single data record from a producer into an Amazon Kinesis stream.

Options

  • Data<~Blob>: The data blob to put into the record, which is base64-encoded when the blob is serialized.

  • ExplicitHashKey<~String>: The hash value used to determine explicitly the shard that the data record is assigned to by overriding the partition key hash.

  • PartitionKey<~String>: Determines which shard in the stream the data record is assigned to.

  • SequenceNumberForOrdering<~String>: Guarantees strictly increasing sequence numbers, for puts from the same client and to the same partition key.

  • StreamName<~String>: The stream name associated with the request.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/fog/aws/requests/kinesis/put_record.rb', line 19

def put_record(options={})
  body = {
    "Data" => options.delete("Data"),
    "ExplicitHashKey" => options.delete("ExplicitHashKey"),
    "PartitionKey" => options.delete("PartitionKey"),
    "SequenceNumberForOrdering" => options.delete("SequenceNumberForOrdering"),
    "StreamName" => options.delete("StreamName")
  }.reject{ |_,v| v.nil? }

  response = request({
                       'X-Amz-Target' => "Kinesis_#{@version}.PutRecord",
                       :body          => body,
                     }.merge(options))
  response.body = Fog::JSON.decode(response.body) unless response.body.nil?
  response
end

#put_records(options = {}) ⇒ Object

Writes multiple data records from a producer into an Amazon Kinesis stream in a single call (also referred to as a PutRecords request).

Options

  • Records<~Array>: The records associated with the request.

    • Record<~Hash>: A record.

      • Data<~Blob>: The data blob to put into the record, which is base64-encoded when the blob is serialized.

      • ExplicitHashKey<~String>: The hash value used to determine explicitly the shard that the data record is assigned to by overriding the partition key hash.

      • PartitionKey<~String>: Determines which shard in the stream the data record is assigned to.

  • StreamName<~String>: The stream name associated with the request.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html



20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/fog/aws/requests/kinesis/put_records.rb', line 20

def put_records(options={})
  body = {
    "Records" => options.delete("Records"),
    "StreamName" => options.delete("StreamName")
  }.reject{ |_,v| v.nil? }

  response = request({
                       'X-Amz-Target' => "Kinesis_#{@version}.PutRecords",
                       :body          => body,
                     }.merge(options))
  response.body = Fog::JSON.decode(response.body) unless response.body.nil?
  response
end

#remove_tags_from_stream(options = {}) ⇒ Object

Deletes tags from the specified Amazon Kinesis stream.

Options

  • StreamName<~String>: The name of the stream.

  • TagKeys<~Array>: A list of tag keys.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_RemoveTagsFromStream.html



16
17
18
19
20
21
22
23
24
25
26
 
# File 'lib/fog/aws/requests/kinesis/remove_tags_from_stream.rb', line 16

def remove_tags_from_stream(options={})
  body = {
    "StreamName" => options.delete("StreamName"),
    "TagKeys" => options.delete("TagKeys")
  }.reject{ |_,v| v.nil? }

  request({
            'X-Amz-Target' => "Kinesis_#{@version}.RemoveTagsFromStream",
            :body          => body,
          }.merge(options))
end

#split_shard(options = {}) ⇒ Object

Splits a shard into two new shards in the stream, to increase the stream’s capacity to ingest and transport data.

Options

  • NewStartingHashKey<~String>: A hash key value for the starting hash key of one of the child shards created by the split.

  • ShardToSplit<~String>: The shard ID of the shard to split.

  • StreamName<~String>: The name of the stream for the shard split.

Returns

  • response<~Excon::Response>:

See Also

docs.aws.amazon.com/kinesis/latest/APIReference/API_SplitShard.html



17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/fog/aws/requests/kinesis/split_shard.rb', line 17

def split_shard(options={})
  body = {
    "NewStartingHashKey" => options.delete("NewStartingHashKey"),
    "ShardToSplit" => options.delete("ShardToSplit"),
    "StreamName" => options.delete("StreamName")
  }.reject{ |_,v| v.nil? }

  request({
            'X-Amz-Target' => "Kinesis_#{@version}.SplitShard",
            :body          => body,
          }.merge(options))
end