From bd0c1296bb23e2afd15da3741fdedc078e7149af Mon Sep 17 00:00:00 2001
From: Victor Chang <vicchang@nvidia.com>
Date: Thu, 6 Oct 2022 15:07:23 -0700
Subject: [PATCH 1/3] gh-199 Include export status for every file

Signed-off-by: Victor Chang <vicchang@nvidia.com>
---
 ...Monai.Deploy.InformaticsGateway.Api.csproj |  2 +-
 ...oy.InformaticsGateway.Configuration.csproj |  4 +-
 .../Logging/Log.500.ExportService.cs          |  2 +-
 .../Monai.Deploy.InformaticsGateway.csproj    |  4 +-
 .../Services/Export/DicomWebExportService.cs  | 21 +++++--
 .../Export/ExportRequestDataMessage.cs        |  9 ++-
 .../Export/ExportRequestEventDetails.cs       |  6 +-
 .../Services/Export/ExportServiceBase.cs      |  6 +-
 .../Services/Export/ScuExportService.cs       | 23 ++++++--
 .../Export/DicomWebExportServiceTest.cs       | 18 ++++--
 .../Services/Export/ExportServiceBaseTest.cs  | 57 ++++++++++++++++++-
 .../Services/Export/ScuExportServiceTest.cs   | 31 ++++++----
 ...InformaticsGateway.Integration.Test.csproj |  4 +-
 13 files changed, 148 insertions(+), 39 deletions(-)

diff --git a/src/Api/Monai.Deploy.InformaticsGateway.Api.csproj b/src/Api/Monai.Deploy.InformaticsGateway.Api.csproj
index 116e3cbfa..ade970aad 100644
--- a/src/Api/Monai.Deploy.InformaticsGateway.Api.csproj
+++ b/src/Api/Monai.Deploy.InformaticsGateway.Api.csproj
@@ -30,7 +30,7 @@
     </PackageReference>
     <PackageReference Include="Macross.Json.Extensions" Version="3.0.0" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.Abstractions" Version="6.0.9" />
-    <PackageReference Include="Monai.Deploy.Messaging" Version="0.1.6" />
+    <PackageReference Include="Monai.Deploy.Messaging" Version="0.1.7-rc0009" />
     <PackageReference Include="Monai.Deploy.Storage" Version="0.2.7" />
   </ItemGroup>
 
diff --git a/src/Configuration/Monai.Deploy.InformaticsGateway.Configuration.csproj b/src/Configuration/Monai.Deploy.InformaticsGateway.Configuration.csproj
index 8b45a0cbb..2ca1f6785 100644
--- a/src/Configuration/Monai.Deploy.InformaticsGateway.Configuration.csproj
+++ b/src/Configuration/Monai.Deploy.InformaticsGateway.Configuration.csproj
@@ -1,4 +1,4 @@
-<!--
+<!--
   ~ Copyright 2021-2022 MONAI Consortium
   ~
   ~ Licensed under the Apache License, Version 2.0 (the "License");
@@ -30,7 +30,7 @@
     </PackageReference>
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="6.0.2" />
     <PackageReference Include="Microsoft.Extensions.Options" Version="6.0.0" />
-    <PackageReference Include="Monai.Deploy.Messaging" Version="0.1.6" />
+    <PackageReference Include="Monai.Deploy.Messaging" Version="0.1.7-rc0009" />
     <PackageReference Include="Monai.Deploy.Storage" Version="0.2.7" />
     <PackageReference Include="System.IO.Abstractions" Version="17.2.3" />
   </ItemGroup>
diff --git a/src/InformaticsGateway/Logging/Log.500.ExportService.cs b/src/InformaticsGateway/Logging/Log.500.ExportService.cs
index 6c2b6a1a8..e270b933f 100644
--- a/src/InformaticsGateway/Logging/Log.500.ExportService.cs
+++ b/src/InformaticsGateway/Logging/Log.500.ExportService.cs
@@ -109,7 +109,7 @@ public static partial class Log
         [LoggerMessage(EventId = 528, Level = LogLevel.Error, Message = "Failed to export with error {status}.")]
         public static partial void DimseExportInstanceError(this ILogger logger, DicomStatus status);
 
-        [LoggerMessage(EventId = 529, Level = LogLevel.Error, Message = "Error while adding DICOM C-STORE request: {message}")]
+        [LoggerMessage(EventId = 529, Level = LogLevel.Error, Message = "{message}")]
         public static partial void DimseExportErrorAddingInstance(this ILogger logger, string message, Exception ex);
 
         [LoggerMessage(EventId = 530, Level = LogLevel.Error, Message = "{message}")]
diff --git a/src/InformaticsGateway/Monai.Deploy.InformaticsGateway.csproj b/src/InformaticsGateway/Monai.Deploy.InformaticsGateway.csproj
index 6f66a08b6..814d2c8be 100644
--- a/src/InformaticsGateway/Monai.Deploy.InformaticsGateway.csproj
+++ b/src/InformaticsGateway/Monai.Deploy.InformaticsGateway.csproj
@@ -1,4 +1,4 @@
-<!--
+<!--
   ~ Copyright 2022 MONAI Consortium
   ~
   ~ Licensed under the Apache License, Version 2.0 (the "License");
@@ -48,7 +48,7 @@
     <PackageReference Include="Microsoft.Extensions.Logging" Version="6.0.0" />
     <PackageReference Include="Microsoft.Extensions.Logging.Console" Version="6.0.0" />
     <PackageReference Include="Microsoft.Extensions.Options" Version="6.0.0" />
-    <PackageReference Include="Monai.Deploy.Messaging.RabbitMQ" Version="0.1.6" />
+    <PackageReference Include="Monai.Deploy.Messaging.RabbitMQ" Version="0.1.7-rc0009" />
     <PackageReference Include="Monai.Deploy.Storage" Version="0.2.7" />
     <PackageReference Include="Monai.Deploy.Storage.MinIO" Version="0.2.7" />
     <PackageReference Include="Polly" Version="7.2.3" />
diff --git a/src/InformaticsGateway/Services/Export/DicomWebExportService.cs b/src/InformaticsGateway/Services/Export/DicomWebExportService.cs
index 317455da6..691393e5e 100644
--- a/src/InformaticsGateway/Services/Export/DicomWebExportService.cs
+++ b/src/InformaticsGateway/Services/Export/DicomWebExportService.cs
@@ -35,6 +35,7 @@
 using Monai.Deploy.InformaticsGateway.Logging;
 using Monai.Deploy.InformaticsGateway.Repositories;
 using Monai.Deploy.InformaticsGateway.Services.Common;
+using Monai.Deploy.Messaging.Events;
 using Polly;
 
 namespace Monai.Deploy.InformaticsGateway.Services.Export
@@ -98,7 +99,7 @@ private async Task HandleTransaction(ExportRequestDataMessage exportRequestData,
             {
                 var errorMessage = $"The specified inference request '{transaction}' cannot be found and will not be exported.";
                 _logger.InferenceRequestExportDestinationNotFound(transaction);
-                exportRequestData.SetFailed(errorMessage);
+                exportRequestData.SetFailed(FileExportStatus.ConfigurationError, errorMessage);
                 return;
             }
 
@@ -108,7 +109,7 @@ private async Task HandleTransaction(ExportRequestDataMessage exportRequestData,
             {
                 var errorMessage = "The inference request '{transaction}' contains no `outputResources` nor any DICOMweb export destinations.";
                 _logger.InferenceRequestExportNoDestinationNotFound();
-                exportRequestData.SetFailed(errorMessage);
+                exportRequestData.SetFailed(FileExportStatus.ConfigurationError, errorMessage);
                 return;
             }
 
@@ -126,9 +127,21 @@ private async Task HandleTransaction(ExportRequestDataMessage exportRequestData,
 
         private async Task ExportToDicomWebDestination(IDicomWebClient dicomWebClient, ExportRequestDataMessage exportRequestData, RequestOutputDataResource destination, CancellationToken cancellationToken)
         {
+            DicomFile dicomFile;
+            try
+            {
+                dicomFile = _dicomToolkit.Load(exportRequestData.FileContent);
+            }
+            catch (Exception ex)
+            {
+                var errorMessage = $"Error reading DICOM file: {ex.Message}.";
+                _logger.ExportException(errorMessage, ex);
+                exportRequestData.SetFailed(FileExportStatus.UnsupportedDataType, errorMessage);
+                return;
+            }
+
             try
             {
-                var dicomFile = _dicomToolkit.Load(exportRequestData.FileContent);
                 await Policy
                    .Handle<Exception>()
                    .WaitAndRetryAsync(
@@ -147,7 +160,7 @@ await Policy
             {
                 var errorMessage = ex.Message;
                 _logger.ExportException(errorMessage, ex);
-                exportRequestData.SetFailed(errorMessage);
+                exportRequestData.SetFailed(FileExportStatus.ServiceError, errorMessage);
             }
         }
 
diff --git a/src/InformaticsGateway/Services/Export/ExportRequestDataMessage.cs b/src/InformaticsGateway/Services/Export/ExportRequestDataMessage.cs
index 0b02b5f1c..4b0a3baf0 100644
--- a/src/InformaticsGateway/Services/Export/ExportRequestDataMessage.cs
+++ b/src/InformaticsGateway/Services/Export/ExportRequestDataMessage.cs
@@ -28,6 +28,8 @@ public class ExportRequestDataMessage
         public byte[] FileContent { get; private set; }
         public bool IsFailed { get; private set; }
         public IList<string> Messages { get; init; }
+        public FileExportStatus ExportStatus { get; private set; }
+        public string Filename { get; }
 
         public string ExportTaskId
         {
@@ -44,8 +46,6 @@ public string[] Destinations
             get { return _exportRequest.Destinations; }
         }
 
-        public string Filename { get; }
-
         public ExportRequestDataMessage(ExportRequestEvent exportRequest, string filename)
         {
             IsFailed = false;
@@ -53,6 +53,7 @@ public ExportRequestDataMessage(ExportRequestEvent exportRequest, string filenam
 
             _exportRequest = exportRequest ?? throw new System.ArgumentNullException(nameof(exportRequest));
             Filename = filename ?? throw new System.ArgumentNullException(nameof(filename));
+            ExportStatus = FileExportStatus.Success;
         }
 
         public void SetData(byte[] data)
@@ -61,9 +62,11 @@ public void SetData(byte[] data)
             FileContent = data;
         }
 
-        public void SetFailed(string errorMessage)
+        public void SetFailed(FileExportStatus fileExportStatus, string errorMessage)
         {
             Guard.Against.NullOrWhiteSpace(errorMessage, nameof(errorMessage));
+
+            ExportStatus = fileExportStatus;
             IsFailed = true;
             Messages.Add(errorMessage);
         }
diff --git a/src/InformaticsGateway/Services/Export/ExportRequestEventDetails.cs b/src/InformaticsGateway/Services/Export/ExportRequestEventDetails.cs
index 250f43a2c..2bbda686a 100644
--- a/src/InformaticsGateway/Services/Export/ExportRequestEventDetails.cs
+++ b/src/InformaticsGateway/Services/Export/ExportRequestEventDetails.cs
@@ -50,8 +50,10 @@ public ExportRequestEventDetails(ExportRequestEvent exportRequest)
         /// <summary>
         /// Gets whether the export task is completed or not based on file count.
         /// </summary>
-        public bool IsCompleted
-        { get { return (SucceededFiles + FailedFiles) == Files.Count(); } }
+        public bool IsCompleted { get { return (SucceededFiles + FailedFiles) == Files.Count(); } }
+
+        public Dictionary<string, FileExportStatus> FileStatuses { get; private set; } = new Dictionary<string, FileExportStatus>();
+
 
         public ExportStatus Status
         {
diff --git a/src/InformaticsGateway/Services/Export/ExportServiceBase.cs b/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
index 07d9e0741..f4eced5b1 100644
--- a/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
+++ b/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
@@ -229,7 +229,7 @@ private IEnumerable<ExportRequestDataMessage> DownloadPayloadActionCallback(Expo
                 {
                     var errorMessage = $"Error downloading payload.";
                     _logger.ErrorDownloadingPayload(ex);
-                    exportRequestData.SetFailed(errorMessage);
+                    exportRequestData.SetFailed(FileExportStatus.DownloadError, errorMessage);
                 }
 
                 yield return exportRequestData;
@@ -243,6 +243,7 @@ private void ReportingActionBlock(ExportRequestDataMessage exportRequestData)
             var exportRequest = _exportRequests[exportRequestData.ExportTaskId];
             lock (SyncRoot)
             {
+                exportRequest.FileStatuses.Add(exportRequestData.Filename, exportRequestData.ExportStatus);
                 if (exportRequestData.IsFailed)
                 {
                     exportRequest.FailedFiles++;
@@ -265,7 +266,8 @@ private void ReportingActionBlock(ExportRequestDataMessage exportRequestData)
 
             _logger.ExportCompleted(exportRequest.FailedFiles, exportRequest.Files.Count());
 
-            var exportCompleteEvent = new ExportCompleteEvent(exportRequest, exportRequest.Status);
+            var exportCompleteEvent = new ExportCompleteEvent(exportRequest, exportRequest.Status, exportRequest.FileStatuses);
+
             var jsonMessage = new JsonMessage<ExportCompleteEvent>(exportCompleteEvent, MessageBrokerConfiguration.InformaticsGatewayApplicationId, exportRequest.CorrelationId, exportRequest.DeliveryTag);
 
             Policy
diff --git a/src/InformaticsGateway/Services/Export/ScuExportService.cs b/src/InformaticsGateway/Services/Export/ScuExportService.cs
index ea740990e..60372b993 100644
--- a/src/InformaticsGateway/Services/Export/ScuExportService.cs
+++ b/src/InformaticsGateway/Services/Export/ScuExportService.cs
@@ -20,6 +20,7 @@
 using System.Threading;
 using System.Threading.Tasks;
 using Ardalis.GuardClauses;
+using FellowOakDicom;
 using FellowOakDicom.Network;
 using FellowOakDicom.Network.Client;
 using Microsoft.Extensions.DependencyInjection;
@@ -30,6 +31,7 @@
 using Monai.Deploy.InformaticsGateway.Configuration;
 using Monai.Deploy.InformaticsGateway.Logging;
 using Monai.Deploy.InformaticsGateway.Repositories;
+using Monai.Deploy.Messaging.Events;
 using Polly;
 
 namespace Monai.Deploy.InformaticsGateway.Services.Export
@@ -86,7 +88,7 @@ private async Task HandleDesination(ExportRequestDataMessage exportRequestData,
             catch (ConfigurationException ex)
             {
                 _logger.ScuExportConfigurationError(ex.Message, ex);
-                exportRequestData.SetFailed(ex.Message);
+                exportRequestData.SetFailed(FileExportStatus.ConfigurationError, ex.Message);
                 return;
             }
 
@@ -155,10 +157,21 @@ private async Task<bool> GenerateRequestsAsync(
             IDicomClient client,
             ManualResetEvent manualResetEvent)
         {
+            DicomFile dicomFile;
             try
             {
-                var dicomFile = _dicomToolkit.Load(exportRequestData.FileContent);
+                dicomFile = _dicomToolkit.Load(exportRequestData.FileContent);
+            }
+            catch (Exception ex)
+            {
+                var errorMessage = $"Error reading DICOM file: {ex.Message}";
+                _logger.ExportException(errorMessage, ex);
+                exportRequestData.SetFailed(FileExportStatus.UnsupportedDataType, errorMessage);
+                return false;
+            }
 
+            try
+            {
                 var request = new DicomCStoreRequest(dicomFile);
 
                 request.OnResponseReceived += (req, response) =>
@@ -171,7 +184,7 @@ private async Task<bool> GenerateRequestsAsync(
                     {
                         var errorMessage = $"Failed to export with error {response.Status}";
                         _logger.DimseExportInstanceError(response.Status);
-                        exportRequestData.SetFailed(errorMessage);
+                        exportRequestData.SetFailed(FileExportStatus.ServiceError, errorMessage);
                     }
                     manualResetEvent.Set();
                 };
@@ -183,7 +196,7 @@ private async Task<bool> GenerateRequestsAsync(
             {
                 var errorMessage = $"Error while adding DICOM C-STORE request: {exception.Message}";
                 _logger.DimseExportErrorAddingInstance(exception.Message, exception);
-                exportRequestData.SetFailed(errorMessage);
+                exportRequestData.SetFailed(FileExportStatus.ServiceError, errorMessage);
                 return false;
             }
         }
@@ -213,7 +226,7 @@ private void HandleCStoreException(Exception ex, ExportRequestDataMessage export
             }
 
             _logger.ExportException(errorMessage, ex);
-            exportRequestData.SetFailed(errorMessage);
+            exportRequestData.SetFailed(FileExportStatus.ServiceError, errorMessage);
         }
     }
 }
diff --git a/src/InformaticsGateway/Test/Services/Export/DicomWebExportServiceTest.cs b/src/InformaticsGateway/Test/Services/Export/DicomWebExportServiceTest.cs
index 2ef90730c..3d4e46c40 100644
--- a/src/InformaticsGateway/Test/Services/Export/DicomWebExportServiceTest.cs
+++ b/src/InformaticsGateway/Test/Services/Export/DicomWebExportServiceTest.cs
@@ -22,6 +22,7 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Ardalis.GuardClauses;
 using FellowOakDicom;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
@@ -160,7 +161,7 @@ public async Task ExportDataBlockCallback_ReturnsNullIfInferenceRequestCannotBeF
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ConfigurationError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -215,7 +216,7 @@ public async Task ExportDataBlockCallback_ReturnsNullIfInferenceRequestContainsN
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ConfigurationError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -294,7 +295,7 @@ public async Task ExportDataBlockCallback_RecordsStowFailuresAndReportFailure()
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ServiceError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -382,7 +383,7 @@ public async Task CompletesDataflow(HttpStatusCode httpStatusCode)
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == (httpStatusCode == HttpStatusCode.OK ? ExportStatus.Success : ExportStatus.Failure))), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, (httpStatusCode == HttpStatusCode.OK ? ExportStatus.Success : ExportStatus.Failure), (httpStatusCode == HttpStatusCode.OK ? FileExportStatus.Success : FileExportStatus.ServiceError)))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -410,6 +411,15 @@ public async Task CompletesDataflow(HttpStatusCode httpStatusCode)
                ItExpr.IsAny<CancellationToken>());
         }
 
+        private bool CheckMessage(Message message, ExportStatus exportStatus, FileExportStatus fileExportStatus)
+        {
+            Guard.Against.Null(message, nameof(message));
+
+            var exportEvent = message.ConvertTo<ExportCompleteEvent>();
+            return exportEvent.Status == exportStatus &&
+                    exportEvent.FileStatuses.First().Value == fileExportStatus;
+        }
+
         private static MessageReceivedEventArgs CreateMessageReceivedEventArgs(string transactionId)
         {
             var exportRequestEvent = new ExportRequestEvent
diff --git a/src/InformaticsGateway/Test/Services/Export/ExportServiceBaseTest.cs b/src/InformaticsGateway/Test/Services/Export/ExportServiceBaseTest.cs
index 5025c241b..a014b496b 100644
--- a/src/InformaticsGateway/Test/Services/Export/ExportServiceBaseTest.cs
+++ b/src/InformaticsGateway/Test/Services/Export/ExportServiceBaseTest.cs
@@ -62,7 +62,7 @@ protected override Task<ExportRequestDataMessage> ExportDataBlockCallback(Export
 
             if (ExportShallFail || exportRequestData is null)
             {
-                exportRequestData.SetFailed("Failed");
+                exportRequestData.SetFailed(FileExportStatus.ServiceError, "Failed");
             }
 
             return Task.FromResult(exportRequestData);
@@ -195,6 +195,61 @@ public async Task DataflowTest_PayloadDownlaodFailure()
                                                               It.IsAny<ushort>()), Times.Once());
         }
 
+
+        [RetryFact(1, 10, DisplayName = "Data flow test - end to end workflow with partial failure")]
+        public async Task DataflowTest_EndToEnd_WithPartialFailure()
+        {
+            var messageCount = 5;
+            var testData = "this is a test";
+
+            _messagePublisherService.Setup(p => p.Publish(It.IsAny<string>(), It.IsAny<Message>()));
+            _messageSubscriberService.Setup(p => p.Acknowledge(It.IsAny<MessageBase>()));
+            _messageSubscriberService.Setup(p => p.RequeueWithDelay(It.IsAny<MessageBase>()));
+            _messageSubscriberService.Setup(
+                p => p.Subscribe(It.IsAny<string>(),
+                                 It.IsAny<string>(),
+                                 It.IsAny<Action<MessageReceivedEventArgs>>(),
+                                 It.IsAny<ushort>()))
+                .Callback<string, string, Action<MessageReceivedEventArgs>, ushort>((topic, queue, messageReceivedCallback, prefetchCount) =>
+                {
+                    while (messageCount-- > 0)
+                    {
+                        messageReceivedCallback(CreateMessageReceivedEventArgs());
+                    }
+                });
+
+            _storageService.Setup(p => p.GetObjectAsync(It.IsAny<string>(), It.IsAny<string>(), It.IsAny<CancellationToken>()))
+                .ReturnsAsync(new MemoryStream(Encoding.UTF8.GetBytes(testData)));
+
+            var countdownEvent = new CountdownEvent(5 * 3);
+            var service = new TestExportService(_logger.Object, _configuration, _serviceScopeFactory.Object);
+            service.ReportActionCompleted += (sender, e) =>
+            {
+                countdownEvent.Signal();
+            };
+            service.ExportDataBlockCalled += (sender, e) =>
+            {
+                var data = sender as ExportRequestDataMessage;
+                Assert.Equal(testData, Encoding.UTF8.GetString(data.FileContent));
+
+                service.ExportShallFail = data.Filename == "file1";
+                countdownEvent.Signal();
+            };
+            await service.StartAsync(_cancellationTokenSource.Token);
+            Assert.True(countdownEvent.Wait(1000000));
+            await StopAndVerify(service);
+
+            _messagePublisherService.Verify(
+                p => p.Publish(It.IsAny<string>(),
+                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.PartialFailure)), Times.Exactly(5));
+            _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Exactly(5));
+            _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
+            _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
+                                                              It.IsAny<string>(),
+                                                              It.IsAny<Action<MessageReceivedEventArgs>>(),
+                                                              It.IsAny<ushort>()), Times.Once());
+        }
+
         [RetryFact(1, 10, DisplayName = "Data flow test - end to end workflow")]
         public async Task DataflowTest_EndToEnd()
         {
diff --git a/src/InformaticsGateway/Test/Services/Export/ScuExportServiceTest.cs b/src/InformaticsGateway/Test/Services/Export/ScuExportServiceTest.cs
index e53f812be..06cfd5c51 100644
--- a/src/InformaticsGateway/Test/Services/Export/ScuExportServiceTest.cs
+++ b/src/InformaticsGateway/Test/Services/Export/ScuExportServiceTest.cs
@@ -16,9 +16,11 @@
 
 using System;
 using System.IO;
+using System.Linq;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using Ardalis.GuardClauses;
 using FellowOakDicom;
 using FellowOakDicom.Network;
 using Microsoft.Extensions.DependencyInjection;
@@ -146,7 +148,7 @@ public async Task ShallFailWhenNoDestinationIsDefined()
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ConfigurationError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -191,7 +193,7 @@ public async Task ShallFailWhenDestinationIsNotConfigured()
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ConfigurationError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -241,7 +243,7 @@ public async Task AssociationRejected()
             await StopAndVerify(service);
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ServiceError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -293,7 +295,7 @@ public async Task SimulateAbort()
             await StopAndVerify(service);
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ServiceError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -344,7 +346,7 @@ public async Task CStoreFailure()
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ServiceError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -395,7 +397,7 @@ public async Task ErrorLoadingDicomContent()
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.UnsupportedDataType))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -403,7 +405,7 @@ public async Task ErrorLoadingDicomContent()
                                                               It.IsAny<Action<MessageReceivedEventArgs>>(),
                                                               It.IsAny<ushort>()), Times.Once());
 
-            _logger.VerifyLogging("Error while adding DICOM C-STORE request: error", LogLevel.Error, Times.Once());
+            _logger.VerifyLoggingMessageBeginsWith("Error reading DICOM file: error", LogLevel.Error, Times.Once());
         }
 
         [RetryFact(10, 250, DisplayName = "Unreachable Server")]
@@ -446,7 +448,7 @@ public async Task UnreachableServer()
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Failure)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Failure, FileExportStatus.ServiceError))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -456,7 +458,7 @@ public async Task UnreachableServer()
             _logger.VerifyLoggingMessageBeginsWith("Association aborted with error", LogLevel.Error, Times.Once());
         }
 
-        [RetryFact(10, 250, DisplayName = "C-STORe success")]
+        [RetryFact(10, 250, DisplayName = "C-STORE success")]
         public async Task ExportCompletes()
         {
             _scpLogger.Invocations.Clear();
@@ -496,7 +498,7 @@ public async Task ExportCompletes()
 
             _messagePublisherService.Verify(
                 p => p.Publish(It.IsAny<string>(),
-                               It.Is<Message>(match => (match.ConvertTo<ExportCompleteEvent>()).Status == ExportStatus.Success)), Times.Once());
+                               It.Is<Message>(match => CheckMessage(match, ExportStatus.Success, FileExportStatus.Success))), Times.Once());
             _messageSubscriberService.Verify(p => p.Acknowledge(It.IsAny<MessageBase>()), Times.Once());
             _messageSubscriberService.Verify(p => p.RequeueWithDelay(It.IsAny<MessageBase>()), Times.Never());
             _messageSubscriberService.Verify(p => p.Subscribe(It.IsAny<string>(),
@@ -507,6 +509,15 @@ public async Task ExportCompletes()
             _logger.VerifyLogging($"Instance sent successfully.", LogLevel.Information, Times.Once());
         }
 
+        private bool CheckMessage(Message message, ExportStatus exportStatus, FileExportStatus fileExportStatus)
+        {
+            Guard.Against.Null(message, nameof(message));
+
+            var exportEvent = message.ConvertTo<ExportCompleteEvent>();
+            return exportEvent.Status == exportStatus &&
+                    exportEvent.FileStatuses.First().Value == fileExportStatus;
+        }
+
         private static MessageReceivedEventArgs CreateMessageReceivedEventArgs(string destination)
         {
             var exportRequestEvent = new ExportRequestEvent
diff --git a/tests/Integration.Test/Monai.Deploy.InformaticsGateway.Integration.Test.csproj b/tests/Integration.Test/Monai.Deploy.InformaticsGateway.Integration.Test.csproj
index 606f97f99..152857690 100644
--- a/tests/Integration.Test/Monai.Deploy.InformaticsGateway.Integration.Test.csproj
+++ b/tests/Integration.Test/Monai.Deploy.InformaticsGateway.Integration.Test.csproj
@@ -1,4 +1,4 @@
-<!--
+<!--
   ~ Copyright 2022 MONAI Consortium
   ~
   ~ Licensed under the Apache License, Version 2.0 (the "License");
@@ -33,7 +33,7 @@
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="6.0.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.3.2" />
     <PackageReference Include="Minio" Version="4.0.5" />
-    <PackageReference Include="Monai.Deploy.Messaging.RabbitMQ" Version="0.1.6" />
+    <PackageReference Include="Monai.Deploy.Messaging.RabbitMQ" Version="0.1.7-rc0009" />
     <PackageReference Include="Monai.Deploy.Storage.MinIO" Version="0.2.7" />
     <PackageReference Include="Moq" Version="4.18.2" />
     <PackageReference Include="Polly" Version="7.2.3" />

From 8ff68fa9c47d1c9b5be1cdc83c12cc3c4eb04f03 Mon Sep 17 00:00:00 2001
From: Victor Chang <vicchang@nvidia.com>
Date: Thu, 6 Oct 2022 15:36:17 -0700
Subject: [PATCH 2/3] gh-199 Improve logging

Signed-off-by: Victor Chang <vicchang@nvidia.com>
---
 .../Logging/Log.500.ExportService.cs                  | 11 ++++++++---
 .../Services/Export/DicomWebExportService.cs          |  2 +-
 .../Services/Export/ExportServiceBase.cs              |  3 ++-
 3 files changed, 11 insertions(+), 5 deletions(-)

diff --git a/src/InformaticsGateway/Logging/Log.500.ExportService.cs b/src/InformaticsGateway/Logging/Log.500.ExportService.cs
index e270b933f..da3997bd8 100644
--- a/src/InformaticsGateway/Logging/Log.500.ExportService.cs
+++ b/src/InformaticsGateway/Logging/Log.500.ExportService.cs
@@ -28,8 +28,8 @@ public static partial class Log
         [LoggerMessage(EventId = 501, Level = LogLevel.Warning, Message = "{ServiceName} paused due to insufficient storage space.  Available storage space: {availableFreeSpace:D}.")]
         public static partial void ExportPausedDueToInsufficientStorageSpace(this ILogger logger, string serviceName, long availableFreeSpace);
 
-        [LoggerMessage(EventId = 502, Level = LogLevel.Warning, Message = "The export request {exportTaskId} is already queued for export.")]
-        public static partial void ExportRequestAlreadyQueued(this ILogger logger, string exportTaskId);
+        [LoggerMessage(EventId = 502, Level = LogLevel.Warning, Message = "Correlation ID={correlationId}. The export request {exportTaskId} is already queued for export.")]
+        public static partial void ExportRequestAlreadyQueued(this ILogger logger, string correlationId, string exportTaskId);
 
         [LoggerMessage(EventId = 503, Level = LogLevel.Debug, Message = "Downloading {file}.")]
         public static partial void DownloadingFile(this ILogger logger, string file);
@@ -49,7 +49,7 @@ public static partial class Log
         [LoggerMessage(EventId = 508, Level = LogLevel.Error, Message = "Error acknowledging message. Waiting {timeSpan} before next retry. Retry attempt {retryCount}.")]
         public static partial void ErrorAcknowledgingMessageWithRetry(this ILogger logger, Exception ex, TimeSpan timeSpan, int retryCount);
 
-        [LoggerMessage(EventId = 509, Level = LogLevel.Information, Message = "Sending acknowledgement.")]
+        [LoggerMessage(EventId = 509, Level = LogLevel.Information, Message = "Sending acknowledgment.")]
         public static partial void SendingAckowledgement(this ILogger logger);
 
         [LoggerMessage(EventId = 510, Level = LogLevel.Error, Message = "Error publishing message. Waiting {timeSpan} before next retry. Retry attempt {retryCount}.")]
@@ -117,5 +117,10 @@ public static partial class Log
 
         [LoggerMessage(EventId = 531, Level = LogLevel.Warning, Message = "Export service paused due to insufficient storage space.  Available storage space: {availableFreeSpace:D}")]
         public static partial void ExportServiceStoppedDueToLowStorageSpace(this ILogger logger, long availableFreeSpace);
+
+        [LoggerMessage(EventId = 532, Level = LogLevel.Information, Message = "Correlation ID={correlationId}. Export request {exportTaskId} received & queued for processing.")]
+        public static partial void ExportRequestQueuedForProcessing(this ILogger logger, string correlationId, string exportTaskId);
+
+
     }
 }
diff --git a/src/InformaticsGateway/Services/Export/DicomWebExportService.cs b/src/InformaticsGateway/Services/Export/DicomWebExportService.cs
index 691393e5e..22d51bce7 100644
--- a/src/InformaticsGateway/Services/Export/DicomWebExportService.cs
+++ b/src/InformaticsGateway/Services/Export/DicomWebExportService.cs
@@ -75,7 +75,7 @@ public DicomWebExportService(
 
         protected override async Task<ExportRequestDataMessage> ExportDataBlockCallback(ExportRequestDataMessage exportRequestData, CancellationToken cancellationToken)
         {
-            using var loggerScope = _logger.BeginScope(new LoggingDataDictionary<string, object> { { "ExportTaskId", exportRequestData.ExportTaskId }, { "CorrelationId", exportRequestData.CorrelationId } });
+            using var loggerScope = _logger.BeginScope(new LoggingDataDictionary<string, object> { { "ExportTaskId", exportRequestData.ExportTaskId }, { "CorrelationId", exportRequestData.CorrelationId }, { "Filename", exportRequestData.Filename } });
 
             using var scope = _serviceScopeFactory.CreateScope();
             var repository = scope.ServiceProvider.GetRequiredService<IInferenceRequestRepository>();
diff --git a/src/InformaticsGateway/Services/Export/ExportServiceBase.cs b/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
index f4eced5b1..41dca6aed 100644
--- a/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
+++ b/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
@@ -163,7 +163,7 @@ private void OnMessageReceivedCallback(MessageReceivedEventArgs eventArgs)
                     var exportRequest = eventArgs.Message.ConvertTo<ExportRequestEvent>();
                     if (_exportRequests.ContainsKey(exportRequest.ExportTaskId))
                     {
-                        _logger.ExportRequestAlreadyQueued(exportRequest.ExportTaskId);
+                        _logger.ExportRequestAlreadyQueued(exportRequest.CorrelationId, exportRequest.ExportTaskId);
                         return;
                     }
 
@@ -174,6 +174,7 @@ private void OnMessageReceivedCallback(MessageReceivedEventArgs eventArgs)
 
                     _exportRequests.Add(exportRequest.ExportTaskId, exportRequestWithDetails);
                     exportFlow.Post(exportRequestWithDetails);
+                    _logger.ExportRequestQueuedForProcessing(exportRequest.CorrelationId, exportRequest.ExportTaskId);
                 }
 
                 exportFlow.Complete();

From f504bdb4881e5585bbd43d0196b140b71d44e202 Mon Sep 17 00:00:00 2001
From: Victor Chang <vicchang@nvidia.com>
Date: Thu, 6 Oct 2022 16:22:16 -0700
Subject: [PATCH 3/3] gh-199 Fix typo

Signed-off-by: Victor Chang <vicchang@nvidia.com>
---
 src/InformaticsGateway/Logging/Log.500.ExportService.cs     | 2 +-
 src/InformaticsGateway/Services/Export/ExportServiceBase.cs | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/InformaticsGateway/Logging/Log.500.ExportService.cs b/src/InformaticsGateway/Logging/Log.500.ExportService.cs
index da3997bd8..3e26a6832 100644
--- a/src/InformaticsGateway/Logging/Log.500.ExportService.cs
+++ b/src/InformaticsGateway/Logging/Log.500.ExportService.cs
@@ -50,7 +50,7 @@ public static partial class Log
         public static partial void ErrorAcknowledgingMessageWithRetry(this ILogger logger, Exception ex, TimeSpan timeSpan, int retryCount);
 
         [LoggerMessage(EventId = 509, Level = LogLevel.Information, Message = "Sending acknowledgment.")]
-        public static partial void SendingAckowledgement(this ILogger logger);
+        public static partial void SendingAcknowledgement(this ILogger logger);
 
         [LoggerMessage(EventId = 510, Level = LogLevel.Error, Message = "Error publishing message. Waiting {timeSpan} before next retry. Retry attempt {retryCount}.")]
         public static partial void ErrorPublishingExportCompleteEventWithRetry(this ILogger logger, Exception ex, TimeSpan timeSpan, int retryCount);
diff --git a/src/InformaticsGateway/Services/Export/ExportServiceBase.cs b/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
index 41dca6aed..21df1feb1 100644
--- a/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
+++ b/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
@@ -281,7 +281,7 @@ private void ReportingActionBlock(ExportRequestDataMessage exportRequestData)
                    })
                .Execute(() =>
                {
-                   _logger.SendingAckowledgement();
+                   _logger.SendingAcknowledgement();
                    _messageSubscriber.Acknowledge(jsonMessage);
                });