refactor: Better usage of Utils in PushQueue (#441)

aneoconsulting · Jul 11, 2023 · b0265e1 · b0265e1
2 parents 8ee0cdc + 49b2671
commit b0265e1
Show file tree

Hide file tree

Showing 2 changed files with 31 additions and 55 deletions.
diff --git a/Adaptors/Amqp/src/PushQueueStorage.cs b/Adaptors/Amqp/src/PushQueueStorage.cs
@@ -27,35 +27,12 @@
 
 using ArmoniK.Core.Base;
 using ArmoniK.Core.Base.DataStructures;
+using ArmoniK.Utils;
 
 using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.ObjectPool;
 
 namespace ArmoniK.Core.Adapters.Amqp;
 
-/// <summary>
-///   Policy for creating a <see cref="Session" /> for the <see cref="ObjectPool{Session}" />
-/// </summary>
-internal sealed class SessionPooledObjectPolicy : IPooledObjectPolicy<Session>
-{
-  private readonly IConnectionAmqp connectionAmqp_;
-
-  /// <summary>
-  ///   Initializes a <see cref="SessionPooledObjectPolicy" />
-  /// </summary>
-  /// <param name="connectionAmqp">AMQP connection that will be used to create new sessions</param>
-  public SessionPooledObjectPolicy(IConnectionAmqp connectionAmqp)
-    => connectionAmqp_ = connectionAmqp;
-
-  /// <inheritdoc />
-  public Session Create()
-    => new(connectionAmqp_.Connection);
-
-  /// <inheritdoc />
-  public bool Return(Session obj)
-    => !obj.IsClosed;
-}
-
 public class PushQueueStorage : QueueStorage, IPushQueueStorage
 {
   private const int MaxInternalQueuePriority = 10;
@@ -70,8 +47,10 @@ public PushQueueStorage(QueueCommon.Amqp          options,
     : base(options,
            connectionAmqp)
   {
-    logger_      = logger;
-    sessionPool_ = new DefaultObjectPool<Session>(new SessionPooledObjectPolicy(ConnectionAmqp));
+    logger_ = logger;
+    sessionPool_ = new ObjectPool<Session>(200,
+                                           () => new Session(connectionAmqp.Connection),
+                                           session => !session.IsClosed);
   }
 
   /// <inheritdoc />
@@ -80,11 +59,12 @@ public async Task PushMessagesAsync(IEnumerable<MessageData> messages,
                                       CancellationToken        cancellationToken = default)
   {
     var priorityGroups = messages.GroupBy(msgData => msgData.Options.Priority);
-    await Task.WhenAll(priorityGroups.Select(group => PushMessagesAsync(group,
-                                                                        partitionId,
-                                                                        group.Key,
-                                                                        cancellationToken)))
-              .ConfigureAwait(false);
+    await priorityGroups.ParallelForEach(new ParallelTaskOptions(cancellationToken),
+                                         group => PushMessagesAsync(group,
+                                                                    partitionId,
+                                                                    group.Key,
+                                                                    cancellationToken))
+                        .ConfigureAwait(false);
   }
 
   private async Task PushMessagesAsync(IEnumerable<MessageData> messages,
@@ -114,29 +94,25 @@ private async Task PushMessagesAsync(IEnumerable<MessageData> messages,
                      whichQueue,
                      internalPriority);
 
-    var session = sessionPool_.Get();
-    try
-    {
-      var sender = new SenderLink(session,
-                                  $"{partitionId}###SenderLink{whichQueue}",
-                                  $"{partitionId}###q{whichQueue}");
-
-      await Task.WhenAll(messages.Select(msgData => sender.SendAsync(new Message(Encoding.UTF8.GetBytes(msgData.TaskId))
-                                                                     {
-                                                                       Header = new Header
-                                                                                {
-                                                                                  Priority = (byte)internalPriority,
-                                                                                },
-                                                                       Properties = new Properties(),
-                                                                     })))
-                .ConfigureAwait(false);
-
-      await sender.CloseAsync()
+    await using var session = await sessionPool_.GetAsync(cancellationToken)
+                                                .ConfigureAwait(false);
+
+    var sender = new SenderLink(session,
+                                $"{partitionId}###SenderLink{whichQueue}",
+                                $"{partitionId}###q{whichQueue}");
+
+    await messages.ParallelForEach(new ParallelTaskOptions(cancellationToken),
+                                   msgData => sender.SendAsync(new Message(Encoding.UTF8.GetBytes(msgData.TaskId))
+                                                               {
+                                                                 Header = new Header
+                                                                          {
+                                                                            Priority = (byte)internalPriority,
+                                                                          },
+                                                                 Properties = new Properties(),
+                                                               }))
                   .ConfigureAwait(false);
-    }
-    finally
-    {
-      sessionPool_.Return(session);
-    }
+
+    await sender.CloseAsync()
+                .ConfigureAwait(false);
   }
 }
diff --git a/Utils/src/ArmoniK.Core.Utils.csproj b/Utils/src/ArmoniK.Core.Utils.csproj
@@ -25,7 +25,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="ArmoniK.Utils" Version="0.2.0" />
+    <PackageReference Include="ArmoniK.Utils" Version="0.3.0" />
     <PackageReference Include="JetBrains.Annotations" Version="2022.3.1" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="7.0.0" />
     <PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="7.0.0" />