Packages
ex_aws
0.3.1
2.7.0
2.6.1
2.6.0
2.5.11
2.5.10
2.5.9
2.5.8
2.5.7
2.5.6
2.5.5
2.5.4
2.5.3
2.5.2
2.5.1
2.5.0
2.4.4
2.4.3
2.4.2
2.4.1
2.4.0
2.3.4
2.3.3
2.3.2
2.3.1
2.3.0
2.2.10
2.2.9
2.2.8
2.2.7
2.2.6
2.2.5
2.2.4
2.2.3
2.2.2
2.2.1
2.2.0
2.1.9
2.1.8
2.1.7
2.1.6
2.1.5
2.1.4
2.1.3
2.1.2
2.1.1
2.1.0
2.0.2
2.0.1
2.0.0
1.1.5
1.1.4
1.1.3
1.1.2
1.1.1
1.1.0
1.0.0
1.0.0-rc.4
1.0.0-rc.3
1.0.0-rc.1
1.0.0-beta3
1.0.0-beta2
1.0.0-beta1
1.0.0-beta0
0.5.0
0.4.19
0.4.18
0.4.17
0.4.15
0.4.14
0.4.13
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.1
0.3.0
0.2.0
0.1.2
0.1.1
0.1.0
0.0.5
0.0.4
0.0.3
AWS client for Elixir. Currently supports Dynamo, DynamoStreams, EC2, Firehose, Kinesis, KMS, Lambda, RRDS, Route53, S3, SES, SNS, SQS, STS and others.
Current section
Files
Jump to
Current section
Files
lib/ex_aws/kinesis/lazy.ex
defmodule ExAws.Kinesis.Lazy do
alias ExAws.Kinesis
@moduledoc false
# Implimentation of lazy functions surfaced by ExAws.Kinesis.Client
def stream_shards(client, stream, opts \\ []) do
request_fun = fn
{:initial, initial} -> initial
fun_opts -> Kinesis.Impl.describe_stream(client, stream, Map.merge(opts, fun_opts))
end
client
|> Kinesis.Impl.describe_stream(stream, opts)
|> do_stream_shards(request_fun)
end
defp do_stream_shards({:error, results}, _), do: {:error, results}
defp do_stream_shards({:ok, results}, request_fun) do
stream = build_shard_stream({:ok, results}, request_fun)
{:ok, put_in(results["StreamDescription"], %{"Shards" => stream})}
end
defp build_shard_stream(initial, request_fun) do
Stream.resource(fn -> {request_fun, {:initial, initial}} end, fn
:quit -> {:halt, nil}
{fun, args} -> case fun.(args) do
{:ok, %{"StreamDescription" => %{"Shards" => shards, "HasMoreShards" => true}}} ->
opts = %{"ExclusiveStartShardId" => shards |> List.last |> Map.get("ShardId")}
{shards, {fun, opts}}
{:ok, %{"StreamDescription" => %{"Shards" => shards}}} ->
{shards, :quit}
end
end, &pass/1)
end
defp pass(x), do: x
@doc """
Returns a stream of record shard iterator tuples.
NOTE: This stream is basically INFINITE, in that it runs
until the shard it is reading from closes, which may be never.
"""
def stream_records(client, shard_iterator, opts \\ [], each_req_fun \\ &pass/1) do
sleep_time = Keyword.get(opts, :sleep_between_req_time, 200)
Stream.resource(fn ->
pid = spawn_link(__MODULE__, :kinesis_request_handler, [self, client, sleep_time, opts, each_req_fun])
{pid, shard_iterator}
end,
fn
{:quit, pid} -> {:halt, pid}
{pid, shard_iter} ->
send(pid, {:request, shard_iter})
result = receive do
value -> value
end
case result do
{:ok, %{"Records" => records, "NextShardIterator" => shard_iter}} ->
{
records |> each_req_fun.(),
{pid, shard_iter}
}
{:ok, %{"Records" => records}} ->
{each_req_fun.(records), {:quit, pid}}
end
end,
fn(pid) -> send pid, :quit end
)
end
def kinesis_request_handler(parent_pid, client, sleep_time, opts, each_req_fun) do
receive do
{:request, shard_iter} ->
send parent_pid, Kinesis.Impl.get_records(client, shard_iter, opts)
:timer.sleep(sleep_time)
:wait
:quit ->
exit(:normal)
end
kinesis_request_handler(parent_pid, client, sleep_time, opts, each_req_fun)
end
end