Packages
ex_aws
0.0.3
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/mix/tasks/tail.ex
defmodule Mix.Tasks.Kinesis.Tail do
alias ExAws.Kinesis
require Logger
use Mix.Task
@shortdoc "tails a stream"
@moduledoc """
Tails a Stream
## Usage
kinesis.tail [stream_name] [options]
## Options
--poll N Time in seconds between polling. Default: 5
--debug Sets debug_requests: true on ex_aws. Logs all kinesis requests
## Examples
$mix kinesis.tail Elixir.Jetstream.Messages.Timeseries
$mix kinesis.tail logs --debug --poll 10
"""
def run(argv) do
{:ok, _} = Application.ensure_all_started(:ex_aws)
{opts, [stream_name|_], _} = OptionParser.parse(argv)
sleep_time = Keyword.get(opts, :poll, "5") |> String.to_integer
debug = Keyword.get(opts, :debug, false)
Application.put_env(:ex_aws, :debug_requests, debug)
Application.put_env(:ex_aws, :kinesis_namespace, nil)
Logger.info "Streaming from #{stream_name |> ExAws.Config.namespace(:kinesis)}"
stream_name
|> get_shards
|> Enum.map(&Kinesis.get_shard_iterator(stream_name, &1["ShardId"], "LATEST"))
|> Enum.map(&get_records(&1, sleep_time))
end
def get_shards(name) do
case Kinesis.Lazy.describe_stream(name) do
{:ok, %{"StreamDescription" => %{"Shards" => shards}}} -> shards
error -> raise error
end
end
def get_records({:ok, %{"ShardIterator" => iterator}}, wait_time) do
iterator
|> Kinesis.Lazy.get_records(%{}, fn
[] -> :timer.sleep(wait_time * 1000); []
val -> val
end)
|> Stream.map(&format_msg/1)
|> Stream.run
end
defp format_msg(msg) do
IO.ANSI.format_fragment([:blue, msg["PartitionKey"], :bright, " | ",
:reset, msg["Data"] |> ensure_new_line ])
|> IO.chardata_to_string
|> IO.write
end
defp ensure_new_line(data) do
case String.last(data) do
"\n" -> data
_ -> [data, "\n"]
end
end
end