gh-56 fix to read all DICOM header when loading from stream

mocsharp · mocsharp · commit 50245ab6126e · 2022-03-10T09:56:20.000-08:00
- fix issue when exporting multiple files, only one get exported
diff --git a/src/InformaticsGateway/Common/DicomToolkit.cs b/src/InformaticsGateway/Common/DicomToolkit.cs
@@ -101,7 +101,8 @@ public DicomFile Load(byte[] fileContent)
             Guard.Against.NullOrEmpty(fileContent, nameof(fileContent));
 
             using var stream = new MemoryStream(fileContent);
-            var dicomFile = DicomFile.Open(stream);
+            var dicomFile = DicomFile.Open(stream, FileReadOption.ReadAll);
+
             if (dicomFile is null)
             {
                 throw new DicomDataException("Invalid DICOM content");
diff --git a/src/InformaticsGateway/Services/Export/DicomWebExportService.cs b/src/InformaticsGateway/Services/Export/DicomWebExportService.cs
@@ -59,7 +59,7 @@ internal class DicomWebExportService : ExportServiceBase
         private readonly IOptions<InformaticsGatewayConfiguration> _configuration;
         private readonly IDicomToolkit _dicomToolkit;
 
-        protected override int Concurrentcy { get; }
+        protected override int Concurrency { get; }
         public override string RoutingKey { get; }
         public override string ServiceName => "DICOMweb Export Service";
 
@@ -81,7 +81,7 @@ public DicomWebExportService(
             _dicomToolkit = dicomToolkit ?? throw new ArgumentNullException(nameof(dicomToolkit));
 
             RoutingKey = $"{configuration.Value.Messaging.Topics.ExportRequestPrefix}.{configuration.Value.DicomWeb.AgentName}";
-            Concurrentcy = configuration.Value.DicomWeb.MaximumNumberOfConnection;
+            Concurrency = configuration.Value.DicomWeb.MaximumNumberOfConnection;
         }
 
         protected override async Task<ExportRequestDataMessage> ExportDataBlockCallback(ExportRequestDataMessage exportRequestData, CancellationToken cancellationToken)
@@ -137,10 +137,10 @@ await Policy
                            _logger.Log(LogLevel.Error, exception, $"Error exporting to DICOMweb destination. Waiting {timeSpan} before next retry. Retry attempt {retryCount}.");
                        })
                    .ExecuteAsync(async () =>
-                       {
-                           var result = await dicomWebClient.Stow.Store(new List<DicomFile> { dicomFile }, cancellationToken);
-                           CheckAndLogResult(result);
-                       });
+                   {
+                       var result = await dicomWebClient.Stow.Store(new List<DicomFile> { dicomFile }, cancellationToken);
+                       CheckAndLogResult(result);
+                   });
             }
             catch (Exception ex)
             {
diff --git a/src/InformaticsGateway/Services/Export/ExportRequestDataMessage.cs b/src/InformaticsGateway/Services/Export/ExportRequestDataMessage.cs
@@ -55,12 +55,15 @@ public string Destination
             get { return _exportRequest.Destination; }
         }
 
-        public ExportRequestDataMessage(ExportRequestMessage exportRequest)
+        public string Filename { get; }
+
+        public ExportRequestDataMessage(ExportRequestMessage exportRequest, string filename)
         {
             IsFailed = false;
             Messages = new List<string>();
 
             _exportRequest = exportRequest ?? throw new System.ArgumentNullException(nameof(exportRequest));
+            Filename = filename ?? throw new System.ArgumentNullException(nameof(filename));
         }
 
         public void SetData(byte[] data)
diff --git a/src/InformaticsGateway/Services/Export/ExportServiceBase.cs b/src/InformaticsGateway/Services/Export/ExportServiceBase.cs
@@ -63,12 +63,12 @@ public abstract class ExportServiceBase : IHostedService, IMonaiService
         private readonly IMessageBrokerSubscriberService _messageSubscriber;
         private readonly IMessageBrokerPublisherService _messagePublisher;
         private readonly IServiceScope _scope;
-        private readonly Dictionary<string, ExportRequestMessage> _exportRequets;
+        private readonly Dictionary<string, ExportRequestMessage> _exportRequests;
         private readonly string _exportQueueName;
         private TransformManyBlock<ExportRequestMessage, ExportRequestDataMessage> _exportFlow;
 
         public abstract string RoutingKey { get; }
-        protected abstract int Concurrentcy { get; }
+        protected abstract int Concurrency { get; }
         public ServiceStatus Status { get; set; } = ServiceStatus.Unknown;
         public abstract string ServiceName { get; }
 
@@ -102,7 +102,7 @@ public ExportServiceBase(
             _messageSubscriber = _scope.ServiceProvider.GetRequiredService<IMessageBrokerSubscriberService>();
             _messagePublisher = _scope.ServiceProvider.GetRequiredService<IMessageBrokerPublisherService>();
 
-            _exportRequets = new Dictionary<string, ExportRequestMessage>();
+            _exportRequests = new Dictionary<string, ExportRequestMessage>();
             _exportQueueName = _configuration.Messaging.Topics.ExportRequestQueue;
         }
 
@@ -131,7 +131,7 @@ private void SetupExportFlow(CancellationToken cancellationToken)
             {
                 var executionOptions = new ExecutionDataflowBlockOptions
                 {
-                    MaxDegreeOfParallelism = Concurrentcy,
+                    MaxDegreeOfParallelism = Concurrency,
                     MaxMessagesPerTask = 1,
                     CancellationToken = cancellationToken
                 };
@@ -184,11 +184,22 @@ private void OnMessageReceivedCallback(MessageReceivedEventArgs eventArgs)
                 _messageSubscriber.Reject(eventArgs.Message);
                 return;
             }
-            var exportRequest = eventArgs.Message.ConvertTo<ExportRequestMessage>();
-            exportRequest.MessageId = eventArgs.Message.MessageId;
 
-            _exportRequets.Add(exportRequest.ExportTaskId, exportRequest);
-            _exportFlow.Post(exportRequest);
+            lock (SyncRoot)
+            {
+                var exportRequest = eventArgs.Message.ConvertTo<ExportRequestMessage>();
+                if (_exportRequests.ContainsKey(exportRequest.ExportTaskId))
+                {
+                    _logger.Log(LogLevel.Warning, $"The export request {exportRequest.ExportTaskId} is already queued for export.");
+                    return;
+                }
+
+                exportRequest.MessageId = eventArgs.Message.MessageId;
+                exportRequest.DeliveryTag = eventArgs.Message.DeliveryTag;
+
+                _exportRequests.Add(exportRequest.ExportTaskId, exportRequest);
+                _exportFlow.Post(exportRequest);
+            }
         }
 
         private IEnumerable<ExportRequestDataMessage> DownloadPayloadActionCallback(ExportRequestMessage exportRequest, CancellationToken cancellationToken)
@@ -198,31 +209,33 @@ private IEnumerable<ExportRequestDataMessage> DownloadPayloadActionCallback(Expo
             var scope = _serviceScopeFactory.CreateScope();
             var storageService = scope.ServiceProvider.GetRequiredService<IStorageService>();
 
-            var exportRequestData = new ExportRequestDataMessage(exportRequest);
             foreach (var file in exportRequest.Files)
             {
+                var exportRequestData = new ExportRequestDataMessage(exportRequest, file);
                 try
                 {
                     _logger.Log(LogLevel.Debug, $"Downloading file {file}...");
                     Policy
-                      .Handle<Exception>()
-                      .WaitAndRetry(
-                          _configuration.Export.Retries.RetryDelays,
-                          (exception, timeSpan, retryCount, context) =>
-                          {
-                              _logger.Log(LogLevel.Error, exception, $"Error downloading payload. Waiting {timeSpan} before next retry. Retry attempt {retryCount}.");
-                          })
-                      .Execute(() =>
-                      {
-                          storageService.GetObject(_configuration.Storage.StorageServiceBucketName, file, (stream) =>
-                          {
-                              _logger.Log(LogLevel.Debug, $"Copying file {file}...");
-                              using var memoryStream = new MemoryStream();
-                              stream.CopyTo(memoryStream);
-                              exportRequestData.SetData(memoryStream.ToArray());
-                              _logger.Log(LogLevel.Debug, $"File {file} ready for export...");
-                          }, cancellationToken).Wait();
-                      });
+                       .Handle<Exception>()
+                       .WaitAndRetry(
+                           _configuration.Export.Retries.RetryDelays,
+                           (exception, timeSpan, retryCount, context) =>
+                           {
+                               _logger.Log(LogLevel.Error, exception, $"Error downloading payload. Waiting {timeSpan} before next retry. Retry attempt {retryCount}.");
+                           })
+                       .Execute(() =>
+                       {
+                           _logger.Log(LogLevel.Debug, $"Downloading {file}...");
+                           var task = storageService.GetObject(_configuration.Storage.StorageServiceBucketName, file, (stream) =>
+                           {
+                               using var memoryStream = new MemoryStream();
+                               stream.CopyTo(memoryStream);
+                               exportRequestData.SetData(memoryStream.ToArray());
+                           }, cancellationToken);
+
+                           task.Wait();
+                           _logger.Log(LogLevel.Debug, $"File {file} ready for export.");
+                       });
                 }
                 catch (Exception ex)
                 {
@@ -239,7 +252,7 @@ private void ReportingActionBlock(ExportRequestDataMessage exportRequestData)
         {
             using var loggerScope = _logger.BeginScope(new LoggingDataDictionary<string, object> { { "ExportTaskId", exportRequestData.ExportTaskId }, { "CorrelationId", exportRequestData.CorrelationId } });
 
-            var exportRequest = _exportRequets[exportRequestData.ExportTaskId];
+            var exportRequest = _exportRequests[exportRequestData.ExportTaskId];
             lock (SyncRoot)
             {
                 if (exportRequestData.IsFailed)
@@ -300,6 +313,11 @@ private void ReportingActionBlock(ExportRequestDataMessage exportRequestData)
                    _logger.Log(LogLevel.Information, $"Publishing export complete message.");
                    _messagePublisher.Publish(_configuration.Messaging.Topics.ExportComplete, jsonMessage.ToMessage());
                });
+
+            lock(SyncRoot)
+            {
+                _exportRequests.Remove(exportRequestData.ExportTaskId);
+            }
         }
     }
 }
diff --git a/src/InformaticsGateway/Services/Export/ScuExportService.cs b/src/InformaticsGateway/Services/Export/ScuExportService.cs
@@ -52,7 +52,7 @@ internal class ScuExportService : ExportServiceBase
         private readonly IOptions<InformaticsGatewayConfiguration> _configuration;
         private readonly IDicomToolkit _dicomToolkit;
 
-        protected override int Concurrentcy { get; }
+        protected override int Concurrency { get; }
         public override string RoutingKey { get; }
         public override string ServiceName => "DICOM Export Service";
 
@@ -70,12 +70,12 @@ public ScuExportService(
             _dicomToolkit = dicomToolkit ?? throw new ArgumentNullException(nameof(dicomToolkit));
 
             RoutingKey = $"{configuration.Value.Messaging.Topics.ExportRequestPrefix}.{_configuration.Value.Dicom.Scu.AgentName}";
-            Concurrentcy = _configuration.Value.Dicom.Scu.MaximumNumberOfAssociations;
+            Concurrency = _configuration.Value.Dicom.Scu.MaximumNumberOfAssociations;
         }
 
         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 } });
 
             var manualResetEvent = new ManualResetEvent(false);
             IDicomClient client = null;
diff --git a/src/InformaticsGateway/Test/Services/Export/ExportServiceBaseTest.cs b/src/InformaticsGateway/Test/Services/Export/ExportServiceBaseTest.cs
@@ -36,7 +36,7 @@ public class TestExportService : ExportServiceBase
         public event EventHandler ExportDataBlockCalled;
 
         public bool ExportShallFail = false;
-        protected override int Concurrentcy => 1;
+        protected override int Concurrency => 1;
         public override string RoutingKey => AgentName;
         public override string ServiceName { get => "Test Export Service"; }
 
@@ -260,4 +260,4 @@ private async Task StopAndVerify(TestExportService service)
             Thread.Sleep(250);
         }
     }
-}
+}

Original file line number	Diff line number	Diff line change
`@@ -101,7 +101,8 @@ public DicomFile Load(byte[] fileContent)`
`101`	`101`	`Guard.Against.NullOrEmpty(fileContent, nameof(fileContent));`
`102`	`102`
`103`	`103`	`using var stream = new MemoryStream(fileContent);`
`104`		`- var dicomFile = DicomFile.Open(stream);`
	`104`	`+ var dicomFile = DicomFile.Open(stream, FileReadOption.ReadAll);`
	`105`	`+`
`105`	`106`	`if (dicomFile is null)`
`106`	`107`	`{`
`107`	`108`	`throw new DicomDataException("Invalid DICOM content");`
Original file line number	Diff line number	Diff line change
`@@ -55,12 +55,15 @@ public string Destination`
`55`	`55`	`get { return _exportRequest.Destination; }`
`56`	`56`	`}`
`57`	`57`
`58`		`- public ExportRequestDataMessage(ExportRequestMessage exportRequest)`
	`58`	`+ public string Filename { get; }`
	`59`	`+`
	`60`	`+ public ExportRequestDataMessage(ExportRequestMessage exportRequest, string filename)`
`59`	`61`	`{`
`60`	`62`	`IsFailed = false;`
`61`	`63`	`Messages = new List<string>();`
`62`	`64`
`63`	`65`	`_exportRequest = exportRequest ?? throw new System.ArgumentNullException(nameof(exportRequest));`
	`66`	`+ Filename = filename ?? throw new System.ArgumentNullException(nameof(filename));`
`64`	`67`	`}`
`65`	`68`
`66`	`69`	`public void SetData(byte[] data)`
Original file line number	Diff line number	Diff line change
`@@ -36,7 +36,7 @@ public class TestExportService : ExportServiceBase`
`36`	`36`	`public event EventHandler ExportDataBlockCalled;`
`37`	`37`
`38`	`38`	`public bool ExportShallFail = false;`
`39`		`- protected override int Concurrentcy => 1;`
	`39`	`+ protected override int Concurrency => 1;`
`40`	`40`	`public override string RoutingKey => AgentName;`
`41`	`41`	`public override string ServiceName { get => "Test Export Service"; }`
`42`	`42`
`@@ -260,4 +260,4 @@ private async Task StopAndVerify(TestExportService service)`
`260`	`260`	`Thread.Sleep(250);`
`261`	`261`	`}`
`262`	`262`	`}`
`263`		`-}`
	`263`	`+}`