Current section
Files
Jump to
Current section
Files
src/erlcloud_kinesis.erl
%% Amazon Kinesis Service (Kinesis)
-module(erlcloud_kinesis).
%%% Library initialization.
-export([configure/2, configure/3, configure/4, new/2, new/3]).
-export([create_stream/2, create_stream/3,
delete_stream/1, delete_stream/2,
list_streams/0, list_streams/1, list_streams/2, list_streams/3,
describe_stream/1, describe_stream/2, describe_stream/3, describe_stream/4,
describe_stream_summary/1, describe_stream_summary/2,
enable_enhanced_monitoring/1, enable_enhanced_monitoring/2, enable_enhanced_monitoring/3,
disable_enhanced_monitoring/1, disable_enhanced_monitoring/2, disable_enhanced_monitoring/3,
get_shard_iterator/3, get_shard_iterator/4, get_shard_iterator/5,
get_records/1, get_records/2, get_records/3, get_records/4,
put_record/3, put_record/4, put_record/5, put_record/6, put_record/7,
put_records/2, put_records/3, put_records/4,
merge_shards/3, merge_shards/4,
split_shards/3, split_shards/4,
add_tags_to_stream/2, add_tags_to_stream/3,
list_tags_for_stream/1, list_tags_for_stream/2,
list_tags_for_stream/3, list_tags_for_stream/4,
list_all_tags_for_stream/1, list_all_tags_for_stream/2,
remove_tags_from_stream/2, remove_tags_from_stream/3
]).
-include("erlcloud.hrl").
-include("erlcloud_aws.hrl").
-type get_records_limit() :: 1..10000.
-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,
kinesis_host=Host
}.
-spec new(string(), string(), string(), non_neg_integer()) -> aws_config().
new(AccessKeyID, SecretAccessKey, Host, Port) ->
#aws_config{
access_key_id=AccessKeyID,
secret_access_key=SecretAccessKey,
kinesis_host=Host,
kinesis_port=Port
}.
-spec configure(string(), string()) -> ok.
configure(AccessKeyID, SecretAccessKey) ->
erlcloud_config:configure(AccessKeyID, SecretAccessKey, fun new/2).
-spec configure(string(), string(), string()) -> ok.
configure(AccessKeyID, SecretAccessKey, Host) ->
erlcloud_config:configure(AccessKeyID, SecretAccessKey, Host, fun new/3).
-spec configure(string(), string(), string(), non_neg_integer()) -> ok.
configure(AccessKeyID, SecretAccessKey, Host, Port) ->
erlcloud_config:configure(AccessKeyID, SecretAccessKey, Host, Port, fun new/4).
default_config() -> erlcloud_aws:default_config().
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_CreateStream.html]
%%
%% ===Example===
%%
%% This operation adds a new Amazon Kinesis stream to your AWS account.
%%
%% `
%% erlcloud_kinesis:create_stream(<<"test">>, 2).
%% {ok,{incomplete,#Fun<jsx_decoder.1.688044>}}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec create_stream(binary(), 1..100000) -> erlcloud_kinesis_impl:json_return().
create_stream(StreamName, ShardCount) when is_integer(ShardCount), ShardCount > 0, ShardCount =< 100000 ->
Json = [{<<"StreamName">>, StreamName}, {<<"ShardCount">>, ShardCount}],
erlcloud_kinesis_impl:request(default_config(), "Kinesis_20131202.CreateStream", Json).
-spec create_stream(binary(), 1..100000, aws_config()) -> erlcloud_kinesis_impl:json_return().
create_stream(StreamName, ShardCount, Config) when is_record(Config, aws_config), is_integer(ShardCount), ShardCount > 0, ShardCount =< 100000 ->
Json = [{<<"StreamName">>, StreamName}, {<<"ShardCount">>, ShardCount}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.CreateStream", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_DeleteStream.html]
%%
%% ===Example===
%%
%% This operation deletes a stream and all of its shards and data.
%%
%% `
%% erlcloud_kinesis:delete_stream(<<"test">>).
%% {ok,{incomplete,#Fun<jsx_decoder.1.688044>}}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec delete_stream(binary()) -> erlcloud_kinesis_impl:json_return().
delete_stream(StreamName) ->
Json = [{<<"StreamName">>, StreamName}],
erlcloud_kinesis_impl:request(default_config(), "Kinesis_20131202.DeleteStream", Json).
-spec delete_stream(binary(), aws_config()) -> erlcloud_kinesis_impl:json_return().
delete_stream(StreamName, Config) when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.DeleteStream", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_ListStreams.html]
%%
%% ===Example===
%%
%% This operation returns an array of the names of all the streams that are associated with the AWS account making the ListStreams request.
%%
%% `
%% erlcloud_kinesis:list_streams().
%% {ok,[{<<"HasMoreStreams">>,false},
%% {<<"StreamNames">>,[<<"staging">>]}]}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec list_streams() -> erlcloud_kinesis_impl:json_return().
list_streams() ->
list_streams(default_config()).
-spec list_streams(binary() | aws_config()) -> erlcloud_kinesis_impl:json_return().
list_streams(Config) when is_record(Config, aws_config) ->
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.ListStreams", []);
list_streams(ExclusiveStartStreamName) ->
Json = [{<<"ExclusiveStartStreamName">>, ExclusiveStartStreamName}],
erlcloud_kinesis_impl:request(default_config(), "Kinesis_20131202.ListStreams", Json).
-spec list_streams(binary(), 1..100 | aws_config()) -> erlcloud_kinesis_impl:json_return().
list_streams(ExclusiveStartStreamName, Config) when is_record(Config, aws_config) ->
Json = [{<<"ExclusiveStartStreamName">>, ExclusiveStartStreamName}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.ListStreams", Json);
list_streams(ExclusiveStartStreamName, Limit) when is_integer(Limit), Limit > 0, Limit =< 100 ->
Json = [{<<"ExclusiveStartStreamName">>, ExclusiveStartStreamName}, {<<"Limit">>, Limit}],
erlcloud_kinesis_impl:request(default_config(), "Kinesis_20131202.ListStreams", Json).
-spec list_streams(binary(), 1..100, aws_config()) -> erlcloud_kinesis_impl:json_return().
list_streams(ExclusiveStartStreamName, Limit, Config) when is_record(Config, aws_config), is_integer(Limit), Limit > 0, Limit =< 100 ->
Json = [{<<"ExclusiveStartStreamName">>, ExclusiveStartStreamName}, {<<"Limit">>, Limit}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.ListStreams", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStream.html]
%%
%% ===Example===
%%
%% This operation returns the following information about the stream: the current status of the stream, the stream Amazon Resource Name (ARN), and an array of shard objects that comprise the stream.
%%
%% `
%% erlcloud_kinesis:describe_stream(<<"staging">>).
%% {ok,[{<<"StreamDescription">>,
%% [{<<"HasMoreShards">>,false},
%% {<<"Shards">>,
%% [[{<<"HashKeyRange">>,
%% [{<<"EndingHashKey">>,
%% <<"170141183460469231731687303715884105727">>},
%% {<<"StartingHashKey">>,<<"0">>}]},
%% {<<"SequenceNumberRange">>,
%% [{<<"StartingSequenceNumber">>,
%% <<"495372647485535624187345081927970814089871018992"...>>}]},
%% {<<"ShardId">>,<<"shardId-000000000000">>}],
%% [{<<"HashKeyRange">>,
%% [{<<"EndingHashKey">>,
%% <<"340282366920938463463374607431768211455">>},
%% {<<"StartingHashKey">>,
%% <<"170141183460469231731687303715884105728">>}]},
%% {<<"SequenceNumberRange">>,
%% [{<<"StartingSequenceNumber">>,
%% <<"49537264748575863163933038815938617127259750"...>>}]},
%% {<<"ShardId">>,<<"shardId-000000000001">>}]]},
%% {<<"StreamARN">>,
%% <<"arn:aws:kinesis:us-east-1:821148768124:stream/staging">>},
%% {<<"StreamName">>,<<"staging">>},
%% {<<"StreamStatus">>,<<"ACTIVE">>}]}]}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec describe_stream(binary()) -> erlcloud_kinesis_impl:json_return().
describe_stream(StreamName) ->
describe_stream(StreamName, default_config()).
-spec describe_stream(binary(), get_records_limit() | aws_config()) -> erlcloud_kinesis_impl:json_return().
describe_stream(StreamName, Config) when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.DescribeStream", Json);
describe_stream(StreamName, Limit) ->
describe_stream(StreamName, Limit, default_config()).
-spec describe_stream(binary(), get_records_limit(), string() | aws_config()) -> erlcloud_kinesis_impl:json_return().
describe_stream(StreamName, Limit, Config)
when is_record(Config, aws_config),
is_integer(Limit),
Limit >= 1, Limit =< 10000 ->
Json = [{<<"StreamName">>, StreamName}, {<<"Limit">>, Limit}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.DescribeStream", Json);
describe_stream(StreamName, Limit, ExcludeShard) ->
describe_stream(StreamName, Limit, ExcludeShard, default_config()).
-spec describe_stream(binary(), get_records_limit(), string(), aws_config()) -> erlcloud_kinesis_impl:json_return().
describe_stream(StreamName, Limit, ExcludeShard, Config)
when is_record(Config, aws_config),
is_integer(Limit),
Limit >= 1, Limit =< 10000 ->
Json = [{<<"StreamName">>, StreamName}, {<<"Limit">>, Limit}, {<<"ExclusiveStartShardId">>, ExcludeShard}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.DescribeStream", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_DescribeStreamSummary.html]
%%
%% ===Example===
%%
%% Provides a summarized description of the specified Kinesis data stream without the shard list.
%%
%% `
%% erlcloud_kinesis:describe_stream_summary(<<"staging">>).
%% {ok,[{<<"StreamDescriptionSummary">>,
%% [{<<EncryptionType">>,<<"NONE">>},
%% {<<"EnhancedMonitoring">>,[[{<<"ShardLevelMetrics">>,[<<"ALL">>]}]]},
%% {<<"KeyId">>,<<"staging">>},
%% {<<"OpenShardCount">>,10},
%% {<<"RetentionPeriodHours">>,24},
%% {<<"StreamARN">>,
%% <<"arn:aws:kinesis:us-east-1:821148768124:stream/staging">>},
%% {<<"StreamName">>,<<"staging">>},
%% {<<"StreamStatus">>,<<"ACTIVE">>}]}]}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec describe_stream_summary(binary()) -> erlcloud_kinesis_impl:json_return().
describe_stream_summary(StreamName) ->
describe_stream_summary(StreamName, default_config()).
-spec describe_stream_summary(binary(), aws_config()) -> erlcloud_kinesis_impl:json_return().
describe_stream_summary(StreamName, Config) when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.DescribeStreamSummary", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_EnableEnhancedMonitoring.html]
%%
%% ===Example===
%%
%% nables enhanced Kinesis data stream monitoring for shard-level metrics.
%%
%% `
%% erlcloud_kinesis:enable_enhanced_monitoring(<<"staging">>, [<<"ALL">>]).
%% {ok,[
%% {<<"CurrentShardLevelMetrics">>,[]},
%% {<<"DesiredShardLevelMetrics">>,[{<<"ShardLevelMetrics">>,[<<"ALL">>]}]},
%% {<<"StreamName">>,<<"staging">>}
%% ]}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec enable_enhanced_monitoring(binary()) -> erlcloud_kinesis_impl:json_return().
enable_enhanced_monitoring(StreamName) ->
enable_enhanced_monitoring(StreamName, default_config()).
-spec enable_enhanced_monitoring(binary(), list(binary()) | aws_config()) -> erlcloud_kinesis_impl:json_return().
enable_enhanced_monitoring(StreamName, Config) when is_record(Config, aws_config) ->
enable_enhanced_monitoring(StreamName, [<<"ALL">>], Config);
enable_enhanced_monitoring(StreamName, Metrics) when is_list(Metrics) ->
enable_enhanced_monitoring(StreamName, Metrics, default_config()).
-spec enable_enhanced_monitoring(binary(), list(binary()), aws_config()) -> erlcloud_kinesis_impl:json_return().
enable_enhanced_monitoring(StreamName, Metrics, Config)
when is_record(Config, aws_config), is_list(Metrics) ->
Json = [{<<"StreamName">>, StreamName}, {<<"ShardLevelMetrics">>, Metrics}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.EnableEnhancedMonitoring", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_DisableEnhancedMonitoring.html]
%%
%% ===Example===
%%
%% Disables enhanced monitoring.
%%
%% `
%% erlcloud_kinesis:disable_enhanced_monitoring(<<"staging">>, [<<"ALL">>]).
%% {ok,[
%% {<<"CurrentShardLevelMetrics">>,[{<<"ShardLevelMetrics">>,[<<"ALL">>]}]},
%% {<<"DesiredShardLevelMetrics">>,[]},
%% {<<"StreamName">>,<<"staging">>}
%% ]}
%% '
%% @end
%%------------------------------------------------------------------------------
-spec disable_enhanced_monitoring(binary()) -> erlcloud_kinesis_impl:json_return().
disable_enhanced_monitoring(StreamName) ->
disable_enhanced_monitoring(StreamName, default_config()).
-spec disable_enhanced_monitoring(binary(), list(binary()) | aws_config()) -> erlcloud_kinesis_impl:json_return().
disable_enhanced_monitoring(StreamName, Config) when is_record(Config, aws_config) ->
disable_enhanced_monitoring(StreamName, [<<"ALL">>], Config);
disable_enhanced_monitoring(StreamName, Metrics) when is_list(Metrics) ->
disable_enhanced_monitoring(StreamName, Metrics, default_config()).
-spec disable_enhanced_monitoring(binary(), list(binary()), aws_config()) -> erlcloud_kinesis_impl:json_return().
disable_enhanced_monitoring(StreamName, Metrics, Config)
when is_record(Config, aws_config), is_list(Metrics) ->
Json = [{<<"StreamName">>, StreamName}, {<<"ShardLevelMetrics">>, Metrics}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.DisableEnhancedMonitoring", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetShardIterator.html]
%%
%% ===Example===
%%
%% This operation returns a shard iterator in ShardIterator. The shard iterator specifies the position in the shard from which you want to start reading data records sequentially.
%%
%% `
%% erlcloud_kinesis:get_shard_iterator(<<"test">>, <<"shardId-000000000001">>, <<"TRIM_HORIZON">>).
%% {ok,[{<<"ShardIterator">>,
%% <<"AAAAAAAAAAFHJejL6/AjDShV3pIXsxYZT7Xj2G6EHxokHqT2D1stIOVYUEyprlUGWUepKqUDaR0+hB6qTlKvZa+fsBRqgHi4"...>>}]}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec get_shard_iterator(binary(), binary(), binary()) ->
erlcloud_kinesis_impl:json_return().
get_shard_iterator(StreamName, ShardId, ShardIteratorType) ->
get_shard_iterator(StreamName, ShardId, ShardIteratorType, default_config()).
-spec get_shard_iterator(binary(), binary(), binary(),
binary() | integer() | float() | aws_config()) ->
erlcloud_kinesis_impl:json_return().
get_shard_iterator(StreamName, ShardId, ShardIteratorType, Config)
when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName},
{<<"ShardId">>, ShardId},
{<<"ShardIteratorType">>, ShardIteratorType}],
get_shard_iterator_request(Config, Json);
get_shard_iterator(StreamName, ShardId, ShardIteratorType, SeqNOrTs) ->
get_shard_iterator(StreamName, ShardId, ShardIteratorType, SeqNOrTs,
default_config()).
-spec get_shard_iterator(binary(), binary(), binary(),
binary() | integer() | float(), aws_config()) ->
erlcloud_kinesis_impl:json_return().
get_shard_iterator(StreamName, ShardId, ShardIteratorType = <<"AT_TIMESTAMP">>,
Timestamp, Config) ->
Json = [{<<"StreamName">>, StreamName},
{<<"ShardId">>, ShardId},
{<<"ShardIteratorType">>, ShardIteratorType},
{<<"Timestamp">>, Timestamp}],
get_shard_iterator_request(Config, Json);
get_shard_iterator(StreamName, ShardId, ShardIteratorType,
StartingSequenceNumber, Config) ->
Json = [{<<"StreamName">>, StreamName},
{<<"ShardId">>, ShardId},
{<<"ShardIteratorType">>, ShardIteratorType},
{<<"StartingSequenceNumber">>, StartingSequenceNumber}],
get_shard_iterator_request(Config, Json).
get_shard_iterator_request(Config, Json) ->
Operation = "Kinesis_20131202.GetShardIterator",
erlcloud_kinesis_impl:request(Config, Operation, Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_GetRecords.html]
%%
%% ===Example===
%%
%% This operation returns one or more data records from a shard. A GetRecords operation request can retrieve up to 10 MB of data.
%%
%% `
%% {ok, [{_, A2}]} = erlcloud_kinesis:get_shard_terator(<<"test">>, <<"shardId-000000000000">>, <<"TRIM_HORIZON">>).
%% {ok,[{<<"ShardIterator">>,
%% <<"AAAAAAAAAAEuncwaAk+GTC2TIdmdg5w6dIuZ4Scu6vaMGPtaPUfopvw9cBm2NM3Rlj9WyI5JFJr2ahuSh3Z187AdW4Lug86E"...>>}]}
%% erlcloud_kinesis:get_records(A2).
%% {ok,[{<<"NextShardIterator">>,
%% <<"AAAAAAAAAAEkuCmrC+QDW1gUywyu7G8GxvRyM6GSMkcHQ9wrvCJBW87mjn9C8YEckkipaoJySwgKXMmn1BwSPjnjiUCsu6pc"...>>},
%% {<<"Records">>,
%% [[{<<"Data">>,<<"asdasd">>},
%% {<<"PartitionKey">>,<<"key">>},
%% {<<"SequenceNumber">>,
%% <<"49537292605574028653758531131893428543501381406818304001">>}],
%% [{<<"Data">>,<<"asdasd 213123123">>},
%% {<<"PartitionKey">>,<<"key">>},
%% {<<"SequenceNumber">>,
%% <<"49537292605574028653758541428570459745183078607853977601">>}]]}]}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec get_records(string()) -> erlcloud_kinesis_impl:json_return().
get_records(ShardIterator) ->
Json = [{<<"ShardIterator">>, ShardIterator}],
get_normalized_records(default_config(), Json).
-spec get_records(string(), get_records_limit()| aws_config()) -> erlcloud_kinesis_impl:json_return().
get_records(ShardIterator, Config) when is_record(Config, aws_config) ->
Json = [{<<"ShardIterator">>, ShardIterator}],
get_normalized_records(Config, Json);
get_records(ShardIterator, Limit)
when is_integer(Limit), Limit > 0, Limit =< 10000 ->
get_records(ShardIterator, Limit, default_config()).
-spec get_records(binary(), get_records_limit(), aws_config()) ->
erlcloud_kinesis_impl:json_return().
get_records(ShardIterator, Limit, Config) ->
get_records(ShardIterator, Limit, [], Config).
-spec get_records(binary(), get_records_limit(), proplist(), aws_config()) ->
erlcloud_kinesis_impl:json_return().
get_records(ShardIterator, Limit, Options, Config)
when is_record(Config, aws_config),
is_integer(Limit),
Limit > 0, Limit =< 10000 ->
Json = [{<<"ShardIterator">>, ShardIterator},
{<<"Limit">>, Limit}],
ShouldDecode = proplists:get_value(decode, Options, true),
get_normalized_records(Config, Json, ShouldDecode).
get_normalized_records(Config, Json) ->
get_normalized_records(Config, Json, true).
get_normalized_records(Config, Json, ShouldDecode)
when is_record(Config, aws_config) ->
Request = "Kinesis_20131202.GetRecords",
case erlcloud_kinesis_impl:request(Config, Request, Json, ShouldDecode) of
{ok, Resp} when is_binary(Resp) -> {ok, Resp};
{ok, Resp} -> {ok, normalize_response(Resp)};
{error, Reason} -> {error, Reason}
end.
normalize_record([{K,V} | T]) when K == <<"Data">> -> [ {K, base64:decode(V)} | normalize_record(T) ];
normalize_record([K | T]) -> [K | normalize_record(T) ];
normalize_record([]) -> [].
normalize_records([K | V]) -> [ normalize_record(K) | normalize_records(V) ];
normalize_records([]) -> [].
normalize_response([{K,V} | T]) when K == <<"Records">> -> [ {K, normalize_records(V)} | normalize_response(T)];
normalize_response([K | T]) -> [K | normalize_response(T)];
normalize_response([]) -> [].
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecord.html]
%%
%% ===Example===
%%
%% This operation puts a data record into an Amazon Kinesis stream from a producer.
%%
%% `
%% erlcloud_kinesis:put_record(<<"test">>, <<"key">>, <<"asdasd">>).
%% {ok,[{<<"SequenceNumber">>,
%% <<"49537292605574028653758531131893428543501381406818304001">>},
%% {<<"ShardId">>,<<"shardId-000000000000">>}]}
%% erlcloud_kinesis:put_record(<<"test">>, <<"key">>, <<"asdasd 213123123">>).
%% {ok,[{<<"SequenceNumber">>,
%% <<"49537292605574028653758541428570459745183078607853977601">>},
%% {<<"ShardId">>,<<"shardId-000000000000">>}]}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-type partition_key() :: binary().
-type payload() :: binary() | string().
-type explicit_hash_key() :: binary() | undefined.
-type ordering() :: binary() | undefined.
-spec put_record(binary(), partition_key(), payload()) ->
erlcloud_kinesis_impl:json_return().
put_record(StreamName, PartitionKey, Data) ->
put_record(StreamName, PartitionKey, Data, undefined).
-spec put_record(binary(), partition_key(), payload(), explicit_hash_key() | aws_config()) ->
erlcloud_kinesis_impl:json_return().
put_record(StreamName, PartitionKey, Data, Config) when is_record(Config, aws_config) ->
put_record(StreamName, PartitionKey, Data, undefined, Config);
put_record(StreamName, PartitionKey, Data, ExplicitHashKey) ->
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, undefined).
-spec put_record(binary(), partition_key(), payload(), explicit_hash_key(), ordering() | aws_config()) ->
erlcloud_kinesis_impl:json_return().
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, Config) when is_record(Config, aws_config) ->
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, undefined, Config);
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, Ordering) ->
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, Ordering, []).
-spec put_record(binary(), partition_key(), payload(), explicit_hash_key(), ordering(), proplist() | aws_config()) ->
erlcloud_kinesis_impl:json_return().
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, Ordering, Config) when is_record(Config, aws_config) ->
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, Ordering, [], Config);
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, Ordering, Options) ->
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, Ordering, Options, default_config()).
-spec put_record(binary(), partition_key(), payload(), explicit_hash_key(), ordering(), proplist(), aws_config()) ->
erlcloud_kinesis_impl:json_return().
put_record(StreamName, PartitionKey, Data, ExplicitHashKey, Ordering, Options, Config) when is_record(Config, aws_config) ->
Encoded = case proplists:get_value(encode, Options, true) of
true -> base64:encode(Data);
false -> Data
end,
Optional = [{<<"ExplicitHashKey">>, ExplicitHashKey},
{<<"SequenceNumberForOrdering">>, Ordering}],
Json = [{<<"StreamName">>, StreamName},
{<<"PartitionKey">>, PartitionKey},
{<<"Data">>, Encoded}
|[ KV || {_, V} = KV <- Optional, V /= undefined ]],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.PutRecord", Json).
%%------------------------------------------------------------------------------
%% Kinesis API:
%% [https://docs.aws.amazon.com/kinesis/latest/APIReference/API_PutRecords.html]
%%
%% ===Example===
%%
%% This operation puts data records into an Amazon Kinesis stream from a producer.
%%
%% `
%% erlcloud_kinesis:put_records(<<"StreamName">>, [{<<"data">>, <<"partitionKey1">>}]).
%% {ok, [{<<"FailedRecordCount">>, 0},
%% {<<"Records">>,
%% [
%% [{<<"SequenceNumber">>,
%% <<"49537292605574028653758531131893428543501381406818304001">>},
%% {<<"ShardId">>,<<"shardId-000000000000">>}]
%% ]}
%% ]
%% }
%% erlcloud_kinesis:put_records(<<"StreamName">>, [{<<"data">>, <<"explicitHashKey">>, <<"partitionKey1">>}]).
%% {ok, [{<<"FailedRecordCount">>, 0},
%% {<<"Records">>,
%% [
%% [{<<"SequenceNumber">>,
%% <<"49537292605574028653758531131893428543501381406818304001">>},
%% {<<"ShardId">>,<<"shardId-000000000000">>}]
%% ]}
%% ]
%% }
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-type put_records_item() :: {Data :: string(), PartitionKey :: string()}
| {Data :: string(), ExplicitHashKey :: string(), PartitionKey :: string()}.
-type put_records_items() :: [put_records_item()].
-spec put_records(binary(), put_records_items()) -> erlcloud_kinesis_impl:json_return().
put_records(StreamName, Items) ->
put_records(StreamName, Items, default_config()).
-spec put_records(binary(), put_records_items(), function() | aws_config()) -> erlcloud_kinesis_impl:json_return().
put_records(StreamName, Items, EncodingFun) when is_function(EncodingFun, 1) ->
put_records(StreamName, Items, EncodingFun, default_config());
put_records(StreamName, Items, Config) ->
put_records(StreamName, Items, fun default_put_encoding/1, Config).
-spec put_records(binary(), put_records_items(),
function(), Config :: aws_config()) -> erlcloud_kinesis_impl:json_return().
put_records(StreamName, Items, EncodingFun, Config) when is_function(EncodingFun, 1) ->
Operation = put_records_operation(),
Json = prepare_put_records_data(StreamName, Items, EncodingFun),
erlcloud_kinesis_impl:request(Config, Operation, Json).
default_put_encoding(D) -> base64:encode(D).
put_records_operation() ->
"Kinesis_20131202.PutRecords".
prepare_put_records_data(StreamName, Items, Fun) ->
Records = [prepare_put_records_item(X, Fun) || X <- Items],
[{<<"StreamName">>, StreamName}, {<<"Records">>, Records}].
prepare_put_records_item({Data, PartitionKey}, Fun) ->
[{<<"PartitionKey">>, PartitionKey},
{<<"Data">>, Fun(Data)}];
prepare_put_records_item({Data, ExplicitHashKey, PartitionKey}, Fun) ->
[{<<"PartitionKey">>, PartitionKey},
{<<"ExplicitHashKey">>, ExplicitHashKey},
{<<"Data">>, Fun(Data)}].
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_MergeShards.html]
%%
%% ===Example===
%%
%% This operation 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. Two shards are considered adjacent if the union of the hash key ranges for the two shards form a contiguous set with no gaps.
%%
%% `
%% erlcloud_kinesis:merge_shards(<<"test">>, <<"shardId-000000000001">>, <<"shardId-000000000003">>).
%% {ok,{incomplete,#Fun<jsx_decoder.1.688044>}}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec merge_shards(binary(), string(), string()) -> erlcloud_kinesis_impl:json_return().
merge_shards(StreamName, AdjacentShardToMerge, ShardToMerge) ->
Json = [{<<"StreamName">>, StreamName}, {<<"AdjacentShardToMerge">>, AdjacentShardToMerge}, {<<"ShardToMerge">>, ShardToMerge}],
erlcloud_kinesis_impl:request(default_config(), "Kinesis_20131202.MergeShards", Json).
-spec merge_shards(binary(), string(), string(), aws_config()) -> erlcloud_kinesis_impl:json_return().
merge_shards(StreamName, AdjacentShardToMerge, ShardToMerge, Config) when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName}, {<<"AdjacentShardToMerge">>, AdjacentShardToMerge}, {<<"ShardToMerge">>, ShardToMerge}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.MergeShards", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_SplitShard.html]
%%
%% ===Example===
%%
%% This operation splits a shard into two new shards in the stream, to increase the stream's capacity to ingest and transport data.
%%
%% `
%% erlcloud_kinesis:split_shards(<<"test">>, <<"shardId-000000000000">>, <<"10">>).
%% {ok,{incomplete,#Fun<jsx_decoder.1.688044>}}
%% '
%%
%% @end
%%------------------------------------------------------------------------------
-spec split_shards(binary(), string(), string()) -> erlcloud_kinesis_impl:json_return().
split_shards(StreamName, ShardToSplit, NewStartingHashKey) ->
Json = [{<<"StreamName">>, StreamName}, {<<"ShardToSplit">>, ShardToSplit}, {<<"NewStartingHashKey">>, NewStartingHashKey}],
erlcloud_kinesis_impl:request(default_config(), "Kinesis_20131202.SplitShard", Json).
-spec split_shards(binary(), string(), string(), aws_config()) -> erlcloud_kinesis_impl:json_return().
split_shards(StreamName, ShardToSplit, NewStartingHashKey, Config) when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName}, {<<"ShardToSplit">>, ShardToSplit}, {<<"NewStartingHashKey">>, NewStartingHashKey}],
erlcloud_kinesis_impl:request(Config, "Kinesis_20131202.SplitShard", Json).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_AddTagsToStream.html]
%%
%% Adds or updates tags for the specified Amazon Kinesis stream.
%%
%% erlcloud_kinesis:add_tags_to_stream(<<"stream_name">>,
%% [{<<"tag_key">>, <<"tag_value">>}]).
%% ok.
%%
%% @end
%%------------------------------------------------------------------------------
-spec add_tags_to_stream(binary(), [{binary(), binary()}, ...]) ->
ok | {error, any()}.
add_tags_to_stream(StreamName, Tags) ->
add_tags_to_stream(StreamName, Tags, default_config()).
-spec add_tags_to_stream(binary(), [{binary(), binary()}, ...], aws_config()) ->
ok | {error, any()}.
add_tags_to_stream(StreamName, Tags, Config)
when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName},
{<<"Tags">>, Tags}],
Operation = "Kinesis_20131202.AddTagsToStream",
Response = erlcloud_kinesis_impl:request(Config, Operation, Json),
normalize_tags_response(Response).
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_ListTagsForStream.html]
%%
%% Lists the tags for the specified Amazon Kinesis stream.
%%
%% erlcloud_kinesis:list_tags_for_stream(<<"stream_name">>, <<"key2">>, 2).
%% {ok, [
%% {<<"HasMoreTags">>, true},
%% {<<"Tags">>, [[{<<"Key">>, <<"k1">>}, {<<"Value">>, <<"v1">>}],
%% [{<<"Key">>, <<"k2">>}, {<<"Value">>, <<"v2">>}]]}
%% ]
%% }.
%%
%% @end
%%------------------------------------------------------------------------------
-spec list_tags_for_stream(binary()) -> {ok, proplist()} | {error, any()}.
list_tags_for_stream(StreamName) ->
list_tags_for_stream(StreamName, default_config()).
-spec list_tags_for_stream(binary(), aws_config()) ->
{ok, proplist()} | {error, any()}.
list_tags_for_stream(StreamName, Config) when is_record(Config, aws_config) ->
list_tags_for_stream(StreamName, undefined, undefined, Config).
-spec list_tags_for_stream(binary(),
binary() | undefined,
integer() | undefined) ->
{ok, proplist()} | {error, any()}.
list_tags_for_stream(StreamName, ESK, Limit) ->
list_tags_for_stream(StreamName, ESK, Limit, default_config()).
-spec list_tags_for_stream(binary(),
binary() | undefined,
integer() | undefined,
aws_config()) ->
erlcloud_kinesis_impl:json_return().
list_tags_for_stream(StreamName, ESK, Limit, Config)
when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName},
{<<"ExclusiveStartTagKey">>, ESK},
{<<"Limit">>, Limit}],
JsonUpd = lists:filter(fun filter_undefined_params/1, Json),
Operation = "Kinesis_20131202.ListTagsForStream",
erlcloud_kinesis_impl:request(Config, Operation, JsonUpd).
filter_undefined_params({_, undefined}) -> false;
filter_undefined_params(_) -> true.
%%------------------------------------------------------------------------------
%% @doc
%% Wrapper around 'list_tags_for_stream' to return {ok, <list_of_all_tags>}.
%%
%% erlcloud_kinesis:list_all_tags_for_stream(<<"stream_name">>).
%% {ok, [{<<"k1">>, <<"v1">>},
%% {<<"k2">>, <<"v2">>}]
%% }.
%%
%% @end
%%------------------------------------------------------------------------------
-spec list_all_tags_for_stream(binary()) -> {ok, proplist()} | {error, any()}.
list_all_tags_for_stream(StreamName) ->
list_all_tags_for_stream(StreamName, default_config()).
-spec list_all_tags_for_stream(binary(), aws_config()) ->
{ok, proplist()} | {error, any()}.
list_all_tags_for_stream(StreamName, Config) ->
list_all_tags_for_stream(StreamName, undefined, undefined, Config, []).
list_all_tags_for_stream(StreamName, ESK, Limit, Config, Acc) ->
case list_tags_for_stream(StreamName, ESK, Limit, Config) of
{ok, [{<<"HasMoreTags">>, true}, {<<"Tags">>, Tags}]} ->
NewESK = get_last_tag_key(Tags),
UpdAcc = Acc ++ Tags,
list_all_tags_for_stream(StreamName, NewESK, Limit, Config, UpdAcc);
{ok, [{<<"HasMoreTags">>, false}, {<<"Tags">>, Tags}]} ->
{ok, [ {K, V} || [{<<"Key">>, K}, {<<"Value">>, V}] <- Acc ++ Tags ]};
Error ->
Error
end.
get_last_tag_key(Tags) ->
[{<<"Key">>, Key}, _] = lists:last(Tags),
Key.
%%------------------------------------------------------------------------------
%% @doc
%% Kinesis API:
%% [http://docs.aws.amazon.com/kinesis/latest/APIReference/API_RemoveTagsFromStream.html]
%%
%% Removes tags from the specified Amazon Kinesis stream.
%%
%% erlcloud_kinesis:remove_tags_from_stream(<<"stream_name">>, [<<"tag_key">>]).
%% ok.
%%
%% @end
%%------------------------------------------------------------------------------
-spec remove_tags_from_stream(binary(), [binary(), ...]) -> ok | {error, any()}.
remove_tags_from_stream(StreamName, TagKeys) ->
remove_tags_from_stream(StreamName, TagKeys, default_config()).
-spec remove_tags_from_stream(binary(), [binary(), ...], aws_config()) ->
ok | {error, any()}.
remove_tags_from_stream(StreamName, TagKeys, Config)
when is_record(Config, aws_config) ->
Json = [{<<"StreamName">>, StreamName},
{<<"TagKeys">>, TagKeys}],
Operation = "Kinesis_20131202.RemoveTagsFromStream",
Response = erlcloud_kinesis_impl:request(Config, Operation, Json),
normalize_tags_response(Response).
normalize_tags_response({ok, []}) -> ok;
normalize_tags_response(Error) -> Error.
%%------------------------------------------------------------------------------
%% Unit Tests
%%------------------------------------------------------------------------------
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
list_all_tags_pagination_test_() ->
meck:new(EK = erlcloud_kinesis_impl, [passthrough]),
Tags1 = [[{<<"Key">>, <<"k1">>}, {<<"Value">>, <<"v1">>}],
[{<<"Key">>, <<"k2">>}, {<<"Value">>, <<"v2">>}]],
Tags2 = [[{<<"Key">>, <<"k3">>}, {<<"Value">>, <<"v3">>}]],
meck:sequence(EK, request, 3,
[{ok, [{<<"HasMoreTags">>, true}, {<<"Tags">>, Tags1}]},
{ok, [{<<"HasMoreTags">>, false}, {<<"Tags">>, Tags2}]}]),
Result = erlcloud_kinesis:list_all_tags_for_stream(<<"stream">>),
meck:unload(EK),
?_assertEqual({ok, [{<<"k1">>, <<"v1">>},
{<<"k2">>, <<"v2">>},
{<<"k3">>, <<"v3">>}]},
Result).
-endif.