Add reset method of grain.

Author: u-less

src/Ray.Core.Abstractions/Observer/IObserver.cs

@@ -8,5 +8,10 @@ public interface IObserver : IVersion
     {
         Task OnNext(Immutable<byte[]> bytes);
         Task OnNext(Immutable<List<byte[]>> items);
+        /// <summary>
+        /// 重置状态
+        /// </summary>
+        /// <returns></returns>
+        Task Reset();
     }
 }

src/Ray.Core/Core/Abstractions/IObserverUnit.cs

@@ -12,6 +12,12 @@ public interface IObserverUnit<PrimaryKey> : IGrainID
         /// <returns></returns>
         List<string> GetGroups();
         Task<long[]> GetAndSaveVersion(PrimaryKey primaryKey, long srcVersion);
+        /// <summary>
+        /// 重置Grain
+        /// </summary>
+        /// <param name="primaryKey">重置Grain</param>
+        /// <returns></returns>
+        Task Reset(PrimaryKey primaryKey);
         List<Func<byte[], Task>> GetEventHandlers(string observerGroup);
         List<Func<byte[], Task>> GetAllEventHandlers();
         List<Func<List<byte[]>, Task>> GetBatchEventHandlers(string observerGroup);

src/Ray.Core/Core/Grains/ObserverGrain.cs

@@ -667,5 +667,11 @@ protected virtual async ValueTask SaveSnapshotAsync(bool force = false)
                 }
             }
         }
+        public virtual async Task Reset()
+        {
+            await ObserverSnapshotStorage.Delete(GrainId);
+            UnprocessedEventList.Clear();
+            await ReadSnapshotAsync();
+        }
     }
 }

src/Ray.Core/Core/Grains/RayGrain.cs

@@ -395,7 +395,6 @@ protected async Task Over(OverType overType)
             {
                 await ArchiveStorage.Over(Snapshot.Base.StateId, true);
             }
-
         }
         private async Task DeleteArchive(string briefId)
         {
@@ -443,6 +442,12 @@ protected async ValueTask DeleteSnapshot()
                 SnapshotEventVersion = 0;
             }
         }
+        protected async Task Reset()
+        {
+            await Over(OverType.DeleteAll);
+            await ReadSnapshotAsync();
+            await ObserverUnit.Reset(Snapshot.Base.StateId);
+        }
         protected virtual async Task<bool> RaiseEvent(IEvent @event, EventUID eUID = null)
         {
             if (Snapshot.Base.IsOver)

src/Ray.Core/Core/Grains/ShadowGrain.cs

@@ -343,5 +343,9 @@ protected virtual ValueTask OnEventDelivered(FullyEvent<PrimaryKey> @event)
             }
             return Consts.ValueTaskDone;
         }
+        public virtual Task Reset()
+        {
+            return ReadSnapshotAsync();
+        }
     }
 }

src/Ray.Core/Core/Observer/ObserverUnit.cs

@@ -26,6 +26,7 @@ public class ObserverUnit<PrimaryKey> : IObserverUnit<PrimaryKey>
         readonly List<Func<byte[], Task>> eventHandlers = new List<Func<byte[], Task>>();
         readonly List<Func<List<byte[]>, Task>> batchEventHandlers = new List<Func<List<byte[]>, Task>>();
         readonly List<Func<PrimaryKey, long, Task<long>>> observerVersionHandlers = new List<Func<PrimaryKey, long, Task<long>>>();
+        readonly List<Func<PrimaryKey, Task>> observerResetHandlers = new List<Func<PrimaryKey, Task>>();
         protected ILogger Logger { get; private set; }
         public Type GrainType { get; }
 
@@ -46,6 +47,11 @@ public Task<long[]> GetAndSaveVersion(PrimaryKey primaryKey, long srcVersion)
         {
             return Task.WhenAll(observerVersionHandlers.Select(func => func(primaryKey, srcVersion)));
         }
+
+        public Task Reset(PrimaryKey primaryKey)
+        {
+            return Task.WhenAll(observerResetHandlers.Select(func => func(primaryKey)));
+        }
         public List<string> GetGroups() => eventHandlerGroups.Keys.ToList();
         public List<Func<byte[], Task>> GetAllEventHandlers()
         {
@@ -143,6 +149,7 @@ public void Observer(string group, Type observerType)
             eventHandlers.Add(EventHandler);
             batchEventHandlers.Add(BatchEventHandler);
             observerVersionHandlers.Add((actorId, version) => GetObserver(observerType, actorId).GetAndSaveVersion(version));
+            observerResetHandlers.Add((actorId) => GetObserver(observerType, actorId).Reset());
             //内部函数
             Task EventHandler(byte[] bytes)
             {