%% -*- mode: erlang;erlang-indent-level: 4;indent-tabs-mode: nil -*- %% @author Smiler Lee %% @author Nicholas Lundgaard %% @doc %% An Erlang interface to Amazon's DynamoDB Streams. %% %% [http://docs.aws.amazon.com/amazondynamodb/latest/developerguide/Streams.html] %% %% Method names match DynamoDB Streams operations converted to %% lower_case_with_underscores. %% %% Required parameters are passed as function arguments. In addition %% all methods take an options proplist argument which can be used to %% pass optional parameters. %% %% Output is in the form of `{ok, Value}' or `{error, Reason}'. The %% format of `Value' is controlled by the `out' option, which defaults %% to `simple'. The possible values are: %% %% * `simple' - The most interesting part of the output. For example %% `list_streams' will return the streams, `describe_stream' will return %% the stream description. %% %% * `record' - A record containing all the information from the %% DynamoDB Streams response except field types. This is useful if you %% need more detailed information than what is returned with `simple'. %% For example, with `list_streams' the record will contain the last %% evaluated stream ARN which can be used to continue the operation. %% %% * `typed_record' - A record containing all the information from the %% DynamoDB Streams response. All field values are returned with type %% information. This option only makes sense to `get_records'. %% %% * `json' - The output from DynamoDB Streams as processed by %% `jsx:decode' but with no further manipulation. This would rarely be %% useful, unless the DynamoDB Streams API is updated to include data %% that is not yet parsed correctly. %% %% DynamoDB Streams errors are return in the form `{error, {ErrorCode, %% Message}}' where `ErrorCode' and 'Message' are both binary %% strings. So %% to handle limit exceeded exception, match `{error, %% {<<"LimitExceededException">>, _}}'. %% %% As to the error retries, we use `erlcloud_retry:default_retry/1' %% as the default retry strategy, this behaviour can be changed through %% `aws_config()'. %% %% @end -module(erlcloud_ddb_streams). -include("erlcloud.hrl"). -include("erlcloud_aws.hrl"). -include("erlcloud_ddb_streams.hrl"). %%% Library initialization. -export([configure/2, configure/3, new/2, new/3]). %%% DynamoDB Streams API -export([describe_stream/1, describe_stream/2, describe_stream/3, get_records/1, get_records/2, get_records/3, get_shard_iterator/3, get_shard_iterator/4, get_shard_iterator/5, list_streams/0, list_streams/1, list_streams/2 ]). -export_type( [aws_region/0, event_id/0, event_name/0, event_source/0, event_version/0, item/0, key/0, key_schema/0, sequence_number/0, sequence_number_range/0, shard_id/0, shard_iterator/0, shard_iterator_type/0, stream_arn/0, stream_label/0, stream_status/0, stream_view_type/0, table_name/0 ]). %%%------------------------------------------------------------------------------ %%% Library initialization. %%%------------------------------------------------------------------------------ -spec new(string(), string()) -> aws_config(). new(AccessKeyID, SecretAccessKey) -> #aws_config{access_key_id=AccessKeyID, secret_access_key=SecretAccessKey, retry = fun erlcloud_retry:default_retry/1}. -spec new(string(), string(), string()) -> aws_config(). new(AccessKeyID, SecretAccessKey, Host) -> #aws_config{access_key_id=AccessKeyID, secret_access_key=SecretAccessKey, ddb_streams_host=Host, retry = fun erlcloud_retry:default_retry/1}. -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). default_config() -> erlcloud_aws:default_config(). %%%------------------------------------------------------------------------------ %%% Shared Types %%%------------------------------------------------------------------------------ -type table_name() :: binary(). -type attr_name() :: binary(). -type hash_key_name() :: attr_name(). -type range_key_name() :: attr_name(). -type key_schema() :: hash_key_name() | {hash_key_name(), range_key_name()}. -type stream_arn() :: binary(). -type stream_label() :: binary(). -type stream_status() :: enabling | enabled | disabling | disabled. -type stream_view_type() :: keys_only | new_image | old_image | new_and_old_images. -type shard_id() :: binary(). -type shard_iterator() :: binary(). -type shard_iterator_type() :: trim_horizon | latest | at_sequence_number | after_sequence_number. -type sequence_number() :: binary(). -type sequence_number_range() :: {sequence_number(), sequence_number()} | {sequence_number(), undefined}. -type aws_region() :: binary(). -type event_id() :: binary(). -type event_name() :: insert | modify | remove. -type event_source() :: binary(). -type event_version() :: binary(). -type untyped_attr_value() :: binary() | number() | boolean() | undefined | [binary()] | [number()] | [untyped_attr_value()] | [untyped_attr()]. -type typed_attr_value() :: {s, binary()} | {n, number()} | {b, binary()} | {bool, boolean()} | {null, true} | {ss, [binary(),...]} | {ns, [number(),...]} | {bs, [binary(),...]} | {l, [typed_attr_value()]} | {m, [typed_attr()]}. -type untyped_attr() :: {attr_name(), untyped_attr_value()}. -type typed_attr() :: {attr_name(), typed_attr_value()}. -type attr() :: untyped_attr() | typed_attr(). -type item() :: [attr()]. -type key() :: [attr(),...]. -type json_pair() :: {binary(), json_term()}. -type json_term() :: jsx:json_term(). -type ok_return(T) :: {ok, T} | {error, term()}. %%%------------------------------------------------------------------------------ %%% Shared Dynamizers %%%------------------------------------------------------------------------------ -spec dynamize_shard_iterator_type(shard_iterator_type()) -> binary(). dynamize_shard_iterator_type(trim_horizon) -> <<"TRIM_HORIZON">>; dynamize_shard_iterator_type(latest) -> <<"LATEST">>; dynamize_shard_iterator_type(at_sequence_number) -> <<"AT_SEQUENCE_NUMBER">>; dynamize_shard_iterator_type(after_sequence_number) -> <<"AFTER_SEQUENCE_NUMBER">>. %%%------------------------------------------------------------------------------ %%% Shared Undynamizers %%%------------------------------------------------------------------------------ -type undynamize_opt() :: {typed, boolean()}. -type undynamize_opts() :: [undynamize_opt()]. -spec id(X, undynamize_opts()) -> X. id(X, _) -> X. -spec undynamize_number(binary(), undynamize_opts()) -> number(). undynamize_number(Value, _) -> String = binary_to_list(Value), case lists:member($., String) of true -> list_to_float(String); false -> list_to_integer(String) end. -spec undynamize_value_untyped(json_pair(), undynamize_opts()) -> untyped_attr_value(). undynamize_value_untyped({<<"S">>, Value}, _) when is_binary(Value) -> Value; undynamize_value_untyped({<<"N">>, Value}, Opts) -> undynamize_number(Value, Opts); undynamize_value_untyped({<<"B">>, Value}, _) -> base64:decode(Value); undynamize_value_untyped({<<"BOOL">>, Value}, _) when is_boolean(Value) -> Value; undynamize_value_untyped({<<"NULL">>, true}, _) -> undefined; undynamize_value_untyped({<<"SS">>, Values}, _) when is_list(Values) -> Values; undynamize_value_untyped({<<"NS">>, Values}, Opts) -> [undynamize_number(Value, Opts) || Value <- Values]; undynamize_value_untyped({<<"BS">>, Values}, _) -> [base64:decode(Value) || Value <- Values]; undynamize_value_untyped({<<"L">>, List}, Opts) -> [undynamize_value_untyped(Value, Opts) || [Value] <- List]; undynamize_value_untyped({<<"M">>, Map}, Opts) -> [undynamize_attr_untyped(Attr, Opts) || Attr <- Map]. -spec undynamize_attr_untyped(json_pair(), undynamize_opts()) -> untyped_attr(). undynamize_attr_untyped({Name, [ValueJson]}, Opts) -> {Name, undynamize_value_untyped(ValueJson, Opts)}. -spec undynamize_object(fun((json_pair(), undynamize_opts()) -> A), [json_pair()] | [{}], undynamize_opts()) -> [A]. undynamize_object(_, [{}], _) -> %% jsx returns [{}] for empty objects []; undynamize_object(PairFun, List, Opts) -> [PairFun(I, Opts) || I <- List]. -spec undynamize_item(json_term(), undynamize_opts()) -> item(). undynamize_item(Json, Opts) -> case lists:keyfind(typed, 1, Opts) of {typed, true} -> undynamize_object(fun undynamize_attr_typed/2, Json, Opts); _ -> undynamize_object(fun undynamize_attr_untyped/2, Json, Opts) end. -spec undynamize_key(json_term(), undynamize_opts()) -> key(). undynamize_key(Json, Opts) -> case lists:keyfind(typed, 1, Opts) of {typed, true} -> [undynamize_attr_typed(I, Opts) || I <- Json]; _ -> [undynamize_attr_untyped(I, Opts) || I <- Json] end. -spec undynamize_value_typed(json_pair(), undynamize_opts()) -> typed_attr_value(). undynamize_value_typed({<<"S">>, Value}, _) when is_binary(Value) -> {s, Value}; undynamize_value_typed({<<"N">>, Value}, Opts) -> {n, undynamize_number(Value, Opts)}; undynamize_value_typed({<<"B">>, Value}, _) -> {b, base64:decode(Value)}; undynamize_value_typed({<<"BOOL">>, Value}, _) when is_boolean(Value) -> {bool, Value}; undynamize_value_typed({<<"NULL">>, true}, _) -> {null, true}; undynamize_value_typed({<<"SS">>, Values}, _) when is_list(Values) -> {ss, Values}; undynamize_value_typed({<<"NS">>, Values}, Opts) -> {ns, [undynamize_number(Value, Opts) || Value <- Values]}; undynamize_value_typed({<<"BS">>, Values}, _) -> {bs, [base64:decode(Value) || Value <- Values]}; undynamize_value_typed({<<"L">>, List}, Opts) -> {l, [undynamize_value_typed(Value, Opts) || [Value] <- List]}; undynamize_value_typed({<<"M">>, Map}, Opts) -> {m, [undynamize_attr_typed(Attr, Opts) || Attr <- Map]}. -spec undynamize_attr_typed(json_pair(), undynamize_opts()) -> typed_attr(). undynamize_attr_typed({Name, [ValueJson]}, Opts) -> {Name, undynamize_value_typed(ValueJson, Opts)}. key_name(Key) -> proplists:get_value(<<"AttributeName">>, Key). -spec undynamize_key_schema(json_term(), undynamize_opts()) -> key_schema(). undynamize_key_schema([HashKey], _) -> key_name(HashKey); undynamize_key_schema([Key1, Key2], _) -> case proplists:get_value(<<"KeyType">>, Key1) of <<"HASH">> -> {key_name(Key1), key_name(Key2)}; <<"RANGE">> -> {key_name(Key2), key_name(Key1)} end. -spec undynamize_stream_status(binary(), undynamize_opts()) -> stream_status(). undynamize_stream_status(<<"ENABLING">>, _) -> enabling; undynamize_stream_status(<<"ENABLED">>, _) -> enabled; undynamize_stream_status(<<"DISABLING">>, _) -> disabling; undynamize_stream_status(<<"DISABLED">>, _) -> disabled. -spec undynamize_stream_view_type(binary(), undynamize_opts()) -> stream_view_type(). undynamize_stream_view_type(<<"KEYS_ONLY">>, _) -> keys_only; undynamize_stream_view_type(<<"NEW_IMAGE">>, _) -> new_image; undynamize_stream_view_type(<<"OLD_IMAGE">>, _) -> old_image; undynamize_stream_view_type(<<"NEW_AND_OLD_IMAGES">>, _) -> new_and_old_images. -spec undynamize_sequence_number_range(json_term(), undynamize_opts()) -> sequence_number_range(). undynamize_sequence_number_range(Json, _) -> Starting = proplists:get_value(<<"StartingSequenceNumber">>, Json), Ending = proplists:get_value(<<"EndingSequenceNumber">>, Json), {Starting, Ending}. -spec undynamize_event_name(binary(), undynamize_opts()) -> event_name(). undynamize_event_name(<<"INSERT">>, _) -> insert; undynamize_event_name(<<"MODIFY">>, _) -> modify; undynamize_event_name(<<"REMOVE">>, _) -> remove. -type field_table() :: [{binary(), pos_integer(), fun((json_term(), undynamize_opts()) -> term())}]. -spec undynamize_folder(field_table(), json_pair(), undynamize_opts(), tuple()) -> tuple(). undynamize_folder(Table, {Key, Value}, Opts, A) -> case lists:keyfind(Key, 1, Table) of {Key, Index, ValueFun} -> setelement(Index, A, ValueFun(Value, Opts)); false -> A end. -type record_desc() :: {tuple(), field_table()}. -spec undynamize_record(record_desc(), json_term(), undynamize_opts()) -> tuple(). undynamize_record({Record, _}, [{}], _) -> %% jsx returns [{}] for empty objects Record; undynamize_record({Record, Table}, Json, Opts) -> lists:foldl(fun(Pair, A) -> undynamize_folder(Table, Pair, Opts, A) end, Record, Json). %%%------------------------------------------------------------------------------ %%% Shared Options %%%------------------------------------------------------------------------------ -spec id(X) -> X. id(X) -> X. -type out_type() :: json | record | typed_record | simple. -type out_opt() :: {out, out_type()}. -type property() :: proplists:property(). -type aws_opts() :: [json_pair()]. -type ddb_streams_opts() :: [out_opt()]. -type opts() :: {aws_opts(), ddb_streams_opts()}. -spec verify_ddb_streams_opt(atom(), term()) -> ok. verify_ddb_streams_opt(out, Value) -> case lists:member(Value, [json, record, typed_record, simple]) of true -> ok; false -> error({erlcloud_ddb, {invalid_opt, {out, Value}}}) end; verify_ddb_streams_opt(Name, Value) -> error({erlcloud_ddb, {invalid_opt, {Name, Value}}}). -type opt_table_entry() :: {atom(), binary(), fun((_) -> json_term())}. -type opt_table() :: [opt_table_entry()]. -spec opt_folder(opt_table(), property(), opts()) -> opts(). opt_folder(_, {_, undefined}, Opts) -> %% ignore options set to undefined Opts; opt_folder(Table, {Name, Value}, {AwsOpts, DdbOpts}) -> case lists:keyfind(Name, 1, Table) of {Name, Key, ValueFun} -> {[{Key, ValueFun(Value)} | AwsOpts], DdbOpts}; false -> verify_ddb_streams_opt(Name, Value), {AwsOpts, [{Name, Value} | DdbOpts]} end. -spec opts(opt_table(), proplist()) -> opts(). opts(Table, Opts) when is_list(Opts) -> %% remove duplicate options Opts1 = lists:ukeysort(1, proplists:unfold(Opts)), lists:foldl(fun(Opt, A) -> opt_folder(Table, Opt, A) end, {[], []}, Opts1); opts(_, _) -> error({erlcloud_ddb, opts_not_list}). %%%------------------------------------------------------------------------------ %%% Output %%%------------------------------------------------------------------------------ -type ddb_streams_return(Record, Simple) :: {ok, json_term() | Record | Simple} | {error, term()}. -type undynamize_fun() :: fun((json_term(), undynamize_opts()) -> tuple()). -spec out(json_return(), undynamize_fun(), ddb_streams_opts()) -> {ok, json_term() | tuple()} | {simple, term()} | {error, term()}. out({error, Reason}, _, _) -> {error, Reason}; out({ok, Json}, Undynamize, Opts) -> case proplists:get_value(out, Opts, simple) of json -> {ok, Json}; record -> {ok, Undynamize(Json, [])}; typed_record -> {ok, Undynamize(Json, [{typed, true}])}; simple -> {simple, Undynamize(Json, [])} end. %% Returns specified field of tuple for simple return -spec out(json_return(), undynamize_fun(), ddb_streams_opts(), pos_integer()) -> ok_return(term()). out(Result, Undynamize, Opts, Index) -> out(Result, Undynamize, Opts, Index, {error, no_return}). -spec out(json_return(), undynamize_fun(), ddb_streams_opts(), pos_integer(), ok_return(term())) -> ok_return(term()). out(Result, Undynamize, Opts, Index, Default) -> case out(Result, Undynamize, Opts) of {simple, Record} -> case element(Index, Record) of undefined -> Default; Element -> {ok, Element} end; Else -> Else end. %%%------------------------------------------------------------------------------ %%% Shared Records %%%------------------------------------------------------------------------------ -spec stream_record() -> record_desc(). stream_record() -> {#ddb_streams_stream{}, [{<<"StreamArn">>, #ddb_streams_stream.stream_arn, fun id/2}, {<<"StreamLabel">>, #ddb_streams_stream.stream_label, fun id/2}, {<<"TableName">>, #ddb_streams_stream.table_name, fun id/2} ]}. -spec shard_record() -> record_desc(). shard_record() -> {#ddb_streams_shard{}, [{<<"ParentShardId">>, #ddb_streams_shard.parent_shard_id, fun id/2}, {<<"SequenceNumberRange">>, #ddb_streams_shard.sequence_number_range, fun undynamize_sequence_number_range/2}, {<<"ShardId">>, #ddb_streams_shard.shard_id, fun id/2} ]}. -spec stream_description_record() -> record_desc(). stream_description_record() -> {#ddb_streams_stream_description{}, [{<<"CreationRequestDateTime">>, #ddb_streams_stream_description.creation_request_date_time, fun id/2}, {<<"KeySchema">>, #ddb_streams_stream_description.key_schema, fun undynamize_key_schema/2}, {<<"LastEvaluatedShardId">>, #ddb_streams_stream_description.last_evaluated_shard_id, fun id/2}, {<<"Shards">>, #ddb_streams_stream_description.shards, fun(V, Opts) -> [undynamize_record(shard_record(), I, Opts) || I <- V] end}, {<<"StreamArn">>, #ddb_streams_stream_description.stream_arn, fun id/2}, {<<"StreamLabel">>, #ddb_streams_stream_description.stream_label, fun id/2}, {<<"StreamStatus">>, #ddb_streams_stream_description.stream_status, fun undynamize_stream_status/2}, {<<"StreamViewType">>, #ddb_streams_stream_description.stream_view_type, fun undynamize_stream_view_type/2}, {<<"TableName">>, #ddb_streams_stream_description.table_name, fun id/2} ]}. -spec stream_record_record() -> record_desc(). stream_record_record() -> {#ddb_streams_stream_record{}, [{<<"ApproximateCreationDateTime">>, #ddb_streams_stream_record.approximate_creation_date_time, fun id/2}, {<<"Keys">>, #ddb_streams_stream_record.keys, fun undynamize_key/2}, {<<"NewImage">>, #ddb_streams_stream_record.new_image, fun undynamize_item/2}, {<<"OldImage">>, #ddb_streams_stream_record.old_image, fun undynamize_item/2}, {<<"SequenceNumber">>, #ddb_streams_stream_record.sequence_number, fun id/2}, {<<"SizeBytes">>, #ddb_streams_stream_record.size_bytes, fun id/2}, {<<"StreamViewType">>, #ddb_streams_stream_record.stream_view_type, fun undynamize_stream_view_type/2}]}. -spec record_record() -> record_desc(). record_record() -> {#ddb_streams_record{}, [{<<"awsRegion">>, #ddb_streams_record.aws_region, fun id/2}, {<<"dynamodb">>, #ddb_streams_record.dynamodb, fun(V, Opts) -> undynamize_record(stream_record_record(), V, Opts) end}, {<<"eventID">>, #ddb_streams_record.event_id, fun id/2}, {<<"eventName">>, #ddb_streams_record.event_name, fun undynamize_event_name/2}, {<<"eventSource">>, #ddb_streams_record.event_source, fun id/2}, {<<"eventVersion">>, #ddb_streams_record.event_version, fun id/2}]}. %%%------------------------------------------------------------------------------ %%% DescribeStream %%%------------------------------------------------------------------------------ -type describe_stream_opt() :: {exclusive_start_shard_id, shard_id()} | {limit, 1..100} | out_opt(). -type describe_stream_opts() :: [describe_stream_opt()]. -spec describe_stream_opts() -> opt_table(). describe_stream_opts() -> [{exclusive_start_shard_id, <<"ExclusiveStartShardId">>, fun id/1}, {limit, <<"Limit">>, fun id/1}]. -spec describe_stream_record() -> record_desc(). describe_stream_record() -> {#ddb_streams_describe_stream{}, [{<<"StreamDescription">>, #ddb_streams_describe_stream.stream_description, fun(V, Opts) -> undynamize_record(stream_description_record(), V, Opts) end} ]}. -type describe_stream_return() :: ddb_streams_return(#ddb_streams_describe_stream{}, #ddb_streams_stream_description{}). -spec describe_stream(stream_arn()) -> describe_stream_return(). describe_stream(StreamArn) -> describe_stream(StreamArn, [], default_config()). -spec describe_stream(stream_arn(), describe_stream_opts()) -> describe_stream_return(). describe_stream(StreamArn, Opts) -> describe_stream(StreamArn, Opts, default_config()). %%------------------------------------------------------------------------------ %% @doc %% DynamoDB Streams API: %% [http://docs.aws.amazon.com/dynamodbstreams/latest/APIReference/API_DescribeStream.html] %% %% ===Example=== %% %% Describe a stream. %% %% ` %% erlcloud_ddb_streams:describe_stream(<<"arn:aws:dynamodb:us-west-2:111122223333:table/Forum/stream/2015-05-20T20:51:10.252">>). %% {ok, #ddb_streams_stream_description{ %% creation_request_date_time = 1437677671.062, %% key_schema = {<<"ForumName">>, <<"Subject">>}, %% last_evaluated_shard_id = undefined, %% shards = [#ddb_streams_shard{ %% parent_shard_id = <<"shardId-00000001414562045508-2bac9cd1">>, %% sequence_number_range = {<<"20500000000000000910398">>, %% <<"20500000000000000910398">>}, %% shard_id = <<"shardId-00000001414562045508-2bac9cd2">>}], %% stream_arn = <<"arn:aws:dynamodb:us-west-2:111122223333:table/Forum/stream/2015-05-20T20:51:10.252">>, %% stream_label = <<"2015-05-20T20:51:10.252">>, %% stream_status = disabled, %% stream_view_type = new_and_old_images, %% table_name = <<"Forum">>}} %% ' %% @end %%------------------------------------------------------------------------------ -spec describe_stream(stream_arn(), describe_stream_opts(), aws_config()) -> describe_stream_return(). describe_stream(StreamArn, Opts, Config) -> {AwsOpts, DdbStreamsOpts} = opts(describe_stream_opts(), Opts), Return = request( Config, "DynamoDBStreams_20120810.DescribeStream", [{<<"StreamArn">>, StreamArn}] ++ AwsOpts), out(Return, fun(Json, UOpts) -> undynamize_record(describe_stream_record(), Json, UOpts) end, DdbStreamsOpts, #ddb_streams_describe_stream.stream_description). %%%------------------------------------------------------------------------------ %%% GetRecords %%%------------------------------------------------------------------------------ -type get_records_opt() :: {limit, 1..1000} | out_opt(). -type get_records_opts() :: [get_records_opt()]. -spec get_records_opts() -> opt_table(). get_records_opts() -> [{limit, <<"Limit">>, fun id/1}]. -spec get_records_record() -> record_desc(). get_records_record() -> {#ddb_streams_get_records{}, [{<<"NextShardIterator">>, #ddb_streams_get_records.next_shard_iterator, fun id/2}, {<<"Records">>, #ddb_streams_get_records.records, fun(V, Opts) -> [undynamize_record(record_record(), I, Opts) || I <- V] end}]}. -type get_records_return() :: ddb_streams_return(#ddb_streams_get_records{}, [#ddb_streams_record{}]). -spec get_records(shard_iterator()) -> get_records_return(). get_records(ShardIterator) -> get_records(ShardIterator, [], default_config()). -spec get_records(shard_iterator(), get_records_opts()) -> get_records_return(). get_records(ShardIterator, Opts) -> get_records(ShardIterator, Opts, default_config()). %%------------------------------------------------------------------------------ %% @doc %% DynamoDB Streams API: %% [http://docs.aws.amazon.com/dynamodbstreams/latest/APIReference/API_GetRecords.html] %% %% ===Example=== %% %% Get the records from a given shard iterator. %% %% ` %% erlcloud_ddb_streams:get_records(<<"arn:aws:dynamodb:us-west-2:111122223333:table/Forum/stream/2015-05-20T20:51:10.252|1|AAAAAAAAAAEvJp6D+zaQ... ...">>). %% {ok, [#ddb_streams_record{ %% aws_region = <<"us-west-2">>, %% dynamodb = #ddb_streams_stream_record{ %% keys = [{<<"ForumName">>, <<"DynamoDB">>}, %% {<<"Subject">>, <<"DynamoDB Thread 3">>}], %% new_image = undefined, %% old_image = undefined, %% sequence_number = <<"300000000000000499659">>, %% size_bytes = 41, %% stream_view_type = keys_only}, %% event_id = <<"e2fd9c34eff2d779b297b26f5fef4206">>, %% event_name = insert, %% event_source = <<"aws:dynamodb">>, %% event_version = <<"1.0">>}, %% #ddb_streams_record{ %% aws_region = <<"us-west-2">>, %% dynamodb = #ddb_streams_stream_record{ %% keys = [{<<"ForumName">>, <<"DynamoDB">>}, %% {<<"Subject">>, <<"DynamoDB Thread 1">>}], %% new_image = undefined, %% old_image = undefined, %% sequence_number = <<"400000000000000499660">>, %% size_bytes = 41, %% stream_view_type = keys_only}, %% event_id = <<"4b25bd0da9a181a155114127e4837252">>, %% event_name = modify, %% event_source = <<"aws:dynamodb">>, %% event_version = <<"1.0">>}, %% #ddb_streams_record{ %% aws_region = <<"us-west-2">>, %% dynamodb = #ddb_streams_stream_record{ %% keys = [{<<"ForumName">>, <<"DynamoDB">>}, %% {<<"Subject">>, <<"DynamoDB Thread 2">>}], %% new_image = undefined, %% old_image = undefined, %% sequence_number = <<"500000000000000499661">>, %% size_bytes = 41, %% stream_view_type = keys_only}, %% event_id = <<"740280c73a3df7842edab3548a1b08ad">>, %% event_name = remove, %% event_source = <<"aws:dynamodb">>, %% event_version = <<"1.0">>}]} %% ' %% %% @end %%------------------------------------------------------------------------------ -spec get_records(shard_iterator(), get_records_opts(), aws_config()) -> get_records_return(). get_records(ShardIterator, Opts, Config) -> {AwsOpts, DdbStreamsOpts} = opts(get_records_opts(), Opts), Return = request( Config, "DynamoDBStreams_20120810.GetRecords", [{<<"ShardIterator">>, ShardIterator}] ++ AwsOpts), out(Return, fun(Json, UOpts) -> undynamize_record(get_records_record(), Json, UOpts) end, DdbStreamsOpts, #ddb_streams_get_records.records). %%%------------------------------------------------------------------------------ %%% GetShardIterator %%%------------------------------------------------------------------------------ -type get_shard_iterator_opt() :: {sequence_number, sequence_number()} | out_opt(). -type get_shard_iterator_opts() :: [get_shard_iterator_opt()]. -spec get_shard_iterator_opts() -> opt_table(). get_shard_iterator_opts() -> [{sequence_number, <<"SequenceNumber">>, fun id/1}]. -spec get_shard_iterator_record() -> record_desc(). get_shard_iterator_record() -> {#ddb_streams_get_shard_iterator{}, [{<<"ShardIterator">>, #ddb_streams_get_shard_iterator.shard_iterator, fun id/2} ]}. -type get_shard_iterator_return() :: ddb_streams_return(#ddb_streams_get_shard_iterator{}, shard_iterator()). -spec get_shard_iterator(stream_arn(), shard_id(), shard_iterator_type()) -> get_shard_iterator_return(). get_shard_iterator(StreamArn, ShardId, ShardIteratorType) -> get_shard_iterator(StreamArn, ShardId, ShardIteratorType, [], default_config()). -spec get_shard_iterator(stream_arn(), shard_id(), shard_iterator_type(), get_shard_iterator_opts()) -> get_shard_iterator_return(). get_shard_iterator(StreamArn, ShardId, ShardIteratorType, Opts) -> get_shard_iterator(StreamArn, ShardId, ShardIteratorType, Opts, default_config()). %%------------------------------------------------------------------------------ %% @doc %% DynamoDB API: %% [http://docs.aws.amazon.com/dynamodbstreams/latest/APIReference/API_GetShardIterator.html] %% %% ===Example=== %% %% Get a shard iterator. %% %% ` %% erlcloud_ddb_streams:get_shard_iterator(<<"arn:aws:dynamodb:us-west-2:111122223333:table/Forum/stream/2015-05-20T20:51:10.252">>, %% <<"shardId-00000001414576573621-f55eea83">>, %% trim_horizon). %% {ok, <<"arn:aws:dynamodb:us-west-2:111122223333:table/Forum/stream/2015-05-20T20:51:10.252|1|AAAAAAAAAAEvJp6D+zaQ... ...">>} %% ' %% %% @end %%------------------------------------------------------------------------------ -spec get_shard_iterator(stream_arn(), shard_id(), shard_iterator_type(), get_shard_iterator_opts(), aws_config()) -> get_shard_iterator_return(). get_shard_iterator(StreamArn, ShardId, ShardIteratorType, Opts, Config) -> {AwsOpts, DdbStreamsOpts} = opts(get_shard_iterator_opts(), Opts), Return = request( Config, "DynamoDBStreams_20120810.GetShardIterator", [{<<"StreamArn">>, StreamArn}, {<<"ShardId">>, ShardId}, {<<"ShardIteratorType">>, dynamize_shard_iterator_type(ShardIteratorType)}] ++ AwsOpts), out(Return, fun(Json, UOpts) -> undynamize_record(get_shard_iterator_record(), Json, UOpts) end, DdbStreamsOpts, #ddb_streams_get_shard_iterator.shard_iterator). %%%------------------------------------------------------------------------------ %%% ListStreams %%%------------------------------------------------------------------------------ -type list_streams_opt() :: {exclusive_start_stream_arn, stream_arn()} | {limit, 1..100} | {table_name, table_name()} | out_opt(). -type list_streams_opts() :: [list_streams_opt()]. -spec list_streams_opts() -> opt_table(). list_streams_opts() -> [{exclusive_start_stream_arn, <<"ExclusiveStartStreamArn">>, fun id/1}, {limit, <<"Limit">>, fun id/1}, {table_name, <<"TableName">>, fun id/1}]. -spec list_streams_record() -> record_desc(). list_streams_record() -> {#ddb_streams_list_streams{}, [{<<"LastEvaluatedStreamArn">>, #ddb_streams_list_streams.last_evaluated_stream_arn, fun id/2}, {<<"Streams">>, #ddb_streams_list_streams.streams, fun(V, Opts) -> [undynamize_record(stream_record(), I, Opts) || I <- V] end} ]}. -type list_streams_return() :: ddb_streams_return(#ddb_streams_list_streams{}, [#ddb_streams_stream{}]). -spec list_streams() -> list_streams_return(). list_streams() -> list_streams([], default_config()). -spec list_streams(list_streams_opts()) -> list_streams_return(). list_streams(Opts) -> list_streams(Opts, default_config()). %%------------------------------------------------------------------------------ %% @doc %% DynamoDB API: %% [http://docs.aws.amazon.com/dynamodbstreams/latest/APIReference/API_ListStreams.html] %% %% ===Example=== %% %% Get the list of streams with a limit of 3. %% %% ` %% erlcloud_ddb_streams:list_streams([{limit, 3}]). %% {ok, [#ddb_streams_stream{ %% stream_arn = <<"arn:aws:dynamodb:us-west-2:111122223333:table/Forum/stream/2015-05-20T20:51:10.252">>, %% stream_label = <<"2015-05-20T20:51:10.252">>, %% table_name = <<"Forum">>}, %% #ddb_streams_stream{ %% stream_arn = <<"arn:aws:dynamodb:us-west-2:111122223333:table/Forum/stream/2015-05-20T20:50:02.714">>, %% stream_label = <<"2015-05-20T20:50:02.714">>, %% table_name = <<"Forum">>}, %% #ddb_streams_stream{ %% stream_arn = <<"arn:aws:dynamodb:us-west-2:111122223333:table/Forum/stream/2015-05-19T23:03:50.641">>, %% stream_label = <<"2015-05-19T23:03:50.641">>, %% table_name = <<"Forum">>}]} %% ' %% %% @end %%------------------------------------------------------------------------------ -spec list_streams(list_streams_opts(), aws_config()) -> list_streams_return(). list_streams(Opts, Config) -> {AwsOpts, DdbStreamsOpts} = opts(list_streams_opts(), Opts), Return = request( Config, "DynamoDBStreams_20120810.ListStreams", AwsOpts), out(Return, fun(Json, UOpts) -> undynamize_record(list_streams_record(), Json, UOpts) end, DdbStreamsOpts, #ddb_streams_list_streams.streams). %%%------------------------------------------------------------------------------ %%% Request %%%------------------------------------------------------------------------------ -type operation() :: string(). -type json_return() :: {ok, json_term()} | {error, term()}. -spec request(aws_config(), operation(), json_term()) -> json_return(). request(Config, Operation, Json) -> case erlcloud_aws:update_config(Config) of {ok, Config2} -> request2(Config2, Operation, Json); {error, _} = Error -> Error end. -spec request2(aws_config(), operation(), json_term()) -> json_return(). request2(Config, Operation, Json) -> Body = case Json of [] -> <<"{}">>; _ -> jsx:encode(Json) end, Headers = headers(Config, Operation, Body), Request = #aws_request{service = ddb_streams, uri = uri(Config), method = post, request_headers = Headers, request_body = Body}, request_to_return(erlcloud_retry:request(Config, Request, fun result_fun/1)). -spec request_to_return(#aws_request{}) -> json_return(). request_to_return(#aws_request{response_type = ok, response_body = Body}) -> %% TODO check crc {ok, jsx:decode(Body, [{return_maps, false}])}; request_to_return(#aws_request{response_type = error, error_type = aws, httpc_error_reason = undefined, response_status = Status, response_status_line = StatusLine, response_body = Body}) -> {error, {http_error, Status, StatusLine, Body}}; request_to_return(#aws_request{response_type = error, error_type = aws, httpc_error_reason = Reason}) -> {error, Reason}; request_to_return(#aws_request{response_type = error, error_type = httpc, httpc_error_reason = Reason}) -> {error, Reason}. -spec result_fun(#aws_request{}) -> #aws_request{}. result_fun(#aws_request{response_type = ok} = Request) -> Request; result_fun(#aws_request{response_type = error, error_type = aws, response_status = Status} = Request) -> %% == IMPLEMENTATION NOTES == %% Set `httpc_error_reason` to `undefined` in case of retry Request2 = Request#aws_request{httpc_error_reason = undefined}, if Status >= 400 andalso Status < 500 -> client_error(Request2); Status >= 500 -> Request2#aws_request{should_retry = true}; Status < 400 -> Request2#aws_request{should_retry = false} end. -spec client_error(#aws_request{}) -> #aws_request{}. client_error(#aws_request{response_body = Body} = Request) -> %% == IMPLEMENTATION NOTES == %% We store the error reason in `httpc_error_reason` for now, %% this may be changed at any time later case jsx:is_json(Body) of false -> Request#aws_request{should_retry = false}; true -> Json = jsx:decode(Body, [{return_maps, false}]), case proplists:get_value(<<"__type">>, Json) of undefined -> Request#aws_request{should_retry = false}; FullType -> Message = proplists:get_value(<<"message">>, Json, <<>>), case binary:split(FullType, <<"#">>) of [_, Type] when Type =:= <<"LimitExceededException">>; Type =:= <<"ThrottlingException">> -> Request#aws_request{httpc_error_reason = {Type, Message}, should_retry = true}; [_, Type] -> Request#aws_request{httpc_error_reason = {Type, Message}, should_retry = false}; _ -> Request#aws_request{should_retry = false} end end end. -spec headers(aws_config(), string(), binary()) -> [{string(), string()}]. headers(Config, Operation, Body) -> Headers = [{"host", Config#aws_config.ddb_streams_host}, {"x-amz-target", Operation}, {"content-type", "application/x-amz-json-1.0"}], Region = erlcloud_aws:aws_region_from_host(Config#aws_config.ddb_streams_host), erlcloud_aws:sign_v4_headers(Config, Headers, Body, Region, "dynamodb"). uri(#aws_config{ddb_streams_scheme = Scheme, ddb_streams_host = Host} = Config) -> lists:flatten([Scheme, Host, port_spec(Config)]). port_spec(#aws_config{ddb_streams_port=80}) -> ""; port_spec(#aws_config{ddb_streams_port=Port}) -> [":", erlang:integer_to_list(Port)].