func CreateDynamoDBClient() *dynamodb.DynamoDB {
	var dynamoClient *dynamodb.DynamoDB

	localAddr := os.Getenv("LOCAL_DYNAMO_ADDR") //e.g. export LOCAL_DYNAMO_ADDR=http://localhost:8000
	if localAddr != "" {
		log.Printf("Using local dynamodb address - %s", localAddr)
		dynamoClient = dynamodb.New(&aws.Config{Endpoint: aws.String(localAddr), Region: aws.String("here")})
	} else {
		dynamoClient = dynamodb.New(&aws.Config{Region: aws.String("us-east-1")})
	}

	return dynamoClient
}

// Worker service which reads from an SQS queue pulls off job messages, processes
// the jobs, and records the results. The service uses environment variables for
// its configuration.
//
// Requires the following environment variables to be set.
//
// * WORKER_QUEUE_URL - The SQS queue URL where the service will read job messages
// from. Job messages are created when S3 notifies the SQS queue that a file has
// been uploaded to a particular bucket.
//
// * WORKER_RESULT_QUEUE_URL - The SQS queue URL where the job results will be
// sent to.
//
// * WORKER_RESULT_TABLENAME - The name of the DynamoDB table result items should
// be recorded to.
//
// Optionally the follow environment variables can be provided.
//
// * AWS_REGION - The AWS region the worker will use for signing and making all
// requests to. This parameter is only optional if the service is running within
// an EC2 instance. If not running in an EC2 instance AWS_REGION is required.
//
// * WORKER_MESSAGE_VISIBILITY - The ammount of time messges will be hidden in
// the SQS job message queue from other services when a service reads that message.
// Will also be used to extend the visibility timeout for long running jobs.
// Defaults to 60s.
//
// * WORKER_COUNT - The number of workers in the worker pool. Defaults to the
// number of virtual CPUs in the system.
//
func main() {
	doneCh := listenForSigInterrupt()

	cfg, err := getConfig()
	if err != nil {
		log.Println("Unable to get config", err)
		os.Exit(1)
	}

	sqsSvc := sqs.New(nil)
	queue := NewJobMessageQueue(cfg.WorkerQueueURL, cfg.MessageVisibilityTimeout, 5, sqsSvc)
	go queue.Listen(doneCh)

	// Job Workers
	resultsCh := make(chan *wordfreq.JobResult, 10)
	workers := NewWorkerPool(cfg.NumWorkers, resultsCh, queue, s3.New(nil))

	// Notifier to notify a Amazon SNS Topic
	notify := NewResultNotifier(sqsSvc, cfg.ResultQueueURL)
	// Recorder to write results to Amazon DynamoDB
	recorder := NewResultRecorder(cfg.ResultTableName, dynamodb.New(nil))

	// Job Progress Collector
	collector := NewResultCollector(notify, recorder, queue)
	go collector.ProcessJobResult(resultsCh)

	// Wait for the workers to complete before continuing on to exit
	workers.WaitForWorkersDone()
	close(resultsCh)

	// Wait for all results to be completed before continuing
	collector.WaitForResults()
}

// NewDynamoDBClient returns an *dynamodb.Client with a connection to the region
// configured via the AWS_REGION environment variable.
// It returns an error if the connection cannot be made or the table does not exist.
func NewDynamoDBClient(table string) (*Client, error) {
	var c *aws.Config
	if os.Getenv("DYNAMODB_LOCAL") != "" {
		log.Debug("DYNAMODB_LOCAL is set")
		endpoint := "http://localhost:8000"
		c = &aws.Config{
			Endpoint: &endpoint,
		}
	} else {
		c = nil
	}

	session := session.New(c)

	// Fail early, if no credentials can be found
	_, err := session.Config.Credentials.Get()
	if err != nil {
		return nil, err
	}

	d := dynamodb.New(session)

	// Check if the table exists
	_, err = d.DescribeTable(&dynamodb.DescribeTableInput{TableName: &table})
	if err != nil {
		return nil, err
	}
	return &Client{d, table}, nil
}

// here, support a function that maps a given redis key to a dynamodb table key and value field
func NewDynamoModule(keymap KeyMapper) *DynamoModule {

	module := &DynamoModule{}
	cfgdir := "/etc"
	cfg := &module.config
	ok := logging.ReadModuleConfig(cfg, cfgdir, "dynamo") || logging.ReadModuleConfig(cfg, ".", "dynamo")

	sess := session.New(&aws.Config{Region: aws.String("ap-southeast-1")})

	if !ok {
		log.Println("failed to read dynamo config, using defaults")
	} else {
		sess = session.New(&aws.Config{
			Region:     aws.String(cfg.Server.Region),
			Endpoint:   aws.String(cfg.Server.Endpoint),
			DisableSSL: aws.Bool(cfg.Server.DisableSSL),
		})
	}

	module.client = dynamodb.New(sess)
	if keymap != nil {
		module.keyMapper = keymap
	} else {
		module.keyMapper = module.defaultMapper
	}

	if cfg.Server.CacheDuration > 0 {
		logging.Debug.Println("activiating cache, TTL", cfg.Server.CacheDuration)
		module.cache = cache.NewMemoryWithTTL(time.Duration(cfg.Server.CacheDuration) * time.Second)
	}

	return module
}

func CreateDynamoDbBackend(proxy_name string, tablename string, awsConfig *aws.Config) *DynamoDbBackend {
	return &DynamoDbBackend{
		proxy_name: proxy_name,
		tablename:  tablename,
		database:   dynamodb.New(session.New(), awsConfig),
	}
}

func main() {
	svc := dynamodb.New(session.New(), &aws.Config{Region: aws.String("eu-central-1")})

	k := Key{
		Email: "[email protected]",
	}

	item, err := dynamodbattribute.ConvertToMap(k)
	if err != nil {
		panic(err)
	}

	result, err := svc.GetItem(&dynamodb.GetItemInput{
		TableName: aws.String("Users"),
		Key:       item,
	})
	if err != nil {
		panic(err)
	}

	r := Record{}
	err = dynamodbattribute.ConvertFromMap(result.Item, &r)

	fmt.Println(r)
}

func main() {
	svc := dynamodb.New(session.New(), &aws.Config{Region: aws.String("eu-central-1")})

	params := &dynamodb.CreateTableInput{
		TableName: aws.String("ProductCatalog"),
		AttributeDefinitions: []*dynamodb.AttributeDefinition{
			{
				AttributeName: aws.String("Id"),
				AttributeType: aws.String("N"),
			},
		},
		KeySchema: []*dynamodb.KeySchemaElement{
			{
				AttributeName: aws.String("Id"),
				KeyType:       aws.String("HASH"),
			},
		},
		ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
			ReadCapacityUnits:  aws.Int64(5),
			WriteCapacityUnits: aws.Int64(5),
		},
	}
	resp, err := svc.CreateTable(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 ExampleDynamoDB_ListTables() {
	sess, err := session.NewSession()
	if err != nil {
		fmt.Println("failed to create session,", err)
		return
	}

	svc := dynamodb.New(sess)

	params := &dynamodb.ListTablesInput{
		ExclusiveStartTableName: aws.String("TableName"),
		Limit: aws.Int64(1),
	}
	resp, err := svc.ListTables(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)
}

// Svc configures the DynamoDB service to use
func Svc(opts config.Options) *dynamodb.DynamoDB {
	awsConfig := &aws.Config{Region: aws.String(opts.Storage.AWS.Region)}

	// If a session was passed... (AWS Lambda does this)
	if opts.Storage.AWS.SessionToken != "" {
		os.Setenv("AWS_SESSION_TOKEN", opts.Storage.AWS.SessionToken)
	}

	// Look in a variety of places for AWS credentials. First, try the credentials file set by AWS CLI tool.
	// Note the empty string instructs to look under default file path (different based on OS).
	// This file can have multiple profiles and a default profile will be used unless otherwise configured.
	// See: https://godoc.org/github.com/aws/aws-sdk-go/aws/credentials#SharedCredentialsProvider
	creds := credentials.NewSharedCredentials("", opts.Storage.AWS.CredProfile)
	_, err := creds.Get()
	// If that failed, try environment variables.
	if err != nil {
		// The following are checked:
		// Access Key ID: AWS_ACCESS_KEY_ID or AWS_ACCESS_KEY
		// Secret Access Key: AWS_SECRET_ACCESS_KEY or AWS_SECRET_KEY
		creds = credentials.NewEnvCredentials()
	}

	// If credentials were passed via config, then use those. They will take priority over other methods.
	if opts.Storage.AWS.AccessKeyID != "" && opts.Storage.AWS.SecretAccessKey != "" {
		creds = credentials.NewStaticCredentials(opts.Storage.AWS.AccessKeyID, opts.Storage.AWS.SecretAccessKey, "")
	}
	awsConfig.Credentials = creds

	return dynamodb.New(session.New(awsConfig))
}

//New creates a new dynamodb connection
func New(publicKey, secretKey string) *DB {
	creds := credentials.NewStaticCredentials(publicKey, secretKey, "")

	return &DB{
		SVC: dynamodb.New(session.New(), aws.NewConfig().WithCredentials(creds).WithRegion("us-east-1")),
	}
}

func Main() {
	var (
		SQS                        *sqs.SQS
		getUserQueueUrlOutput      *sqs.GetQueueUrlOutput
		getContainerQueueUrlOutput *sqs.GetQueueUrlOutput
		UserQueueUrl               *string
		ContainerQueueUrl          *string

		Dynamo *dynamodb.DynamoDB

		socialWorker *workers.SocialWorker
	)

	SQS = sqs.New(&aws.Config{Region: aws.String("cn-north-1")})
	getUserQueueUrlOutput, err := SQS.GetQueueUrl(&sqs.GetQueueUrlInput{QueueName: aws.String(USER_QUEUE_NAME)})
	if err != nil {
		glog.Errorln("Error on connect user queue url:", err.Error())
		return
	}
	UserQueueUrl = getUserQueueUrlOutput.QueueUrl
	getContainerQueueUrlOutput, err = SQS.GetQueueUrl(&sqs.GetQueueUrlInput{QueueName: aws.String(CONTAINER_QUEUE_NAME)})
	if err != nil {
		glog.Errorln("Error on connect container queue url:", err.Error())
		return
	}
	ContainerQueueUrl = getContainerQueueUrlOutput.QueueUrl

	Dynamo = dynamodb.New(&aws.Config{Region: aws.String("cn-north-1")})

	socialWorker = workers.NewSocialWorker(SQS, UserQueueUrl, ContainerQueueUrl, Dynamo)
	socialWorker.Start()
}

func main() {
	awsSession := session.New()
	awsSession.Config.WithRegion(os.Getenv("AWS_REGION"))

	tree = &dynamotree.Tree{
		TableName: "hstore-example-shortlinks",
		DB:        dynamodb.New(awsSession),
	}
	err := tree.CreateTable()
	if err != nil {
		log.Fatalf("hstore: %s", err)
	}

	goji.Get("/:link", ServeLink)
	goji.Post("/signup", CreateAccount)

	authMux := web.New()
	authMux.Use(RequireAccount)
	authMux.Post("/", CreateLink)
	authMux.Get("/", ListLinks)
	authMux.Delete("/:link", DeleteLink) // TODO(ross): this doesn't work (!)
	goji.Handle("/", authMux)

	goji.Serve()
}

func (suite *StoreImplTest) TestReservedCharacters(c *C) {
	tableName := uniuri.New()
	db := dynamodb.New(session.New(), fakeDynamodbServer.Config)
	s := &Tree{
		TableName:        tableName,
		DB:               db,
		SpecialCharacter: "X",
	}
	err := s.CreateTable()
	c.Assert(err, IsNil)

	v := AccountT{
		ID:    "12345",
		Name:  "alice",
		Email: "[email protected]",
	}
	err = s.Put([]string{"Accounts", "12X345"}, &v)
	c.Assert(err, Equals, ErrReservedCharacterInKey)

	v.Xfoo = "cannot be set"
	err = s.Put([]string{"Accounts", "12345"}, &v)
	c.Assert(err, Equals, ErrReservedCharacterInAttribute)

	v.Xfoo = ""
	v.FooXBar = "can be set"
	err = s.Put([]string{"Accounts", "12345"}, &v)
	c.Assert(err, IsNil)

	err = s.PutLink([]string{"AccountsXEmail", "[email protected]"}, []string{"Accounts", "12345"})
	c.Assert(err, Equals, ErrReservedCharacterInKey)

	err = s.PutLink([]string{"AccountsByEmail", "[email protected]"}, []string{"AccountsX", "12345"})
	c.Assert(err, Equals, ErrReservedCharacterInKey)
}

func (suite *StoreImplTest) TestLinkFailures(c *C) {
	tableName := uniuri.New()
	db := dynamodb.New(session.New(), fakeDynamodbServer.Config)
	s := &Tree{TableName: tableName, DB: db}
	err := s.CreateTable()
	c.Assert(err, IsNil)

	v := AccountT{
		ID:                  "12345",
		Name:                "alice",
		Email:               "[email protected]",
		UnmarshalFailPlease: true,
	}
	err = s.Put([]string{"Accounts", "12345"}, &v)
	c.Assert(err, IsNil)

	err = s.PutLink([]string{"AccountsByEmail", "[email protected]"}, []string{"Accounts", "12345"})
	c.Assert(err, IsNil)

	_, err = s.GetLink([]string{"AccountsByEmail", "missing"})
	c.Assert(err, Equals, ErrNotFound)

	_, err = s.GetLink([]string{"Accounts", "12345"})
	c.Assert(err, Equals, ErrNotLink)
}

func benchPutItemParallel(p, c int, b *testing.B) {
	svc := dynamodb.New(&aws.Config{
		DisableSSL: aws.Bool(true),
	})

	av, err := dynamodbattribute.ConvertToMap(dbItem{Key: "MyKey", Data: "MyData"})
	if err != nil {
		b.Fatal("expect no ConvertToMap errors", err)
	}
	params := &dynamodb.PutItemInput{
		Item:      av,
		TableName: aws.String(testTableName),
	}
	b.N = c

	b.ResetTimer()
	b.SetParallelism(p)
	b.RunParallel(func(pb *testing.PB) {
		for pb.Next() {
			_, err = svc.PutItem(params)
			if err != nil {
				b.Error("expect no request errors", err)
			}
		}
	})
}

func main() {
	svc := dynamodb.New(session.New(&aws.Config{Region: aws.String("us-east-1")}))
	jar := &CookieManager{}
	var cookieCount int
	var sleepTime int64
	flag.IntVar(&cookieCount, "count", 10, "collect this many cookies")
	flag.Int64Var(&sleepTime, "sleep", 2, "sleep this many between executions")
	flag.Parse()
	for i := 0; i <= cookieCount; i++ {
		jar.jar = make(map[string][]*http.Cookie)
		if resp, err := GetCookie(jar); err == nil {
			t, _ := time.Parse(timeLongForm, resp.Header["Date"][0])
			time_string := strconv.FormatInt(t.Unix(), 10)
			body := resp.Body
			params := ProcessCookies(&InputItem{*jar, time_string, table_name, ScrapePage(&body)})
			svc.PutItem(params)
		} else {
			fmt.Println("Failed to get a response body.  Will retry after timeout.")
		}
		if i%5 == 0 && i != 0 {
			fmt.Printf("Got %d cookies.\n", i)
		}
		time.Sleep(time.Duration(sleepTime) * time.Second) // lets hold firm 2s for niceness
	}
}

func (c *ListTablesCommand) Run(args []string) int {
	var exclusiveStartTableName string
	var limit int

	cmdFlags := flag.NewFlagSet("list-tables", flag.ContinueOnError)
	cmdFlags.Usage = func() { c.Ui.Output(c.Help()) }
	cmdFlags.StringVar(&exclusiveStartTableName, "exclusive-start-table-name", "", "exclusive start table name")
	cmdFlags.IntVar(&limit, "limit", 100, "limit")
	if err := cmdFlags.Parse(args); err != nil {
		c.Ui.Error(err.Error())
		return 1
	}

	svc := dynamodb.New(session.New())
	params := &dynamodb.ListTablesInput{
		ExclusiveStartTableName: aws.String(exclusiveStartTableName),
		Limit: limit,
	}

	res, err := svc.ListTables(params)
	if err != nil {
		c.Ui.Error(err.Error())
		return 1
	}
	c.Ui.Output(res)
	return 0
}

func (s *FakeDynamoDBTest) TestStuff(c *C) {
	fakeDB, err := New()
	c.Assert(err, IsNil)
	defer fakeDB.Close()
	c.Assert(fakeDB.Port, Not(Equals), 0)

	db := dynamodb.New(session.New(), fakeDB.Config)

	_, err = db.CreateTable(&dynamodb.CreateTableInput{
		TableName: aws.String("frob"),
		KeySchema: []*dynamodb.KeySchemaElement{
			{
				AttributeName: aws.String("Key"),
				KeyType:       aws.String(dynamodb.KeyTypeHash),
			},
		},
		AttributeDefinitions: []*dynamodb.AttributeDefinition{
			&dynamodb.AttributeDefinition{
				AttributeName: aws.String("Key"),
				AttributeType: aws.String(dynamodb.ScalarAttributeTypeS),
			},
		},
		ProvisionedThroughput: &dynamodb.ProvisionedThroughput{
			ReadCapacityUnits:  aws.Int64(1),
			WriteCapacityUnits: aws.Int64(1),
		},
	})
	c.Assert(err, IsNil)
}

// DynamoTables responses dynamodb tables
func DynamoTables() (clusters []*string, e error) {
	resp, err := dynamodb.New(dynamoCfg).ListTables(nil)
	if err != nil {
		return nil, err
	}
	return resp.TableNames, nil
}

func BenchmarkPutItem(b *testing.B) {
	cfg := aws.Config{
		DisableSSL:  aws.Bool(true),
		Credentials: credentials.NewStaticCredentials("AKID", "SECRET", ""),
	}
	server := successRespServer([]byte(`{}`))
	cfg.Endpoint = aws.String(server.URL)

	svc := dynamodb.New(&cfg)
	svc.Handlers.Send.Clear()
	svc.Handlers.Send.PushBack(func(r *service.Request) {
		r.HTTPResponse = &http.Response{
			StatusCode: http.StatusOK,
			Status:     http.StatusText(http.StatusOK),
			Body:       noopBody,
		}
	})

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		av, err := dynamodbattribute.ConvertToMap(dbItem{Key: "MyKey", Data: "MyData"})
		if err != nil {
			b.Fatal("benchPutItem, expect no ConvertToMap errors", err)
		}
		params := &dynamodb.PutItemInput{
			Item:      av,
			TableName: aws.String("tablename"),
		}
		_, err = svc.PutItem(params)
		if err != nil {
			b.Error("benchPutItem, expect no request errors", err)
		}
	}
}