From 4c8418e3c4b764810889f8476350c8d608745981 Mon Sep 17 00:00:00 2001
From: "renovate[bot]" <29139614+renovate[bot]@users.noreply.github.com>
Date: Thu, 6 Jul 2023 14:55:39 +0000
Subject: [PATCH 1/2] chore(deps): update dependency armonik.utils to v0.3.0

---
 Utils/src/ArmoniK.Core.Utils.csproj | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Utils/src/ArmoniK.Core.Utils.csproj b/Utils/src/ArmoniK.Core.Utils.csproj
index d9724a542..c646553c5 100644
--- 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" />

From 49b2671c0716483ecd8119af7c674d808ff08bf4 Mon Sep 17 00:00:00 2001
From: lemaitre-aneo <flemaitre@aneo.fr>
Date: Tue, 11 Jul 2023 12:07:42 +0200
Subject: [PATCH 2/2] Better usage of Utils in PushQueue

---
 Adaptors/Amqp/src/PushQueueStorage.cs | 84 ++++++++++-----------------
 1 file changed, 30 insertions(+), 54 deletions(-)

diff --git a/Adaptors/Amqp/src/PushQueueStorage.cs b/Adaptors/Amqp/src/PushQueueStorage.cs
index 20b0a8f32..9bd6113d9 100644
--- 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);
   }
 }