refactor: remove submitter from Wait for results method in client part

.github/workflows/build.yml

@@ -181,7 +181,7 @@ jobs:
           tls: ${{ matrix.tls }}
           mtls: ${{ matrix.mtls }}
           ext-csharp-version: ${{ needs.versionning.outputs.version }}
-          core-version: 0.19.3
+          core-version: 0.20.0
 
       - name: Setup hosts file
         run : echo -e "$(kubectl get svc ingress -n armonik -o jsonpath={.status.loadBalancer.ingress[0].ip})\tarmonik.local" | sudo tee -a /etc/hosts
@@ -304,7 +304,7 @@ jobs:
           working-directory: ${{ github.workspace }}/infra
           type: localhost
           ext-csharp-version: ${{ needs.versionning.outputs.version }}
-          core-version: 0.19.3
+          core-version: 0.20.0
 
       - name: Run Test
         timeout-minutes: 20

Client/src/Common/Exceptions/ResultAbortedException.cs

@@ -0,0 +1,36 @@
+// This file is part of the ArmoniK project
+//
+// Copyright (C) ANEO, 2021-$CURRENT_YEAR$. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License")
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+using System;
+
+namespace ArmoniK.DevelopmentKit.Client.Common.Exceptions;
+
+/// <summary>
+///   Exception when result is aborted
+/// </summary>
+public class ResultAbortedException : Exception
+
+{
+  /// <summary>
+  ///   Initializes a new instance of the <see cref="ResultAbortedException" /> with the specified error message
+  /// </summary>
+  /// <param name="message">The error message</param>
+  public ResultAbortedException(string message)
+    : base(message)
+  {
+  }
+}

Client/src/Common/Submitter/BaseClientSubmitter.cs

@@ -1,13 +1,13 @@
 // This file is part of the ArmoniK project
-// 
+//
 // Copyright (C) ANEO, 2021-2023. All rights reserved.
-// 
+//
 // Licensed under the Apache License, Version 2.0 (the "License")
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
-// 
+//
 //     http://www.apache.org/licenses/LICENSE-2.0
-// 
+//
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
@@ -16,7 +16,6 @@
 
 using System;
 using System.Collections.Generic;
-using System.Data;
 using System.IO;
 using System.Linq;
 using System.Threading;
@@ -25,6 +24,7 @@
 using ArmoniK.Api.Client;
 using ArmoniK.Api.Common.Utils;
 using ArmoniK.Api.gRPC.V1;
+using ArmoniK.Api.gRPC.V1.Events;
 using ArmoniK.Api.gRPC.V1.Results;
 using ArmoniK.Api.gRPC.V1.Sessions;
 using ArmoniK.Api.gRPC.V1.SortDirection;
@@ -675,40 +675,15 @@ public byte[] GetResult(string            taskId,
                             ResultId = resultId,
                             Session  = SessionId.Id,
                           };
-
-      Retry.WhileException(5,
-                           2000,
-                           retry =>
-                           {
-                             using var channel          = ChannelPool.GetChannel();
-                             var       submitterService = new Api.gRPC.V1.Submitter.Submitter.SubmitterClient(channel);
-
-                             Logger.LogDebug("Try {try} for {funcName}",
-                                             retry,
-                                             nameof(submitterService.WaitForAvailability));
-                             // TODO: replace with submitterService.TryGetResultStream() => Issue #
-                             var availabilityReply = submitterService.WaitForAvailability(resultRequest,
-                                                                                          cancellationToken: cancellationToken);
-
-                             switch (availabilityReply.TypeCase)
-                             {
-                               case AvailabilityReply.TypeOneofCase.None:
-                                 throw new Exception("Issue with Server !");
-                               case AvailabilityReply.TypeOneofCase.Ok:
-                                 break;
-                               case AvailabilityReply.TypeOneofCase.Error:
-                                 throw new
-                                   ClientResultsException($"Result in Error - {resultId}\nMessage :\n{string.Join("Inner message:\n", availabilityReply.Error.Errors)}",
-                                                          resultId);
-                               case AvailabilityReply.TypeOneofCase.NotCompletedTask:
-                                 throw new DataException($"Result {resultId} was not yet completed");
-                               default:
-                                 throw new InvalidOperationException();
-                             }
-                           },
-                           true,
-                           typeof(IOException),
-                           typeof(RpcException));
+      using var channel      = ChannelPool.GetChannel();
+      var       eventsClient = new Events.EventsClient(channel);
+      eventsClient.WaitForResultsAsync(SessionId.Id,
+                                       new List<string>
+                                       {
+                                         resultId,
+                                       },
+                                       cancellationToken)
+                  .Wait(cancellationToken);
 
       return Retry.WhileException(5,
                                   200,

Client/src/Common/Submitter/EventsClientExt.cs

@@ -0,0 +1,130 @@
+// This file is part of the ArmoniK project
+// 
+// Copyright (C) ANEO, 2021-2023. All rights reserved.
+// 
+// Licensed under the Apache License, Version 2.0 (the "License")
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// 
+//     http://www.apache.org/licenses/LICENSE-2.0
+// 
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using ArmoniK.Api.gRPC.V1;
+using ArmoniK.Api.gRPC.V1.Events;
+using ArmoniK.Api.gRPC.V1.Results;
+using ArmoniK.DevelopmentKit.Client.Common.Exceptions;
+
+using Grpc.Core;
+
+namespace ArmoniK.DevelopmentKit.Client.Common.Submitter;
+
+internal static class EventsClientExt
+{
+  private static FiltersAnd ResultsFilter(string resultId)
+    => new()
+       {
+         And =
+         {
+           new FilterField
+           {
+             Field = new ResultField
+                     {
+                       ResultRawField = new ResultRawField
+                                        {
+                                          Field = ResultRawEnumField.ResultId,
+                                        },
+                     },
+             FilterString = new FilterString
+                            {
+                              Operator = FilterStringOperator.Equal,
+                              Value    = resultId,
+                            },
+           },
+         },
+       };
+
+  internal static async Task WaitForResultsAsync(this Events.EventsClient client,
+                                                 string                   sessionId,
+                                                 ICollection<string>      resultIds,
+                                                 CancellationToken        cancellationToken)
+  {
+    var resultsNotFound = new HashSet<string>(resultIds);
+    while (resultsNotFound.Any())
+    {
+      using var streamingCall = client.GetEvents(new EventSubscriptionRequest
+                                                 {
+                                                   SessionId = sessionId,
+                                                   ReturnedEvents =
+                                                   {
+                                                     EventsEnum.ResultStatusUpdate,
+                                                     EventsEnum.NewResult,
+                                                   },
+                                                   ResultsFilters = new Filters
+                                                                    {
+                                                                      Or =
+                                                                      {
+                                                                        resultsNotFound.Select(ResultsFilter),
+                                                                      },
+                                                                    },
+                                                 },
+                                                 cancellationToken: cancellationToken);
+      try
+      {
+        while (await streamingCall.ResponseStream.MoveNext(cancellationToken))
+        {
+          var resp = streamingCall.ResponseStream.Current;
+          if (resp.UpdateCase == EventSubscriptionResponse.UpdateOneofCase.ResultStatusUpdate && resultsNotFound.Contains(resp.ResultStatusUpdate.ResultId))
+          {
+            if (resp.ResultStatusUpdate.Status == ResultStatus.Completed)
+            {
+              resultsNotFound.Remove(resp.ResultStatusUpdate.ResultId);
+              if (!resultsNotFound.Any())
+              {
+                break;
+              }
+            }
+
+            if (resp.ResultStatusUpdate.Status == ResultStatus.Aborted)
+            {
+              throw new ResultAbortedException($"Result {resp.ResultStatusUpdate.ResultId} has been aborted");
+            }
+          }
+
+          if (resp.UpdateCase == EventSubscriptionResponse.UpdateOneofCase.NewResult && resultsNotFound.Contains(resp.NewResult.ResultId))
+          {
+            if (resp.NewResult.Status == ResultStatus.Completed)
+            {
+              resultsNotFound.Remove(resp.NewResult.ResultId);
+              if (!resultsNotFound.Any())
+              {
+                break;
+              }
+            }
+
+            if (resp.NewResult.Status == ResultStatus.Aborted)
+            {
+              throw new ResultAbortedException($"Result {resp.NewResult.ResultId} has been aborted");
+            }
+          }
+        }
+      }
+      catch (OperationCanceledException)
+      {
+      }
+      catch (RpcException)
+      {
+      }
+    }
+  }
+}