public async Task Save(String endpoint, long position)
        {
            var id = new RiakObjectId($"{Settings.Bucket}.system", endpoint);

            var options = new RiakGetOptions { };
            options.SetRw(Quorum.WellKnown.All);
            var exists = await _riak.Async.Get(id, options);
            if (exists.IsSuccess)
            {
                var checkpoint = exists.Value.GetObject<Checkpoint>();
                checkpoint.Position = position;

                exists.Value.SetObject(checkpoint);

                var putOpt = new RiakPutOptions { IfNotModified = true, IfNoneMatch = false };
                putOpt.SetW(Quorum.WellKnown.All);
                await _riak.Async.Put(exists.Value, putOpt);
            }
            else
            {
                var checkpoint = new Checkpoint
                {
                    Id = endpoint,
                    Position = position,
                };
                var putOpt = new RiakPutOptions { IfNotModified = false, IfNoneMatch = true };
                putOpt.SetW(Quorum.WellKnown.All);
                var o = new RiakObject(id, checkpoint);
                await _riak.Async.Put(o, putOpt);

            }
        }

        public void CustomSerializerWillSerializeJson()
        {
            var testPerson = new Person
            {
                DateOfBirth = new DateTime(1978, 12, 5, 0, 0, 0, DateTimeKind.Utc),
                Email = "[email protected]",
                Name = new Name
                {
                    FirstName = "OJ",
                    Surname = "Reeves"
                },
                PhoneNumbers = new List<PhoneNumber>
                {
                    new PhoneNumber
                    {
                        Number = "12345678",
                        NumberType = PhoneNumberType.Home
                    }
                }
            };

            var sots = new SerializeObjectToString<Person>(JsonConvert.SerializeObject);

            var obj = new RiakObject("bucket", "key");
            obj.SetObject(testPerson, RiakConstants.ContentTypes.ApplicationJson, sots);
            obj.Value.ShouldNotBeNull();
            obj.ContentType.ShouldEqual(RiakConstants.ContentTypes.ApplicationJson);

            var json = obj.Value.FromRiakString();
            json.ShouldEqual("{\"Name\":{\"FirstName\":\"OJ\",\"Surname\":\"Reeves\"},\"PhoneNumbers\":[{\"Number\":\"12345678\",\"NumberType\":1}],\"DateOfBirth\":\"\\/Date(281664000000)\\/\",\"Email\":\"[email protected]\"}");

            var deserialisedPerson = obj.GetObject<Person>();
            deserialisedPerson.ShouldEqual(testPerson);
        }

        public void RangeQuery()
        {
            var Cluster = RiakCluster.FromConfig("riakConfig");
            var Client = Cluster.CreateClient();

            var bucket = "rsds";

            Client.Batch(batch =>
            {
                for (var i = 1; i < 11; i++)
                {
                    var d = DateTime.Now.AddDays(0 - i);
                    var doc = new RiakObject(bucket, i.ToString(), new { value = i, created_date = d });

                    var position = 100 + i;

                    doc.BinIndex("position").Set(position.ToString());
                    
                    batch.Put(doc);
                }
            });

            var query = new RiakMapReduceQuery()
                    .Inputs(RiakIndex.Range("bucket", "position", 100, 200))
                    .MapJs(m => m.Name("Riak.mapValuesJson").Keep(true));

            var result = Client.MapReduce(query);
            //var items = result.Value.PhaseResults.SelectMany(x => x.GetObjects<dynamic>);
        }

 public void InvalidFieldNamesAndTypes()
 {
     id = new RiakObjectId("indexes", "mykey", "test value");
     var obj = new RiakObject("indexes", "mykey", "test value");
     var intIdx = obj.IntIndex("test-int-idx");
     Assert.Throws<FormatException>(() => intIdx.Add("invalid-value"));
 }

        public void ObjectsAreConvertedProperly()
        {
            var testPerson = new Person
            {
                DateOfBirth = new DateTime(1978, 12, 5, 0, 0, 0, DateTimeKind.Utc),
                Email = "[email protected]",
                Name = new Name
                {
                    FirstName = "OJ",
                    Surname = "Reeves"
                },
                PhoneNumbers = new List<PhoneNumber>
                {
                    new PhoneNumber
                    {
                        Number = "12345678",
                        NumberType = PhoneNumberType.Home
                    }
                }
            };
            var obj = new RiakObject("bucket", "key");
            obj.SetObject(testPerson);
            obj.Value.ShouldNotBeNull();
            obj.ContentType.ShouldEqual(RiakConstants.ContentTypes.ApplicationJson);

            var json = obj.Value.FromRiakString();
            json.ShouldEqual("{\"Name\":{\"FirstName\":\"OJ\",\"Surname\":\"Reeves\"},\"PhoneNumbers\":[{\"Number\":\"12345678\",\"NumberType\":1}],\"DateOfBirth\":\"1978-12-05T00:00:00Z\",\"Email\":\"[email protected]\"}");

            var deserialisedPerson = obj.GetObject<Person>();
            deserialisedPerson.ShouldEqual(testPerson);
        }

        public void AsyncDeleteIsSuccessful()
        {
            var riakObject = new RiakObject(TestBucket, TestKey, TestJson, RiakConstants.ContentTypes.ApplicationJson);
            var riakObjectId = riakObject.ToRiakObjectId();

            var putResult = Client.Put(riakObject);
            putResult.IsSuccess.ShouldBeTrue();

            RiakResult theResult = null;
            var resetEvent = new AutoResetEvent(false);

            Client.Async.Delete(riakObjectId, result =>
                                            {
                                                theResult = result;
                                                resetEvent.Set();
                                            });
            resetEvent.WaitOne();

            theResult.IsSuccess.ShouldBeTrue();

            var getResult = Client.Get(riakObjectId);
            getResult.IsSuccess.ShouldBeFalse();
            getResult.ResultCode.ShouldEqual(ResultCode.NotFound);
            getResult.Value.ShouldBeNull();
        }

        public void AsyncDeleteMultipleIsSuccessful()
        {
            var one = new RiakObject(TestBucket, "one", TestJson, RiakConstants.ContentTypes.ApplicationJson);
            var two = new RiakObject(TestBucket, "two", TestJson, RiakConstants.ContentTypes.ApplicationJson);

            Client.Put(one);
            Client.Put(two);

            var oneObjectId = one.ToRiakObjectId();
            var twoObjectId = two.ToRiakObjectId();

            var list = new List<RiakObjectId> { oneObjectId, twoObjectId };

            var results = Client.Async.Delete(list).Result;

            foreach (var riakResult in results)
            {
                riakResult.IsSuccess.ShouldBeTrue(riakResult.ErrorMessage);
            }

            var oneResult = Client.Get(oneObjectId);
            oneResult.IsSuccess.ShouldBeFalse();
            oneResult.ResultCode.ShouldEqual(ResultCode.NotFound);
            oneResult.Value.ShouldBeNull();

            var twoResult = Client.Get(twoObjectId);
            twoResult.IsSuccess.ShouldBeFalse();
            twoResult.ResultCode.ShouldEqual(ResultCode.NotFound);
            twoResult.Value.ShouldBeNull();
        }

        public void AllKeysReturnsListOfKeys()
        {
            var bucket = Bucket + "_" + Guid.NewGuid().ToString();
            var originalKeys = new List<string>();

            for (var i = 0; i < 10; i++)
            {
                var o = new RiakObject(bucket, Guid.NewGuid().ToString(), "{ value: \"this is an object\" }");
                originalKeys.Add(o.Key);

                Client.Put(o);
            }

            var mr = new RiakMapReduceQuery()
                .Inputs(RiakIndex.AllKeys(bucket));

            var result = Client.MapReduce(mr);
            var keys = result.Value.PhaseResults.SelectMany(x => x.GetObjectIds()).ToList();

            result.IsSuccess.ShouldBeTrue(result.ErrorMessage);
            keys.Count.ShouldEqual(10);

            foreach (var key in keys)
            {
                key.Bucket.ShouldNotBeNullOrEmpty();
                key.Key.ShouldNotBeNullOrEmpty();
                originalKeys.Contains(key.Key).ShouldBeTrue();
            }
        }

 public void CreatingBucketInSpecificBucketType()
 {
     id = new RiakObjectId("no_siblings", "sensitive_user_data", "user19735");
     var obj = new RiakObject(id, "{\"name\":\"Bob\"}");
     var rslt = client.Put(obj);
     CheckResult(rslt);
 }

        protected void CreateLinkedObjects(string bucketName)
        {
            var oj = new RiakObject(bucketName, OJ, new Person() { Name = "oj" });
            var jeremiah = new RiakObject(bucketName, Jeremiah, new Person() { Name = "jeremiah" });
            var brent = new RiakObject(bucketName, Brent, new Person() { Name = "brent" });
            var rob = new RiakObject(bucketName, Rob, new Person() { Name = "rob" });

            oj.ContentType = RiakConstants.ContentTypes.ApplicationJson;
            jeremiah.ContentType = RiakConstants.ContentTypes.ApplicationJson;
            brent.ContentType = RiakConstants.ContentTypes.ApplicationJson;
            rob.ContentType = RiakConstants.ContentTypes.ApplicationJson;

#pragma warning disable 618
            oj.LinkTo(jeremiah, "friends");
            oj.LinkTo(jeremiah, "coworkers");

            jeremiah.LinkTo(oj, "friends");
            jeremiah.LinkTo(oj, "coworkers");
            jeremiah.LinkTo(oj, "ozzies");
            jeremiah.LinkTo(brent, "friends");
            jeremiah.LinkTo(brent, "coworkers");
            jeremiah.LinkTo(rob, "ozzies");

            brent.LinkTo(jeremiah, "coworkers");
            brent.LinkTo(jeremiah, "friends");
#pragma warning restore 618
            Client.Put(new[] { oj, jeremiah, brent, rob });
        }

        private static void StoreData()
        {
            Console.WriteLine("[ChaosMonkeyApp] store thread starting");
            IRiakClient client = cluster.CreateClient();
            try
            {
                while (true)
                {
                    var id = new RiakObjectId("chaos-monkey", key.ToString());
                    var obj = new RiakObject(id, Guid.NewGuid().ToString());
                    obj.ContentEncoding = RiakConstants.CharSets.Utf8;
                    obj.ContentType = RiakConstants.ContentTypes.TextPlain;

                    var rslt = client.Put(obj);
                    if (rslt.IsSuccess)
                    {
                        Console.WriteLine("[ChaosMonkeyApp] stored key: {0}", key);
                    }
                    else
                    {
                        Console.WriteLine("[ChaosMonkeyApp] error storing key {0}, {1}", key, rslt.ErrorMessage);
                    }

                    ++key;
                    Thread.Sleep(storeDataInterval);
                    ct.ThrowIfCancellationRequested();
                }
            }
            finally
            {
                Console.WriteLine("[ChaosMonkeyApp] store thread stopping");
            }
        }

        public void TestKVOperations()
        {
            const string key = "bucket_type_test_key";
            var id = new RiakObjectId(TestBucketType, TestBucket, key);
            var obj = new RiakObject(id, Value);

            // put
            var putResult = Client.Put(obj, new RiakPutOptions().SetReturnBody(true).SetDw(3));
            putResult.IsSuccess.ShouldBeTrue();
            putResult.Value.BucketType.ShouldEqual(TestBucketType);

            // get
            var getResult = Client.Get(id);
            getResult.IsSuccess.ShouldBeTrue();
            getResult.Value.BucketType.ShouldEqual(TestBucketType);

            // delete
            var deleteResult = Client.Delete(id, new RiakDeleteOptions().SetDw(3));
            deleteResult.IsSuccess.ShouldBeTrue();

            // multiget
            var ids = new List<RiakObjectId>();
            for (int i = 0; i < 3; i++)
            {
                obj = new RiakObject(new RiakObjectId(TestBucketType, TestBucket, key + i), Value);
                Client.Put(obj, new RiakPutOptions().SetReturnBody(true).SetDw(3));
                ids.Add(obj.ToRiakObjectId());
            }

            var multiGetResult = Client.Get(ids).ToList();
            multiGetResult.All(r => r.IsSuccess).ShouldBeTrue();
            multiGetResult.All(r => r.Value.BucketType == TestBucketType).ShouldBeTrue();
        }

        public void SettingLegacySearchOnRiakBucketMakesBucketSearchable()
        {
            var bucket = Guid.NewGuid().ToString();
            var key = Guid.NewGuid().ToString();
            var props = Client.GetBucketProperties(bucket).Value;
            props.SetLegacySearch(true);

            var setResult = Client.SetBucketProperties(bucket, props);
            setResult.IsSuccess.ShouldBeTrue(setResult.ErrorMessage);

            var obj = new RiakObject(bucket, key, new { name = "OJ", age = 34 });
            var putResult = Client.Put(obj);
            putResult.IsSuccess.ShouldBeTrue(putResult.ErrorMessage);

            var q = new RiakFluentSearch(bucket, "name")
                .Search("OJ")
                .And("age", "34")
                .Build();

            var search = new RiakSearchRequest
            {
                Query = q
            };

            var searchResult = Client.Search(search);
            searchResult.IsSuccess.ShouldBeTrue(searchResult.ErrorMessage);
            searchResult.Value.NumFound.ShouldEqual(1u);
            searchResult.Value.Documents[0].Fields.Count.ShouldEqual(3);
            searchResult.Value.Documents[0].Fields.First(x => x.Key == "id").Value.ShouldEqual(key);
        }

        public void ToRiakObjectIdProducesAValidRiakObjectId()
        {
            var riakObject = new RiakObject(Bucket, Key, "value");
            var riakObjectId = riakObject.ToRiakObjectId();

            riakObjectId.Bucket.ShouldEqual(Bucket);
            riakObjectId.Key.ShouldEqual(Key);
        }

 public void CreatingBucketInSpecificBucketTypeExampleTwo()
 {
     id = new RiakObjectId("no_siblings", "old_memes", "all_your_base");
     var obj = new RiakObject(id, "all your base are belong to us",
         RiakConstants.ContentTypes.TextPlain);
     var rslt = client.Put(obj);
     CheckResult(rslt);
 }

        private RiakObjectId PutCoach(string team, string coach)
        {
            var id = new RiakObjectId("siblings", "coaches", team);
            var obj = new RiakObject(id, coach,
                RiakConstants.ContentTypes.TextPlain);
            var rslt = client.Put(obj);
            CheckResult(rslt);

            return id;
        }

 public void WritingLargeObjectIsSuccessful()
 {
     var text = Enumerable.Range(0, 2000000)
                          .Aggregate(new StringBuilder(),
                                     (sb, i) => sb.Append(i.ToString()))
                          .ToString();
     var riakObject = new RiakObject(TestBucket, "large", text, RiakConstants.ContentTypes.TextPlain);
     var result = Client.Put(riakObject);
     result.IsSuccess.ShouldBeTrue(result.ErrorMessage);
 }

        public void Putting_A_Value_To_A_Write_Once_Bucket_Works()
        {
            string key = Guid.NewGuid().ToString();
            string value = "test value";

            var id = new RiakObjectId(BucketType, Bucket, key);
            var obj = new RiakObject(id, value, RiakConstants.ContentTypes.TextPlain);

            RiakResult<RiakObject> rslt = client.Put(obj);
            Assert.True(rslt.IsSuccess, rslt.ErrorMessage);
        }

        public RiakSessionItem(RiakObject riakObject)
        {
            SessionStoreItems = riakObject.Value.FromRiakString();
            Created = DateTime.Parse(riakObject.UserMetaData["X-Riak-Meta-Created"]);
            Expires = DateTime.Parse(riakObject.UserMetaData["X-Riak-Meta-Expires"]);
            LockDate = DateTime.Parse(riakObject.UserMetaData["X-Riak-Meta-LockDate"]);
            LockId = int.Parse(riakObject.UserMetaData["X-Riak-Meta-LockId"]);
            Locked = bool.Parse(riakObject.UserMetaData["X-Riak-Meta-Locked"]);
            Flags = int.Parse(riakObject.UserMetaData["X-Riak-Meta-Flags"]);

            _timeout = (Expires - Created).Minutes;
        }

        public void LotsOfConcurrentMapRedRequestsShouldWork()
        {
            var keys = new List<string>();

            for (var i = 1; i < 11; i++)
            {
                var key = "key" + i;
                var doc = new RiakObject(MapReduceBucket, key, new { value = i });
                keys.Add(key);

                var result = Client.Put(doc, new RiakPutOptions { ReturnBody = true });
                result.ShouldNotBeNull();
            }

            var input = new RiakBucketKeyInput();
            keys.ForEach(k => input.Add(MapReduceBucket, k));

            var query = new RiakMapReduceQuery()
                .Inputs(input)
                .MapJs(m => m.Source(@"function(o){return[1];}"))
                .ReduceJs(r => r.Name(@"Riak.reduceSum").Keep(true));
            query.Compile();

            var results = new List<RiakMapReduceResult>[ThreadCount];
            var watch = Stopwatch.StartNew();
            Parallel.For(0, ThreadCount, i =>
                {
                    results[i] = DoMapRed(query);
                });
            watch.Stop();
            var executionTime = watch.Elapsed;

            var failures = 0;
            foreach (var r in results.SelectMany(l => l))
            {
                if (r != null)
                {
                    var resultValue = JsonConvert.DeserializeObject<int[]>(r.PhaseResults.ElementAt(1).Values.First().FromRiakString())[0];
                    resultValue.ShouldEqual(10);
                    //r.Value.PhaseResults.ElementAt(1).GetObject<int[]>()[0].ShouldEqual(10);
                }
                else
                {
                    // the only acceptable result is that it ran out of retries when
                    // talking to the cluster (trying to get a connection)
                    //r.ResultCode.ShouldEqual(ResultCode.NoRetries);
                    ++failures;
                }
            }

            Console.WriteLine("Total of {0} out of {1} failed to execute due to connection contention. Execution time = {2} milliseconds, for an average of {3} milliseconds", failures, ThreadCount * ActionCount, executionTime.TotalMilliseconds, (executionTime.TotalMilliseconds/(ThreadCount * ActionCount)));
        }