RabbitMQ client version upgrade

src/WorkflowCore.DSL/Services/DefinitionLoader.cs

@@ -201,13 +201,21 @@ private void AttachInputs(StepSourceV1 source, Type dataType, Type stepType, Wor
                     continue;
                 }
 
-                if ((input.Value is IDictionary<string, object>) || (input.Value is IDictionary<object, object>))
+                if (input.Value is IDictionary<string, object> || input.Value is IDictionary<object, object>)
                 {
                     var acn = BuildObjectInputAction(input, dataParameter, contextParameter, environmentVarsParameter, stepProperty);
                     step.Inputs.Add(new ActionParameter<IStepBody, object>(acn));
                     continue;
                 }
 
+                if (input.Value is IEnumerable<object> list)
+                {
+                    var acn = BuildListInputAction(list, dataParameter, contextParameter, environmentVarsParameter,
+                        stepProperty);
+                    step.Inputs.Add(new ActionParameter<IStepBody, object>(acn));
+                    continue;
+                }
+
                 throw new ArgumentException($"Unknown type for input {input.Key} on {source.Id}");
             }
         }
@@ -253,7 +261,7 @@ private void AttachDirectlyOutput(KeyValuePair<string, string> output, WorkflowS
 
                 Action<IStepBody, object> acn = (pStep, pData) =>
                 {
-                    object resolvedValue = sourceExpr.Compile().DynamicInvoke(pStep); ;
+                    object resolvedValue = sourceExpr.Compile().DynamicInvoke(pStep);
                     propertyInfo.SetValue(pData, resolvedValue, new object[] { output.Key });
                 };
 
@@ -306,7 +314,7 @@ private void AttachNestedOutput(KeyValuePair<string, string> output, WorkflowSte
                     {
                         var targetExpr = Expression.Lambda(memberExpression, dataParameter);
                         object data = targetExpr.Compile().DynamicInvoke(pData);
-                        object resolvedValue = sourceExpr.Compile().DynamicInvoke(pStep); ;
+                        object resolvedValue = sourceExpr.Compile().DynamicInvoke(pStep);
                         propertyInfo.SetValue(data, resolvedValue, new object[] { items[1] });
                     };
 
@@ -379,6 +387,80 @@ void acn(IStepBody pStep, object pData, IStepExecutionContext pContext)
             return acn;
         }
 
+        private static Action<IStepBody, object, IStepExecutionContext> BuildListInputAction(
+            IEnumerable<object> input,
+            ParameterExpression dataParameter,
+            ParameterExpression contextParameter,
+            ParameterExpression environmentVarsParameter,
+            PropertyInfo stepProperty)
+        {
+            void acn(IStepBody pStep, object pData, IStepExecutionContext pContext)
+            {
+                if (input == null)
+                    throw new ArgumentNullException(nameof(input));
+
+                var itemType = stepProperty.PropertyType.IsArray
+                    ? stepProperty.PropertyType.GetElementType()
+                    : stepProperty.PropertyType.GetGenericArguments().FirstOrDefault();
+
+                if (itemType == null)
+                    throw new InvalidOperationException("Unable to determine the item type for stepProperty.");
+
+                var processedItems = new List<object>();
+
+                foreach (var item in input)
+                {
+                    var obj = JObject.FromObject(item);
+                    var stack = new Stack<JObject>();
+                    stack.Push(obj);
+
+                    while (stack.Count > 0)
+                    {
+                        var current = stack.Pop();
+                        foreach (var prop in current.Properties().ToList())
+                        {
+                            if (prop.Name.StartsWith("@"))
+                            {
+                                var expr = DynamicExpressionParser.ParseLambda(
+                                    new[] { dataParameter, contextParameter, environmentVarsParameter },
+                                    typeof(object),
+                                    prop.Value.ToString());
+
+                                var resolved = expr.Compile().DynamicInvoke(pData, pContext,
+                                    Environment.GetEnvironmentVariables());
+                                current.Remove(prop.Name);
+                                current.Add(prop.Name.TrimStart('@'), JToken.FromObject(resolved));
+                            }
+                        }
+
+                        foreach (var child in current.Children<JObject>())
+                            stack.Push(child);
+                    }
+
+                    processedItems.Add(obj.ToObject(itemType));
+                }
+
+                if (stepProperty.PropertyType.IsArray)
+                {
+                    var array = Array.CreateInstance(itemType, processedItems.Count);
+                    for (var i = 0; i < processedItems.Count; i++)
+                        array.SetValue(processedItems[i], i);
+                    stepProperty.SetValue(pStep, array);
+                }
+                else
+                {
+                    var listInstance = Activator.CreateInstance(typeof(List<>).MakeGenericType(itemType));
+                    var addMethod = listInstance.GetType().GetMethod("Add");
+                    foreach (var item in processedItems)
+                        addMethod?.Invoke(listInstance, new[] { item });
+
+                    stepProperty.SetValue(pStep, listInstance);
+                }
+            }
+
+            return acn;
+        }
+
         private static Action<IStepBody, object, IStepExecutionContext> BuildObjectInputAction(KeyValuePair<string, object> input, ParameterExpression dataParameter, ParameterExpression contextParameter, ParameterExpression environmentVarsParameter, PropertyInfo stepProperty)
         {
             void acn(IStepBody pStep, object pData, IStepExecutionContext pContext)
@@ -405,7 +487,7 @@ void acn(IStepBody pStep, object pData, IStepExecutionContext pContext)
                         stack.Push(child);
                 }
 
-                stepProperty.SetValue(pStep, destObj);
+                stepProperty.SetValue(pStep, destObj.ToObject(stepProperty.PropertyType));
             }
             return acn;
         }

src/providers/WorkflowCore.QueueProviders.RabbitMQ/ServiceCollectionExtensions.cs

@@ -2,6 +2,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading.Tasks;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using WorkflowCore.Interface;
 using WorkflowCore.Models;
@@ -19,8 +20,7 @@ public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options, IConnect
             if (options == null) throw new ArgumentNullException(nameof(options));
             if (connectionFactory == null) throw new ArgumentNullException(nameof(connectionFactory));
 
-            return options
-                .UseRabbitMQ((sp, name) => connectionFactory.CreateConnection(name));
+            return options.UseRabbitMQ(async (sp, name) => await connectionFactory.CreateConnectionAsync(name));
         }
 
         public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options,
@@ -31,16 +31,23 @@ public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options,
             if (connectionFactory == null) throw new ArgumentNullException(nameof(connectionFactory));
             if (hostnames == null) throw new ArgumentNullException(nameof(hostnames));
 
-            return options
-                .UseRabbitMQ((sp, name) => connectionFactory.CreateConnection(hostnames.ToList(), name));
+            return options.UseRabbitMQ(async (sp, name) => await connectionFactory.CreateConnectionAsync(hostnames.ToList(), name));
         }
-        
-        public static WorkflowOptions UseRabbitMQ(this WorkflowOptions options, RabbitMqConnectionFactory rabbitMqConnectionFactory)
+
+        private static WorkflowOptions UseRabbitMQ(this WorkflowOptions options, Func<IServiceProvider, string, Task<IConnection>> rabbitMqConnectionFactory)
         {
             if (options == null) throw new ArgumentNullException(nameof(options));
             if (rabbitMqConnectionFactory == null) throw new ArgumentNullException(nameof(rabbitMqConnectionFactory));
 
             options.Services.AddSingleton(rabbitMqConnectionFactory);
+
+            options.Services.AddSingleton<RabbitMqConnectionFactory>(
+                sp => (provider, name) =>
+                {
+                    var connection = rabbitMqConnectionFactory(provider, name).GetAwaiter().GetResult();
+                    return connection;
+                });
+
             options.Services.TryAddSingleton<IRabbitMqQueueNameProvider, DefaultRabbitMqQueueNameProvider>();
             options.UseQueueProvider(RabbitMqQueueProviderFactory);
 

src/providers/WorkflowCore.QueueProviders.RabbitMQ/Services/RabbitMQProvider.cs

@@ -1,5 +1,4 @@
-using Newtonsoft.Json;
-using RabbitMQ.Client;
+using RabbitMQ.Client;
 using System;
 using System.Linq;
 using System.Text;
@@ -17,9 +16,8 @@ public class RabbitMQProvider : IQueueProvider
         private readonly IRabbitMqQueueNameProvider _queueNameProvider;
         private readonly RabbitMqConnectionFactory _rabbitMqConnectionFactory;
         private readonly IServiceProvider _serviceProvider;
-
-        private IConnection _connection = null;
-        private static JsonSerializerSettings SerializerSettings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All };
+
+        private IConnection _connection;
 
         public bool IsDequeueBlocking => false;
 
@@ -37,11 +35,13 @@ public async Task QueueWork(string id, QueueType queue)
             if (_connection == null)
                 throw new InvalidOperationException("RabbitMQ provider not running");
 
-            using (var channel = _connection.CreateModel())
+            using (var channel = await _connection.CreateChannelAsync())
             {
-                channel.QueueDeclare(queue: _queueNameProvider.GetQueueName(queue), durable: true, exclusive: false, autoDelete: false, arguments: null);
+                await channel.QueueDeclareAsync(queue: _queueNameProvider.GetQueueName(queue), durable: true, exclusive: false,
+                    autoDelete: false, arguments: null);
                 var body = Encoding.UTF8.GetBytes(id);
-                channel.BasicPublish(exchange: "", routingKey: _queueNameProvider.GetQueueName(queue), basicProperties: null, body: body);
+
+                await channel.BasicPublishAsync("", _queueNameProvider.GetQueueName(queue), false,body);
             }
         }
 
@@ -50,34 +50,35 @@ public async Task<string> DequeueWork(QueueType queue, CancellationToken cancell
             if (_connection == null)
                 throw new InvalidOperationException("RabbitMQ provider not running");
 
-            using (var channel = _connection.CreateModel())
+            using (var channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken))
             {
-                channel.QueueDeclare(queue: _queueNameProvider.GetQueueName(queue),
-                                     durable: true,
-                                     exclusive: false,
-                                     autoDelete: false,
-                                     arguments: null);
+                await channel.QueueDeclareAsync(queue: _queueNameProvider.GetQueueName(queue),
+                    durable: true,
+                    exclusive: false,
+                    autoDelete: false,
+                    arguments: null, cancellationToken: cancellationToken);
+
+                await channel.BasicQosAsync(prefetchSize: 0, prefetchCount: 1, global: false,
+                    cancellationToken: cancellationToken);
 
-                channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
+                var msg = await channel.BasicGetAsync(_queueNameProvider.GetQueueName(queue), false, cancellationToken);
 
-                var msg = channel.BasicGet(_queueNameProvider.GetQueueName(queue), false);
-                if (msg != null)
+                if (msg == null)
                 {
-                    var data = Encoding.UTF8.GetString(msg.Body.ToArray());
-                    channel.BasicAck(msg.DeliveryTag, false);
-                    return data;
+                    return null;
                 }
-                return null;
+
+                var data = Encoding.UTF8.GetString(msg.Body.ToArray());
+                await channel.BasicAckAsync(msg.DeliveryTag, false, cancellationToken);
+                return data;
             }
         }
-        
+
         public void Dispose()
         {
-            if (_connection != null)
-            {
-                if (_connection.IsOpen)
-                    _connection.Close();
-            }
+            if (_connection == null) return;
+            if (_connection.IsOpen)
+                _connection.CloseAsync();
         }
 
         public async Task Start()
@@ -89,11 +90,10 @@ public async Task Stop()
         {
             if (_connection != null)
             {
-                _connection.Close();
+                await _connection.CloseAsync();
                 _connection = null;
             }
         }
-
     }
 #pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
-}
+}

src/providers/WorkflowCore.QueueProviders.RabbitMQ/WorkflowCore.QueueProviders.RabbitMQ.csproj

@@ -23,7 +23,7 @@
 
   <ItemGroup>
     <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
-    <PackageReference Include="RabbitMQ.Client" Version="6.8.1" />
+    <PackageReference Include="RabbitMQ.Client" Version="7.1.2" />
   </ItemGroup>
 
 </Project>

test/WorkflowCore.IntegrationTests/Scenarios/StoredJsonScenario.cs

@@ -84,5 +84,53 @@ public void should_execute_json_workflow_with_dynamic_data()
             data["Counter6"].Should().Be(1);
             data["Counter10"].Should().Be(1);
         }
+
+
+        [Fact]
+        public void should_execute_json_workflow_with_complex_input_type()
+        {
+            var initialData = new FlowData();
+            var workflowId = StartWorkflow(TestAssets.Utils.GetTestDefinitionJsonComplexInputProperty(), initialData);
+            WaitForWorkflowToComplete(workflowId, TimeSpan.FromSeconds(30));
+
+            var data = GetData<FlowData>(workflowId);
+            GetStatus(workflowId).Should().Be(WorkflowStatus.Complete);
+            UnhandledStepErrors.Count.Should().Be(0);
+            data.Assignee.Should().NotBeNull();
+            data.Assignee.Name.Should().Be("John Doe");
+            data.Assignee.UnitInfo.Should().NotBeNull();
+            data.Assignee.UnitInfo.Name.Should().Be("IT Department");
+
+        }
+
+        [Fact]
+        public void should_execute_json_workflow_with_list_of_complex_input_type()
+        {
+            var initialData = new FlowData();
+            var workflowId = StartWorkflow(TestAssets.Utils.GetTestDefinitionJsonListOfComplexInputProperty(), initialData);
+            WaitForWorkflowToComplete(workflowId, TimeSpan.FromSeconds(30));
+
+            var data = GetData<FlowData>(workflowId);
+            GetStatus(workflowId).Should().Be(WorkflowStatus.Complete);
+            UnhandledStepErrors.Count.Should().Be(0);
+
+            data.AssigneeList.Should().NotBeNullOrEmpty();
+            data.AssigneeList.Count.Should().Be(2);
+
+            data.Assignee.Should().NotBeNull();
+            data.Assignee.Name.Should().Be("John Doe");
+            data.Assignee.UnitInfo?.Name.Should().Be("IT Department");
+
+            data.AssigneeList[0]?.Name.Should().Be(@"Nurlan Mikayilov");
+            data.AssigneeList[0]?.UnitInfo?.Name.Should().Be("IT Department");
+
+            data.AssigneeList[1]?.Name.Should().Be(@"Jala Mammadova");
+            data.AssigneeList[1]?.UnitInfo?.Name.Should().Be("General Department");
+
+            data.AssigneeArray[0]?.Name.Should().Be(@"Amin Nabiyev");
+            data.AssigneeArray[0]?.UnitInfo?.Name.Should().Be("IT Department");
+
+
+        }
     }
 }

test/WorkflowCore.TestAssets/DataTypes/FlowData.cs

@@ -0,0 +1,13 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using WorkflowCore.TestAssets.Steps;
+
+namespace WorkflowCore.TestAssets.DataTypes;
+
+public class FlowData
+{
+    public AssigneeInfo Assignee { get; set; } = new();
+    public List<AssigneeInfo> AssigneeList { get; set; } = [];
+    public AssigneeInfo[] AssigneeArray { get; set; } = [];
+}

test/WorkflowCore.TestAssets/Steps/AssignTask.cs

@@ -0,0 +1,37 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using WorkflowCore.Interface;
+using WorkflowCore.Models;
+using WorkflowCore.TestAssets.DataTypes;
+
+namespace WorkflowCore.TestAssets.Steps;
+
+public class AssignTask : StepBody
+{       
+    public AssigneeInfo Assignee { get; set; }
+    public List<AssigneeInfo> AssigneeList { get; set; } = [];
+
+    public AssigneeInfo[] AssigneeArray { get; set; } = [];
+
+    public override ExecutionResult Run(IStepExecutionContext context)
+    {
+        if (context.Workflow.Data is FlowData flowData)
+        {
+            if (Assignee != null)
+            {
+                flowData.Assignee = new AssigneeInfo
+                {
+                    Id = Assignee.Id,
+                    Name = Assignee.Name,
+                    MemberType = Assignee.MemberType,
+                    UnitInfo = Assignee.UnitInfo
+                };
+            }
+
+            flowData.AssigneeList.AddRange(AssigneeList);
+            flowData.AssigneeArray = AssigneeArray.ToArray();
+        }
+        return ExecutionResult.Next();
+    }
+}