func send(cache []Event) {
	fmt.Println("sending %s", len(cache))
	for index, element := range cache {
		_ = index
		//fmt.Println(element.id, element.event, element.value)
		//fmt.Println(index)
		key, _ := as.NewKey("test", "totals", "total_votes")
		ops := []*as.Operation{
			as.AddOp(as.NewBin("total_votes", 1)), // add the value of the bin to the existing value
			as.GetOp(),                            // get the value of the record after all operations are executed
		}

		_, _ = client.Operate(nil, key, ops...)

		key, _ = as.NewKey("test", "totals", element.value)
		ops = []*as.Operation{
			as.AddOp(as.NewBin(element.value, 1)), // add the value of the bin to the existing value
			as.GetOp(),                            // get the value of the record after all operations are executed
		}

		_, _ = client.Operate(nil, key, ops...)
		//panicOnError(err)
		//fmt.Println(rec)

		//		client.Set(element.id, element.value)
		//		client.Get(element.id)
		//		client.Incr("total_votes")
		//		client.Incr(element.value)
	}
}

func runExample(client *as.Client) {
	// Write initial record.
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "opkey")
	bin1 := as.NewBin("optintbin", 7)
	bin2 := as.NewBin("optstringbin", "string value")
	log.Printf("Put: namespace=%s set=%s key=%s bin1=%s value1=%s bin2=%s value2=%s",
		key.Namespace(), key.SetName(), key.Value(), bin1.Name, bin1.Value, bin2.Name, bin2.Value)
	client.PutBins(shared.WritePolicy, key, bin1, bin2)

	// Add integer, write new string and read record.
	bin3 := as.NewBin(bin1.Name, 4)
	bin4 := as.NewBin(bin2.Name, "new string")
	log.Println("Add: ", bin3.Value)
	log.Println("Write: ", bin4.Value)
	log.Println("Read:")

	record, err := client.Operate(shared.WritePolicy, key, as.AddOp(bin3), as.PutOp(bin4), as.GetOp())
	shared.PanicOnError(err)

	if record == nil {
		log.Fatalf(
			"Failed to get: namespace=%s set=%s key=%s",
			key.Namespace(), key.SetName(), key.Value())
	}

	binExpected := as.NewBin(bin3.Name, 11)
	shared.ValidateBin(key, binExpected, record)
	shared.ValidateBin(key, bin4, record)
}

// When the auction has completed, perform the following logic.
func CloseAuction(offerId int, bid *m.Bid) {
	auctionMap.Remove(offerId)

	// Check if the broker has enough inventory
	offerKeyId := fmt.Sprintf("%s:%d", OFFERS, offerId)
	offerKey, _ := as.NewKey(NAMESPACE, OFFERS, offerKeyId)
	offerRec, _ := db.Get(readPolicy, offerKey)
	sellerId := offerRec.Bins["broker_id"].(int)

	// add to buyer's inventory and reduce credit
	buyerKeyId := fmt.Sprintf("%s:%d", BROKERS, bid.BrokerId)
	buyerKey, _ := as.NewKey(NAMESPACE, BROKERS, buyerKeyId)
	db.Operate(writePolicy, buyerKey,
		as.AddOp(as.NewBin(offerRec.Bins["ticker"].(string), offerRec.Bins["quantity"].(int))),
		as.AddOp(as.NewBin("credit", -1*offerRec.Bins["quantity"].(int)*int(bid.Price))),
	)

	// reduce seller's inventory and add to credit
	sellerKeyId := fmt.Sprintf("%s:%d", BROKERS, sellerId)
	sellerKey, _ := as.NewKey(NAMESPACE, BROKERS, sellerKeyId)
	db.Operate(writePolicy, sellerKey,
		as.AddOp(as.NewBin(offerRec.Bins["ticker"].(string), -1*offerRec.Bins["quantity"].(int))),
		as.AddOp(as.NewBin("credit", offerRec.Bins["quantity"].(int)*int(bid.Price))),
	)

	// mark the bid as winner
	bidKeyId := fmt.Sprintf("%s:%d", BIDS, bid.Id)
	bidKey, _ := as.NewKey(NAMESPACE, BIDS, bidKeyId)
	db.Put(writePolicy, bidKey, as.BinMap{"winner": 1})
	db.Put(writePolicy, offerKey, as.BinMap{"finished": 1, "winner": bid.Id})
}

func runExample(client *as.Client) {
	key, err := as.NewKey(*shared.Namespace, *shared.Set, "addkey")
	shared.PanicOnError(err)

	binName := "addbin"

	// Delete record if it already exists.
	client.Delete(shared.WritePolicy, key)

	// Perform some adds and check results.
	bin := as.NewBin(binName, 10)
	log.Println("Initial add will create record.  Initial value is ", bin.Value, ".")
	client.AddBins(shared.WritePolicy, key, bin)

	bin = as.NewBin(binName, 5)
	log.Println("Add ", bin.Value, " to existing record.")
	client.AddBins(shared.WritePolicy, key, bin)

	record, err := client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)

	if record == nil {
		log.Fatalf(
			"Failed to get: namespace=%s set=%s key=%s",
			key.Namespace(), key.SetName(), key.Value())
	}

	// The value received from the server is an unsigned byte stream.
	// Convert to an integer before comparing with expected.
	received := record.Bins[bin.Name]
	expected := 15

	if received == expected {
		log.Printf("Add successful: ns=%s set=%s key=%s bin=%s value=%s",
			key.Namespace(), key.SetName(), key.Value(), bin.Name, received)
	} else {
		log.Fatalf("Add mismatch: Expected %d. Received %d.", expected, received)
	}

	// Demonstrate add and get combined.
	bin = as.NewBin(binName, 30)
	log.Println("Add ", bin.Value, " to existing record.")
	record, err = client.Operate(shared.WritePolicy, key, as.AddOp(bin), as.GetOp())
	shared.PanicOnError(err)

	expected = 45
	received = record.Bins[bin.Name]

	if received == expected {
		log.Printf("Add successful: ns=%s set=%s key=%s bin=%s value=%s",
			key.Namespace(), key.SetName(), key.Value(), bin.Name, received)
	} else {
		log.Fatalf("Add mismatch: Expected %d. Received %d.", expected, received)
	}
}

func (c *client) writeData(key *as.Key, result interface{}, expire time.Duration) {
	writePolicy := as.NewWritePolicy(0, int32(expire.Seconds()))

	value, err := json.Marshal(result)
	panicOnError(err)

	lockBin := as.NewBin(binLockName, nil)
	dataBin := as.NewBin(binDataName, string(value))

	err = c.asClient.PutBins(writePolicy, key, lockBin, dataBin)
	panicOnError(err)
}

// Increment a bin, being used as a sequence generator.
func nextSeq(seq string) (int, error) {

	keyId := fmt.Sprintf("%s:%s", SEQUENCES, seq)
	key, err := as.NewKey(NAMESPACE, SEQUENCES, keyId)
	if err != nil {
		return 0, err
	}

	rec, err := db.Operate(
		writePolicy,
		key,
		as.AddOp(as.NewBin("seq", 1)),
		as.GetOpForBin("seq"),
	)

	if err != nil {
		return 0, err
	}

	if seq, exists := rec.Bins["seq"]; !exists || int(seq.(int)) <= 0 {
		return 0, fmt.Errorf("sequence not found: %s", seq)
	}

	return rec.Bins["seq"].(int), nil
}

func runExample(client *as.Client) {
	// Set LuaPath
	luaPath, _ := os.Getwd()
	luaPath += "/udf/"
	as.SetLuaPath(luaPath)

	filename := "sum_single_bin"
	regTask, err := client.RegisterUDFFromFile(nil, luaPath+filename+".lua", filename+".lua", as.LUA)
	shared.PanicOnError(err)

	// wait until UDF is created
	shared.PanicOnError(<-regTask.OnComplete())

	sum := 0
	for i := 1; i <= keyCount; i++ {
		sum += i
		key, err := as.NewKey(*shared.Namespace, *shared.Set, i)
		shared.PanicOnError(err)

		bin1 := as.NewBin("bin1", i)
		client.PutBins(nil, key, bin1)
	}

	t := time.Now()
	stm := as.NewStatement(*shared.Namespace, *shared.Set)
	res, err := client.QueryAggregate(nil, stm, filename, filename, "bin1")
	shared.PanicOnError(err)

	for rec := range res.Results() {
		log.Printf("Result %f should equal %d\n", rec.Record.Bins["SUCCESS"], sum)
	}
	log.Println("Map/Reduce took", time.Now().Sub(t))
}

func (c *client) lockKey(key *as.Key) error {
	writePolicy := as.NewWritePolicy(0, -1)
	writePolicy.RecordExistsAction = as.CREATE_ONLY

	lockBin := as.NewBin(binLockName, "true")

	return c.asClient.PutBins(writePolicy, key, lockBin)
}

// Set stores a value by key.
// TODO(nslaughter): change name. Put is a better name because a Set is a thing.
func (db DB) Set(key string, value string) (interface{}, error) {
	k, err := as.NewKey(db.namespace, db.set, key)
	if err != nil {
		return 0, err
	}
	bin := as.NewBin("", value)
	db.client.PutBins(db.wp, k, bin)
	return 1, nil
}

/**
 * Write/Read Array/Map combination directly instead of relying on java serializer.
 */
func testListMapCombined(client *as.Client) {
	log.Printf("Read/Write Array/Map")
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "listmapkey")
	client.Delete(shared.WritePolicy, key)

	blob := []byte{3, 52, 125}
	inner := []interface{}{
		"string2",
		5,
	}

	innerMap := map[interface{}]interface{}{
		"a":    1,
		2:      "b",
		3:      blob,
		"list": inner,
	}

	list := []interface{}{
		"string1",
		8,
		inner,
		innerMap,
	}

	bin := as.NewBin("listmapbin", list)
	client.PutBins(shared.WritePolicy, key, bin)

	record, err := client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)
	received := record.Bins[bin.Name].([]interface{})

	validateSize(4, len(received))
	validate("string1", received[0])
	// Server convert numbers to long, so must expect long.
	validate(8, received[1])

	receivedInner := received[2].([]interface{})
	validateSize(2, len(receivedInner))
	validate("string2", receivedInner[0])
	validate(5, receivedInner[1])

	receivedMap := received[3].(map[interface{}]interface{})
	validateSize(4, len(receivedMap))
	validate(1, receivedMap["a"])
	validate("b", receivedMap[2])
	validateBytes(blob, receivedMap[3].([]byte))

	receivedInner2 := receivedMap["list"].([]interface{})
	validateSize(2, len(receivedInner2))
	validate("string2", receivedInner2[0])
	validate(5, receivedInner2[1])

	log.Printf("Read/Write Array/Map successful")
}

func runReplaceExample(client *as.Client) {
	key, err := as.NewKey(*shared.Namespace, *shared.Set, "replacekey")
	shared.PanicOnError(err)
	bin1 := as.NewBin("bin1", "value1")
	bin2 := as.NewBin("bin2", "value2")
	bin3 := as.NewBin("bin3", "value3")

	log.Printf("Put: namespace=%s set=%s key=%s bin1=%s value1=%s bin2=%s value2=%s",
		key.Namespace(), key.SetName(), key.Value(), bin1.Name, bin1.Value, bin2.Name, bin2.Value)

	client.PutBins(shared.WritePolicy, key, bin1, bin2)

	log.Printf("Replace with: namespace=%s set=%s key=%s bin=%s value=%s",
		key.Namespace(), key.SetName(), key.Value(), bin3.Name, bin3.Value)

	wpolicy := as.NewWritePolicy(0, 0)
	wpolicy.RecordExistsAction = as.REPLACE
	client.PutBins(wpolicy, key, bin3)

	log.Printf("Get: namespace=%s set=%s key=%s", key.Namespace(), key.SetName(), key.Value())

	record, err := client.Get(shared.Policy, key)
	shared.PanicOnError(err)

	if record == nil {
		log.Fatalf(
			"Failed to get: namespace=%s set=%s key=%s", key.Namespace(), key.SetName(), key.Value())
	}

	if record.Bins[bin1.Name] == nil {
		log.Printf(bin1.Name + " was deleted as expected.")
	} else {
		log.Fatalln(bin1.Name + " found when it should have been deleted.")
	}

	if record.Bins[bin2.Name] == nil {
		log.Printf(bin2.Name + " was deleted as expected.")
	} else {
		log.Fatalln(bin2.Name + " found when it should have been deleted.")
	}
	shared.ValidateBin(key, bin3, record)
}

func runExample(client *as.Client) {
	key, err := as.NewKey(*shared.Namespace, *shared.Set, "prependkey")
	shared.PanicOnError(err)

	binName := "prependbin"

	// Delete record if it already exists.
	client.Delete(shared.WritePolicy, key)

	bin := as.NewBin(binName, "World")
	log.Println("Initial prepend will create record.  Initial value is ", bin.Value, ".")
	client.PrependBins(shared.WritePolicy, key, bin)

	bin = as.NewBin(binName, "Hello ")
	log.Println("Prepend \"", bin.Value, "\" to existing record.")
	client.PrependBins(shared.WritePolicy, key, bin)

	record, err := client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)

	if record == nil {
		log.Fatalf(
			"Failed to get: namespace=%s set=%s key=%s",
			key.Namespace(), key.SetName(), key.Value())
	}

	// The value received from the server is an unsigned byte stream.
	// Convert to an integer before comparing with expected.
	received := record.Bins[bin.Name]
	expected := "Hello World"

	if received == expected {
		log.Printf("Prepend successful: ns=%s set=%s key=%s bin=%s value=%s",
			key.Namespace(), key.SetName(), key.Value(), bin.Name, received)
	} else {
		log.Fatalf("Prepend mismatch: Expected %s. Received %s.", expected, received)
	}
}

// Execute put and get on a server configured as multi-bin.  This is the server default.
func runMultiBinExample(client *as.Client) {
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "putgetkey")
	bin1 := as.NewBin("bin1", "value1")
	bin2 := as.NewBin("bin2", "value2")

	log.Printf("Put: namespace=%s set=%s key=%s bin1=%s value1=%s bin2=%s value2=%s",
		key.Namespace(), key.SetName(), key.Value(), bin1.Name, bin1.Value, bin2.Name, bin2.Value)

	client.PutBins(shared.WritePolicy, key, bin1, bin2)

	log.Printf("Get: namespace=%s set=%s key=%s", key.Namespace(), key.SetName(), key.Value())

	record, err := client.Get(shared.Policy, key)
	shared.PanicOnError(err)

	if record == nil {
		panic(errors.New(fmt.Sprintf(
			"Failed to get: namespace=%s set=%s key=%s", key.Namespace(), key.SetName(), key.Value())))
	}

	shared.ValidateBin(key, bin1, record)
	shared.ValidateBin(key, bin2, record)
}

func runExample(client *as.Client) {
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "touchkey")
	bin := as.NewBin("touchbin", "touchvalue")

	log.Printf("Create record with 2 second expiration.")
	writePolicy := as.NewWritePolicy(0, 2)
	client.PutBins(writePolicy, key, bin)

	log.Printf("Touch same record with 5 second expiration.")
	writePolicy.Expiration = 5
	record, err := client.Operate(writePolicy, key, as.TouchOp(), as.GetHeaderOp())

	if record == nil {
		log.Fatalf(
			"Failed to get: namespace=%s set=%s key=%s bin=%s value=%s",
			key.Namespace(), key.SetName(), key.Value(), bin.Name, nil)
	}

	if record.Expiration == 0 {
		log.Fatalf(
			"Failed to get record expiration: namespace=%s set=%s key=%s",
			key.Namespace(), key.SetName(), key.Value())
	}

	log.Printf("Sleep 3 seconds.")
	time.Sleep(3 * time.Second)

	record, err = client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)

	if record == nil {
		log.Fatalf(
			"Failed to get: namespace=%s set=%s key=%s",
			key.Namespace(), key.SetName(), key.Value())
	}

	log.Printf("Success. Record still exists.")
	log.Printf("Sleep 4 seconds.")
	time.Sleep(4 * time.Second)

	record, err = client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)

	if record == nil {
		log.Printf("Success. Record expired as expected.")
	} else {
		log.Fatalf("Found record when it should have expired.")
	}
}

func writeIfGenerationNotChanged(client *as.Client) {
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "udfkey2")
	bin := as.NewBin("udfbin2", "string value")

	// Seed record.
	client.PutBins(shared.WritePolicy, key, bin)

	// Get record generation.
	gen, err := client.Execute(shared.WritePolicy, key, "record_example", "getGeneration")
	shared.PanicOnError(err)

	// Write record if generation has not changed.
	client.Execute(shared.WritePolicy, key, "record_example", "writeIfGenerationNotChanged", as.NewValue(bin.Name), bin.Value, as.NewValue(gen))
	log.Printf("Record written.")
}

/**
 * Write and twice read a non-expiring tuple using the new "NoExpire" value (-1).
 * This example is most effective when the Default Namespace Time To Live (TTL)
 * is set to a small value, such as 5 seconds.  When we sleep beyond that
 * time, we show that the NoExpire TTL flag actually works.
 */
func noExpireExample(client *as.Client) {
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "expirekey")
	bin := as.NewBin("expirebin", "noexpirevalue")

	log.Printf("Put: namespace=%s set=%s key=%s bin=%s value=%s expiration=NoExpire",
		key.Namespace(), key.SetName(), key.Value(), bin.Name, bin.Value)

	// Specify that record NEVER expires.
	// The "Never Expire" value is -1, or 0xFFFFFFFF.
	writePolicy := as.NewWritePolicy(0, 2)
	writePolicy.Expiration = math.MaxUint32
	client.PutBins(writePolicy, key, bin)

	// Read the record, showing it is there.
	log.Printf("Get: namespace=%s set=%s key=%s",
		key.Namespace(), key.SetName(), key.Value())

	record, err := client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)
	if record == nil {
		log.Fatalf(
			"Failed to get record: namespace=%s set=%s key=%s",
			key.Namespace(), key.SetName(), key.Value())
	}

	received := record.Bins[bin.Name]
	expected := bin.Value.String()
	if received == expected {
		log.Printf("Get record successful: namespace=%s set=%s key=%s bin=%s value=%s",
			key.Namespace(), key.SetName(), key.Value(), bin.Name, received)
	} else {
		log.Fatalf("Expire record mismatch: Expected %s. Received %s.",
			expected, received)
	}

	// Read this Record after the Default Expiration, showing it is still there.
	// We should have set the Namespace TTL at 5 sec.
	log.Printf("Sleeping for 10 seconds ...")
	time.Sleep(10 * time.Second)
	record, err = client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)

	if record == nil {
		log.Fatalf("Record expired and should NOT have.")
	} else {
		log.Printf("Found Record (correctly) after default TTL.")
	}
}

/**
 * Write records individually.
 */
func writeRecords(
	client *as.Client,
	keyPrefix string,
	binName string,
	valuePrefix string,
	size int,
) {
	for i := 1; i <= size; i++ {
		key, _ := as.NewKey(*shared.Namespace, *shared.Set, keyPrefix+strconv.Itoa(i))
		bin := as.NewBin(binName, valuePrefix+strconv.Itoa(i))

		log.Printf("Put: ns=%s set=%s key=%s bin=%s value=%s",
			key.Namespace(), key.SetName(), key.Value(), bin.Name, bin.Value)

		client.PutBins(shared.WritePolicy, key, bin)
	}
}

func writeUsingUdf(client *as.Client) {
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "udfkey1")
	bin := as.NewBin("udfbin1", "string value")

	client.Execute(shared.WritePolicy, key, "record_example", "writeBin", as.NewValue(bin.Name), bin.Value)

	record, err := client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)
	expected := bin.Value.String()
	received := record.Bins[bin.Name].(string)

	if received == expected {
		log.Printf("Data matched: namespace=%s set=%s key=%s bin=%s value=%s",
			key.Namespace(), key.SetName(), key.Value(), bin.Name, received)
	} else {
		log.Printf("Data mismatch: Expected %s. Received %s.", expected, received)
	}
}

/**
 * Write and twice read an expiration record.
 */
func expireExample(client *as.Client) {
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "expirekey ")
	bin := as.NewBin("expirebin", "expirevalue")

	log.Printf("Put: namespace=%s set=%s key=%s bin=%s value=%s expiration=2",
		key.Namespace(), key.SetName(), key.Value(), bin.Name, bin.Value)

	// Specify that record expires 2 seconds after it's written.
	writePolicy := as.NewWritePolicy(0, 2)
	client.PutBins(writePolicy, key, bin)

	// Read the record before it expires, showing it is there.
	log.Printf("Get: namespace=%s set=%s key=%s",
		key.Namespace(), key.SetName(), key.Value())

	record, err := client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)
	if record == nil {
		log.Fatalf(
			"Failed to get record: namespace=%s set=%s key=%s",
			key.Namespace(), key.SetName(), key.Value())
	}

	received := record.Bins[bin.Name]
	expected := bin.Value.String()
	if received == expected {
		log.Printf("Get record successful: namespace=%s set=%s key=%s bin=%s value=%s",
			key.Namespace(), key.SetName(), key.Value(), bin.Name, received)
	} else {
		log.Fatalf("Expire record mismatch: Expected %s. Received %s.",
			expected, received)
	}

	// Read the record after it expires, showing it's gone.
	log.Printf("Sleeping for 3 seconds ...")
	time.Sleep(3 * time.Second)
	record, err = client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)
	if record == nil {
		log.Printf("Expiry of record successful. Record not found.")
	} else {
		log.Fatalf("Found record when it should have expired.")
	}
}

/**
 * Write/Read map[interface{}]interface{} directly instead of relying on java serializer.
 */
func testMapComplex(client *as.Client) {
	log.Printf("Read/Write map[interface{}]interface{}")
	key, _ := as.NewKey(*shared.Namespace, *shared.Set, "mapkey2")
	client.Delete(shared.WritePolicy, key)

	blob := []byte{3, 52, 125}
	list := []int{
		100034,
		12384955,
		3,
		512,
	}

	amap := map[interface{}]interface{}{
		"key1": "string1",
		"key2": 2,
		"key3": blob,
		"key4": list,
	}

	bin := as.NewBin("mapbin2", amap)
	client.PutBins(shared.WritePolicy, key, bin)

	record, err := client.Get(shared.Policy, key, bin.Name)
	shared.PanicOnError(err)
	receivedMap := record.Bins[bin.Name].(map[interface{}]interface{})

	validateSize(4, len(receivedMap))
	validate("string1", receivedMap["key1"])
	// Server convert numbers to long, so must expect long.
	validate(2, receivedMap["key2"])
	validateBytes(blob, receivedMap["key3"].([]byte))

	receivedInner := receivedMap["key4"].([]interface{})
	validateSize(4, len(receivedInner))
	validate(100034, receivedInner[0])
	validate(12384955, receivedInner[1])
	validate(3, receivedInner[2])
	validate(512, receivedInner[3])

	log.Printf("Read/Write map[interface{}]interface{} successful")
}