func ExampleKinesis_DecreaseStreamRetentionPeriod() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.DecreaseStreamRetentionPeriodInput{
		RetentionPeriodHours: aws.Int64(1),             // Required
		StreamName:           aws.String("StreamName"), // Required
	}
	resp, err := svc.DecreaseStreamRetentionPeriod(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func KinesisLoop(event <-chan *d2k.Event, callback func(*d2k.Event) interface{}) {
	// main driver for pushing events to Kinesis

	svc := kinesis.New(&aws.Config{Region: "us-west-2"})

	// Wait... for stream to be ready
	for {
		params := &kinesis.DescribeStreamInput{
			StreamName: &KinesisStreamName,
		}
		if resp, err := svc.DescribeStream(params); err != nil {
			if awsErr, ok := err.(awserr.Error); ok {
				log.Fatalln("KinesisLoop:", awsErr.Code(), awsErr.Message())
			} else {
				log.Fatalln("KinesisLoop:", err)
			}
		} else if *resp.StreamDescription.StreamStatus == "ACTIVE" {
			break
		} else {
			log.Println("KinesisLoop:", KinesisStreamName, "not active")
			time.Sleep(1 * time.Second)
		}
	}

	for one_event := range event {
		msg := callback(one_event)
		if data, err := json.Marshal(msg); err != nil {
			log.Fatalln(err)
		} else {
			log.Println("KinesisLoop:", string(data))
			// TODO: send Kinesis event
		}
	}
}

//Consume gets data from a shard of an AWS Kinesis stream and puts them in a Go channel, once every second.
func Consume(shardID string, streamName string, region string, shardIteratorType string, eventChannel chan []byte) {
	svc := kinesis.New(session.New(), &aws.Config{Region: aws.String(region)})
	getShardIteratorParams := &kinesis.GetShardIteratorInput{
		ShardId:           aws.String(shardID),           // Required
		ShardIteratorType: aws.String(shardIteratorType), // Required
		StreamName:        aws.String(streamName),        // Required
	}
	shardIterator, err := svc.GetShardIterator(getShardIteratorParams)
	if err != nil {
		log.Panic(err)
	}

	nextShardIterator := shardIterator.ShardIterator
	for {
		getRecordsParams := &kinesis.GetRecordsInput{
			ShardIterator: nextShardIterator, // Required
			Limit:         aws.Int64(1000),
		}
		getRecordsResp, err := svc.GetRecords(getRecordsParams)
		if err != nil {
			log.Panic(err)
		}
		for _, record := range getRecordsResp.Records {
			eventChannel <- record.Data
		}
		nextShardIterator = getRecordsResp.NextShardIterator
		time.Sleep(time.Second)
	}
}

func ExampleKinesis_SplitShard() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.SplitShardInput{
		NewStartingHashKey: aws.String("HashKey"),    // Required
		ShardToSplit:       aws.String("ShardId"),    // Required
		StreamName:         aws.String("StreamName"), // Required
	}
	resp, err := svc.SplitShard(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleKinesis_UpdateShardCount() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.UpdateShardCountInput{
		ScalingType:      aws.String("ScalingType"), // Required
		StreamName:       aws.String("StreamName"),  // Required
		TargetShardCount: aws.Int64(1),              // Required
	}
	resp, err := svc.UpdateShardCount(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleKinesis_PutRecord() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.PutRecordInput{
		Data:                      []byte("PAYLOAD"),          // Required
		PartitionKey:              aws.String("PartitionKey"), // Required
		StreamName:                aws.String("StreamName"),   // Required
		ExplicitHashKey:           aws.String("HashKey"),
		SequenceNumberForOrdering: aws.String("SequenceNumber"),
	}
	resp, err := svc.PutRecord(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

// NewKinesisBufferFlusher returns a KinesisBufferFlusher configured with
// the stream name and partition key.
func NewKinesisBufferFlusher(name, partitionKey string) *KinesisBufferFlusher {
	sess := session.New()

	// Are we assuming a role?
	roleARN := conf.GetString("role-arn")
	if roleARN != "" {
		sess.Config.Credentials = stscreds.NewCredentials(sess, roleARN, func(o *stscreds.AssumeRoleProvider) {
			rsn := conf.GetString("role-session-name")
			if rsn != "" {
				o.RoleSessionName = rsn
			}
		})
	}

	region := conf.GetString("region")
	if region != "" {
		sess.Config.Region = aws.String(region)
	}

	return &KinesisBufferFlusher{
		Name:         aws.String(name),
		PartitionKey: partitionKey,
		kinesis:      kinesis.New(sess),
	}
}

func ExampleKinesis_GetShardIterator() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.GetShardIteratorInput{
		ShardId:                aws.String("ShardId"),           // Required
		ShardIteratorType:      aws.String("ShardIteratorType"), // Required
		StreamName:             aws.String("StreamName"),        // Required
		StartingSequenceNumber: aws.String("SequenceNumber"),
		Timestamp:              aws.Time(time.Now()),
	}
	resp, err := svc.GetShardIterator(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleKinesis_ListTagsForStream() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.ListTagsForStreamInput{
		StreamName:           aws.String("StreamName"), // Required
		ExclusiveStartTagKey: aws.String("TagKey"),
		Limit:                aws.Int64(1),
	}
	resp, err := svc.ListTagsForStream(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleKinesis_EnableEnhancedMonitoring() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.EnableEnhancedMonitoringInput{
		ShardLevelMetrics: []*string{ // Required
			aws.String("MetricsName"), // Required
			// More values...
		},
		StreamName: aws.String("StreamName"), // Required
	}
	resp, err := svc.EnableEnhancedMonitoring(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleKinesis_GetRecords() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.GetRecordsInput{
		ShardIterator: aws.String("ShardIterator"), // Required
		Limit:         aws.Int64(1),
	}
	resp, err := svc.GetRecords(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

// Set defaults for all fields, initialize channel and client if not provided
func (stream *StreamWorker) initialize() error {
	if stream.IteratorType == "" {
		stream.IteratorType = ShardIteratorTypeLatest
	}

	if stream.BatchSize == 0 {
		stream.BatchSize = DefaultBatchSize
	}

	if stream.SleepTime == 0 {
		stream.SleepTime = DefaultSleepTime
	}

	if stream.Output == nil {
		stream.Output = make(chan Record)
	}

	if stream.State == nil {
		stream.State = make(chan WorkerState)
	}

	if stream.Client == nil {
		stream.Client = kinesis.New(stream.AwsConfig)
	}

	Logger.WithFields(log.Fields{
		"StreamWorker": stream,
		"AWS":          stream.AwsConfig,
	}).Debug("StreamWorker initialized")

	return nil
}

func (k *KinesisOutput) Connect() error {
	// We attempt first to create a session to Kinesis using an IAMS role, if that fails it will fall through to using
	// environment variables, and then Shared Credentials.
	if k.Debug {
		log.Printf("kinesis: Establishing a connection to Kinesis in %+v", k.Region)
	}
	Config := &aws.Config{
		Region: aws.String(k.Region),
	}
	svc := kinesis.New(session.New(Config))

	KinesisParams := &kinesis.ListStreamsInput{
		Limit: aws.Int64(100),
	}

	resp, err := svc.ListStreams(KinesisParams)

	if err != nil {
		log.Printf("kinesis: Error in ListSteams API call : %+v \n", err)
	}

	if checkstream(resp.StreamNames, k.StreamName) {
		if k.Debug {
			log.Printf("kinesis: Stream Exists")
		}
		k.svc = svc
		return nil
	} else {
		log.Printf("kinesis : You have configured a StreamName %+v which does not exist. exiting.", k.StreamName)
		os.Exit(1)
	}
	return err
}

// Put a single record.
func (kr KinesisRelay) putRecord(msg string) {
	kr.logger.Info.Println("Attempting to put record into stream")

	svc := kinesis.New(&kr.awsConfig)

	// TODO real partition key.
	params := &kinesis.PutRecordInput{
		StreamName:   aws.String(kr.config.Kinesis.StreamName),
		PartitionKey: aws.String("1"),
		Data:         []byte(msg),
	}

	resp, err := svc.PutRecord(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		kr.logger.Error.Printf("Error w/ PutRecord: %#v", err.Error())
		return
	}

	kr.logger.Trace.Printf("PutRecord response:  %#v", resp)

	kr.logger.Info.Println("Done putting record!")
}

func TestMigrateStreams(t *testing.T) {
	if os.Getenv("AWS_ACCESS_KEY_ID") == "" || os.Getenv("AWS_SECRET_ACCESS_KEY") == "" {
		Log.Printf("WARNING: NO AWS CREDS SPECIFIED, SKIPPING MIGRATIONS TEST")
		return
	}

	config := &aws.Config{
		Credentials: credentials.NewEnvCredentials(),
		Region:      aws.String("us-east-1"),
	}
	client := kinesis.New(session.New(config))

	sm := StreamMigrator{
		Streams: []kinesis.CreateStreamInput{
			kinesis.CreateStreamInput{
				StreamName: aws.String(testDomain),
				ShardCount: aws.Int64(1),
			},
		},
		Client:  client,
		Timeout: 30,
	}

	sm.Migrate()
	sm.Migrate()

}

func ExampleKinesis_RemoveTagsFromStream() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := kinesis.New(sess)

	params := &kinesis.RemoveTagsFromStreamInput{
		StreamName: aws.String("StreamName"), // Required
		TagKeys: []*string{ // Required
			aws.String("TagKey"), // Required
			// More values...
		},
	}
	resp, err := svc.RemoveTagsFromStream(params)

	if err != nil {
		// Print the error, cast err to awserr.Error to get the Code and
		// Message from an error.
		fmt.Println(err.Error())
		return
	}

	// Pretty-print the response data.
	fmt.Println(resp)
}

func ExampleKinesis_GetShardIterator() {
	svc := kinesis.New(nil)

	params := &kinesis.GetShardIteratorInput{
		ShardID:                aws.String("ShardId"),           // Required
		ShardIteratorType:      aws.String("ShardIteratorType"), // Required
		StreamName:             aws.String("StreamName"),        // Required
		StartingSequenceNumber: aws.String("SequenceNumber"),
	}
	resp, err := svc.GetShardIterator(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			// Generic AWS Error with Code, Message, and original error (if any)
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				// A service error occurred
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			// This case should never be hit, the SDK should always return an
			// error which satisfies the awserr.Error interface.
			fmt.Println(err.Error())
		}
	}

	// Pretty-print the response data.
	fmt.Println(awsutil.StringValue(resp))
}

func main() {
	flag.Parse()
	log.SetHandler(text.New(os.Stderr))
	log.SetLevel(log.DebugLevel)

	// set up producer
	svc := kinesis.New(session.New())
	p := producer.New(producer.Config{
		StreamName:  *stream,
		BacklogSize: 500,
		Client:      svc,
	})
	p.Start()

	// open data file
	f, err := os.Open("/tmp/users.txt")
	if err != nil {
		log.Fatal("Cannot open users.txt file")
	}
	defer f.Close()

	// loop over file data
	b := bufio.NewScanner(f)
	for b.Scan() {
		err := p.Put(b.Bytes(), "site")

		if err != nil {
			log.WithError(err).Fatal("error producing")
		}
	}

	p.Stop()
}

func Kinesis_PutRecord(streamName string, log logrus.Fields) (shardId string, startingSequenceNumber string) {
	svc := kinesis.New(&aws.Config{Region: aws.String(region)})

	value := fromMapToBytes(log)

	params := &kinesis.PutRecordInput{
		Data:         []byte(value),          // Required
		PartitionKey: aws.String("Key"),      // Required
		StreamName:   aws.String(streamName), // Required
		//	ExplicitHashKey:           aws.String("1234"),
		//	SequenceNumberForOrdering: aws.String("123"),
	}
	resp, err := svc.PutRecord(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			fmt.Println(err.Error())
		}
	}
	return *resp.ShardID, *resp.SequenceNumber
}

func ExampleKinesis_PutRecords() {
	svc := kinesis.New(nil)

	params := &kinesis.PutRecordsInput{
		Records: []*kinesis.PutRecordsRequestEntry{ // Required
			{ // Required
				Data:            []byte("PAYLOAD"),          // Required
				PartitionKey:    aws.String("PartitionKey"), // Required
				ExplicitHashKey: aws.String("HashKey"),
			},
			// More values...
		},
		StreamName: aws.String("StreamName"), // Required
	}
	resp, err := svc.PutRecords(params)

	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok {
			// Generic AWS error with Code, Message, and original error (if any)
			fmt.Println(awsErr.Code(), awsErr.Message(), awsErr.OrigErr())
			if reqErr, ok := err.(awserr.RequestFailure); ok {
				// A service error occurred
				fmt.Println(reqErr.Code(), reqErr.Message(), reqErr.StatusCode(), reqErr.RequestID())
			}
		} else {
			// This case should never be hit, the SDK should always return an
			// error which satisfies the awserr.Error interface.
			fmt.Println(err.Error())
		}
	}

	// Pretty-print the response data.
	fmt.Println(awsutil.Prettify(resp))
}