small improvments

Signed-off-by: Neil South <[email protected]>

Author: neildsouth

src/InformaticsGateway/Services/Connectors/PayloadAssembler.cs

@@ -129,30 +129,34 @@ private async void OnTimedEvent(Object source, System.Timers.ElapsedEventArgs e)
             try
             {
                 _timer.Enabled = false;
-                _logger.BucketActive(_payloads.Count);
+                if (_payloads.Any())
+                {
+                    _logger.BucketActive(_payloads.Count);
+                }
                 foreach (var key in _payloads.Keys)
                 {
                     _logger.BucketElapsedTime(key);
                     var payload = await _payloads[key].Task.ConfigureAwait(false);
                     using var loggerScope = _logger.BeginScope(new LoggingDataDictionary<string, object> { { "CorrelationId", payload.CorrelationId } });
-                    if (payload.HasTimedOut)
+
+                    if (payload.IsUploadCompleted())
+                    {
+                        if (_payloads.TryRemove(key, out _))
+                        {
+                            await QueueBucketForNotification(key, payload).ConfigureAwait(false);
+                        }
+                        else
+                        {
+                            _logger.BucketRemoveError(key);
+                        }
+                    }
+                    else if (payload.HasTimedOut)
                     {
                         if (payload.ContainerUploadFailures())
                         {
                             _payloads.TryRemove(key, out _);
                             _logger.PayloadRemovedWithFailureUploads(key);
                         }
-                        else if (payload.IsUploadCompleted())
-                        {
-                            if (_payloads.TryRemove(key, out _))
-                            {
-                                await QueueBucketForNotification(key, payload).ConfigureAwait(false);
-                            }
-                            else
-                            {
-                                _logger.BucketRemoveError(key);
-                            }
-                        }
                     }
                 }
             }

src/InformaticsGateway/Services/Storage/IObjectUploadQueue.cs

@@ -16,6 +16,7 @@
  */
 
 using System.Threading;
+using System.Threading.Tasks;
 using Monai.Deploy.InformaticsGateway.Api.Storage;
 
 namespace Monai.Deploy.InformaticsGateway.Services.Storage
@@ -36,6 +37,6 @@ internal interface IObjectUploadQueue
         /// The default implementation blocks the call until a file is available from the queue.
         /// </summary>
         /// <param name="cancellationToken">Propagates notification that operations should be canceled.</param>
-        FileStorageMetadata Dequeue(CancellationToken cancellationToken);
+        Task<FileStorageMetadata> Dequeue(CancellationToken cancellationToken);
     }
 }

src/InformaticsGateway/Services/Storage/ObjectUploadQueue.cs

@@ -16,9 +16,9 @@
 
 using System;
 using System.Collections.Concurrent;
-using System.Collections.Generic;
 using System.Diagnostics;
 using System.Threading;
+using System.Threading.Tasks;
 using Ardalis.GuardClauses;
 using Microsoft.Extensions.Logging;
 using Monai.Deploy.InformaticsGateway.Api.Storage;
@@ -47,7 +47,7 @@ public void Queue(FileStorageMetadata file)
             _logger.InstanceAddedToUploadQueue(file.Id, _workItems.Count, process.WorkingSet64 / 1024.0);
         }
 
-        public FileStorageMetadata Dequeue(CancellationToken cancellationToken)
+        public async Task<FileStorageMetadata> Dequeue(CancellationToken cancellationToken)
         {
             while (!cancellationToken.IsCancellationRequested)
             {
@@ -56,6 +56,10 @@ public FileStorageMetadata Dequeue(CancellationToken cancellationToken)
                     _logger.InstanceInUploadQueue(_workItems.Count);
                     return reuslt;
                 }
+                if (_workItems.IsEmpty)
+                {
+                    await Task.Delay(100, cancellationToken).ConfigureAwait(false);
+                }
             }
             throw new OperationCanceledException("Cancellation requested.");
         }

src/InformaticsGateway/Services/Storage/ObjectUploadService.cs

@@ -20,7 +20,6 @@
 using System.Diagnostics;
 using System.Threading;
 using System.Threading.Tasks;
-using System.Threading.Tasks.Dataflow;
 using Ardalis.GuardClauses;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -32,7 +31,6 @@
 using Monai.Deploy.InformaticsGateway.Common;
 using Monai.Deploy.InformaticsGateway.Configuration;
 using Monai.Deploy.InformaticsGateway.Logging;
-using Monai.Deploy.InformaticsGateway.Repositories;
 using Monai.Deploy.InformaticsGateway.Services.Common;
 using Monai.Deploy.Storage.API;
 using Polly;
@@ -106,7 +104,7 @@ private async Task StartWorker(int thread, CancellationToken cancellationToken)
             {
                 try
                 {
-                    var item = _uplaodQueue.Dequeue(cancellationToken);
+                    var item = await _uplaodQueue.Dequeue(cancellationToken);
                     await ProcessObject(item);
                 }
                 catch (OperationCanceledException ex)