public async Task GetClusterList()
        {
            ClusterConfig config = new ClusterConfig();
            MockReliableStateManager stateManager = new MockReliableStateManager();
            ClusterService target = new ClusterService(null, null, new MockApplicationDeployService(), stateManager, this.CreateServiceParameters(), config);

            int readyClusters = 10;
            int deletingCluster = 4;
            int newClusters = 2;
            int removeClusters = 1;

            IReliableDictionary<int, Cluster> dictionary =
                await stateManager.GetOrAddAsync<IReliableDictionary<int, Cluster>>(ClusterService.ClusterDictionaryName);

            using (ITransaction tx = stateManager.CreateTransaction())
            {
                await this.AddClusters(tx, dictionary, readyClusters, ClusterStatus.Ready);
                await this.AddClusters(tx, dictionary, deletingCluster, ClusterStatus.Deleting);
                await this.AddClusters(tx, dictionary, newClusters, ClusterStatus.New);
                await this.AddClusters(tx, dictionary, removeClusters, ClusterStatus.Remove);
                await tx.CommitAsync();
            }

            IEnumerable<ClusterView> actual = await target.GetClusterListAsync();

            Assert.AreEqual(readyClusters, actual.Count());
        }

        public void Read(BinaryReader reader)
        {
            if (reader == null)
                throw new ArgumentNullException("reader");

            Config = ClusterConfig.Read(reader);
        }

 // todo : Schedule class
 public void AssignTo(Estimation estimation)
 {
     if (estimation == null)
     {
         this.Time.Estimated = null;
         _assignedTo = new ClusterConfig(null, null);
         _state = TaskState.NotAssigned;
     }
     else
         AssignTo(estimation.ClusterName, estimation.Cores, estimation.Time);
 }

        public void Read(BinaryReader reader)
        {
            if (reader == null)
                throw new ArgumentNullException("reader");

            byte stepMode = reader.ReadByte();

            if (!Enum.IsDefined(typeof(StepMode), stepMode))
                throw new InvalidDataException("Unknown step mode: " + stepMode);

            Config = ClusterConfig.GetCompatibility((StepMode)stepMode,
                reader.ReadInt32(),
                reader.ReadByte() > 0);
        }

        public async Task BalanceClusters()
        {
            ClusterConfig config = new ClusterConfig();
            MockReliableStateManager stateManager = new MockReliableStateManager();
            ClusterService target = new ClusterService(null, null, new MockApplicationDeployService(), stateManager, this.CreateServiceParameters(), config);

            await target.BalanceClustersAsync(config.MinimumClusterCount);

            ConditionalResult<IReliableDictionary<int, Cluster>> result =
                await stateManager.TryGetAsync<IReliableDictionary<int, Cluster>>(ClusterService.ClusterDictionaryName);

            Assert.IsTrue(result.HasValue);
            Assert.AreEqual(config.MinimumClusterCount, await result.Value.GetCountAsync());
            Assert.IsTrue(result.Value.All(x => x.Value.Status == ClusterStatus.New));
        }

        public async Task TestBalanceClusters()
        {
            ClusterConfig config = new ClusterConfig();
            MockReliableStateManager stateManager = new MockReliableStateManager();
            ClusterService target = new ClusterService(null, stateManager)
            {
                Config = config
            };

            await target.BalanceClustersAsync(config.MinimumClusterCount);

            var result = await stateManager.TryGetAsync<IReliableDictionary<int, Cluster>>(ClusterService.ClusterDictionaryName);

            Assert.IsTrue(result.HasValue);
            Assert.AreEqual(config.MinimumClusterCount, await result.Value.GetCountAsync());
            Assert.IsTrue(result.Value.All(x => x.Value.Status == ClusterStatus.New));
        }

        private InternalState InitVBucket(ClusterConfig config, ISaslAuthenticationProvider auth)
        {
            // we have a vbucket config, which has its own server list
            // it's supposed to be the same as the cluster config's list,
            // but the order is significicant (because of the bucket indexes),
            // so we we'll use this for initializing the locator
            var vbsm = config.vBucketServerMap;

            if (log.IsInfoEnabled) log.Info("Has vbucket. Server count: " + (vbsm.serverList == null ? 0 : vbsm.serverList.Length));

            // parse the ip addresses of the servers in the vbucket map
            // make sure we have a propert vbucket map
            ValidateVBucketMap(vbsm, vbsm.serverList.Length);

            // create vbuckets from the int[][] arrays
            var buckets = vbsm.vBucketMap.Select(a => new VBucket(a[0], a.Skip(1).ToArray())).ToArray();

            var locator = new VBucketNodeLocator(vbsm.hashAlgorithm, buckets);

            // create a (host=>node) lookup from the node info objects,
            // so we can pass the extra config data to the factory method
            // (the vbucket map only contains 'host:port' strings)
            // this expects that all nodes listed in the vbucket map are listed in the config.nodes member as well
            var realNodes = config.nodes.ToDictionary(node => node.HostName + ":" + node.Port);
            var nodes = new List<IMemcachedNode>();

            foreach (var hostSpec in vbsm.serverList)
            {
                ClusterNode node;

                if (!realNodes.TryGetValue(hostSpec, out node))
                    throw new InvalidOperationException(String.Format("VBucket map contains a node {0} whihc was not found in the cluster info's node list.", hostSpec));

                var ip = GetFirstAddress(node.HostName);
                var endpoint = new IPEndPoint(ip, node.Port);

                nodes.Add(this.CreateNode(endpoint, auth, node.ConfigurationData));
            }

            return new InternalState
            {
                CurrentNodes = nodes.ToArray(),
                Locator = locator,
                OpFactory = new VBucketAwareOperationFactory(locator),
                IsVbucket = true
            };
        }

        private void ReconfigurePool(ClusterConfig config)
        {
            if (log.IsDebugEnabled)
            {
                if (Thread.CurrentThread.Name == null)
                {
                    Thread.CurrentThread.Name = "cbp_thread";
                }
            }

            // kill the timer first
            this.isTimerActive = false;
            if (this.resurrectTimer != null)
                this.resurrectTimer.Change(Timeout.Infinite, Timeout.Infinite);

            if (config == null)
            {
                if (log.IsInfoEnabled) log.Info("Config is empty, all nodes are down.");
                return; //continue to use current state and wait for config update
            }

            var currentState = this.state;

            // these should be disposed after we've been reinitialized
            var oldNodes = currentState == null ? null : currentState.CurrentNodes;

            // default bucket does not require authentication
            // Couchbase 1.6 tells us if a bucket needs authentication,
            // so let's try to use the config's password
            var password = config.authType == "sasl"
                            ? config.saslPassword
                            : this.configuration.BucketPassword;

            var authenticator = this.configuration.Bucket == null
                                   ? null
                                   : new PlainTextAuthenticator(null, this.configuration.Bucket, password);

            try
            {
                var state = (config == null || config.vBucketServerMap == null)
                                ? this.InitBasic(config, authenticator)
                                : this.InitVBucket(config, authenticator);

                var nodes = state.CurrentNodes;
                state.Locator.Initialize(nodes);

                // we need to subscribe the failed event,
                // so we can periodically check the dead
                // nodes, since we do not get a config
                // update every time a node dies
                foreach (IMemcachedNode node in nodes)
                {
                    node.Failed += this.NodeFail;
                }

                Interlocked.Exchange(ref this.state, state);
            }
            catch (Exception e)
            {
                log.Error("Failed to initialize the pool.", e);

                Interlocked.Exchange(ref this.state, InternalState.Empty);
            }

            //Clean up the old nodes
            CloseNodes(oldNodes);
        }

        private InternalState InitBasic(ClusterConfig config, ISaslAuthenticationProvider auth)
        {
            if (log.IsInfoEnabled) log.Info("No vbucket. Server count: " + (config.nodes == null ? 0 : config.nodes.Length));

            // the cluster can return host names in the server list, so
            // we ha ve to make sure they are converted to IP addresses
            var nodes = config == null
                        ? Enumerable.Empty<IMemcachedNode>()
                            : (from node in config.nodes
                               let ip = new IPEndPoint(GetFirstAddress(node.HostName), node.Port)
                               where node.Status == "healthy"
                               select CreateNode(ip, auth, node.ConfigurationData));

            return new InternalState
            {
                CurrentNodes = nodes.ToArray(),
                Locator = configuration.CreateNodeLocator() ?? new KetamaNodeLocator(),
                OpFactory = BasicCouchbaseOperationFactory.Instance
            };
        }

        public async Task ProcessNewCluster()
        {
            bool calledActual = false;
            string nameTemplate = "Test:{0}";
            string nameActual = null;

            MockReliableStateManager stateManager = new MockReliableStateManager();
            MockClusterOperator clusterOperator = new MockClusterOperator()
            {
                CreateClusterAsyncFunc = name =>
                {
                    nameActual = name;
                    calledActual = true;
                    return Task.FromResult(String.Format(nameTemplate, name));
                }
            };

            ClusterConfig config = new ClusterConfig();
            ClusterService target = new ClusterService(
                clusterOperator,
                null,
                new MockApplicationDeployService(),
                stateManager,
                this.CreateServiceParameters(),
                config);

            Cluster cluster = this.CreateCluster(ClusterStatus.New);

            Cluster actual = await target.ProcessClusterStatusAsync(cluster);

            Assert.IsTrue(calledActual);
            Assert.AreEqual(ClusterStatus.Creating, actual.Status);
            Assert.AreEqual(String.Format(nameTemplate, nameActual), actual.Address);
        }

        public async Task JoinClusterUserAlreadyExists()
        {
            ClusterConfig config = new ClusterConfig() {MaximumUsersPerCluster = 2};
            MockReliableStateManager stateManager = new MockReliableStateManager();
            MockMailer mockMailer = new MockMailer();
            ClusterService target = new ClusterService(
                null,
                mockMailer,
                new MockApplicationDeployService(),
                stateManager,
                this.CreateServiceParameters(),
                config);

            int idWithUser = 5;
            string email = "[email protected]";
            Cluster clusterWithUser = new Cluster(
                "test",
                ClusterStatus.Ready,
                0,
                0,
                "",
                new[] {80, 81},
                new[] {new ClusterUser(email, 80)},
                DateTimeOffset.UtcNow);

            int idWithoutUser = 6;
            Cluster clusterWithoutUser = new Cluster(
                "test2",
                ClusterStatus.Ready,
                0,
                0,
                "",
                new[] {80, 81},
                new ClusterUser[0],
                DateTimeOffset.UtcNow);


            IReliableDictionary<int, Cluster> dictionary =
                await stateManager.GetOrAddAsync<IReliableDictionary<int, Cluster>>(ClusterService.ClusterDictionaryName);

            using (ITransaction tx = stateManager.CreateTransaction())
            {
                await dictionary.AddAsync(tx, idWithUser, clusterWithUser);
                await dictionary.AddAsync(tx, idWithoutUser, clusterWithoutUser);
                await tx.CommitAsync();
            }

            try
            {
                await target.JoinClusterAsync(idWithoutUser, email);
                Assert.Fail("JoinClusterFailedException not thrown.");
            }
            catch (JoinClusterFailedException result)
            {
                Assert.AreEqual(JoinClusterFailedReason.UserAlreadyJoined, result.Reason);
            }
        }

		private InternalState InitBasic(ClusterConfig config, ISaslAuthenticationProvider auth)
		{
			if (log.IsInfoEnabled) log.Info("No vbucket. Server count: " + (config.nodes == null ? 0 : config.nodes.Length));

			// no vbucket config, use the node list and the ports
			var portType = this.configuration.Port;

			var tmp = config == null
					? Enumerable.Empty<IMemcachedNode>()
						: (from node in config.nodes
						   let ip = new IPEndPoint(IPAddress.Parse(node.hostname),
													(portType == BucketPortType.Proxy
														? node.ports.proxy
														: node.ports.direct))
						   where node.status == "healthy"
						   select (IMemcachedNode)(new BinaryNode(ip, this.configuration.SocketPool, auth)));

			return new InternalState
			{
				CurrentNodes = tmp.ToArray(),
				Locator = this.configuration.CreateNodeLocator() ?? new KetamaNodeLocator(),
				OpFactory = new Enyim.Caching.Memcached.Protocol.Binary.BinaryOperationFactory()
			};
		}

        /// <summary>
        /// When cluster config changes, update the static config
        /// list of bootstrap URIs with known working nodes from cluster config
        /// </summary>
        public void UpdateNodes(ClusterConfig config)
        {
            var currentUri = this.urls[this.urlIndex]; //last known good Uri
            var serverConfigUris = config.nodes.Where(n => n.Status == "healthy") //healthy nodes from config
                                      .Select(n => new UriBuilder(
                                                    currentUri.Scheme, n.HostName,
                                                    currentUri.Port, currentUri.PathAndQuery).Uri).ToList();

            serverConfigUris.ForEach(u => statusPool[u] = true);
            var newConfigUris = serverConfigUris.Union(this.urls).ToArray(); //allow any client supplied URIs to remain

            Interlocked.Exchange(ref this.urls, newConfigUris);
            Interlocked.Exchange(ref this.urlIndex, 0);
            Interlocked.Exchange(ref this.realUrls, this.urls.Distinct().ToDictionary(u => u, u => (Uri)null));
        }

        public async Task ProcessRemoveTimeLimit()
        {
            ClusterConfig config = new ClusterConfig()
            {
                MaximumClusterUptime = TimeSpan.FromHours(2)
            };

            MockReliableStateManager stateManager = new MockReliableStateManager();
            ClusterService target = new ClusterService(null, null, new MockApplicationDeployService(), stateManager, this.CreateServiceParameters(), config);

            Cluster cluster = new Cluster(
                "test",
                ClusterStatus.Ready,
                0,
                0,
                String.Empty,
                new int[0],
                new ClusterUser[0],
                DateTimeOffset.UtcNow - config.MaximumClusterUptime);

            Cluster actual = await target.ProcessClusterStatusAsync(cluster);

            Assert.AreEqual(ClusterStatus.Remove, actual.Status);
        }

        public async Task ProcessRemove()
        {
            bool calledActual = false;
            string nameActual = null;

            MockReliableStateManager stateManager = new MockReliableStateManager();
            MockClusterOperator clusterOperator = new MockClusterOperator()
            {
                DeleteClusterAsyncFunc = name =>
                {
                    nameActual = name;
                    calledActual = true;
                    return Task.FromResult(true);
                }
            };

            ClusterConfig config = new ClusterConfig();
            ClusterService target = new ClusterService(
                clusterOperator,
                null,
                new MockApplicationDeployService(),
                stateManager,
                this.CreateServiceParameters(),
                config);

            Cluster cluster = this.CreateCluster(ClusterStatus.Remove);

            Cluster actual = await target.ProcessClusterStatusAsync(cluster);

            Assert.IsTrue(calledActual);
            Assert.AreEqual(ClusterStatus.Deleting, actual.Status);
        }

        public async Task ProcessDeletingSuccessful()
        {
            MockReliableStateManager stateManager = new MockReliableStateManager();
            MockClusterOperator clusterOperator = new MockClusterOperator()
            {
                GetClusterStatusAsyncFunc = domain => Task.FromResult(ClusterOperationStatus.ClusterNotFound)
            };

            ClusterConfig config = new ClusterConfig();
            ClusterService target = new ClusterService(
                clusterOperator,
                null,
                new MockApplicationDeployService(),
                stateManager,
                this.CreateServiceParameters(),
                config);

            Cluster cluster = this.CreateCluster(ClusterStatus.Deleting);

            Cluster actual = await target.ProcessClusterStatusAsync(cluster);

            Assert.AreEqual(ClusterStatus.Deleted, actual.Status);
        }

        public async Task ProcessCreatingClusterFailed()
        {
            MockReliableStateManager stateManager = new MockReliableStateManager();
            MockClusterOperator clusterOperator = new MockClusterOperator()
            {
                GetClusterStatusAsyncFunc = name => Task.FromResult(ClusterOperationStatus.CreateFailed)
            };

            ClusterConfig config = new ClusterConfig();
            ClusterService target = new ClusterService(
                clusterOperator,
                null,
                new MockApplicationDeployService(),
                stateManager,
                this.CreateServiceParameters(),
                config);
            Cluster cluster = this.CreateCluster(ClusterStatus.Creating);

            Cluster actual = await target.ProcessClusterStatusAsync(cluster);

            Assert.AreEqual(ClusterStatus.Remove, actual.Status);
            Assert.AreEqual(0, actual.Ports.Count());
            Assert.AreEqual(0, actual.Users.Count());
        }

        public async Task ProcessCreatingClusterSuccess()
        {
            MockReliableStateManager stateManager = new MockReliableStateManager();
            MockClusterOperator clusterOperator = new MockClusterOperator()
            {
                GetClusterStatusAsyncFunc = name => Task.FromResult(ClusterOperationStatus.Ready)
            };

            ClusterConfig config = new ClusterConfig();
            ClusterService target = new ClusterService(
                clusterOperator,
                null,
                new MockApplicationDeployService(),
                stateManager,
                this.CreateServiceParameters(),
                config);
            Cluster cluster = this.CreateCluster(ClusterStatus.Creating);

            Cluster actual = await target.ProcessClusterStatusAsync(cluster);

            Assert.AreEqual(ClusterStatus.Ready, actual.Status);
            Assert.IsTrue(actual.CreatedOn.ToUniversalTime() <= DateTimeOffset.UtcNow);
            actual.Ports.SequenceEqual(await clusterOperator.GetClusterPortsAsync(""));
        }

        public async Task ProcessClustersAsyncSaveChanges()
        {
            int key = 1;
            string nameTemplate = "Test:{0}";
            MockReliableStateManager stateManager = new MockReliableStateManager();
            ClusterConfig config = new ClusterConfig();
            MockClusterOperator clusterOperator = new MockClusterOperator()
            {
                CreateClusterAsyncFunc = name => { return Task.FromResult(String.Format(nameTemplate, name)); }
            };

            IReliableDictionary<int, Cluster> dictionary =
                await stateManager.GetOrAddAsync<IReliableDictionary<int, Cluster>>(ClusterService.ClusterDictionaryName);

            Cluster original = this.CreateCluster(ClusterStatus.New);
            using (ITransaction tx = stateManager.CreateTransaction())
            {
                await dictionary.SetAsync(tx, key, original);
            }

            ClusterService target = new ClusterService(
                clusterOperator,
                null,
                new MockApplicationDeployService(),
                stateManager,
                this.CreateServiceParameters(),
                config);

            await target.ProcessClustersAsync();

            using (ITransaction tx = stateManager.CreateTransaction())
            {
                Cluster actual = (await dictionary.TryGetValueAsync(tx, key)).Value;

                Assert.AreNotEqual(original, actual);
            }
        }

		private void ReconfigurePool(ClusterConfig config)
		{
			// kill the timer first
			this.isTimerActive = false;
			if (this.resurrectTimer != null)
				this.resurrectTimer.Change(Timeout.Infinite, Timeout.Infinite);

			if (config == null)
			{
				if (log.IsInfoEnabled) log.Info("Config is empty, all nodes are down.");

				Interlocked.Exchange(ref this.state, InternalState.Empty);

				return;
			}

			// these should be disposed after we've been reinitialized
			var oldNodes = this.state == null ? null : this.state.CurrentNodes;

			// default bucket does not require authentication
			// membase 1.6 tells us if a bucket needs authentication,
			// so let's try to use the config's password
			var password = config.authType == "sasl"
										? config.saslPassword
										: this.bucketPassword;

			var authenticator = this.bucketName == null
						   ? null
						   : new PlainTextAuthenticator(null, this.bucketName, password);

			try
			{
				var state = (config == null || config.vBucketServerMap == null)
								? this.InitBasic(config, authenticator)
								: this.InitVBucket(config, authenticator);

				var nodes = state.CurrentNodes;

				state.Locator.Initialize(nodes);

				// we need to subscribe the failed event, 
				// so we can periodically check the dead 
				// nodes, since we do not get a config 
				// update every time a node dies
				for (var i = 0; i < nodes.Length; i++) nodes[i].Failed += this.NodeFail;

				Interlocked.Exchange(ref this.state, state);
			}
			catch (Exception e)
			{
				log.Error("Failed to initialize the pool.", e);

				Interlocked.Exchange(ref this.state, InternalState.Empty);
			}

			// kill the old nodes
			if (oldNodes != null)
				for (var i = 0; i < oldNodes.Length; i++)
					try
					{
						oldNodes[i].Failed -= this.NodeFail;
						oldNodes[i].Dispose();
					}
					catch { }
		}