public async Task WriteStateAsync(string grainType, GrainReference grainReference, IGrainState grainState)
        {
            try
            {
                var collection = await this.EnsureCollection(grainType);
                var documents = await this.Client.ReadDocumentFeedAsync(collection.DocumentsLink);
                var documentId = grainReference.ToKeyString();

                var document = documents.Where(d => d.Id == documentId).FirstOrDefault();

                if(document != null)
                {
                    document.State = grainState.AsDictionary();
                    await this.Client.ReplaceDocumentAsync(document);
                }
                else
                {
                    await this.Client.CreateDocumentAsync(collection.DocumentsLink,
                        new GrainStateDocument { Id = documentId, State = grainState.AsDictionary() });
                }
            }
            catch (Exception ex)
            {
                Log.Error(0, "Error in WriteStateAsync", ex);
            }
        }

 public ReminderTableData ReadRows(GrainReference grainRef)
 {
     Dictionary<string, ReminderEntry> reminders;
     reminderTable.TryGetValue(grainRef, out reminders);
     return reminders == null ? new ReminderTableData() :
         new ReminderTableData(reminders.Values.ToList());
 }

 public string ToElasticValue(GrainReference gref)
 {
     ///"GrainReference=00000000000000000000000000000000060000006aa96326+abc"
     /// to
     /// "GrainReference^00000000000000000000000000000000060000006aa96326_abc"
     return gref.ToKeyString().Replace("=", "__").Replace("_", "+");
 }

		public async Task ReadStateAsync(string grainType, GrainReference grainReference, IGrainState grainState)
		{
			var tableResult = await _table.ExecuteAsync(TableOperation.Retrieve<DynamicTableEntity>(grainReference.ToKeyString(), grainType));
			if (tableResult.Result == null)
			{
				return;
			}
			var entity = tableResult.Result as DynamicTableEntity;

			var serializer = new JsonSerializer();
			using (var memoryStream = new MemoryStream())
			{
				foreach (var propertyName in entity.Properties.Keys.Where(p => p.StartsWith("d")).OrderBy(p => p))
				{
					var dataPart = entity.Properties[propertyName];
					await memoryStream.WriteAsync(dataPart.BinaryValue, 0, dataPart.BinaryValue.Length);
				}

				memoryStream.Position = 0;
				using (var bsonReader = new BsonReader(memoryStream))
				{
					var data = serializer.Deserialize<Dictionary<string, object>>(bsonReader);
					grainState.SetAll(data);
				}
			}
		}

        public async Task ReadStateAsync(string grainType, GrainReference grainId, IGrainState grainState)
        {
            try
            {
                var blobName = BlobStorageProvider.GetBlobName(grainType, grainId);
                var blob = container.GetBlockBlobReference(blobName);
                var text = await blob.DownloadTextAsync();
                if (string.IsNullOrWhiteSpace(text))
                {
                    return;
                }

                var data = JsonConvert.DeserializeObject(text, grainState.GetType());
                var dict = ((IGrainState)data).AsDictionary();
                grainState.SetAll(dict);
            }
            catch (StorageException ex)
            {
                ;
            }
            catch (Exception ex)
            {
                Log.Error(0, ex.ToString());
            }
        }

        public async Task ReadStateAsync(string grainType, GrainReference grainReference, IGrainState grainState)
        {
            if (!(grainState is IAggregateState))
                throw new NotAggregateStateException(grainState.GetType());

            var stream = this.GetStreamName(grainType, grainReference);

            var sliceStart = 0;
            StreamEventsSlice currentSlice;

            do
            {
                var sliceCount = sliceStart + ReadPageSize;

                currentSlice = await this.Connection.ReadStreamEventsForwardAsync(stream, sliceStart, sliceCount, true);

                if (currentSlice.Status == SliceReadStatus.StreamNotFound)
                    return;

                if (currentSlice.Status == SliceReadStatus.StreamDeleted)
                    throw new StreamDeletedException();

                sliceStart = currentSlice.NextEventNumber;

                foreach (var @event in currentSlice.Events)
                {
                    dynamic deserialisedEvent = DeserializeEvent(@event.Event);
                    StateTransformer.ApplyEvent(deserialisedEvent, grainState as IAggregateState);
                }

            } while (!currentSlice.IsEndOfStream);
        }

 public async Task<ReminderEntry> ReadRow(GrainReference grainRef, string reminderName)
 {
     using (var conn = new SqlConnection(connectionString))
     {
         conn.Open();
         return await ReadRowInternal(grainRef, reminderName, conn, null);
     }
 }

 public static string CreateIdFrom(GrainReference grainRef,string reminderName)
 {
     if (String.IsNullOrEmpty(reminderName))
         throw new ArgumentNullException("reminderName");
     if (grainRef==null)
         throw new ArgumentNullException("grainRef");
     return grainRef.ToKeyString() + "," + reminderName;
 }

 internal static void OnStorageActivate(IStorageProvider storage, string grainType, GrainReference grain, TimeSpan latency)
 {
     StorageActivateTotal.Increment();
     if (latency > TimeSpan.Zero)
     {
         StorageReadLatency.AddSample(latency);
     }
 }

 /// <summary>
 /// Reads persisted state from the backing store and deserializes it into the the target
 /// grain state object.
 /// </summary>
 /// <param name="grainType">A string holding the name of the grain class.</param>
 /// <param name="grainReference">Represents the long-lived identity of the grain.</param>
 /// <param name="grainState">A reference to an object to hold the persisted state of the grain.</param>
 /// <returns>Completion promise for this operation.</returns>
 public async Task ReadStateAsync(string grainType, GrainReference grainReference, IGrainState grainState)
 {
     if (DataManager == null) throw new ArgumentException("DataManager property not initialized");
     var entityData = await DataManager.Read(grainState.GetType().Name, grainReference.ToKeyString());
     if (entityData != null)
     {
         ConvertFromStorageFormat(grainState, entityData);
     }
 }

        public Task ClearStateAsync(string grainType, GrainReference grainReference, IGrainState grainState)
        {
            if (!(grainState is IAggregateState))
                throw new NotAggregateStateException(grainState.GetType());

            var state = grainState as IAggregateState;
            var stream = this.GetStreamName(grainType, grainReference);

            return this.Connection.DeleteStreamAsync(stream, state.Version);
        }

 /// <summary>
 /// Throws a storage exception if one has been added for the grain reference for reading.
 /// </summary>
 /// <param name="grainReference"></param>
 /// <returns></returns>
 public Task OnRead(GrainReference grainReference)
 {
     Exception exception;
     if (readFaults.TryGetValue(grainReference, out exception))
     {
         readFaults.Remove(grainReference);
         throw exception;
     }
     return TaskDone.Done;
 }

 private static ReminderEntry CreateReminder(GrainReference grainRef, string reminderName)
 {
     return new ReminderEntry
     {
         GrainRef = grainRef,
         Period = TimeSpan.FromMinutes(1),
         StartAt = DateTime.UtcNow.Add(TimeSpan.FromMinutes(1)),
         ReminderName = reminderName
     };
 }

 private static async Task<string> UpsertReminder(IReminderTable reminder, GrainReference grainRef, string reminderName, DateTime startAt, TimeSpan period)
 {
     var reminderRow = new ReminderEntry
                       {
                           GrainRef = grainRef,
                           Period = period,
                           StartAt = startAt,
                           ReminderName = reminderName
                       };
     return await reminder.UpsertRow(reminderRow);
 }

 internal static async Task<ElasticReminderEntry> Get(ElasticClient elastic, GrainReference grainRef, string reminderName)
 {
     var op = await elastic.GetAsync<ElasticReminderEntry>(CreateIdFrom(grainRef, reminderName));
     if (op.IsValid)
     {
         op.Source.ETag = op.Version;
         return op.Source;
     }
     else
         throw new ElasticsearchStorageException();
 }

 private IDictionary<string, object> GetLastState(string grainType, GrainReference grainReference)
 {
     lock (StateStore)
     {
         var keys = MakeGrainStateKeys(grainType, grainReference);
         var storedState = StateStore.ReadRow(keys);
         LastId = GetId(grainReference);
         LastState = storedState;
         return storedState;
     }
 }

 /// <summary>
 /// Remove a row from the table
 /// </summary>
 /// <param name="grainRef"></param>
 /// <param name="reminderName"></param>
 /// <param name="eTag"></param>
 /// <returns>true if a row with <paramref name="grainRef"/> and <paramref name="reminderName"/> existed and was removed successfully, false otherwise</returns>
 public Task<bool> RemoveRow(GrainReference grainRef, string reminderName, string eTag)
 {
     if (logger.IsVerbose) logger.Verbose("RemoveRow entry grainRef = {0}, reminderName = {1}, eTag = {2}", grainRef, reminderName, eTag);
     bool result = remTable.RemoveRow(grainRef, reminderName, eTag);
     if (result == false)
     {
         logger.Warn(ErrorCode.RS_Table_Remove, "RemoveRow failed for grainRef = {0}, ReminderName = {1}, eTag = {2}. Table now is: {3}",
             grainRef.ToDetailedString(), reminderName, eTag, remTable.ReadAll());
     }
     return Task.FromResult(result);
 }

        public async Task WriteStateAsync(string grainType, GrainReference grainReference, GrainState grainState)
        {
            var stateName = grainState.GetType().Name;
            var key = grainReference.ToKeyString();
            var id = string.Format(CultureInfo.InvariantCulture, "{0}/{1}", stateName, key);

            using (IAsyncDocumentSession session = this.documentStore.OpenAsyncSession())
            {
                await session.StoreAsync(grainState, id);
                await session.SaveChangesAsync();
            }
        }

        /// <summary>
        /// Createa a GrainIdentity from a grain reference and type name
        /// </summary>
        /// <param name="grainType"></param>
        /// <param name="grainReference"></param>
        /// <returns></returns>
        public static GrainIdentity FromGrainReference(string grainType, GrainReference grainReference)
        {
            Guard.NotNullOrEmpty(grainType, "grainType");
            Guard.NotNull(grainReference, "grainReference");

            return new GrainIdentity()
            {
                GrainType = grainType,
                ShardKey = (int)grainReference.GetUniformHashCode(),
                GrainKey = grainReference.ToKeyString()
            };
        }

 public async Task WriteStateAsync(string grainType, GrainReference grainId, IGrainState grainState)
 {
     try
     {
         var blobName = BlobStorageProvider.GetBlobName(grainType, grainId);
         var storedData = JsonConvert.SerializeObject(grainState.AsDictionary());
         var blob = container.GetBlockBlobReference(blobName);
         await blob.UploadTextAsync(storedData);
     }
     catch (Exception ex)
     {
         Log.Error(0, ex.ToString());
     }
 }