#!/usr/bin/env ruby # frozen_string_literal: true require File.expand_path("../config/boot", __dir__) require "aws-sdk-kinesis" require "json" ENDPOINT = ARGV[0] STREAM_NAME = ARGV[1] REGION = ARGV[2] @kinesis = Aws::Kinesis::Client.new( access_key_id: "key", secret_access_key: "secret", endpoint: ENDPOINT, region: REGION || "us-east-1" ) res = @kinesis.get_shard_iterator( stream_name: STREAM_NAME, shard_id: "shardId-000000000000", shard_iterator_type: "TRIM_HORIZON" ) iterator = res.data[:shard_iterator] loop do res = @kinesis.get_records(shard_iterator: iterator) res.data[:records].each do |record| body = record[:data] pp JSON.parse(body) puts "---" end iterator = res.data[:next_shard_iterator] sleep 1 end