A Script to Get a Kinesis Stream Test Message

1
#!/bin/bash
2

3
JQ_CHECK=$(which jq)
4
if [ -z "$JQ_CHECK" ]; then
5
  echo
6
  echo "This script requires the jq JSON processor. Please install for your OS from https://stedolan.github.io/jq/download/"
7
  echo
8
  exit 1
9
fi
10

11
if [ $# -ne 1 ]; then
12
  echo
13
  echo "usage: $0 <stream_name>"
14
  echo
15
  exit 1
16
fi
17

18
# Set the stream name
19
STREAM_NAME=$1
20

21
# Choose the iterator type:
22
# TRIM HORIZON is for starting at the begining of the Kinesis Stream.
23
# This can take a while if you have a lot of records.
24
# To use TRIM HORIZON, uncomment the following line:
25
# TYPE=TRIM_HORIZON
26

27
# AT_TIMESTAMP allows you to go back to a point in time. This is set for going back one hour
28
# To use AT_TIMESTAMP, uncomment the following two lines:
29
# TIMESTAMP=$(($(date +%s) - 3600)).000
30
# TYPE="AT_TIMESTAMP --timestamp $TIMESTAMP"
31

32
# LATEST means start at the most current point in the stream and read forward
33
TYPE=LATEST
34

35
# Get a list of shards
36
SHARDS=$(aws kinesis list-shards --stream-name $STREAM_NAME | jq -r .Shards[].ShardId)
37

38
# Get all the starting points
39
SHARD_ITERATOR=()
40
i=0
41
for shard in $SHARDS ; do
42
  SHARD_ITERATOR[$i]=$(aws kinesis get-shard-iterator --shard-id $shard --shard-iterator-type $TYPE --stream-name $STREAM_NAME --query 'ShardIterator')
43
  i=$((i+1))
44
done
45

46
# Start getting events from all shards and display them
47
while [ 1 ] ; do
48
  len=${#SHARD_ITERATOR[@]}
49
  for (( i=0; i < $len; i++ )); do
50
    DATA=$(aws kinesis get-records --limit 50 --shard-iterator ${SHARD_ITERATOR[$i]})
51
    SHARD_ITERATOR[$i]=$(echo $DATA | jq -r .NextShardIterator)
52
    ROWS=$(echo $DATA | jq -r .Records[].Data?)
53
    for row in $ROWS; do
54
      echo $row | base64 -d | jq .
55
    done
56
  done
57
done