func main() {
	flag.Parse()
	svc := firehose.New(session.New())

	cfg := connector.Config{
		MaxRecordCount: 400,
	}

	c := connector.NewConsumer(*app, *stream, cfg)

	c.Start(connector.HandlerFunc(func(b connector.Buffer) {
		params := &firehose.PutRecordBatchInput{
			DeliveryStreamName: aws.String(*delivery),
			Records:            convertToFirehoseRecrods(b.GetRecords()),
		}

		_, err := svc.PutRecordBatch(params)

		if err != nil {
			fmt.Println(err.Error())
			os.Exit(1)
		}

		fmt.Println("Put records to Firehose")
	}))

	select {} // run forever
}

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

	svc := firehose.New(sess)

	params := &firehose.PutRecordBatchInput{
		DeliveryStreamName: aws.String("DeliveryStreamName"), // Required
		Records: []*firehose.Record{ // Required
			{ // Required
				Data: []byte("PAYLOAD"), // Required
			},
			// More values...
		},
	}
	resp, err := svc.PutRecordBatch(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 ExampleFirehose_ListDeliveryStreams() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := firehose.New(sess)

	params := &firehose.ListDeliveryStreamsInput{
		ExclusiveStartDeliveryStreamName: aws.String("DeliveryStreamName"),
		Limit: aws.Int64(1),
	}
	resp, err := svc.ListDeliveryStreams(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 newFirehoseService(filename, profile, region string) *firehose.Firehose {
	var creds *credentials.Credentials
	if _, err := os.Stat(filename); err == nil {
		log.Printf("Connecting to AWS using credentials from '%s'", filename)
		creds = credentials.NewSharedCredentials(filename, profile)
	} else {
		log.Printf("AWS credentials file '%s' dosen't exists, I will be using EC2 Role credentials", filename)
		sess := session.New()
		creds = credentials.NewCredentials(&ec2rolecreds.EC2RoleProvider{Client: ec2metadata.New(sess)})
	}
	sess := session.New(&aws.Config{Credentials: creds, Region: aws.String(region)})
	return firehose.New(sess)
}

func ExampleFirehose_DeleteDeliveryStream() {
	svc := firehose.New(session.New())

	params := &firehose.DeleteDeliveryStreamInput{
		DeliveryStreamName: aws.String("DeliveryStreamName"), // Required
	}
	resp, err := svc.DeleteDeliveryStream(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 ExampleFirehose_DescribeDeliveryStream() {
	svc := firehose.New(nil)

	params := &firehose.DescribeDeliveryStreamInput{
		DeliveryStreamName:          aws.String("DeliveryStreamName"), // Required
		ExclusiveStartDestinationId: aws.String("DestinationId"),
		Limit: aws.Int64(1),
	}
	resp, err := svc.DescribeDeliveryStream(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 ExampleFirehose_PutRecord() {
	svc := firehose.New(session.New())

	params := &firehose.PutRecordInput{
		DeliveryStreamName: aws.String("DeliveryStreamName"), // Required
		Record: &firehose.Record{ // Required
			Data: []byte("PAYLOAD"), // Required
		},
	}
	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)
}

func main() {
	var options = Options{
		Address: ":3000",
		Backlog: 100,
	}

	config.MustResolve(&options)

	if options.Version {
		fmt.Printf("%s\n", version)
		os.Exit(0)
	}

	c := &client.Client{
		Firehose:   firehose.New(session.New()),
		StreamName: options.StreamName,
		Backlog:    make(chan []byte, options.Backlog),
	}

	s := &server.Server{
		Client: c,
	}

	c.Start()

	var h http.Handler = s

	if options.Username != "" && options.Password != "" {
		h = basicauth.BasicAuth{
			Username: options.Username,
			Password: options.Password,
			Handler:  h,
		}
	}

	h = handlers.LoggingHandler(os.Stderr, h)

	log.Fatalln(http.ListenAndServe(options.Address, h))
}

// Client configures and returns a fully initialized AWSClient
func (c *Config) Client() (interface{}, error) {
	// Get the auth and region. This can fail if keys/regions were not
	// specified and we're attempting to use the environment.
	var errs []error

	log.Println("[INFO] Building AWS region structure")
	err := c.ValidateRegion()
	if err != nil {
		errs = append(errs, err)
	}

	var client AWSClient
	if len(errs) == 0 {
		// store AWS region in client struct, for region specific operations such as
		// bucket storage in S3
		client.region = c.Region

		log.Println("[INFO] Building AWS auth structure")
		creds := GetCredentials(c.AccessKey, c.SecretKey, c.Token, c.Profile, c.CredsFilename)
		// Call Get to check for credential provider. If nothing found, we'll get an
		// error, and we can present it nicely to the user
		cp, err := creds.Get()
		if err != nil {
			if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == "NoCredentialProviders" {
				errs = append(errs, fmt.Errorf(`No valid credential sources found for AWS Provider.
  Please see https://terraform.io/docs/providers/aws/index.html for more information on
  providing credentials for the AWS Provider`))
			} else {
				errs = append(errs, fmt.Errorf("Error loading credentials for AWS Provider: %s", err))
			}
			return nil, &multierror.Error{Errors: errs}
		}

		log.Printf("[INFO] AWS Auth provider used: %q", cp.ProviderName)

		awsConfig := &aws.Config{
			Credentials: creds,
			Region:      aws.String(c.Region),
			MaxRetries:  aws.Int(c.MaxRetries),
			HTTPClient:  cleanhttp.DefaultClient(),
		}

		if logging.IsDebugOrHigher() {
			awsConfig.LogLevel = aws.LogLevel(aws.LogDebugWithHTTPBody)
			awsConfig.Logger = awsLogger{}
		}

		if c.Insecure {
			transport := awsConfig.HTTPClient.Transport.(*http.Transport)
			transport.TLSClientConfig = &tls.Config{
				InsecureSkipVerify: true,
			}
		}

		// Set up base session
		sess := session.New(awsConfig)
		sess.Handlers.Build.PushFrontNamed(addTerraformVersionToUserAgent)

		// Some services exist only in us-east-1, e.g. because they manage
		// resources that can span across multiple regions, or because
		// signature format v4 requires region to be us-east-1 for global
		// endpoints:
		// http://docs.aws.amazon.com/general/latest/gr/sigv4_changes.html
		usEast1Sess := sess.Copy(&aws.Config{Region: aws.String("us-east-1")})

		// Some services have user-configurable endpoints
		awsEc2Sess := sess.Copy(&aws.Config{Endpoint: aws.String(c.Ec2Endpoint)})
		awsElbSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.ElbEndpoint)})
		awsIamSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.IamEndpoint)})
		dynamoSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.DynamoDBEndpoint)})
		kinesisSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.KinesisEndpoint)})

		// These two services need to be set up early so we can check on AccountID
		client.iamconn = iam.New(awsIamSess)
		client.stsconn = sts.New(sess)

		err = c.ValidateCredentials(client.stsconn)
		if err != nil {
			errs = append(errs, err)
			return nil, &multierror.Error{Errors: errs}
		}
		accountId, err := GetAccountId(client.iamconn, client.stsconn, cp.ProviderName)
		if err == nil {
			client.accountid = accountId
		}

		authErr := c.ValidateAccountId(client.accountid)
		if authErr != nil {
			errs = append(errs, authErr)
		}

		client.apigateway = apigateway.New(sess)
		client.autoscalingconn = autoscaling.New(sess)
		client.cfconn = cloudformation.New(sess)
		client.cloudfrontconn = cloudfront.New(sess)
		client.cloudtrailconn = cloudtrail.New(sess)
		client.cloudwatchconn = cloudwatch.New(sess)
		client.cloudwatcheventsconn = cloudwatchevents.New(sess)
		client.cloudwatchlogsconn = cloudwatchlogs.New(sess)
//.........这里部分代码省略.........

// Client configures and returns a fully initialized AWSClient
func (c *Config) Client() (interface{}, error) {
	var client AWSClient

	// Get the auth and region. This can fail if keys/regions were not
	// specified and we're attempting to use the environment.
	var errs []error

	log.Println("[INFO] Building AWS region structure")
	err := c.ValidateRegion()
	if err != nil {
		errs = append(errs, err)
	}

	if len(errs) == 0 {
		// store AWS region in client struct, for region specific operations such as
		// bucket storage in S3
		client.region = c.Region

		log.Println("[INFO] Building AWS auth structure")
		creds := getCreds(c.AccessKey, c.SecretKey, c.Token)
		// Call Get to check for credential provider. If nothing found, we'll get an
		// error, and we can present it nicely to the user
		_, err = creds.Get()
		if err != nil {
			errs = append(errs, fmt.Errorf("Error loading credentials for AWS Provider: %s", err))
			return nil, &multierror.Error{Errors: errs}
		}
		awsConfig := &aws.Config{
			Credentials: creds,
			Region:      aws.String(c.Region),
			MaxRetries:  aws.Int(c.MaxRetries),
			HTTPClient:  cleanhttp.DefaultClient(),
		}

		log.Println("[INFO] Initializing IAM Connection")
		sess := session.New(awsConfig)
		client.iamconn = iam.New(sess)

		err = c.ValidateCredentials(client.iamconn)
		if err != nil {
			errs = append(errs, err)
		}

		// Some services exist only in us-east-1, e.g. because they manage
		// resources that can span across multiple regions, or because
		// signature format v4 requires region to be us-east-1 for global
		// endpoints:
		// http://docs.aws.amazon.com/general/latest/gr/sigv4_changes.html
		usEast1AwsConfig := &aws.Config{
			Credentials: creds,
			Region:      aws.String("us-east-1"),
			MaxRetries:  aws.Int(c.MaxRetries),
			HTTPClient:  cleanhttp.DefaultClient(),
		}
		usEast1Sess := session.New(usEast1AwsConfig)

		awsDynamoDBConfig := *awsConfig
		awsDynamoDBConfig.Endpoint = aws.String(c.DynamoDBEndpoint)

		log.Println("[INFO] Initializing DynamoDB connection")
		dynamoSess := session.New(&awsDynamoDBConfig)
		client.dynamodbconn = dynamodb.New(dynamoSess)

		log.Println("[INFO] Initializing ELB connection")
		client.elbconn = elb.New(sess)

		log.Println("[INFO] Initializing S3 connection")
		client.s3conn = s3.New(sess)

		log.Println("[INFO] Initializing SQS connection")
		client.sqsconn = sqs.New(sess)

		log.Println("[INFO] Initializing SNS connection")
		client.snsconn = sns.New(sess)

		log.Println("[INFO] Initializing RDS Connection")
		client.rdsconn = rds.New(sess)

		awsKinesisConfig := *awsConfig
		awsKinesisConfig.Endpoint = aws.String(c.KinesisEndpoint)

		log.Println("[INFO] Initializing Kinesis Connection")
		kinesisSess := session.New(&awsKinesisConfig)
		client.kinesisconn = kinesis.New(kinesisSess)

		authErr := c.ValidateAccountId(client.iamconn)
		if authErr != nil {
			errs = append(errs, authErr)
		}

		log.Println("[INFO] Initializing Kinesis Firehose Connection")
		client.firehoseconn = firehose.New(sess)

		log.Println("[INFO] Initializing AutoScaling connection")
		client.autoscalingconn = autoscaling.New(sess)

		log.Println("[INFO] Initializing EC2 Connection")
		client.ec2conn = ec2.New(sess)

//.........这里部分代码省略.........

// Client configures and returns a fully initialized AWSClient
func (c *Config) Client() (interface{}, error) {
	// Get the auth and region. This can fail if keys/regions were not
	// specified and we're attempting to use the environment.
	log.Println("[INFO] Building AWS region structure")
	err := c.ValidateRegion()
	if err != nil {
		return nil, err
	}

	var client AWSClient
	// store AWS region in client struct, for region specific operations such as
	// bucket storage in S3
	client.region = c.Region

	log.Println("[INFO] Building AWS auth structure")
	creds, err := GetCredentials(c)
	if err != nil {
		return nil, err
	}
	// Call Get to check for credential provider. If nothing found, we'll get an
	// error, and we can present it nicely to the user
	cp, err := creds.Get()
	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == "NoCredentialProviders" {
			return nil, errors.New(`No valid credential sources found for AWS Provider.
  Please see https://terraform.io/docs/providers/aws/index.html for more information on
  providing credentials for the AWS Provider`)
		}

		return nil, fmt.Errorf("Error loading credentials for AWS Provider: %s", err)
	}

	log.Printf("[INFO] AWS Auth provider used: %q", cp.ProviderName)

	awsConfig := &aws.Config{
		Credentials:      creds,
		Region:           aws.String(c.Region),
		MaxRetries:       aws.Int(c.MaxRetries),
		HTTPClient:       cleanhttp.DefaultClient(),
		S3ForcePathStyle: aws.Bool(c.S3ForcePathStyle),
	}

	if logging.IsDebugOrHigher() {
		awsConfig.LogLevel = aws.LogLevel(aws.LogDebugWithHTTPBody)
		awsConfig.Logger = awsLogger{}
	}

	if c.Insecure {
		transport := awsConfig.HTTPClient.Transport.(*http.Transport)
		transport.TLSClientConfig = &tls.Config{
			InsecureSkipVerify: true,
		}
	}

	// Set up base session
	sess, err := session.NewSession(awsConfig)
	if err != nil {
		return nil, errwrap.Wrapf("Error creating AWS session: {{err}}", err)
	}

	// Removes the SDK Version handler, so we only have the provider User-Agent
	// Ex: "User-Agent: APN/1.0 HashiCorp/1.0 Terraform/0.7.9-dev"
	sess.Handlers.Build.Remove(request.NamedHandler{Name: "core.SDKVersionUserAgentHandler"})
	sess.Handlers.Build.PushFrontNamed(addTerraformVersionToUserAgent)

	if extraDebug := os.Getenv("TERRAFORM_AWS_AUTHFAILURE_DEBUG"); extraDebug != "" {
		sess.Handlers.UnmarshalError.PushFrontNamed(debugAuthFailure)
	}

	// Some services exist only in us-east-1, e.g. because they manage
	// resources that can span across multiple regions, or because
	// signature format v4 requires region to be us-east-1 for global
	// endpoints:
	// http://docs.aws.amazon.com/general/latest/gr/sigv4_changes.html
	usEast1Sess := sess.Copy(&aws.Config{Region: aws.String("us-east-1")})

	// Some services have user-configurable endpoints
	awsEc2Sess := sess.Copy(&aws.Config{Endpoint: aws.String(c.Ec2Endpoint)})
	awsElbSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.ElbEndpoint)})
	awsIamSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.IamEndpoint)})
	awsS3Sess := sess.Copy(&aws.Config{Endpoint: aws.String(c.S3Endpoint)})
	dynamoSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.DynamoDBEndpoint)})
	kinesisSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.KinesisEndpoint)})

	// These two services need to be set up early so we can check on AccountID
	client.iamconn = iam.New(awsIamSess)
	client.stsconn = sts.New(sess)

	if !c.SkipCredsValidation {
		err = c.ValidateCredentials(client.stsconn)
		if err != nil {
			return nil, err
		}
	}

	if !c.SkipRequestingAccountId {
		partition, accountId, err := GetAccountInfo(client.iamconn, client.stsconn, cp.ProviderName)
		if err == nil {
			client.partition = partition
//.........这里部分代码省略.........

func ExampleFirehose_UpdateDestination() {
	svc := firehose.New(session.New())

	params := &firehose.UpdateDestinationInput{
		CurrentDeliveryStreamVersionId: aws.String("DeliveryStreamVersionId"), // Required
		DeliveryStreamName:             aws.String("DeliveryStreamName"),      // Required
		DestinationId:                  aws.String("DestinationId"),           // Required
		ElasticsearchDestinationUpdate: &firehose.ElasticsearchDestinationUpdate{
			BufferingHints: &firehose.ElasticsearchBufferingHints{
				IntervalInSeconds: aws.Int64(1),
				SizeInMBs:         aws.Int64(1),
			},
			CloudWatchLoggingOptions: &firehose.CloudWatchLoggingOptions{
				Enabled:       aws.Bool(true),
				LogGroupName:  aws.String("LogGroupName"),
				LogStreamName: aws.String("LogStreamName"),
			},
			DomainARN:           aws.String("ElasticsearchDomainARN"),
			IndexName:           aws.String("ElasticsearchIndexName"),
			IndexRotationPeriod: aws.String("ElasticsearchIndexRotationPeriod"),
			RetryOptions: &firehose.ElasticsearchRetryOptions{
				DurationInSeconds: aws.Int64(1),
			},
			RoleARN: aws.String("RoleARN"),
			S3Update: &firehose.S3DestinationUpdate{
				BucketARN: aws.String("BucketARN"),
				BufferingHints: &firehose.BufferingHints{
					IntervalInSeconds: aws.Int64(1),
					SizeInMBs:         aws.Int64(1),
				},
				CloudWatchLoggingOptions: &firehose.CloudWatchLoggingOptions{
					Enabled:       aws.Bool(true),
					LogGroupName:  aws.String("LogGroupName"),
					LogStreamName: aws.String("LogStreamName"),
				},
				CompressionFormat: aws.String("CompressionFormat"),
				EncryptionConfiguration: &firehose.EncryptionConfiguration{
					KMSEncryptionConfig: &firehose.KMSEncryptionConfig{
						AWSKMSKeyARN: aws.String("AWSKMSKeyARN"), // Required
					},
					NoEncryptionConfig: aws.String("NoEncryptionConfig"),
				},
				Prefix:  aws.String("Prefix"),
				RoleARN: aws.String("RoleARN"),
			},
			TypeName: aws.String("ElasticsearchTypeName"),
		},
		RedshiftDestinationUpdate: &firehose.RedshiftDestinationUpdate{
			CloudWatchLoggingOptions: &firehose.CloudWatchLoggingOptions{
				Enabled:       aws.Bool(true),
				LogGroupName:  aws.String("LogGroupName"),
				LogStreamName: aws.String("LogStreamName"),
			},
			ClusterJDBCURL: aws.String("ClusterJDBCURL"),
			CopyCommand: &firehose.CopyCommand{
				DataTableName:    aws.String("DataTableName"), // Required
				CopyOptions:      aws.String("CopyOptions"),
				DataTableColumns: aws.String("DataTableColumns"),
			},
			Password: aws.String("Password"),
			RoleARN:  aws.String("RoleARN"),
			S3Update: &firehose.S3DestinationUpdate{
				BucketARN: aws.String("BucketARN"),
				BufferingHints: &firehose.BufferingHints{
					IntervalInSeconds: aws.Int64(1),
					SizeInMBs:         aws.Int64(1),
				},
				CloudWatchLoggingOptions: &firehose.CloudWatchLoggingOptions{
					Enabled:       aws.Bool(true),
					LogGroupName:  aws.String("LogGroupName"),
					LogStreamName: aws.String("LogStreamName"),
				},
				CompressionFormat: aws.String("CompressionFormat"),
				EncryptionConfiguration: &firehose.EncryptionConfiguration{
					KMSEncryptionConfig: &firehose.KMSEncryptionConfig{
						AWSKMSKeyARN: aws.String("AWSKMSKeyARN"), // Required
					},
					NoEncryptionConfig: aws.String("NoEncryptionConfig"),
				},
				Prefix:  aws.String("Prefix"),
				RoleARN: aws.String("RoleARN"),
			},
			Username: aws.String("Username"),
		},
		S3DestinationUpdate: &firehose.S3DestinationUpdate{
			BucketARN: aws.String("BucketARN"),
			BufferingHints: &firehose.BufferingHints{
				IntervalInSeconds: aws.Int64(1),
				SizeInMBs:         aws.Int64(1),
			},
			CloudWatchLoggingOptions: &firehose.CloudWatchLoggingOptions{
				Enabled:       aws.Bool(true),
				LogGroupName:  aws.String("LogGroupName"),
				LogStreamName: aws.String("LogStreamName"),
			},
			CompressionFormat: aws.String("CompressionFormat"),
			EncryptionConfiguration: &firehose.EncryptionConfiguration{
				KMSEncryptionConfig: &firehose.KMSEncryptionConfig{
					AWSKMSKeyARN: aws.String("AWSKMSKeyARN"), // Required
				},
//.........这里部分代码省略.........

func createFirehoseService(profile, region string) *firehose.Firehose {
	creds := credentials.NewSharedCredentials("", profile)
	return firehose.New(&aws.Config{Region: aws.String(region), Credentials: creds})
}

func ExampleFirehose_CreateDeliveryStream() {
	svc := firehose.New(session.New())

	params := &firehose.CreateDeliveryStreamInput{
		DeliveryStreamName: aws.String("DeliveryStreamName"), // Required
		RedshiftDestinationConfiguration: &firehose.RedshiftDestinationConfiguration{
			ClusterJDBCURL: aws.String("ClusterJDBCURL"), // Required
			CopyCommand: &firehose.CopyCommand{ // Required
				DataTableName:    aws.String("DataTableName"), // Required
				CopyOptions:      aws.String("CopyOptions"),
				DataTableColumns: aws.String("DataTableColumns"),
			},
			Password: aws.String("Password"), // Required
			RoleARN:  aws.String("RoleARN"),  // Required
			S3Configuration: &firehose.S3DestinationConfiguration{ // Required
				BucketARN: aws.String("BucketARN"), // Required
				RoleARN:   aws.String("RoleARN"),   // Required
				BufferingHints: &firehose.BufferingHints{
					IntervalInSeconds: aws.Int64(1),
					SizeInMBs:         aws.Int64(1),
				},
				CompressionFormat: aws.String("CompressionFormat"),
				EncryptionConfiguration: &firehose.EncryptionConfiguration{
					KMSEncryptionConfig: &firehose.KMSEncryptionConfig{
						AWSKMSKeyARN: aws.String("AWSKMSKeyARN"), // Required
					},
					NoEncryptionConfig: aws.String("NoEncryptionConfig"),
				},
				Prefix: aws.String("Prefix"),
			},
			Username: aws.String("Username"), // Required
		},
		S3DestinationConfiguration: &firehose.S3DestinationConfiguration{
			BucketARN: aws.String("BucketARN"), // Required
			RoleARN:   aws.String("RoleARN"),   // Required
			BufferingHints: &firehose.BufferingHints{
				IntervalInSeconds: aws.Int64(1),
				SizeInMBs:         aws.Int64(1),
			},
			CompressionFormat: aws.String("CompressionFormat"),
			EncryptionConfiguration: &firehose.EncryptionConfiguration{
				KMSEncryptionConfig: &firehose.KMSEncryptionConfig{
					AWSKMSKeyARN: aws.String("AWSKMSKeyARN"), // Required
				},
				NoEncryptionConfig: aws.String("NoEncryptionConfig"),
			},
			Prefix: aws.String("Prefix"),
		},
	}
	resp, err := svc.CreateDeliveryStream(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 main() {
	// load the configuration
	file, _ := os.Open("config.json")
	decoder := json.NewDecoder(file)
	configuration := Configuration{}
	err := decoder.Decode(&configuration)
	if err != nil {
		fmt.Println("Error reading config.json:", err)
	}

	// declare our outgoing queue
	outgoing := fifo.NewQueue()

	// initialize aws sdk and credentials from configuration
	svc := firehose.New(session.New(), &aws.Config{Region: aws.String(configuration.Aws_region),
		Credentials: credentials.NewStaticCredentials(configuration.Aws_access_key, configuration.Aws_secret_key, "")})

	_, err = svc.ListDeliveryStreams(&firehose.ListDeliveryStreamsInput{})
	if err != nil {
		fmt.Println("Error listing delivery streams:", err)
		return
	}

	// start Filewatchers for the files
	for _, comp := range configuration.Components {
		fmt.Println("Watching", comp.Name, "at", comp.File)
		go Filewatcher(comp, configuration, outgoing)
	}

	// loop
	for {
		outlen := outgoing.Len()
		if outlen > 0 {
			// create a request body
			var data = bytes.NewBuffer([]byte(``))
			encoder := json.NewEncoder(data)

			// add queue contents as payload, not exceeding Firehose_record_size
			for {
				// get the next item in the queue
				item := outgoing.Peek()
				if item == nil {
					// no more items in queue
					break
				}

				json_value, err := json.Marshal(item)
				if err != nil {
					fmt.Println("Encoding error:", err)
					break
				}

				// AWS currently limits Kinesis Firehose records to 1024KB
				if (bytes.NewBuffer(json_value).Len() + data.Len()) > 1024000 {
					// record size would be exceeded
					break
				}

				// encode the item to payload buffer, while removing it from the queue
				error := encoder.Encode(outgoing.Next())

				if error != nil {
					fmt.Println("Encoding error:", error)
				}

			}

			// create the Firehose record
			params := &firehose.PutRecordInput{
				DeliveryStreamName: aws.String(configuration.Firehose_stream_name),
				Record: &firehose.Record{
					Data: data.Bytes(),
				},
			}

			// send the record to Firehose
			resp, err := svc.PutRecord(params)

			if err != nil {
				fmt.Println(err.Error())
				return
			} else {
				fmt.Println(resp)
			}
		}
		// sleep Shipping_interval before the next iteration
		time.Sleep(time.Duration(configuration.Shipping_interval) * time.Second)
	}
}

func TestInterface(t *testing.T) {
	assert.Implements(t, (*firehoseiface.FirehoseAPI)(nil), firehose.New(nil))
}