Packages
ex_aws
0.2.0
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
--from Sequence number to start at. If unspecified, LATEST is used
## Examples
$mix kinesis.tail my-kinesis-stream
$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)
seq = Keyword.get(opts, :from)
{shard_type, opts} = case seq do
nil -> {"LATEST", %{}}
val -> {"AT_SEQUENCE_NUMBER", %{StartingSequenceNumber: val}}
end
Application.put_env(:ex_aws, :debug_requests, debug)
Logger.info "Streaming from #{stream_name}"
stream_name
|> get_shards
|> Enum.map(&Kinesis.get_shard_iterator(stream_name, &1["ShardId"], shard_type, opts))
|> Enum.map(&get_records(&1, sleep_time))
end
defp get_shards(name) do
case Kinesis.describe_stream(name) do
{:ok, %{"StreamDescription" => %{"Shards" => shards}}} -> shards
error -> raise inspect(error)
end
end
def get_records({:ok, %{"ShardIterator" => iterator}}, wait_time) do
iterator
|> Kinesis.stream_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