[Deprecated] Extract streaming from GAgentBase.

samples/ArtifactGAgent/ArtifactGAgent.Silo/Program.cs

@@ -17,7 +17,7 @@
             .AddLogStorageBasedLogConsistencyProvider("LogStorage")
             .UseLocalhostClustering()
             .UseAevatar()
-            .ConfigureLogging(logging => logging.AddConsole());
+            .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Information).AddConsole());
     })
     .UseConsoleLifetime();
 

samples/PluginGAgent/PluginGAgent.Grains/CommanderGAgent.cs

@@ -1,6 +1,5 @@
 using Aevatar.Core;
 using Aevatar.Core.Abstractions;
-using Microsoft.Extensions.Logging;
 
 namespace PluginGAgent.Grains;
 

samples/PluginGAgent/PluginGAgent.Grains/WorkerGAgent.cs

@@ -6,16 +6,10 @@ namespace PluginGAgent.Grains;
 
 [GenerateSerializer]
 
-public class WorkerGAgentState : StateBase
-{
-
-}
+public class WorkerGAgentState : StateBase;
 
 [GenerateSerializer]
-public class WorkerStateLogEvent : StateLogEventBase<WorkerStateLogEvent>
-{
-
-}
+public class WorkerStateLogEvent : StateLogEventBase<WorkerStateLogEvent>;
 
 [GAgent("worker", "pluginTest")]
 public class WorkerGAgent : GAgentBase<WorkerGAgentState, WorkerStateLogEvent>

samples/PluginGAgent/PluginGAgent.Silo/Extensions/OrleansHostExtension.cs

@@ -4,6 +4,7 @@
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using MongoDB.Driver;
+using Orleans.Configuration;
 
 namespace PluginGAgent.Silo.Extensions;
 
@@ -30,6 +31,11 @@ public static IHostBuilder UseOrleansConfiguration(this IHostBuilder hostBuilder
                     })
                     .AddMemoryStreams("Aevatar")
                     .UseAevatarPermissionManagement()
+                    .Configure<ClusterOptions>(options =>
+                    {
+                        options.ClusterId = "default";
+                        options.ServiceId = "testServiceId";
+                    })
                     .UseAevatar()
                     ;
             })

samples/PluginGAgent/PluginGAgent.Silo/PluginGAgentTestHostedService.cs

@@ -24,17 +24,6 @@ public PluginGAgentTestHostedService(
     public async Task StartAsync(CancellationToken cancellationToken)
     {
         await _application.InitializeAsync(_serviceProvider);
-        var tenantId = "test".ToGuid();
-        var pluginManager = _serviceProvider.GetRequiredService<IPluginGAgentManager>();
-        var plugins = await pluginManager.GetPluginAssembliesAsync(tenantId);
-        Console.WriteLine("Plugins:");
-        foreach (var plugin in plugins)
-        {
-            Console.WriteLine(plugin.FullName);
-        }
-        var permissionManager = _serviceProvider.GetRequiredService<IPermissionManager>();
-        var userId = "TestUser".ToGuid().ToString();
-        await permissionManager.SetAsync("DoSomething", "User", userId, true);
     }
 
     public Task StopAsync(CancellationToken cancellationToken)

samples/PluginGAgent/PluginGAgent.Silo/PluginGAgentTestModule.cs

@@ -1,23 +1,22 @@
-using System.Reflection;
+using Aevatar;
 using Aevatar.Core.Abstractions;
 using Aevatar.PermissionManagement;
 using Microsoft.Extensions.DependencyInjection;
-using Orleans.Serialization;
 using Serilog;
 using Volo.Abp.AspNetCore.Serilog;
 using Volo.Abp.Autofac;
 using Volo.Abp.AutoMapper;
 using Volo.Abp.Modularity;
 using Volo.Abp.PermissionManagement;
-using Volo.Abp.PermissionManagement.MongoDB;
 
 namespace PluginGAgent.Silo;
 
 [DependsOn(
     typeof(AbpAspNetCoreSerilogModule),
     typeof(AbpAutofacModule),
     typeof(AbpAutoMapperModule),
-    typeof(AevatarPermissionManagementModule)
+    // typeof(AevatarPermissionManagementModule),
+    typeof(AevatarModule)
 )]
 public class PluginGAgentTestModule : AbpModule
 {
@@ -29,5 +28,9 @@ public override void ConfigureServices(ServiceConfigurationContext context)
             true, writeToProviders: true);
         context.Services.AddHttpClient();
         context.Services.AddSingleton<IEventDispatcher, DefaultEventDispatcher>();
+        // Configure<PermissionManagementOptions>(options =>
+        // {
+        //     options.IsDynamicPermissionStoreEnabled = true;
+        // });
     }
 }

samples/PluginGAgent/PluginGAgent.Silo/appsettings.json

@@ -33,10 +33,12 @@
     ]
   },
   "Plugins": {
-    "TenantId": "cd6b8f09214673d3cade4e832627b4f6"
+    "TenantId": "cd6b8f09214673d3cade4e832627b4f6",
+    "ConnectionString": "mongodb://localhost:27017/AevatarDb"
   },
   "ConnectionStrings": {
-    "Default": "mongodb://localhost:27017/Aevatar"
+    "Default": "mongodb://localhost:27017/Aevatar",
+    "Orleans": "mongodb://localhost:27017/AevatarDb"
   },
   "Aevatar": {
     "StreamNamespace": "Aevatar"

src/Aevatar.Core.Abstractions/IStateAgent.cs

@@ -1,4 +1,5 @@
 using Orleans.Concurrency;
+using Orleans.Streams;
 
 namespace Aevatar.Core.Abstractions;
 
@@ -79,6 +80,18 @@ public interface IGAgent : IGrainWithGuidKey
     /// <param name="configuration"></param>
     /// <returns></returns>
     Task ConfigAsync(ConfigurationBase configuration);
+
+    /// <summary>
+    /// Get GAgentAsyncObserver
+    /// </summary>
+    /// <returns></returns>
+    Task<IAsyncObserver<EventWrapperBase>> GetGAgentAsyncObserverAsync();
+
+    /// <summary>
+    /// Resume subscription of parent's stream.
+    /// </summary>
+    /// <returns></returns>
+    Task ResumeSubscriptionAsync(IAsyncStream<EventWrapperBase> stream);
 }
 
 public interface IStateGAgent<TState> : IGAgent

src/Aevatar.Core.Abstractions/Projections/IProjectionGrain.cs

@@ -1,7 +1,6 @@
 namespace Aevatar.Core.Abstractions.Projections;
 
-public interface IProjectionGrain<TState> : IGrainWithGuidKey
-    where TState : StateBase, new()
+public interface IProjectionGrain : IGrainWithStringKey
 {
     Task ActivateAsync();
 }

src/Aevatar.Core/AevatarGAgentConstants.cs

@@ -6,4 +6,5 @@ public static class AevatarGAgentConstants
     public const string StateHandlerDefaultMethodName = "HandleStateAsync";
     public const string ConfigDefaultMethodName = "PerformConfigAsync";
     public const string ForwardEventMethodName = "ForwardEventAsync";
+    public const int EventDispatcherMaxBatchSize = 20;
 }

src/Aevatar.Core/GAgentBase.Publish.cs

@@ -85,7 +85,7 @@ private async Task SendEventUpwardsAsync<T>(EventWrapper<T> eventWrapper) where
 
         try
         {
-            var stream = GetEventBaseStream(State.Parent.Value);
+            var stream = GetEventBaseStreamForOwn(State.Parent.Value);
             await stream.OnNextAsync(eventWrapper);
         }
         catch (Exception ex)
@@ -102,7 +102,7 @@ private async Task SendEventToSelfAsync<T>(EventWrapper<T> eventWrapper) where T
             $"{GrainId.ToString()} is sending event to self: {JsonConvert.SerializeObject(eventWrapper)}");
         try
         {
-            var streamOfThisGAgent = GetEventBaseStream(GrainId);
+            var streamOfThisGAgent = GetEventBaseStreamForOwn(GrainId);
             await streamOfThisGAgent.OnNextAsync(eventWrapper);
         }
         catch (Exception ex)
@@ -124,10 +124,8 @@ private async Task SendEventDownwardsAsync<T>(EventWrapper<T> eventWrapper) wher
 
         try
         {
-            foreach (var stream in State.Children.Select(GetEventBaseStream))
-            {
-                await stream.OnNextAsync(eventWrapper);
-            }
+            var streamForChildren = GetEventBaseStreamForChildren(GrainId);
+            await streamForChildren.OnNextAsync(eventWrapper);
         }
         catch (Exception ex)
         {

src/Aevatar.Core/GAgentBase.cs

@@ -1,4 +1,3 @@
-using System.Collections.Concurrent;
 using Aevatar.Core.Abstractions;
 using Aevatar.Core.Abstractions.Projections;
 using Microsoft.Extensions.DependencyInjection;
@@ -45,7 +44,11 @@ public abstract partial class
     private Lazy<IStreamProvider> LazyStreamProvider => new(()
         => this.GetStreamProvider(AevatarCoreConstants.StreamProvider));
 
+    private Lazy<IGAgentFactory> LazyGAgentFactory => new(()
+        => ServiceProvider.GetRequiredService<IGAgentFactory>());
+
     protected IStreamProvider StreamProvider => LazyStreamProvider.Value;
+    protected IGAgentFactory GAgentFactory => LazyGAgentFactory.Value;
 
     public ILogger Logger { get; set; } = NullLogger.Instance;
 
@@ -74,12 +77,16 @@ public async Task RegisterAsync(IGAgent gAgent)
 
     public async Task SubscribeToAsync(IGAgent gAgent)
     {
+        var parentGrainId = gAgent.GetGrainId();
+        var parentStream = GetEventBaseStreamForChildren(parentGrainId);
+        var asyncObserver = new GAgentAsyncObserver(_observers);
+        await ResumeOrSubscribeAsync(parentStream, asyncObserver);
         await SetParentAsync(gAgent.GetGrainId());
     }
 
-    public Task UnsubscribeFromAsync(IGAgent gAgent)
+    public async Task UnsubscribeFromAsync(IGAgent gAgent)
     {
-        return ClearParentAsync(gAgent.GetGrainId());
+        await ClearParentAsync(gAgent.GetGrainId());
     }
 
     public async Task UnregisterAsync(IGAgent gAgent)
@@ -89,7 +96,7 @@ public async Task UnregisterAsync(IGAgent gAgent)
         await OnUnregisterAgentAsync(gAgent.GetGrainId());
     }
 
-    public virtual Task<List<Type>?> GetAllSubscribedEventsAsync(bool includeBaseHandlers = false)
+    public async virtual Task<List<Type>?> GetAllSubscribedEventsAsync(bool includeBaseHandlers = false)
     {
         var eventHandlerMethods = GetEventHandlerMethods(GetType());
         eventHandlerMethods = eventHandlerMethods.Where(m =>
@@ -101,7 +108,7 @@ public async Task UnregisterAsync(IGAgent gAgent)
             handlingTypes = handlingTypes.Where(t => t != typeof(RequestAllSubscriptionsEvent));
         }
 
-        return Task.FromResult(handlingTypes.ToList())!;
+        return handlingTypes.ToList();
     }
 
     public Task<List<GrainId>> GetChildrenAsync()
@@ -127,6 +134,18 @@ public async Task ConfigAsync(ConfigurationBase configuration)
         }
     }
 
+    public async Task<IAsyncObserver<EventWrapperBase>> GetGAgentAsyncObserverAsync()
+    {
+        var asyncObserver = new GAgentAsyncObserver(_observers);
+        return asyncObserver;
+    }
+
+    public async Task ResumeSubscriptionAsync(IAsyncStream<EventWrapperBase> stream)
+    {
+        var asyncObserver = new GAgentAsyncObserver(_observers);
+        await ResumeOrSubscribeAsync(stream, asyncObserver);
+    }
+
     protected virtual Task PerformConfigAsync(TConfiguration configuration)
     {
         return Task.CompletedTask;
@@ -238,21 +257,35 @@ private async Task BaseOnActivateAsync(CancellationToken cancellationToken)
         var initTasks = new[]
         {
             InitializeOrResumeEventBaseStreamAsync(),
+            ResumeEventBaseStreamForChildrenAsync(),
             ActivateProjectionGrainAsync()
         };
         await Task.WhenAll(initTasks);
     }
 
     private async Task InitializeOrResumeEventBaseStreamAsync()
     {
-        var streamOfThisGAgent = GetEventBaseStream(this.GetGrainId());
+        var streamOfThisGAgent = GetEventBaseStreamForOwn(this.GetGrainId());
         var asyncObserver = new GAgentAsyncObserver(_observers);
         await ResumeOrSubscribeAsync(streamOfThisGAgent, asyncObserver);
     }
 
+    private async Task ResumeEventBaseStreamForChildrenAsync()
+    {
+        var streamForChildren = GetEventBaseStreamForChildren(this.GetGrainId());
+        var tasks = new List<Task>();
+        foreach (var childGrainId in State.Children)
+        {
+            var child = await GAgentFactory.GetGAgentAsync(childGrainId);
+            tasks.Add(child.ResumeSubscriptionAsync(streamForChildren));
+        }
+
+        await Task.WhenAll(tasks);
+    }
+
     private async Task ActivateProjectionGrainAsync()
     {
-        var projectionGrain = GrainFactory.GetGrain<IProjectionGrain<TState>>(Guid.Empty);
+        var projectionGrain = GrainFactory.GetGrain<IProjectionGrain>(typeof(TState).FullName);
         await projectionGrain.ActivateAsync();
     }
 
@@ -330,16 +363,21 @@ protected virtual Task HandleRaiseEventAsync()
         return Task.CompletedTask;
     }
 
-    private IAsyncStream<EventWrapperBase> GetEventBaseStream(GrainId grainId)
+    private IAsyncStream<EventWrapperBase> GetEventBaseStreamForOwn(GrainId grainId)
     {
         var grainIdString = grainId.ToString();
-        var streamId = StreamId.Create(AevatarOptions!.StreamNamespace, grainIdString);
-        return StreamProvider.GetStream<EventWrapperBase>(streamId);
+        return GetEventBaseStream(grainIdString);
     }
 
-    private IAsyncStream<StateWrapper<TState>> GetStateProjectionStream()
+    private IAsyncStream<EventWrapperBase> GetEventBaseStreamForChildren(GrainId grainId)
     {
-        var streamId = StreamId.Create(AevatarOptions!.StreamNamespace, typeof(StateWrapper<TState>).FullName!);
-        return StreamProvider.GetStream<StateWrapper<TState>>(streamId);
+        var grainIdString = $"{grainId.ToString()}/Children";
+        return GetEventBaseStream(grainIdString);
+    }
+
+    private IAsyncStream<EventWrapperBase> GetEventBaseStream(string streamKey)
+    {
+        var streamId = StreamId.Create(AevatarOptions!.StreamNamespace, streamKey);
+        return StreamProvider.GetStream<EventWrapperBase>(streamId);
     }
 }