Muhannad/xchange created on multi queue

SW.Bitween.Api/Data/BitweenDbContext.cs

@@ -82,6 +82,15 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                 sc.Property(i => i.Id).ValueGeneratedOnAdd();
                 sc.HasIndex(i => i.Code).IsUnique();
             });
+
+            modelBuilder.Entity<WorkGroup>(wg =>
+            {
+                wg.HasKey(i => i.Id);
+                wg.Property(i => i.Id).ValueGeneratedOnAdd();
+                wg.Property(p => p.BusMessageName).IsRequired().IsUnicode(false).HasMaxLength(100);
+                wg.Property(p => p.Options).StoreAsJson();
+
+            });
 
             modelBuilder.Entity<Partner>(b =>
             {
@@ -148,6 +157,7 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
                 b.HasOne<Subscription>().WithMany().HasForeignKey(p => p.AggregationForId).IsRequired(false)
                     .HasConstraintName("FK_Subscriptions_AggFor").OnDelete(DeleteBehavior.Restrict);
                 b.HasOne(i => i.Category).WithMany().HasForeignKey(i => i.CategoryId);
+                b.HasOne(i => i.WorkGroup).WithMany().HasForeignKey(i => i.WorkGroupId);
                 b.Property(p => p.MatchExpression).HasConversion(
                     domainObject =>
                         domainObject == null ? null : MatchSpecValueConverter.SerializeMatchSpec(domainObject),
@@ -302,8 +312,25 @@ async public override Task<int> SaveChangesAsync(CancellationToken cancellationT
             ChangeTracker.ApplyAuditValues(requestContext.GetNameIdentifier());
             //using var transaction = await Database.BeginTransactionAsync();
             var affectedRecords = await base.SaveChangesAsync(cancellationToken);
-            await ChangeTracker.PublishDomainEvents(publish);
-            //await transaction.CommitAsync();
+            //await ChangeTracker.PublishDomainEvents(publish);
+            var entitiesWithEvents = ChangeTracker.Entries<IGeneratesDomainEvents>()
+                .Select(e => e.Entity)
+                .Where(e => e.Events.Any())
+                .ToArray();
+
+            foreach (var entity in entitiesWithEvents)
+            {
+                var events = entity.Events.ToArray();
+                entity.Events.Clear();
+                foreach (var domainEvent in events)
+                    if (domainEvent is IHasWorkGroup hasWorkGroup)
+                        await publish.Publish(hasWorkGroup.GetBusMessageName(),
+                            JsonConvert.SerializeObject(new XchangeMessage { Id = hasWorkGroup.Id }));
+                    else
+                        await publish.Publish(domainEvent.GetType().Name, JsonConvert.SerializeObject(domainEvent));
+            }
+
+
             return affectedRecords;
         }
     }

SW.Bitween.Api/Domain/Subscription/Subscription.cs

@@ -14,28 +14,28 @@ public Subscription()
         }
 
         //receiving
-        public Subscription(string name, int documentId) : this(name, documentId, SubscriptionType.Receiving)
+        public Subscription(string name, int documentId) : this(WorkGroup.None, name, documentId, SubscriptionType.Receiving)
         {
             Inactive = true;
         }
 
         //aggregation
-        public Subscription(string name, int aggregationFor, int partnerId) : this(name, Document.AggregationDocumentId,
+        public Subscription(string name, int aggregationFor, int partnerId) : this(WorkGroup.None,name, Document.AggregationDocumentId,
             SubscriptionType.Aggregation, partnerId, aggregationFor)
         {
             Inactive = true;
         }
 
         //apiresult or filter
-        public Subscription(string name, int documentId, SubscriptionType type, int partnerId) : this(name, documentId,
+        public Subscription(string name, int documentId, SubscriptionType type, int partnerId) : this(WorkGroup.None,name, documentId,
             type, partnerId, null)
         {
             Inactive = true;
             if (!(type == SubscriptionType.ApiCall || type == SubscriptionType.Internal))
                 throw new ArgumentException();
         }
 
-        private Subscription(string name, int documentId, SubscriptionType type, int? partnerId = null,
+        private Subscription(WorkGroup workGroup, string name, int documentId, SubscriptionType type, int? partnerId = null,
             int? aggregationForId = null, bool temporary = false)
         {
             Inactive = true;
@@ -51,6 +51,7 @@ private Subscription(string name, int documentId, SubscriptionType type, int? pa
             ValidatorProperties = new Dictionary<string, string>();
             DocumentFilter = new Dictionary<string, string>();
             Temporary = temporary;
+            WorkGroup = workGroup;
         }
 
         public string Name { get; set; }
@@ -59,6 +60,8 @@ private Subscription(string name, int documentId, SubscriptionType type, int? pa
         public int? PartnerId { get; private set; }
         public int? CategoryId { get; set; }
         public SubscriptionCategory Category { get; set; }
+        public int? WorkGroupId { get; set; }
+        public WorkGroup WorkGroup { get; set; }
         public bool Temporary { get; private set; }
         public DateTime? PausedOn { get; private set; }
         public string ValidatorId { get; set; }

SW.Bitween.Api/Domain/WorkGroup/WorkGroup.cs

@@ -0,0 +1,24 @@
+using SW.Bitween.Domain;
+using SW.Bitween.Model;
+using SW.Bus.RabbitMqExtensions;
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Domain;
+
+public interface IWorkGroup
+{
+    string BusMessageName { get; }
+    string GetBusMessageName();
+    WorkGroupOptions Options { get; }
+}
+
+public class WorkGroup : BaseEntity,IWorkGroup
+{
+    public string Name { get; set; }
+    public string BusMessageName { get; set; }
+
+    public string GetBusMessageName() => $"{Id}{BusMessageName}";
+    //public string 
+    public static WorkGroup None => new() { BusMessageName = "Ungrouped"};
+    public WorkGroupOptions Options { get; set; }
+}

SW.Bitween.Api/Domain/Xchange/Xchange.cs

@@ -11,7 +11,7 @@ private Xchange()
         {
         }
 
-        public Xchange(int documentId, XchangeFile file, string[] references = null, SubscriptionType subscriptionType = SubscriptionType.Internal, string correlationId = null)
+        public Xchange(int documentId, IWorkGroup workGroup, XchangeFile file, string[] references = null, SubscriptionType subscriptionType = SubscriptionType.Internal, string correlationId = null)
         {
             Id = Guid.NewGuid().ToString("N"); 
             DocumentId = documentId;
@@ -34,11 +34,12 @@ public Xchange(int documentId, XchangeFile file, string[] references = null, Sub
             };
 
             xchangeEvent.Id = Id;
+            xchangeEvent.WorkGroup = workGroup ?? WorkGroup.None;
             Events.Add(xchangeEvent);
         }
 
         public Xchange(Subscription subscription, XchangeFile file, string[] references = null, string correlationId = null) : 
-            this(subscription.DocumentId, file, references, subscription.Type)
+            this(subscription.DocumentId, subscription.WorkGroup, file, references, subscription.Type)
         {
             SubscriptionId = subscription.Id;
             MapperId = subscription.MapperId;
@@ -51,8 +52,8 @@ public Xchange(Subscription subscription, XchangeFile file, string[] references
         }
 
         //retry xchange
-        public Xchange(Xchange xchange, XchangeFile file) : 
-            this(xchange.DocumentId, file, xchange.References)
+        public Xchange(Xchange xchange, XchangeFile file,IWorkGroup workGroup) : 
+            this(xchange.DocumentId,workGroup, file, xchange.References)
         {
             SubscriptionId = xchange.SubscriptionId;
             MapperId = xchange.MapperId;
@@ -65,7 +66,7 @@ public Xchange(Xchange xchange, XchangeFile file) :
         }
         //retry with reset subscription properties
         public Xchange(Subscription subscription, Xchange xchange, XchangeFile file) : 
-            this(xchange.DocumentId, file, xchange.References)
+            this(xchange.DocumentId,subscription.WorkGroup, file, xchange.References)
         {
             SubscriptionId = xchange.SubscriptionId;
             MapperId = subscription.MapperId;

SW.Bitween.Api/Domain/Xchange/XchangeCreatedEvent.cs

@@ -3,12 +3,18 @@
 
 namespace SW.Bitween.Domain
 {
-    internal abstract class XchangeCreatedEvent : BaseDomainEvent
+    internal class XchangeMessage
     {
         public string Id { get; set; }
-
     }
+    internal abstract class XchangeCreatedEvent : BaseDomainEvent,IHasWorkGroup
+    {
+        public string Id { get; set; }
+        public string GetBusMessageName()=> WorkGroup.GetBusMessageName();
 
+        public IWorkGroup WorkGroup { get; set; }
+
+    }
     internal class ApiXchangeCreatedEvent : XchangeCreatedEvent
     {
     }

SW.Bitween.Api/Domain/XchangeResult/XchangeResult.cs

@@ -9,7 +9,7 @@ private XchangeResult()
         {
         }
 
-        public XchangeResult(string xchangeId, XchangeFile outputFile, XchangeFile responseFile = null, string responseXchangeId = null, string exception = null)
+        public XchangeResult(string xchangeId,WorkGroup workGroup, XchangeFile outputFile, XchangeFile responseFile = null, string responseXchangeId = null, string exception = null)
         {
             Id = xchangeId;
             Success = exception == null;
@@ -40,7 +40,8 @@ public XchangeResult(string xchangeId, XchangeFile outputFile, XchangeFile respo
             {
                 Id = Id,
                 Success = Success,
-                ResponseBad = ResponseBad
+                ResponseBad = ResponseBad,
+                WorkGroup =  workGroup ?? WorkGroup.None,
             });
         }
 

SW.Bitween.Api/Domain/XchangeResult/XchangeResultCreatedEvent.cs

@@ -2,10 +2,12 @@
 
 namespace SW.Bitween.Domain
 {
-    public class XchangeResultCreatedEvent : BaseDomainEvent
+    public class XchangeResultCreatedEvent : BaseDomainEvent,IHasWorkGroup
     {
         public string Id { get; set; }
         public bool Success { get; set; }
         public bool ResponseBad { get; set; }
+        public IWorkGroup WorkGroup { get; set; } = Domain.WorkGroup.None;
+        public string GetBusMessageName()=>  $"{WorkGroup.GetBusMessageName()}{XchangeService.ResultQueueSuffix}";
     }
 }

SW.Bitween.Api/Extensions/InfolinkDbContextExtensions.cs

@@ -27,5 +27,9 @@ where partner.ApiCredentials.Any(cred => cred.Key == partnerKey)
 
             return (par, par.ApiCredentials.First(c => c.Key == partnerKey).Name);
         }
+
+        public static IQueryable<Subscription> Subscriptions(this BitweenDbContext dbContext) =>
+            dbContext.Set<Subscription>().Include(s => s.WorkGroup);
     }
+
 }

SW.Bitween.Api/Interfaces/IHasWorkGroup.cs

@@ -0,0 +1,7 @@
+namespace SW.Bitween;
+
+public interface IHasWorkGroup
+{
+    public string Id { get; }
+    string GetBusMessageName();
+}

SW.Bitween.Api/Interfaces/IInfolinkCache.cs

@@ -14,6 +14,9 @@ public interface IInfolinkCache
 
 
     void Revoke();
-    void BroadcastRevoke();
+    Task BroadcastRevoke();
 
+    Task<WorkGroup[]> ListWorkGroupsAsync();
+    Task<WorkGroup> WorkGroupByIdAsync(int workGroupId);
+    Task<WorkGroup> WorkGroupBySubscriptionIdAsync(int subscriptionId);
 }

SW.Bitween.Api/Resources/Subscriptions/Get.cs

@@ -52,6 +52,7 @@ public async Task<object> Handle(int key)
                     CategoryDescription = subscriber.Category?.Description,
                     CategoryCode = subscriber.Category?.Code,
                     CategoryId = subscriber.CategoryId,
+                    WorkGroupId =  subscriber.WorkGroupId,
                     Schedules = subscriber.Schedules.Select(s => new ScheduleView
                     {
                         Backwards = s.Backwards,

SW.Bitween.Api/Resources/Subscriptions/Search.cs

@@ -59,8 +59,10 @@ join document in _dbContext.Set<Document>() on subscriber.DocumentId equals docu
                     MatchExpression = subscriber.MatchExpression,
                     PartnerId = subscriber.PartnerId,
                     CategoryId = subscriber.CategoryId,
+                    WorkGroupId =  subscriber.WorkGroupId,
                     CategoryDescription = subscriber.Category.Description,
-                    CategoryCode = subscriber.Category.Code
+                    CategoryCode = subscriber.Category.Code,
+
                 };
 
             query = query.AsNoTracking().AsQueryable();

SW.Bitween.Api/Resources/WorkGroups/Create.cs

@@ -0,0 +1,37 @@
+using System.Threading.Tasks;
+using SW.Bitween.Domain;
+using SW.Bitween.Model; 
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Resources.WorkGroups;
+
+public class Create(BitweenDbContext dbContext, RequestContext requestContext,IInfolinkCache _BitweenCache, IBroadcast _broadcast)
+    : ICommandHandler<CreateWorkGroupModel, object>
+{
+    private readonly RequestContext _requestContext = requestContext;
+
+    public async Task<object> Handle(CreateWorkGroupModel request)
+    {
+        var workgroup = new WorkGroup()
+        {
+            Name = request.Name,
+            BusMessageName =  request.BusMessageName,
+            Options = new WorkGroupOptions()
+            {
+                RabbitMqOptions = new ConsumerSettings
+                {
+                    Prefetch = request.Options?.RabbitMqOptions?.Prefetch,
+                    Priority = request.Options?.RabbitMqOptions?.Priority
+                }
+            }
+        };
+        dbContext.Add(workgroup);
+        await dbContext.SaveChangesAsync();
+        _BitweenCache.BroadcastRevoke();
+        await _broadcast.RefreshConsumers();
+        return new
+        {
+            workgroup.Id
+        };
+    }
+}

SW.Bitween.Api/Resources/WorkGroups/Delete.cs

@@ -0,0 +1,31 @@
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using SW.Bitween.Domain;
+using SW.Bitween.Model;
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Resources.WorkGroups;
+
+[HandlerName(nameof(Delete))]
+public class Delete(BitweenDbContext dbContext, RequestContext requestContext, IBroadcast _broadcast, IInfolinkCache _infolinkCache)
+    : ICommandHandler<int, DeleteWorkGroupModel, object>
+{
+    private readonly RequestContext _requestContext = requestContext;
+
+    public async Task<object> Handle(int key, DeleteWorkGroupModel _)
+    {
+        var category = await dbContext.Set<WorkGroup>().FindAsync(key);
+        if (category is null)
+            throw new SWValidationException("CATEGORY_NOT_FOUND", $"Workgroup with id {key} was not found");
+
+        if (await dbContext.Set<Subscription>().AnyAsync(i => i.WorkGroupId.Value == category.Id))
+            throw new SWValidationException("CANT_BE_DELETED", "Workgroup with Subscriptions cant be deleted");
+
+        //Todo chek rabbitMq
+        dbContext.Remove(category);
+        await dbContext.SaveChangesAsync();
+        _infolinkCache.BroadcastRevoke();
+        await _broadcast.RefreshConsumers();
+        return null;
+    }
+}