Partitioned action block.

src/Squidex.Infrastructure/Tasks/PartitionedActionBlock.cs

@@ -0,0 +1,110 @@
+// ==========================================================================
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex UG (haftungsbeschraenkt)
+//  All rights reserved. Licensed under the MIT license.
+// ==========================================================================
+
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
+using Squidex.Infrastructure.Reflection;
+
+namespace Squidex.Infrastructure.Tasks
+{
+    public class PartitionedActionBlock<TInput> : ITargetBlock<TInput>
+    {
+        private readonly ITargetBlock<TInput> distributor;
+        private readonly ActionBlock<TInput>[] workers;
+
+        public Task Completion
+        {
+            get { return Task.WhenAll(workers.Select(x => x.Completion)); }
+        }
+
+        public PartitionedActionBlock(Action<TInput> action, Func<TInput, int> partitioner)
+            : this (ToAsync(action), partitioner, new ExecutionDataflowBlockOptions())
+        {
+        }
+
+        public PartitionedActionBlock(Func<TInput, Task> action, Func<TInput, int> partitioner)
+            : this(action, partitioner, new ExecutionDataflowBlockOptions())
+        {
+        }
+
+        public PartitionedActionBlock(Action<TInput> action, Func<TInput, int> partitioner, ExecutionDataflowBlockOptions dataflowBlockOptions)
+            : this(ToAsync(action), partitioner, dataflowBlockOptions)
+        {
+        }
+
+        public PartitionedActionBlock(Func<TInput, Task> action, Func<TInput, int> partitioner, ExecutionDataflowBlockOptions dataflowBlockOptions)
+        {
+            Guard.NotNull(action, nameof(action));
+            Guard.NotNull(partitioner, nameof(partitioner));
+            Guard.NotNull(dataflowBlockOptions, nameof(dataflowBlockOptions));
+            Guard.GreaterThan(dataflowBlockOptions.MaxDegreeOfParallelism, 1, nameof(dataflowBlockOptions.MaxDegreeOfParallelism));
+
+            workers = new ActionBlock<TInput>[dataflowBlockOptions.MaxDegreeOfParallelism];
+
+            for (var i = 0; i < dataflowBlockOptions.MaxDegreeOfParallelism; i++)
+            {
+                var workerOption = SimpleMapper.Map(dataflowBlockOptions, new ExecutionDataflowBlockOptions());
+
+                workerOption.MaxDegreeOfParallelism = 1;
+                workerOption.MaxMessagesPerTask = 1;
+
+                workers[i] = new ActionBlock<TInput>(action, workerOption);
+            }
+
+            var distributorOption = new ExecutionDataflowBlockOptions
+            {
+                MaxDegreeOfParallelism = 1,
+                MaxMessagesPerTask = 1,
+                BoundedCapacity = 1
+            };
+
+            distributor = new ActionBlock<TInput>(x =>
+            {
+                var partition = partitioner(x) % workers.Length;
+
+                return workers[partition].SendAsync(x);
+            }, distributorOption);
+
+            distributor.Completion.ContinueWith(x =>
+            {
+                foreach (var worker in workers)
+                {
+                    worker.Complete();
+                }
+            });
+        }
+
+        public DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, TInput messageValue, ISourceBlock<TInput> source, bool consumeToAccept)
+        {
+            return distributor.OfferMessage(messageHeader, messageValue, source, consumeToAccept);
+        }
+
+        public void Complete()
+        {
+            distributor.Complete();
+        }
+
+        public void Fault(Exception exception)
+        {
+            distributor.Fault(exception);
+        }
+
+        private static Func<TInput, Task> ToAsync(Action<TInput> action)
+        {
+            Guard.NotNull(action, nameof(action));
+
+            return x =>
+            {
+                action(x);
+
+                return TaskHelper.Done;
+            };
+        }
+    }
+}

src/Squidex/Areas/Api/Controllers/Rules/Models/RuleActionDto.cs

@@ -7,12 +7,14 @@
 
 using System;
 using System.Linq;
+using System.Runtime.Serialization;
 using Newtonsoft.Json;
 using Squidex.Domain.Apps.Core.Rules;
 
 namespace Squidex.Areas.Api.Controllers.Rules.Models
 {
     [JsonConverter(typeof(JsonInheritanceConverter), "actionType", typeof(RuleActionDto))]
+    [KnownType(nameof(Subtypes))]
     public abstract class RuleActionDto
     {
         public abstract RuleAction ToAction();

src/Squidex/Areas/Api/Controllers/Rules/Models/RuleTriggerDto.cs

@@ -14,7 +14,7 @@ using Squidex.Domain.Apps.Core.Rules;
 namespace Squidex.Areas.Api.Controllers.Rules.Models
 {
     [JsonConverter(typeof(JsonInheritanceConverter), "triggerType", typeof(RuleTriggerDto))]
-    [KnownType("Subtypes")]
+    [KnownType(nameof(Subtypes))]
     public abstract class RuleTriggerDto
     {
         public abstract RuleTrigger ToTrigger();

src/Squidex/Areas/Api/Controllers/Schemas/Models/FieldPropertiesDto.cs

@@ -8,12 +8,14 @@
 using System;
 using System.ComponentModel.DataAnnotations;
 using System.Linq;
+using System.Runtime.Serialization;
 using Newtonsoft.Json;
 using Squidex.Domain.Apps.Core.Schemas;
 
 namespace Squidex.Areas.Api.Controllers.Schemas.Models
 {
     [JsonConverter(typeof(JsonInheritanceConverter), "fieldType", typeof(FieldPropertiesDto))]
+    [KnownType(nameof(Subtypes))]
     public abstract class FieldPropertiesDto
     {
         /// <summary>

tests/Squidex.Domain.Apps.Core.Tests/Model/InvariantPartitionTests.cs

@@ -14,7 +14,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Core.Model
 {
-    public sealed class InvariantPartitionTests
+    public class InvariantPartitionTests
     {
         [Fact]
         public void Should_provide_single_value()

tests/Squidex.Domain.Apps.Core.Tests/Model/PartitioningTests.cs

@@ -9,7 +9,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Core.Model
 {
-    public sealed class PartitioningTests
+    public class PartitioningTests
     {
         [Fact]
         public void Should_consider_null_as_valid_partitioning()

tests/Squidex.Domain.Apps.Core.Tests/Operations/HandleRules/Triggers/AssetChangedTriggerTests.cs

@@ -17,7 +17,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Core.Operations.HandleRules.Triggers
 {
-    public sealed class AssetChangedTriggerTests
+    public class AssetChangedTriggerTests
     {
         private readonly IRuleTriggerHandler sut = new AssetChangedTriggerHandler();
 

tests/Squidex.Domain.Apps.Core.Tests/Operations/HandleRules/Triggers/ContentChangedTriggerTests.cs

@@ -22,7 +22,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Core.Operations.HandleRules.Triggers
 {
-    public sealed class ContentChangedTriggerTests
+    public class ContentChangedTriggerTests
     {
         private readonly IRuleTriggerHandler sut = new ContentChangedTriggerHandler();
         private static readonly NamedId<Guid> SchemaMatch = new NamedId<Guid>(Guid.NewGuid(), "my-schema1");

tests/Squidex.Domain.Apps.Core.Tests/Operations/Scripting/ContentDataObjectTests.cs

@@ -14,7 +14,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Core.Operations.Scripting
 {
-    public sealed class ContentDataObjectTests
+    public class ContentDataObjectTests
     {
         [Fact]
         public void Should_update_data_when_setting_field()

tests/Squidex.Domain.Apps.Core.Tests/Operations/ValidateContent/Validators/RequiredStringValidatorTests.cs

@@ -13,7 +13,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Core.Operations.ValidateContent.Validators
 {
-    public sealed class RequiredStringValidatorTests
+    public class RequiredStringValidatorTests
     {
         private readonly List<string> errors = new List<string>();
 

tests/Squidex.Domain.Apps.Core.Tests/Operations/ValidateContent/Validators/RequiredValidatorTests.cs

@@ -13,7 +13,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Core.Operations.ValidateContent.Validators
 {
-    public sealed class RequiredValidatorTests
+    public class RequiredValidatorTests
     {
         private readonly List<string> errors = new List<string>();
 

tests/Squidex.Domain.Apps.Entities.Tests/Rules/Guards/Actions/AlgoliaActionTests.cs

@@ -11,7 +11,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Entities.Rules.Guards.Actions
 {
-    public sealed class AlgoliaActionTests
+    public class AlgoliaActionTests
     {
         [Fact]
         public async Task Should_add_error_if_app_id_not_defined()

tests/Squidex.Domain.Apps.Entities.Tests/Rules/Guards/Actions/AzureQueueActionTests.cs

@@ -11,7 +11,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Entities.Rules.Guards.Actions
 {
-    public sealed class AzureQueueActionTests
+    public class AzureQueueActionTests
     {
         [Fact]
         public async Task Should_add_error_if_connection_string_is_null()

tests/Squidex.Domain.Apps.Entities.Tests/Rules/Guards/Actions/FastlyActionTests.cs

@@ -11,7 +11,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Entities.Rules.Guards.Actions
 {
-    public sealed class FastlyActionTests
+    public class FastlyActionTests
     {
         [Fact]
         public async Task Should_add_error_if_service_id_not_defined()

tests/Squidex.Domain.Apps.Entities.Tests/Rules/Guards/Actions/SlackActionTests.cs

@@ -12,7 +12,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Entities.Rules.Guards.Actions
 {
-    public sealed class SlackActionTests
+    public class SlackActionTests
     {
         [Fact]
         public async Task Should_add_error_if_webhook_url_is_null()

tests/Squidex.Domain.Apps.Entities.Tests/Rules/Guards/Actions/WebhookActionTests.cs

@@ -12,7 +12,7 @@ using Xunit;
 
 namespace Squidex.Domain.Apps.Entities.Rules.Guards.Actions
 {
-    public sealed class WebhookActionTests
+    public class WebhookActionTests
     {
         [Fact]
         public async Task Should_add_error_if_url_is_null()

tests/Squidex.Infrastructure.Tests/Commands/EnrichWithTimestampCommandMiddlewareTests.cs

@@ -13,7 +13,7 @@ using Xunit;
 
 namespace Squidex.Infrastructure.Commands
 {
-    public sealed class EnrichWithTimestampCommandMiddlewareTests
+    public class EnrichWithTimestampCommandMiddlewareTests
     {
         private readonly IClock clock = A.Fake<IClock>();
 

tests/Squidex.Infrastructure.Tests/DispatchingTests.cs

@@ -12,7 +12,7 @@ using Xunit;
 
 namespace Squidex.Infrastructure
 {
-    public sealed class DispatchingTests
+    public class DispatchingTests
     {
         private interface IMyEvent
         {

tests/Squidex.Infrastructure.Tests/EventSourcing/DefaultEventNotifierTests.cs

@@ -10,7 +10,7 @@ using Xunit;
 
 namespace Squidex.Infrastructure.EventSourcing
 {
-    public sealed class DefaultEventNotifierTests
+    public class DefaultEventNotifierTests
     {
         private readonly DefaultEventNotifier sut = new DefaultEventNotifier(new InMemoryPubSub());
 

tests/Squidex.Infrastructure.Tests/Json/InstantConverterTests.cs

@@ -12,7 +12,7 @@ using Xunit;
 
 namespace Squidex.Infrastructure.Json
 {
-    public sealed class InstantConverterTests
+    public class InstantConverterTests
     {
         [Fact]
         public void Should_serialize_and_deserialize()

tests/Squidex.Infrastructure.Tests/Migrations/MigratorTests.cs

@@ -16,7 +16,7 @@ using Xunit;
 
 namespace Squidex.Infrastructure.Migrations
 {
-    public sealed class MigratorTests
+    public class MigratorTests
     {
         private readonly IMigrationStatus status = A.Fake<IMigrationStatus>();
         private readonly ISemanticLog log = A.Fake<ISemanticLog>();

tests/Squidex.Infrastructure.Tests/Tasks/PartitionedActionBlockTests.cs

@@ -0,0 +1,62 @@
+// ==========================================================================
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex UG (haftungsbeschraenkt)
+//  All rights reserved. Licensed under the MIT license.
+// ==========================================================================
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
+using Xunit;
+
+namespace Squidex.Infrastructure.Tasks
+{
+    public class PartitionedActionBlockTests
+    {
+        [Fact]
+        public async Task Should_propagate_in_order()
+        {
+            var random = new Random();
+            var partitions = 10;
+
+            var lists = new List<int>[partitions];
+
+            for (var i = 0; i < partitions; i++)
+            {
+                lists[i] = new List<int>();
+            }
+
+            var block = new PartitionedActionBlock<(int P, int V)>(x =>
+            {
+                random.Next(10);
+
+                lists[x.P].Add(x.V);
+            }, x => x.P, new ExecutionDataflowBlockOptions
+            {
+                MaxDegreeOfParallelism = 100,
+                MaxMessagesPerTask = 1,
+                BoundedCapacity = 100
+            });
+
+            for (var i = 0; i < partitions; i++)
+            {
+                for (var j = 0; j < 10; j++)
+                {
+                    await block.SendAsync((i, j));
+                }
+            }
+
+            block.Complete();
+
+            await block.Completion;
+
+            foreach (var list in lists)
+            {
+                Assert.Equal(Enumerable.Range(0, 10).ToList(), list);
+            }
+        }
+    }
+}

tests/Squidex.Infrastructure.Tests/ValidationExtensionsTests.cs

@@ -10,7 +10,7 @@ using Xunit;
 
 namespace Squidex.Infrastructure
 {
-    public sealed class ValidationExtensionsTests
+    public class ValidationExtensionsTests
     {
         [Fact]
         public void Should_return_true_if_is_between()