[On Hold] Feature steam improvements and Extract streaming from GAgentBase.

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()
                     ;
             })

src/Aevatar.Core.Abstractions/AevatarCoreConstants.cs

@@ -3,5 +3,6 @@ namespace Aevatar.Core.Abstractions;
 public class AevatarCoreConstants
 {
     public const string StreamProvider = "Aevatar";
+    public const string DefaultStreamNamespace = "Aevatar";
     public const char GAgentNamespaceSeparator = '.';
 }

src/Aevatar.Core.Abstractions/AevatarCoreStreamConfig.cs

@@ -0,0 +1,14 @@
+namespace Aevatar.Core.Abstractions;
+
+public static class AevatarCoreStreamConfig
+{
+    public static string Prefix { get; private set; } = AevatarCoreConstants.DefaultStreamNamespace;
+
+    public static void Initialize(string? prefix)
+    {
+        if (prefix != null)
+        {
+            Prefix = prefix;
+        }
+    }
+}

src/Aevatar.Core.Abstractions/EventPublish/IEventPubChildrenGroupGrain.cs

@@ -0,0 +1,16 @@
+namespace Aevatar.Core.Abstractions.EventPublish;
+
+using Orleans.Concurrency;
+
+public interface IEventPubChildrenGroupGrain : IGrainWithIntegerCompoundKey
+{
+    Task DownwardsEventAsync(EventWrapperBase eventWrapper);
+    Task UpwardsEventAsync(EventWrapperBase eventWrapper);
+    Task AddChildAsync(GrainId childGrainId);
+    Task AddManyChildAsync(List<GrainId> childrenGrainIds);
+    Task RemoveChildAsync(GrainId childId);
+    [ReadOnly]
+    Task<List<GrainId>> GetChildrenAsync();
+    [ReadOnly]
+    Task<int> GetChildrenCountAsync();
+}

src/Aevatar.Core.Abstractions/EventPublish/IStreamCoordinatorGrain.cs

@@ -0,0 +1,22 @@
+using Orleans.Concurrency;
+
+namespace Aevatar.Core.Abstractions.EventPublish;
+
+public interface IStreamCoordinatorGrain : IGrainWithStringKey
+{
+    Task<bool> SetParentAsync(GrainId parentGrainId);
+    Task ClearParentAsync(GrainId parentGrainId);
+    Task SetGroupIndexAsync(int groupIndex);
+    [ReadOnly]
+    Task<int> GetGroupIndexAsync();
+    [ReadOnly]
+    Task<GrainId> GetParentAsync();
+    Task<int> RegisterChildAsync(GrainId childGrainId);
+    Task<int> RegisterManyChildAsync(List<GrainId> childrenGrainIds);
+    Task UnregisterChildAsync(GrainId childGrainId);
+    [ReadOnly]
+    Task<List<GrainId>> GetChildrenAsync();
+    Task PublishEventAsync(EventWrapperBase eventWrapper);
+    Task DownwardsEventAsync(EventWrapperBase eventWrapper);
+    Task UpwardsEventAsync(EventWrapperBase eventWrapper);
+}

src/Aevatar.Core.Abstractions/IStateAgent.cs

@@ -1,4 +1,5 @@
 using Orleans.Concurrency;
+using Orleans.Streams;
 
 namespace Aevatar.Core.Abstractions;
 
@@ -80,6 +81,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/StateBase.cs

@@ -6,6 +6,7 @@ public abstract class StateBase
     [Id(0)] public List<GrainId> Children { get; set; } = [];
     [Id(1)] public GrainId? Parent { get; set; }
     [Id(2)] public string? GAgentCreator { get; set; }
+    [Id(3)] public GrainId StreamCoordinatorGrainId { get; set; }
 
     public void Apply(StateLogEventBase @stateLogEvent)
     {

src/Aevatar.Core/AevatarGAgentConstants.cs

@@ -7,4 +7,5 @@ public static class AevatarGAgentConstants
     public const string ConfigDefaultMethodName = "PerformConfigAsync";
     public const string ForwardEventMethodName = "ForwardEventAsync";
     public const int MaxSyncWorkConcurrency = 4;
-}
+    public const int MaxChildrenPerGroup = 1500;
+}

src/Aevatar.Core/EventPublish/EventPubChildrenGroupGrain.cs

@@ -0,0 +1,141 @@
+using Aevatar.Core.Abstractions;
+using Aevatar.Core.Abstractions.EventPublish;
+using Aevatar.Core.Extensions;
+using Microsoft.Extensions.Logging;
+using Newtonsoft.Json;
+using Orleans.EventSourcing;
+using Orleans.Providers;
+using Orleans.Streams;
+using Orleans.Concurrency;
+
+namespace Aevatar.Core.EventPublish;
+
+[GenerateSerializer]
+public class EventPubChildrenGroupState : StateBase
+{
+    [Id(0)] public List<GrainId> Children { get; set; } = new();
+    [Id(1)] public int ChildrenCount { get; set; }
+}
+
+[GenerateSerializer]
+public class EventPubChildrenGroupStateLogEvent : StateLogEventBase<EventPubChildrenGroupStateLogEvent>;
+
+[StorageProvider(ProviderName = "PubSubStore")]
+[LogConsistencyProvider(ProviderName = "LogStorage")]
+[Reentrant]
+public class EventPubChildrenGroupGrain :
+    JournaledGrain<EventPubChildrenGroupState, StateLogEventBase<EventPubChildrenGroupStateLogEvent>>,
+    IEventPubChildrenGroupGrain
+{
+    private readonly ILogger<EventPubChildrenGroupGrain> _logger;
+    private readonly IStreamProvider _streamProvider;
+
+    public EventPubChildrenGroupGrain(ILogger<EventPubChildrenGroupGrain> logger)
+    {
+        _logger = logger;
+        _streamProvider = this.GetStreamProvider(AevatarCoreConstants.StreamProvider);
+    }
+
+    public async Task DownwardsEventAsync(EventWrapperBase eventWrapper)
+    {
+        _logger.LogInformation("[{grainId}] Starting DownwardsEventAsync for Event: {eventWrapper}",
+            this.GetGrainId().ToString(),
+            JsonConvert.SerializeObject(eventWrapper));
+
+        foreach (var childGrainId in State.Children)
+        {
+            var stream = _streamProvider.GetEventWrapperBaseStream(childGrainId.ToString());
+            await stream.OnNextAsync(eventWrapper);
+        }
+    }
+
+    public async Task UpwardsEventAsync(EventWrapperBase eventWrapper)
+    {
+        _logger.LogInformation("[{grainId}] Starting UpwardsEventAsync for Event: {eventWrapper}",
+            this.GetGrainId().ToString(),
+            JsonConvert.SerializeObject(eventWrapper));
+
+        var stream = _streamProvider.GetEventWrapperBaseStream(this.GetPrimaryKeyString());
+        await stream.OnNextAsync(eventWrapper);
+    }
+
+    public async Task AddChildAsync(GrainId childGrainId)
+    {
+        _logger.LogInformation("[{grainId}] Adding child GrainId: {GrainId}", this.GetGrainId().ToString(),
+            childGrainId);
+
+        RaiseEvent(new AddChildStateLogEvent { ChildId = childGrainId });
+        await ConfirmEvents();
+    }
+
+    public async Task AddManyChildAsync(List<GrainId> childrenGrainIds)
+    {
+        _logger.LogInformation("[{grainId}] Adding multiple children GrainIds: {childrenGrainIds}",
+            this.GetGrainId().ToString(), string.Join(", ", childrenGrainIds.ToString()));
+
+        base.RaiseEvent(new AddChildManyStateLogEvent
+        {
+            ChildrenIds = childrenGrainIds
+        });
+        await ConfirmEvents();
+    }
+
+    public async Task RemoveChildAsync(GrainId childId)
+    {
+        _logger.LogInformation("[{grainId}] Removing child GrainId: {GrainId}", this.GetGrainId().ToString(),
+            childId);
+
+        RaiseEvent(new RemoveChildStateLogEvent { ChildId = childId });
+        await ConfirmEvents();
+    }
+
+    public Task<List<GrainId>> GetChildrenAsync()
+    {
+        return Task.FromResult(State.Children);
+    }
+
+    public Task<int> GetChildrenCountAsync()
+    {
+        return Task.FromResult(State.ChildrenCount);
+    }
+
+    protected override void TransitionState(EventPubChildrenGroupState state,
+        StateLogEventBase<EventPubChildrenGroupStateLogEvent> @event)
+    {
+        switch (@event)
+        {
+            case AddChildStateLogEvent addChildEvent:
+                state.Children.Add(addChildEvent.ChildId);
+                state.ChildrenCount++;
+                break;
+            case AddChildManyStateLogEvent addChildManyEvent:
+                state.Children.AddRange(addChildManyEvent.ChildrenIds);
+                state.ChildrenCount += addChildManyEvent.ChildrenIds.Count;
+                break;
+            case RemoveChildStateLogEvent removeChildEvent:
+                state.Children.Remove(removeChildEvent.ChildId);
+                state.ChildrenCount--;
+                break;
+        }
+
+        base.TransitionState(state, @event);
+    }
+
+    [GenerateSerializer]
+    public class AddChildStateLogEvent : StateLogEventBase<EventPubChildrenGroupStateLogEvent>
+    {
+        [Id(0)] public required GrainId ChildId { get; set; }
+    }
+
+    [GenerateSerializer]
+    public class AddChildManyStateLogEvent : StateLogEventBase<EventPubChildrenGroupStateLogEvent>
+    {
+        [Id(0)] public required List<GrainId> ChildrenIds { get; set; }
+    }
+
+    [GenerateSerializer]
+    public class RemoveChildStateLogEvent : StateLogEventBase<EventPubChildrenGroupStateLogEvent>
+    {
+        [Id(0)] public required GrainId ChildId { get; set; }
+    }
+}